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

"""问题记录的构造与规范化（§2.7.4）。

集中在一处生成，保证所有工具产出的问题字段一致：finding_type、severity、
time_alignment_status、evidence_ids、manual_review_required 缺一不可。
"""
from __future__ import annotations

from typing import Any, Iterable

from algorithms.airflow_review.contracts import (
    ALIGN_ALIGNED,
    FINDING_TYPES,
    SEV_MANUAL,
    SEVERITIES,
    as_list,
    unique,
)


class FindingFactory:
    """按调用顺序编号 F001、F002……；编号在单次工具调用内稳定可引用。"""

    def __init__(self) -> None:
        self._seq = 0

    def make(
        self,
        *,
        finding_type: str,
        conclusion: str,
        target_id: str = "",
        target_type: str = "",
        target_name: str = "",
        severity: str = SEV_MANUAL,
        mine_id: str = "",
        actual_value: float | None = None,
        required_value: float | None = None,
        difference: float | None = None,
        ratio: float | None = None,
        plan_id: str = "",
        plan_version: str = "",
        plan_verification_status: str = "",
        actual_window: dict[str, Any] | None = None,
        required_effective_window: dict[str, Any] | None = None,
        time_alignment_status: str = ALIGN_ALIGNED,
        rule_version: str = "",
        evidence_ids: Iterable[str] = (),
        calculation_trace: list[Any] | None = None,
        manual_review_required: bool = False,
        extra: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        if finding_type not in FINDING_TYPES:
            raise ValueError(f"未知 finding_type：{finding_type}")
        if severity not in SEVERITIES:
            raise ValueError(f"未知 severity：{severity}")
        self._seq += 1
        payload: dict[str, Any] = {
            "finding_id": f"F{self._seq:03d}",
            "finding_type": finding_type,
            "mine_id": mine_id,
            "target_id": target_id,
            "target_type": target_type,
            "target_name": target_name,
            "actual_value": actual_value,
            "required_value": required_value,
            "difference": difference,
            "ratio": ratio,
            "plan_id": plan_id,
            "plan_version": plan_version,
            "plan_verification_status": plan_verification_status,
            "actual_window": actual_window or {},
            "required_effective_window": required_effective_window or {},
            "time_alignment_status": time_alignment_status,
            "severity": severity,
            "conclusion": conclusion,
            "rule_version": rule_version,
            "evidence_ids": unique(list(evidence_ids)),
            "calculation_trace": calculation_trace or [],
            "manual_review_required": manual_review_required,
        }
        if extra:
            payload.update(extra)
        return payload


def severity_rank(severity: str) -> int:
    """排序用权重：重大 → 一般 → 关注 → 待复核。"""
    order = {"major": 0, "general": 1, "attention": 2, "manual_review": 3}
    return order.get(severity, 9)


def sort_findings(findings: list[dict[str, Any]]) -> list[dict[str, Any]]:
    return sorted(findings, key=lambda item: (severity_rank(item.get("severity", "")), item.get("finding_id", "")))


def summarize(findings: list[dict[str, Any]]) -> dict[str, Any]:
    counts = {severity: 0 for severity in SEVERITIES}
    for finding in findings:
        counts[finding.get("severity", "manual_review")] = counts.get(finding.get("severity", "manual_review"), 0) + 1
    by_type: dict[str, int] = {}
    for finding in findings:
        by_type[finding.get("finding_type", "")] = by_type.get(finding.get("finding_type", ""), 0) + 1
    return {
        "total_findings": len(findings),
        "major_count": counts.get("major", 0),
        "general_count": counts.get("general", 0),
        "attention_count": counts.get("attention", 0),
        "manual_review_count": counts.get("manual_review", 0),
        "by_finding_type": by_type,
        "evidence_ids": unique([eid for finding in findings for eid in as_list(finding.get("evidence_ids"))]),
    }