SupplyAgent/supply_agent/logging/publish.py

"""Publish agent run logs: HTML visualize → OSS upload → MySQL record."""

from __future__ import annotations

import logging
from pathlib import Path
from typing import TYPE_CHECKING

from supply_agent.logging.logger import slugify_agent_name
from supply_agent.logging.visualize import generate_visualization
from supply_infra.config import get_infra_settings
from supply_infra.db.repositories.oss_log_repo import OssLogRepository
from supply_infra.db.session import get_session
from supply_infra.oss.client import OssClient

if TYPE_CHECKING:
    from supply_agent.logging.logger import AgentLogger

_log = logging.getLogger("supply_agent.publish")


def publish_run_artifacts(agent_logger: AgentLogger) -> str | None:
    """
    After a run finishes: generate HTML, upload .log/.jsonl/.html to OSS,
    and insert a row into ``oss_logs``.

    Returns the public HTML URL, or None if skipped / failed.
    Failures are logged and do not raise (so agent runs are not broken).
    """
    settings = get_infra_settings()
    if not settings.log_oss_upload_enabled:
        return None
    if not settings.aliyun_oss_access_key_id or not settings.aliyun_oss_access_key_secret:
        _log.debug("OSS credentials missing; skip log publish")
        return None

    log_file = agent_logger.log_file
    jsonl_file = agent_logger.jsonl_file
    if not log_file or not log_file.exists():
        _log.warning("No log file to publish")
        return None

    try:
        source = jsonl_file if jsonl_file and jsonl_file.exists() else log_file
        html_path = generate_visualization(source)

        agent_slug = slugify_agent_name(agent_logger.agent_name) or "agent"
        log_name = html_path.stem

        client = OssClient(settings)
        files = [p for p in (log_file, jsonl_file, html_path) if p and Path(p).exists()]
        html_url: str | None = None

        for path in files:
            key = client.object_key(agent_slug, Path(path).name)
            url = client.upload_file(path, key)
            if Path(path).suffix.lower() == ".html":
                html_url = url
            _log.info("Uploaded %s → %s", path.name, url)

        if not html_url:
            raise RuntimeError("HTML was not uploaded")

        with get_session() as session:
            OssLogRepository(session).create(
                log_name=log_name,
                agent_name=agent_logger.agent_name or agent_slug,
                oss_path=html_url,
            )

        _log.info("oss_logs saved: log_name=%s url=%s", log_name, html_url)
        print(f"[publish] 可视化已上传: {html_url}")
        return html_url
    except Exception:
        _log.exception("Failed to publish run artifacts")
        print("[publish] 日志上传/入库失败，详见日志；不影响本次 Agent 结果")
        return None