script-build-and-framework-tmp/visualization-old/backend/app/retrieval_detail_projection.py

from __future__ import annotations

import json
import re
from typing import Any

from .runtime_tool_catalog import (
    MAIN_DECISION_READ_TOOLS,
    RETRIEVAL_AGENT_LABELS,
    business_tool_label,
)


QUERY_TOOL_LABELS = {
    "search_script_decode_case": "解构 Case 查询",
    "external_search_case": "外部内容查询",
    "search_knowledge": "创作知识查询",
}
QUERY_TOOL_NAMES = frozenset(QUERY_TOOL_LABELS)
RETRIEVAL_TOOL_NAMES = frozenset({*QUERY_TOOL_NAMES, *MAIN_DECISION_READ_TOOLS})
RETRIEVAL_RESULT_COLLECTION_KEYS: dict[str, str | None] = {
    "search_script_decode_case": "results",
    "external_search_case": "data",
    "search_knowledge": None,
    "get_domain_info": "domain_info",
}
TERMINAL_RUN_STATUSES = frozenset(
    {"success", "partial", "failed", "stopped", "completed", "cancelled", "cancel", "cancle"}
)

_CONDITION_LABELS = {
    "keyword": "关键词",
    "account_name": "账号",
    "match_fields": "匹配范围",
    "top_k": "最多返回",
    "return_field": "返回内容",
    "platform_channel": "平台",
    "max_count": "最多返回",
    "message": "查询需求",
    "script_build_id": "构建记录",
}
_HIDDEN_CONDITIONS = {"execution_id", "account_name_guard"}


def is_retrieval_detail_event(event: dict[str, Any]) -> bool:
    return (
        str(event.get("event_type") or "") == "tool_call"
        and (
            str(event.get("event_name") or "") in RETRIEVAL_TOOL_NAMES
            or str(event.get("agent_role") or "") in RETRIEVAL_AGENT_LABELS
        )
    )


def summarize_retrieval_output(
    event_name: str,
    *,
    raw_status: Any,
    ended_at: Any,
    output: Any,
    run_status: Any = None,
) -> dict[str, Any]:
    """Return only the bounded status/count needed by the execution view."""

    parsed = _parse_jsonish(output)
    state, count, error = _outcome(
        event_name,
        raw_status=raw_status,
        ended_at=ended_at,
        output=parsed,
        run_status=run_status,
    )
    return {
        "state": state,
        "count": count,
        "error": error,
    }


def project_retrieval_event(
    event: dict[str, Any], *, run_status: Any = None
) -> dict[str, Any]:
    name = str(event.get("event_name") or "")
    source_kind = "direct-tool" if name in MAIN_DECISION_READ_TOOLS else "agent-query"
    input_value = _event_side(event, "input")
    output_value = _event_side(event, "output")
    output_side = event.get("output")
    output_truncated = isinstance(output_side, dict) and bool(output_side.get("truncated"))
    summary_output = (
        _parse_jsonish(event.get("output_preview"))
        if output_truncated
        else output_value
    )
    summary = summarize_retrieval_output(
        name,
        raw_status=event.get("status"),
        ended_at=event.get("ended_at"),
        output=summary_output,
        run_status=run_status,
    )
    state = summary["state"]
    count = summary.get("count")
    error = summary.get("error")
    items = _result_items(name, output_value) if state == "hit" and not output_truncated else []
    completeness = _completeness(event.get("output"), output_value)
    return {
        "detailKind": "retrieval-event",
        "eventId": _integer(event.get("id")) or 0,
        "sourceKind": source_kind,
        "toolName": name,
        "businessLabel": QUERY_TOOL_LABELS.get(name)
        or business_tool_label(name),
        "queryConditions": _query_conditions(input_value),
        "outcome": {
            "state": state,
            **({"count": count} if count is not None else {}),
            "message": _outcome_message(state, count, error, source_kind),
            **({"errorKind": _error_kind(error)} if error else {}),
        },
        "items": items,
        "completeness": completeness,
    }


def _outcome(
    name: str,
    *,
    raw_status: Any,
    ended_at: Any,
    output: Any,
    run_status: Any,
) -> tuple[str, int | None, str | None]:
    status = str(raw_status or "").lower()
    terminal_run = str(run_status or "").lower() in TERMINAL_RUN_STATUSES
    if terminal_run and (status == "running" or not ended_at):
        return "interrupted", None, None

    error = _error_text(output)
    if status in {"error", "failed", "failure"} or error:
        return "failure", None, error

    count = _result_count(name, output)
    if count is not None:
        return ("hit" if count > 0 else "empty"), count, None

    if status == "running" or not ended_at:
        return "running", None, None

    if name in MAIN_DECISION_READ_TOOLS and output not in (None, "", [], {}):
        return "hit", None, None
    return "unknown", None, None


def _error_text(value: Any) -> str | None:
    if isinstance(value, dict):
        success = value.get("success")
        if success is False or (isinstance(success, str) and success.lower() == "false"):
            candidate = value.get("error") or value.get("message") or "返回结果标记为失败"
            return _clean_error(candidate)
        error = value.get("error")
        if error not in (None, "", False, [], {}):
            return _clean_error(error)
        status = str(value.get("status") or "").lower()
        if status in {"error", "failed", "failure"}:
            return _clean_error(value.get("message") or value.get("output") or status)
    if isinstance(value, str):
        text = value.strip()
        if re.match(r"^(?:错误|失败)\s*[:：]", text) or text.startswith("❌"):
            return _clean_error(text)
    return None


def _clean_error(value: Any) -> str:
    text = str(value or "").strip()
    text = re.sub(r"^(?:错误|失败)\s*[:：]\s*", "", text)
    text = re.sub(r"^❌\s*", "", text)
    return text or "执行失败，未保存更多错误说明"


def _result_count(name: str, value: Any) -> int | None:
    if name == "search_knowledge" and isinstance(value, list):
        return len(value)
    if isinstance(value, list):
        return len(value)
    if not isinstance(value, dict):
        return None
    for key in ("count", "returned_count", "result_count"):
        count = _integer(value.get(key))
        if count is not None:
            return count
    for key in ("results", "data", "items", "domain_info"):
        if isinstance(value.get(key), list):
            return len(value[key])
    return None


def _result_items(name: str, output: Any) -> list[dict[str, Any]]:
    if name == "search_script_decode_case":
        values = _result_collection(name, output)
        return [_case_item(item, index) for index, item in enumerate(values or [], 1)]
    if name == "external_search_case":
        values = _result_collection(name, output)
        return [_external_item(item, index, output) for index, item in enumerate(values or [], 1)]
    if name == "search_knowledge":
        return [
            _knowledge_item(item, index)
            for index, item in enumerate(_result_collection(name, output) or [], 1)
        ]
    return _direct_items(name, output)


def _result_collection(name: str, output: Any) -> Any:
    key = RETRIEVAL_RESULT_COLLECTION_KEYS.get(name)
    if key is None:
        return output
    return output.get(key) if isinstance(output, dict) else None


def _case_item(item: Any, index: int) -> dict[str, Any]:
    data = item if isinstance(item, dict) else {"data": item}
    account = str(data.get("account") or "未知账号")
    return {
        "id": str(data.get("post_id") or index),
        "title": f"{account} · Case {index}",
        "subtitle": str(data.get("post_id") or "来源标识未记录"),
        "meta": _meta(
            ("匹配分数", data.get("score")),
            ("返回内容", data.get("return_field")),
        ),
        "sections": [{"title": "完整解构内容", "value": data.get("data")}],
    }


def _external_item(item: Any, index: int, output: dict[str, Any]) -> dict[str, Any]:
    data = item if isinstance(item, dict) else {"body_text": item}
    return {
        "id": str(data.get("channel_content_id") or index),
        "title": str(data.get("title") or f"外部内容 {index}"),
        "subtitle": str(data.get("channel") or output.get("platform_channel") or "外部平台"),
        "href": data.get("link"),
        "meta": _meta(
            ("内容类型", data.get("content_type")),
            ("点赞", data.get("like_count")),
            ("发布时间", data.get("publish_timestamp")),
        ),
        "sections": [
            value
            for value in (
                _value_section("正文", data.get("body_text")),
                _value_section("图片内容理解", data.get("images_text_escape")),
                _value_section("图片", data.get("images")),
                _value_section("视频", data.get("videos")),
            )
            if value
        ],
    }


def _knowledge_item(item: Any, index: int) -> dict[str, Any]:
    data = item if isinstance(item, dict) else {"content": item}
    return {
        "id": str(data.get("id") or index),
        "title": str(data.get("title") or f"知识条目 {index}"),
        "subtitle": str(data.get("purpose") or ""),
        "meta": [],
        "sections": [section for section in [_value_section("完整内容", data.get("content"))] if section],
    }


def _direct_items(name: str, output: Any) -> list[dict[str, Any]]:
    if name == "get_domain_info" and isinstance(output, dict):
        values = _result_collection(name, output)
        return [
            {
                "id": str(item.get("id") or index),
                "title": f"领域事实 {index}",
                "subtitle": str(item.get("note") or ""),
                "meta": _meta(("轮次", item.get("round_index"))),
                "sections": [
                    section
                    for section in (
                        _value_section("事实内容", item.get("content")),
                        _value_section("核实说明", item.get("note")),
                        _value_section("来源", item.get("source")),
                    )
                    if section
                ],
            }
            for index, item in enumerate(values or [], 1)
            if isinstance(item, dict)
        ]

    if output in (None, "", [], {}):
        return []
    data = output if isinstance(output, dict) else {"完整返回": output}
    title = {
        "get_topic_detail": "选题完整信息",
        "get_script_direction": "当前创作方向",
        "get_script_snapshot": "当前主脚本快照",
        "get_account_script_section_patterns": "账号段落模式",
        "get_account_script_persona_points": "账号人设",
    }.get(name, business_tool_label(name))
    preferred = {
        "get_topic_detail": [("选题", "topic"), ("创作点", "points"), ("内容关系", "item_relations")],
        "get_script_snapshot": [("脚本概况", "script_build"), ("段落", "paragraphs"), ("元素", "elements"), ("段落元素关联", "paragraph_element_links")],
        "get_account_script_section_patterns": [("分段规律摘要", "分段规律摘要"), ("典型分段依据", "典型分段依据"), ("段落结构模式", "段落结构模式列表"), ("分析规模", "分析帖子数")],
    }.get(name)
    if preferred:
        sections = [_value_section(label, data.get(key)) for label, key in preferred]
    else:
        sections = [_value_section(_condition_label(key), value) for key, value in data.items()]
    return [{
        "id": name,
        "title": title,
        "subtitle": "",
        "meta": [],
        "sections": [section for section in sections if section],
    }]


def _query_conditions(value: Any) -> list[dict[str, Any]]:
    if not isinstance(value, dict):
        if value in (None, ""):
            return []
        return [{"key": "input", "label": "调用输入", "value": value}]
    return [
        {"key": str(key), "label": _condition_label(str(key)), "value": item}
        for key, item in value.items()
        if key not in _HIDDEN_CONDITIONS and item not in (None, "", [], {})
    ]


def _condition_label(key: str) -> str:
    return _CONDITION_LABELS.get(key, key.replace("_", " "))


def _outcome_message(
    state: str, count: int | None, error: str | None, source_kind: str
) -> str:
    if state == "failure":
        return f"查询执行失败：{error}" if error else "查询执行失败。"
    if state == "empty":
        return "查询执行成功，没有匹配数据。"
    if state == "hit":
        if source_kind == "direct-tool":
            return "读取成功。" if count is None else f"读取成功，共返回 {count} 条记录。"
        return f"查询成功，返回 {count} 条结果。" if count is not None else "查询成功。"
    if state == "interrupted":
        return "构建已经结束，但这次活动没有完成。"
    if state == "running":
        return "查询仍在执行。"
    return "现有记录不足以可靠判断查询结果。"


def _error_kind(error: str | None) -> str:
    text = str(error or "").lower()
    if any(token in text for token in ("参数", "仅支持", "必须", "不能为空", "return_field")):
        return "parameter"
    if any(token in text for token in ("http", "timeout", "timed out", "connection", "请求失败")):
        return "transport"
    if any(token in text for token in ("upstream", "openrouter", "403", "502", "503")):
        return "upstream"
    return "unknown"


def _completeness(side: Any, output: Any) -> dict[str, Any]:
    meta = side if isinstance(side, dict) else {}
    truncated = bool(meta.get("truncated"))
    omitted = _integer(meta.get("omittedCharacters"))
    if isinstance(output, str):
        match = re.search(r"…\[truncated\s+(\d+)\s+chars\]", output)
        if match:
            truncated = True
            omitted = int(match.group(1))
    return {
        "bodyAvailable": side is not None and output is not None,
        "truncated": truncated,
        **({"omittedCharacters": omitted} if omitted is not None else {}),
    }


def _event_side(event: dict[str, Any], side: str) -> Any:
    wrapped = event.get(side)
    if isinstance(wrapped, dict) and "content" in wrapped:
        return _parse_jsonish(wrapped.get("content"))
    if wrapped is not None:
        return _parse_jsonish(wrapped)
    fallback = event.get("inputData" if side == "input" else "agentOutputData")
    if fallback is not None:
        return _parse_jsonish(fallback)
    return _parse_jsonish(event.get(f"{side}_preview"))


def _parse_jsonish(value: Any) -> Any:
    if not isinstance(value, str):
        return value
    text = value.strip()
    if not text or text[0] not in "[{":
        return value
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        return value


def _value_section(title: str, value: Any) -> dict[str, Any] | None:
    if value in (None, "", [], {}):
        return None
    return {"title": title, "value": value}


def _meta(*items: tuple[str, Any]) -> list[dict[str, Any]]:
    return [
        {"label": label, "value": value}
        for label, value in items
        if value not in (None, "", [], {})
    ]


def _integer(value: Any) -> int | None:
    if isinstance(value, bool):
        return None
    try:
        return int(value)
    except (TypeError, ValueError):
        return None