vent_agent_2.0/algorithms/airflow_review/engine/trace.py

"""证据链追溯（工具 7 的确定性内核）。

把 evidence_id 解析成"原始文件 + 位置 + 数值 + 计算链 + 规则版本"。证据不完整时
必须显式报告缺失 id 并把问题状态置为 manual_review_required（需求 §2.3 工具 7 要求 3），
不允许把缺证据的结论当成完整结论输出。
"""
from __future__ import annotations

from typing import Any, Iterable, Mapping, Sequence

from algorithms.airflow_review.contracts import as_mapping, unique


def trace_evidence(
    *,
    evidence: Mapping[str, Mapping[str, Any]] | None = None,
    evidence_ids: Iterable[str] | None = None,
    finding_id: str = "",
    trace_level: str = "full",
    calculation_trace: Sequence[Any] | None = None,
    rule_versions: Sequence[str] | None = None,
) -> dict[str, Any]:
    """解析证据链。evidence 是各工具返回的 evidence 映射的合并结果。"""
    store = dict(evidence or {})
    requested = [eid for eid in (evidence_ids or [])]
    chain: list[dict[str, Any]] = []
    missing: list[str] = []
    for evidence_id in requested:
        record = store.get(evidence_id)
        if record is None:
            missing.append(evidence_id)
            continue
        chain.append(_to_chain_entry(record, trace_level))

    versions = unique([str(item) for item in (rule_versions or []) if item])
    versions.extend(unique([str(record.get("rule_version")) for record in store.values() if record.get("rule_version")]))
    complete = bool(chain) and not missing
    return {
        "finding_id": finding_id,
        "evidence_chain": chain,
        "calculation_chain": list(calculation_trace or []),
        "rule_versions": unique(versions),
        "requested_count": len(requested),
        "resolved_count": len(chain),
        "missing_evidence_ids": missing,
        "is_complete": complete,
        "trace_level": trace_level,
        "manual_review_required": not complete,
        "conclusion": ("证据链完整" if complete else
                       ("该结论未关联任何原始证据" if not chain else
                        f"证据链不完整，缺失 {len(missing)} 条证据：{'、'.join(missing)}；"
                        f"该问题状态必须为 manual_review_required")),
    }


def _to_chain_entry(record: Mapping[str, Any], trace_level: str) -> dict[str, Any]:
    locator = as_mapping(record.get("locator"))
    entry: dict[str, Any] = {
        "evidence_id": record.get("evidence_id"),
        "source_file": record.get("source_file", ""),
        "source_type": record.get("source_type", ""),
        "value": record.get("normalized_value") if record.get("normalized_value") is not None
        else record.get("raw_value"),
        "source_url": f"evidence://{record.get('evidence_id')}",
    }
    if locator:
        entry["locator"] = locator
        if "page" in locator:
            entry["page"] = locator["page"]
        if "sheet" in locator:
            entry["sheet"] = locator["sheet"]
        if "cell" in locator:
            entry["cell"] = locator["cell"]
    if trace_level == "full":
        entry["raw_value"] = record.get("raw_value")
        if record.get("unit"):
            entry["unit"] = record["unit"]
        if record.get("snippet"):
            entry["snippet"] = record["snippet"]
        if record.get("confidence") is not None:
            entry["confidence"] = record["confidence"]
        if record.get("note"):
            entry["note"] = record["note"]
    return entry