SupplyAgent/api/services/oss_logs.py

"""Query oss_logs for agent run visualizations."""
from __future__ import annotations

from typing import Any

from supply_infra.db.repositories.oss_log_repo import OssLogRepository
from supply_infra.db.session import get_session

_DEMAND_AGENT = "demand_belong_category_agent"


def _serialize_log(row: Any) -> dict[str, Any]:
    return {
        "id": row.id,
        "log_name": row.log_name,
        "agent_name": row.agent_name,
        "oss_path": row.oss_path,
        "create_time": row.create_time.isoformat(sep=" ", timespec="seconds")
        if row.create_time
        else None,
    }


def list_agent_oss_logs(agent_name: str | None = None) -> dict[str, Any]:
    """List all Agent OSS logs, optionally filtered by exact agent name."""
    with get_session() as session:
        repo = OssLogRepository(session)
        rows = repo.list_by_agent_name(agent_name) if agent_name else repo.list_all()
        return {
            "items": [_serialize_log(row) for row in rows],
            "agents": repo.list_agent_names(),
        }


def list_demand_belong_oss_logs() -> list[dict[str, Any]]:
    """List demand_belong_category_agent oss logs, newest first."""
    with get_session() as session:
        repo = OssLogRepository(session)
        rows = repo.list_by_agent_name(_DEMAND_AGENT)
        return [_serialize_log(row) for row in rows]