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

"""问题去重与合并（需求 §2.3 工具 6 要求 4、§2.4 Skill 6 第 2–4 条）。

两条规则：
1. 同一地点、同一问题类型的多条记录只保留最严重的一条，被合并的记入 related_ids；
2. 单一回风系统与矿井总需风量复用同一基准（required_reuse_key 相同）时，保留
   两类对象的层级关系，但同一风险不重复计数。
"""
from __future__ import annotations

from typing import Any, Iterable, Mapping

from algorithms.airflow_review.contracts import unique
from algorithms.airflow_review.engine.findings import severity_rank

# 复用同一需风基准的等价对象：回风系统与矿井总进风/总回风都指向同一份权威需风量，
# 因此按 basis 相等去重，只保留最严重的一条。
_REUSE_EQUIVALENT_TYPES = ("return_air_system", "mine_total",
                           "mine_total_intake", "mine_total_return")


def dedupe_findings(findings: Iterable[Mapping[str, Any]]) -> dict[str, Any]:
    """返回去重后的问题清单与合并说明。"""
    ordered = sorted(findings, key=lambda item: (severity_rank(item.get("severity", "")),
                                                 str(item.get("finding_id") or "")))
    kept: list[dict[str, Any]] = []
    index: dict[tuple[str, str], int] = {}
    reused: dict[str, list[str]] = {}
    merged_count = 0
    suppressed: list[dict[str, Any]] = []

    for finding in ordered:
        item = dict(finding)
        key = (str(item.get("target_id") or ""), str(item.get("finding_type") or ""))
        reuse_key = str(item.get("required_reuse_key") or "")
        if reuse_key and str(item.get("target_type")) in _REUSE_EQUIVALENT_TYPES:
            # 同一复用基准只报一次，另一类对象只记录层级关系
            if reuse_key in reused:
                reused[reuse_key].append(str(item.get("target_id") or ""))
                merged_count += 1
                suppressed.append({"finding_id": item.get("finding_id"),
                                   "merged_into": reused[reuse_key][0],
                                   "reason": f"与 {reused[reuse_key][0]} 复用同一需风基准（{reuse_key}），不重复计数",
                                   "target_id": item.get("target_id")})
                continue
            reused[reuse_key] = [str(item.get("target_id") or "")]
        if key in index and key[0]:
            existing = kept[index[key]]
            existing.setdefault("related_finding_ids", []).append(item.get("finding_id"))
            existing["duplicate_count"] = existing.get("duplicate_count", 1) + 1
            merged_count += 1
            suppressed.append({"finding_id": item.get("finding_id"),
                               "merged_into": existing.get("finding_id"),
                               "reason": f"与 {existing.get('finding_id')} 同一地点同一问题类型",
                               "target_id": item.get("target_id")})
            continue
        index[key] = len(kept)
        kept.append(item)

    return {
        "findings": kept,
        "merged_count": merged_count,
        "merged_details": suppressed,
        "reuse_groups": {key: value for key, value in reused.items() if len(value) > 1},
        "evidence_ids": unique([eid for finding in kept for eid in finding.get("evidence_ids") or []]),
    }


def reuse_summary(reuse_groups: Mapping[str, list[str]]) -> list[dict[str, Any]]:
    """把复用分组整理成报告可读的说明：同一基准下有哪些对象、只计一次风险。"""
    return [
        {"required_reuse_key": key, "target_ids": list(targets),
         "reason": "单一回风系统与矿井总需风量等值复用，两种对象的结论分列但风险只计一次"}
        for key, targets in sorted(reuse_groups.items())
    ]