Agent/src/infra/trace/logging/tool_logging.py

"""工具调用日志的通用封装。"""

from __future__ import annotations

import json
from typing import Any, Dict

from .log_capture import log, log_fold


def _pretty_json_if_possible(text: str) -> str:
    """如果文本是合法 JSON，则返回带缩进的可读格式；否则原样返回。"""
    raw = (text or "").strip()
    if not raw:
        return text
    if not (raw.startswith("{") or raw.startswith("[")):
        return text
    try:
        parsed = json.loads(raw)
    except Exception:
        return text
    return json.dumps(parsed, ensure_ascii=False, indent=2)


def _truncate_deep(obj: Any, str_limit: int = 2000) -> Any:
    """递归遍历对象，对超长字符串做截断，其余结构原样保留。"""
    if isinstance(obj, str):
        return obj if len(obj) <= str_limit else obj[:str_limit] + f"...(truncated, total {len(obj)} chars)"
    if isinstance(obj, dict):
        return {k: _truncate_deep(v, str_limit) for k, v in obj.items()}
    if isinstance(obj, list):
        return [_truncate_deep(item, str_limit) for item in obj]
    return obj


def _structure_metadata(md: Dict[str, Any], body_limit: int = 200) -> Dict[str, Any]:
    """对 metadata 做结构化精简，剥离 raw_data / 完整正文等大字段。

    - 含 article_info 的结果：提取标题、统计、正文预览，丢弃 raw HTML / 图片列表。
    - 含 account_info 的结果：保留账号关键字段。
    - 含 search_results 的结果：每条只保留标题和 URL。
    - 其他情况：递归截断超长字符串。
    """
    # --- 文章详情 ---
    article_info = md.get("article_info")
    if isinstance(article_info, dict):
        body = str(article_info.get("body_text", "") or "")
        body_preview = body[:body_limit] + "..." if len(body) > body_limit else body
        # 去掉图片标记行
        body_preview = "\n".join(
            line for line in body_preview.splitlines()
            if not line.strip().startswith("[image:")
        )
        images = article_info.get("image_url_list") or []
        return {
            "article_info": {
                "title": article_info.get("title", ""),
                "content_link": article_info.get("content_link", ""),
                "publish_timestamp": article_info.get("publish_timestamp"),
                "statistics": {
                    "view_count": article_info.get("view_count"),
                    "like_count": article_info.get("like_count"),
                    "share_count": article_info.get("share_count"),
                    "looking_count": article_info.get("looking_count"),
                    "comment_count": article_info.get("comment_count"),
                    "collect_count": article_info.get("collect_count"),
                },
                "is_original": article_info.get("is_original", False),
                "image_count": len(images),
                "body_length": len(body),
                "body_preview": body_preview,
            }
        }

    # --- 账号信息 ---
    account_info = md.get("account_info")
    if isinstance(account_info, dict):
        return {
            "account_info": {
                "account_name": account_info.get("account_name", ""),
                "wx_gh": account_info.get("wx_gh", ""),
                "channel_account_id": account_info.get("channel_account_id", ""),
            }
        }

    # --- 搜索结果列表 ---
    search_results = md.get("search_results")
    if isinstance(search_results, list):
        brief = [
            {"title": item.get("title", ""), "url": item.get("url", "")}
            for item in search_results[:20]
        ]
        return {"search_results": brief, "total": len(search_results)}

    # --- 兜底：递归截断 ---
    return _truncate_deep(md)


def format_tool_result_for_log(result: Any) -> str:
    """将 ToolResult 或普通字符串格式化为可写入日志的文本。

    对文章详情类结果，输出结构化摘要（标题/统计/正文预览），
    剥离 raw_data 和完整正文，避免日志被大段内容淹没。
    """
    if result is None:
        return ""
    if isinstance(result, str):
        s = result
        return s if len(s) <= 8000 else s[:8000] + "\n...(truncated)"
    title = getattr(result, "title", "") or ""
    output = getattr(result, "output", None) or ""
    err = getattr(result, "error", None)
    payload: Dict[str, Any] = {"title": title, "output": output}
    if err:
        payload["error"] = err
    md = getattr(result, "metadata", None)
    if isinstance(md, dict) and md:
        payload["metadata"] = _structure_metadata(md)
    return json.dumps(payload, ensure_ascii=False)


def log_tool_call(tool_name: str, params: Dict[str, Any], result: str) -> None:
    """以折叠块结构化输出工具调用参数与返回内容。"""
    with log_fold(f"🔧 {tool_name}"):
        with log_fold("📥 调用参数"):
            log(json.dumps(params, ensure_ascii=False, indent=2))
        with log_fold("📤 返回内容"):
            log(_pretty_json_if_possible(result))