vent_agent_2.0/extensions/vent_airflow/tool_support.py

"""风量研判工具的支撑代码：参数解析、大对象落盘、地点匹配、证据链展开与来源抽取。

与 tools.py 分开的原因：tools.py 是**对外契约**（工具名、参数、摘要字段），改动要
克制；这里是内部实现细节，跟着解析器与引擎的接口演进。混在一起会让"哪些是契约"
看不出来。
"""
from __future__ import annotations

import functools
import json
from typing import Any, Iterable, Mapping

import config

from algorithms.airflow_review.contracts import MINE_TZ, mappings, now_iso
from algorithms.airflow_review.engine.matching import match_locations, match_return_systems
from algorithms.airflow_review.engine.trace import trace_evidence
from extensions.vent_airflow import workspace_paths
from obs import get_logger
from tools.registry import source

logger = get_logger(__name__)

_INLINE = config.AIRFLOW_MAX_INLINE_ITEMS


def _readable_errors(func):
    """把工具内部的意外异常转成可读错误返回，而不是抛出去打断整轮对话。

    工具抛异常会一路穿过 ToolNode 打掉整个 SSE 流（实测：比对工具因为入参里时间窗被
    写成字符串，崩掉的是整轮研判）。AGENTS §四的约定是"工具失败返回可读原因，不抛裸
    异常"——返回说明后模型能自己改参数重试，日志里也留了完整堆栈。
    """
    @functools.wraps(func)
    def wrapper(*args: Any, **kwargs: Any) -> dict[str, Any]:
        try:
            return func(*args, **kwargs)
        except Exception as exc:  # noqa: BLE001 - 这里就是要兜住一切，见上
            logger.exception("工具执行失败 tool=%s", getattr(func, "__name__", "?"))
            return {
                "ok": False,
                "tool": func.__name__,
                "error": f"工具内部错误（已记录日志）：{type(exc).__name__}: {exc}",
                "hint": ("请核对入参形态：时间窗要传 {start, end} 对象（字符串也能被识别，"
                         "但请优先传对象）；数值字段不要带单位；数组里不要混入字符串。"
                         "若反复失败，如实告知用户并转人工复核。"),
            }
    return wrapper


def _build_targets(
    dataset: Mapping[str, Any],
    verification: Mapping[str, Any],
    *,
    mine_id: str,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    """把数据集与核验通过的需风量组装成比对对象；地点匹配全部在工具内确定性完成。"""
    verified = list(verification.get("verified_items") or [])
    locations = list(dataset.get("locations") or [])
    workplace_match = match_locations(locations, verified)
    system_match = match_return_systems(list(dataset.get("return_systems") or []), verified)
    observed = _observed_window(dataset)
    mine_total = next((item for item in verified if str(item.get("target_type")) == "mine_total"), None)
    reuse_key = str((mine_total or {}).get("required_reuse_key") or "")

    targets: list[dict[str, Any]] = []
    for pair in workplace_match["pairs"]:
        required = next((item for item in verified if item.get("target_id") == pair["target_id"]), {})
        targets.append(_target(pair["target_id"], pair["target_type"], pair.get("target_name"),
                               actual=pair["actual_airflow_m3_min"], required=required,
                               location_type=pair.get("location_type"), observed=observed,
                               measured_evidence=locations))
    system_evidence = {str(item.get("location_name")): item.get("evidence_id")
                       for item in (dataset.get("return_systems") or [])}
    for pair in system_match["pairs"]:
        required = next((item for item in verified if item.get("target_id") == pair["target_id"]), {})
        target = _target(pair["target_id"], "return_air_system", pair.get("target_name"),
                         actual=pair["actual_airflow_m3_min"], required=required,
                         location_type="", observed=observed, measured_evidence=[])
        measured_id = system_evidence.get(str(pair.get("location_name")))
        if measured_id:
            target["evidence_ids"] = [measured_id, *target["evidence_ids"]]
        targets.append(target)
    if mine_total:
        intake = dataset.get("mine_total_intake") or {}
        returned = dataset.get("mine_total_return") or {}
        for target_id, label, measured, evidence_id in (
            ("MINE_TOTAL_INTAKE", "矿井总进风量", intake, intake.get("evidence_id")),
            ("MINE_TOTAL_RETURN", "矿井总回风量", returned, returned.get("evidence_id")),
        ):
            if measured.get("airflow_m3_min") is None:
                continue
            target = _target(target_id, "mine_total_intake" if "INTAKE" in target_id else "mine_total_return",
                             label, actual=measured.get("airflow_m3_min"), required=mine_total,
                             location_type="", observed=observed, measured_evidence=[],
                             extra_reuse=reuse_key)
            if evidence_id:
                target["evidence_ids"] = [evidence_id, *target["evidence_ids"]]
            targets.append(target)
    report = {
        "workplace": workplace_match,
        "return_system": system_match,
        "mine_total_available": mine_total is not None,
    }
    return targets, report


def _target(
    target_id: str,
    target_type: str,
    target_name: Any,
    *,
    actual: Any,
    required: Mapping[str, Any],
    location_type: Any,
    observed: Mapping[str, Any],
    measured_evidence: list[Mapping[str, Any]],
    extra_reuse: str = "",
) -> dict[str, Any]:
    evidence_ids = [item.get("evidence_id") for item in measured_evidence
                    if item.get("location_id") == target_id and item.get("evidence_id")]
    return {
        "target_id": target_id,
        "target_type": target_type,
        "target_name": target_name,
        "location_type": location_type or "",
        "actual_airflow_m3_min": actual,
        "actual_source": "measurement_report",
        "actual_window": observed,
        "required_airflow_m3_min": required.get("required_airflow_m3_min"),
        "required_source": required.get("required_source", "verified_monthly_air_distribution_plan"),
        "required_reuse_key": required.get("required_reuse_key") or extra_reuse,
        "plan_id": required.get("plan_id", ""),
        "plan_version": required.get("plan_version", ""),
        "plan_verification_status": "valid",
        "required_effective_window": required.get("required_effective_window") or {},
        "evidence_ids": [eid for eid in evidence_ids if eid] + list(required.get("evidence_ids") or []),
    }


def _observed_window(dataset: Mapping[str, Any]) -> dict[str, Any]:
    """实测窗口：旬报按观测时点取点窗口；无时点时用报表月份首日。"""
    records = dataset.get("records") or []
    observed = next((record.get("observed_at") for record in records if record.get("observed_at")), "")
    if not observed:
        period = str(dataset.get("report_period") or "")
        observed = f"{period}-01T09:00:00+08:00" if period else ""
    if not observed:
        return {}
    return {"start": observed, "end": observed, "aggregation": "point"}


def _match_summary(report: Mapping[str, Any]) -> dict[str, Any]:
    workplace = report.get("workplace") or {}
    system = report.get("return_system") or {}
    return {
        "workplace_matched": workplace.get("matched_count", 0),
        "workplace_match_rate": workplace.get("match_rate", 0.0),
        "unmatched_dataset_count": len(workplace.get("unmatched_dataset") or []),
        "unmatched_plan_workplace_count": len(workplace.get("unmatched_plan_workplaces") or []),
        "return_system_matched": len(system.get("pairs") or []),
        "unmatched_return_system_count": len(system.get("unmatched_dataset") or []),
    }


def _manual_summary(details: list[Mapping[str, Any]]) -> dict[str, Any]:
    breakdown: dict[str, int] = {}
    for detail in details:
        key = str(detail.get("location_type") or "未分类")
        breakdown[key] = breakdown.get(key, 0) + 1
    return {"count": len(details), "by_location_type": breakdown,
            "sample": [detail.get("target_name") for detail in details[:8]]}


def _collect_evidence(payload: Any) -> dict[str, Mapping[str, Any]]:
    """从任意层级的产物里收集 evidence 映射（工具产物结构不同，统一扫一遍）。"""
    found: dict[str, Mapping[str, Any]] = {}

    def walk(node: Any, depth: int = 0) -> None:
        if depth > 6:
            return
        if isinstance(node, Mapping):
            evidence = node.get("evidence")
            if isinstance(evidence, Mapping):
                for key, value in evidence.items():
                    if isinstance(value, Mapping) and value.get("evidence_id"):
                        found.setdefault(str(key), value)
            for value in node.values():
                walk(value, depth + 1)
        elif isinstance(node, list):
            for item in node:
                walk(item, depth + 1)

    walk(payload)
    return found


def _evidence_ids_for_finding(payload: Any, finding_id: str) -> list[str]:
    for finding in (payload.get("findings") if isinstance(payload, Mapping) else None) or []:
        if isinstance(finding, Mapping) and finding.get("finding_id") == finding_id:
            return list(finding.get("evidence_ids") or [])
    return []


def _attach_evidence_chains(
    findings: list[Mapping[str, Any]],
    evidence_path: str,
    verified_items: list[Mapping[str, Any]],
) -> tuple[list[dict[str, Any]], list[str]]:
    """把问题的证据链展开到 findings 上，并补上权威需风量引用的证据。

    报告中"证据索引"与"证据完整率"都依赖这一步：问题只有 evidence_ids 是不够的，
    报告要能看到"哪个文件、哪一页/哪个单元格、什么值"。
    """
    if not evidence_path:
        return mappings(findings), []
    payload, error = _load_argument(evidence_path)
    if error:
        return mappings(findings), [f"证据文件读取失败，报告未展开证据链：{error}"]
    store = _collect_evidence(payload)
    if not store:
        return mappings(findings), ["证据文件中没有 evidence 映射，报告未展开证据链"]
    for item in verified_items:
        for evidence_id in item.get("evidence_ids") or []:
            store.setdefault(str(evidence_id), {"evidence_id": evidence_id,
                                                "source_file": "配风计划", "source_type": "plan",
                                                "note": "权威需风量引用"})
    enriched: list[dict[str, Any]] = []
    incomplete: list[str] = []
    for finding in findings:
        item = dict(finding)
        chain = trace_evidence(evidence=store, evidence_ids=item.get("evidence_ids") or [],
                               finding_id=str(item.get("finding_id") or ""))
        item["evidence_chain"] = chain["evidence_chain"]
        if not chain["is_complete"]:
            item["manual_review_required"] = True
            item["evidence_status"] = chain["conclusion"]
            incomplete.append(f"{item.get('finding_id')}：{chain['conclusion']}")
        enriched.append(item)
    return enriched, incomplete[:5]


def _sources(evidence: Any, evidence_ids: Iterable[str] | None, limit: int = 10) -> list[dict[str, Any]]:
    """把证据转成来源标注；条数截断，避免刷屏占用上下文。"""
    if not isinstance(evidence, Mapping):
        return []
    result: list[dict[str, Any]] = []
    for evidence_id in list(evidence_ids or [])[:limit]:
        record = evidence.get(evidence_id)
        if not isinstance(record, Mapping):
            continue
        locator = record.get("locator") or {}
        where = " ".join(str(locator.get(key)) for key in ("sheet", "cell", "page") if locator.get(key))
        result.append(source(title=str(record.get("source_file") or "原始资料"),
                             url=f"evidence://{record.get('evidence_id')}",
                             snippet=f"{where} {record.get('normalized_value') or record.get('raw_value')}".strip(),
                             score=record.get("confidence")))
    return result


def _save(payload: Any, name: str, *, suffix: str = ".json") -> str:
    return workspace_paths.write_artifact(json.dumps(payload, ensure_ascii=False, default=str, indent=2),
                                          name, suffix=suffix)


def _cleaned_text(document: Mapping[str, Any]) -> str:
    """把清洗后的文本落成文件，并在每页前加页标记，便于模型引用页码。"""
    lines: list[str] = []
    for block in document.get("blocks") or []:
        if block.get("line") == 1:
            lines.append(f"\n===== 第 {block.get('page')} 页 =====")
        lines.append(str(block.get("text") or ""))
    header = (f"# 配风计划清洗文本：{document.get('source_file')}\n"
              f"# 共 {document.get('page_count')} 页；已去页眉页脚与重复行；"
              f"表格行内的制表符表示不同列\n")
    return header + "\n".join(lines)


def _load_argument(value: Any, *, default: Any = None) -> tuple[Any, str]:
    """参数既可以是 JSON 字符串，也可以是文件路径（模型经 write_file 落盘的产物）。"""
    if value is None or value == "":
        return (default, "") if default is not None else (None, "参数为空")
    if isinstance(value, (dict, list)):
        return value, ""
    text = str(value).strip()
    if text.startswith("{") or text.startswith("["):
        try:
            return json.loads(text), ""
        except json.JSONDecodeError as exc:
            return None, f"JSON 解析失败：{exc.msg}（第 {exc.lineno} 行）"
    path = text
    try:
        real = workspace_paths.resolve_input(path)
    except workspace_paths.WorkspacePathError as exc:
        return None, (f"既不是合法 JSON，也无法作为文件路径解析：{exc}"
                      if default is None else "")
    try:
        return json.loads(real.read_text(encoding="utf-8")), ""
    except (OSError, json.JSONDecodeError) as exc:
        return None, f"读取 {path} 失败：{exc}"


def _error(tool: str, message: str, **extra: Any) -> dict[str, Any]:
    """工具失败返回可读原因，不抛裸异常（AGENTS.md 约定）。"""
    return {"ok": False, "tool": tool, "error": message, "at": now_iso(MINE_TZ), **extra}