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

"""配风计划核验（工具 4 的确定性内核）。

按「用风地点 → 回风系统 → 矿井总需风量」自下而上复算，比对计划值、复算值、
层级汇总关系与最低风量要求，产出 verified_items / blocked_items 与安全隐患问题。
"""
from __future__ import annotations

from typing import Any, Mapping

from algorithms.airflow_review.contracts import (
    FINDING_DATA_INSUFFICIENT,
    as_window,
    mappings,
    FINDING_PLAN_MISMATCH,
    SEV_ATTENTION,
    SEV_MAJOR,
    SEV_MANUAL,
    VERIFY_INVALID,
    VERIFY_MANUAL,
    VERIFY_VALID,
    WORKPLACE_LOCATION_TYPES,
    unique,
)
from algorithms.airflow_review.engine.findings import FindingFactory
from algorithms.airflow_review.engine.hierarchy import ancestors_of, build_hierarchy
from algorithms.airflow_review.engine.recalc import (
    STATUS_OK,
    recalc_all,
    within_tolerance,
)
from algorithms.airflow_review.rules.loader import RuleSet

STATUS_VALID = "valid"
STATUS_INVALID = "invalid"
STATUS_MANUAL = "manual_review"


def verify_plan(
    plan: Mapping[str, Any],
    rules: RuleSet,
    tolerance: Mapping[str, Any] | None = None,
    finding_factory: FindingFactory | None = None,
) -> dict[str, Any]:
    """核验配风计划，返回逐项结论、权威需风量与问题清单。

    阻断规则：只有**计算错误**沿层级向上传播（错误会污染父级汇总值）；依据不足的
    对象只是自身不进权威集，不牵连父级——父级的汇总正确性可以独立用子项计划值校验。
    """
    items = mappings(plan.get("items"))
    factory = finding_factory or FindingFactory()
    effective_tolerance = rules.tolerance(tolerance)
    hierarchy = build_hierarchy(items)
    nodes = hierarchy["nodes"]
    recalc = recalc_all(items, rules)
    mine_id = str(plan.get("mine_id") or "")
    plan_id = str(plan.get("plan_id") or "")
    plan_version = str(plan.get("plan_version") or "")
    effective_window = as_window(plan.get("effective_window"))

    findings: list[dict[str, Any]] = []
    results: list[dict[str, Any]] = []
    blocked: list[str] = []
    verified: list[dict[str, Any]] = []
    manual_entries: list[dict[str, Any]] = []

    for issue in hierarchy["issues"]:
        if issue["severity"] != "major":
            continue
        findings.append(factory.make(
            finding_type=FINDING_DATA_INSUFFICIENT,
            conclusion=f"计划层级关系存在问题：{issue['message']}",
            target_id=issue["target_id"],
            mine_id=mine_id,
            plan_id=plan_id,
            plan_version=plan_version,
            severity=SEV_MANUAL,
            rule_version=rules.rule_version,
            manual_review_required=True,
            extra={"hierarchy_issue": issue["code"]},
        ))

    for item in items:
        target_id = str(item.get("target_id") or "")
        trace = recalc.get(target_id) or {}
        status, reason = _decide_status(item, trace, tolerance=effective_tolerance)
        reuse_key = _reuse_key(target_id, hierarchy)
        entry = {
            "target_id": target_id,
            "target_type": item.get("target_type"),
            "location_type": trace.get("location_type"),
            "target_name": trace.get("target_name"),
            "verification_status": status,
            "verification_reason": reason,
            "planned_required_airflow_m3_min": trace.get("planned_required_airflow_m3_min"),
            "calculated_required_airflow_m3_min": trace.get("calculated_required_airflow_m3_min"),
            "difference_m3_min": None,
            "difference_percent": None,
            "min_airflow_m3_min": trace.get("min_airflow_m3_min"),
            "max_airflow_m3_min": trace.get("max_airflow_m3_min"),
            "governing_method_id": trace.get("governing_method_id"),
            "aggregation_basis": trace.get("aggregation_basis") or "",
            "calculation_trace": trace.get("methods") or [],
            "verification_checks": trace.get("checks") or [],
            "stated_reproduction": trace.get("stated_reproduction"),
            "recalc_warnings": trace.get("warnings") or [],
            "rule_version": rules.rule_version,
            "is_authoritative": status == VERIFY_VALID,
            "required_reuse_key": reuse_key,
            "is_single_return_system_equivalent": bool(
                hierarchy["single_return_equivalence"].get("equivalent")) and reuse_key != "",
            "effective_window": effective_window,
            "evidence_ids": trace.get("evidence_ids") or [],
        }
        planned = entry["planned_required_airflow_m3_min"]
        calculated = entry["calculated_required_airflow_m3_min"]
        if isinstance(planned, (int, float)) and isinstance(calculated, (int, float)):
            _, difference, _ = within_tolerance(planned, calculated, effective_tolerance)
            entry["difference_m3_min"] = difference
            if planned:
                entry["difference_percent"] = round(difference / float(planned) * 100.0, 4)
        results.append(entry)

        if status == VERIFY_VALID:
            verified.append(_verified_entry(entry, plan, rules, effective_window))
            continue
        if status == VERIFY_INVALID:
            blocked.extend([target_id, *ancestors_of(target_id, nodes)])
            findings.append(_item_finding(item, entry, status, reason, plan, rules, factory, mine_id))
            continue
        # 依据不足：工作面类逐条列问题（承载 75% 红线，必须可逐项复核），
        # 硐室与其他巷道数量可达数十条，汇总为一条计划级问题并附明细。
        manual_entries.append({"target_id": target_id,
                               "target_name": entry.get("target_name"),
                               "location_type": entry.get("location_type"),
                               "reason": reason})
        if _is_workplace(entry):
            findings.append(_item_finding(item, entry, status, reason, plan, rules, factory, mine_id))

    blocked = unique(blocked)
    manual_items = unique([entry["target_id"] for entry in manual_entries])
    # 层级归属问题单独影响计划级结论：归属不明时整个计划不能算"核验通过"
    hierarchy_major_targets = unique([issue["target_id"] for issue in hierarchy["issues"]
                                      if issue["severity"] == "major" and issue["target_id"]])
    invalid = [entry["target_id"] for entry in results if entry["verification_status"] == VERIFY_INVALID]
    if invalid:
        plan_status = STATUS_INVALID
    elif manual_items or hierarchy_major_targets:
        plan_status = STATUS_MANUAL
    else:
        plan_status = STATUS_VALID
    plan_authoritative = plan_status != STATUS_INVALID

    if plan_status != STATUS_VALID or blocked:
        findings.append(_plan_level_finding(plan_status, blocked, manual_entries, results,
                                            plan, rules, factory, mine_id,
                                            hierarchy_targets=hierarchy_major_targets))

    return {
        "plan_id": plan_id,
        "plan_version": plan_version,
        "plan_calculation_status": plan_status,
        "plan_authoritative": plan_authoritative,
        "rule_version": rules.rule_version,
        "tolerance": effective_tolerance,
        "items": results,
        "verified_items": verified,
        "blocked_items": blocked,
        "manual_review_items": manual_items,
        "manual_review_details": manual_entries,
        "findings": findings,
        "hierarchy": {
            "levels": hierarchy["levels"],
            "issues": hierarchy["issues"],
            "blocking_targets": hierarchy["blocking_targets"],
            "single_return_system": hierarchy["single_return_system"],
            "single_return_equivalence": hierarchy["single_return_equivalence"],
        },
        "effective_window": effective_window,
        "evidence_ids": unique([eid for entry in results for eid in entry["evidence_ids"]]),
    }


def _decide_status(
    item: Mapping[str, Any],
    trace: Mapping[str, Any],
    tolerance: Mapping[str, float],
) -> tuple[str, str]:
    """判定单项核验状态；先看依据是否充分，再看是否算错。"""
    if not trace:
        return VERIFY_MANUAL, "该对象未参与复算"
    if trace.get("calculation_status") != STATUS_OK:
        return VERIFY_MANUAL, "缺少必要参数或所有计算方法均不适用，无法复算需风量"

    planned = trace.get("planned_required_airflow_m3_min")
    calculated = trace.get("calculated_required_airflow_m3_min")
    if not isinstance(planned, (int, float)) or not isinstance(calculated, (int, float)):
        return VERIFY_MANUAL, "计划值或复算值缺失"

    within, difference, limit = within_tolerance(planned, calculated, tolerance)
    if not within:
        return VERIFY_INVALID, (
            f"计划需风量 {planned} m³/min 与复算值 {calculated} m³/min 相差 {difference} m³/min，"
            f"超出容差 {limit} m³/min"
        )

    stated = trace.get("stated_reproduction") or {}
    if stated.get("status") == "inconsistent":
        return VERIFY_INVALID, f"计划写出的代入值与结果不自洽：{stated.get('reason', '')}"
    stated_result = stated.get("stated_result_m3_min")
    recomputed = stated.get("recomputed_m3_min")
    if isinstance(stated_result, (int, float)) and isinstance(recomputed, (int, float)):
        stated_ok, stated_diff, _ = within_tolerance(stated_result, recomputed, tolerance)
        if not stated_ok:
            return VERIFY_INVALID, (
                f"计划写出的代入值与结果不自洽：按 {stated.get('expression')} 应为 {recomputed} m³/min，"
                f"计划写为 {stated_result} m³/min（差 {stated_diff}）"
            )

    for check in trace.get("verification_checks") or []:
        if check.get("status") == "fail":
            return VERIFY_INVALID, f"{check.get('name')}不满足：{check.get('left_value')} {check.get('operator')} {check.get('limit_value')}"

    min_required = trace.get("min_airflow_m3_min")
    if isinstance(min_required, (int, float)) and isinstance(calculated, (int, float)):
        if calculated < min_required and abs(calculated - min_required) > max(
            float(tolerance.get("absolute_m3_min", 1.0)), abs(min_required) * 0.005
        ):
            return VERIFY_INVALID, f"复算需风量 {calculated} m³/min 低于法定最低风量 {min_required} m³/min"

    for check in trace.get("checks") or []:
        if check.get("status") == "insufficient" and check.get("check_id") in ("VF-CFI-MIN", "VF-HFI-MIN-COAL", "VF-HFI-MIN-ROCK"):
            return VERIFY_MANUAL, f"{check.get('name')}缺少参数：{check.get('reason')}"
    return VERIFY_VALID, "计划值、复算值、层级汇总关系与验算项均一致"


def _item_finding(
    item: Mapping[str, Any],
    entry: Mapping[str, Any],
    status: str,
    reason: str,
    plan: Mapping[str, Any],
    rules: RuleSet,
    factory: FindingFactory,
    mine_id: str,
) -> dict[str, Any]:
    name = entry.get("target_name") or entry.get("target_id")
    if status == VERIFY_INVALID:
        conclusion = f"配风计划中{name}需风量计算不正确：{reason}"
        severity = SEV_MAJOR
    else:
        conclusion = f"{name}的需风量无法核验，暂不作为正式比对基准：{reason}"
        severity = SEV_MANUAL
    return factory.make(
        finding_type=FINDING_PLAN_MISMATCH,
        conclusion=conclusion,
        target_id=str(entry.get("target_id") or ""),
        target_type=str(entry.get("target_type") or ""),
        target_name=str(name),
        severity=severity,
        mine_id=mine_id,
        required_value=entry.get("planned_required_airflow_m3_min"),
        actual_value=entry.get("calculated_required_airflow_m3_min"),
        difference=entry.get("difference_m3_min"),
        plan_id=str(plan.get("plan_id") or ""),
        plan_version=str(plan.get("plan_version") or ""),
        plan_verification_status=status,
        required_effective_window=entry.get("effective_window"),
        rule_version=rules.rule_version,
        evidence_ids=entry.get("evidence_ids") or [],
        calculation_trace=[entry.get("calculation_trace"), entry.get("verification_checks"),
                           entry.get("stated_reproduction")],
        manual_review_required=status == VERIFY_MANUAL,
        extra={"is_hazard": status == VERIFY_INVALID,
               "verification_reason": reason,
               "governing_method_id": entry.get("governing_method_id")},
    )


def _plan_level_finding(
    plan_status: str,
    blocked: list[str],
    manual_entries: list[dict[str, Any]],
    results: list[dict[str, Any]],
    plan: Mapping[str, Any],
    rules: RuleSet,
    factory: FindingFactory,
    mine_id: str,
    hierarchy_targets: list[str] | None = None,
) -> dict[str, Any]:
    invalid = [entry["target_id"] for entry in results if entry["verification_status"] == VERIFY_INVALID]
    manual_items = unique([entry["target_id"] for entry in manual_entries])
    by_type: dict[str, int] = {}
    for entry in manual_entries:
        key = str(entry.get("location_type") or "未分类")
        by_type[key] = by_type.get(key, 0) + 1
    version = plan.get("plan_version") or ""
    if plan_status == STATUS_INVALID:
        conclusion = (f"配风计划 {version} 存在计算错误，受影响对象"
                      f"（{('、'.join(blocked)) or '无'}）暂不作为正式比对基准")
        severity = SEV_MAJOR
        manual_review = False
    elif plan_status == STATUS_MANUAL:
        breakdown = "、".join(f"{name} {count} 项" for name, count in sorted(by_type.items()))
        conclusion = (f"配风计划 {version} 计划依据不足：{len(manual_items)} 个对象缺少计算过程或参数"
                      f"（{breakdown}），未列入权威需风量；其余对象核验通过")
        severity = SEV_MANUAL
        manual_review = True
    else:
        conclusion = f"配风计划 {version} 整体核验通过"
        severity = SEV_ATTENTION
        manual_review = False
    if plan_status == STATUS_MANUAL and not manual_items and hierarchy_targets:
        conclusion = (f"配风计划 {version} 的层级归属关系不完整（涉及 "
                      f"{('、'.join(hierarchy_targets))}），无法确认汇总基准，需人工复核")
        manual_review = True
    return factory.make(
        finding_type=FINDING_DATA_INSUFFICIENT if plan_status == STATUS_MANUAL else FINDING_PLAN_MISMATCH,
        conclusion=conclusion,
        target_id="MINE_TOTAL" if plan_status != STATUS_VALID else "",
        target_type="mine_total",
        severity=severity,
        mine_id=mine_id,
        plan_id=str(plan.get("plan_id") or ""),
        plan_version=str(version),
        plan_verification_status=plan_status,
        required_effective_window=as_window(plan.get("effective_window")),
        rule_version=rules.rule_version,
        evidence_ids=unique([eid for entry in results for eid in entry.get("evidence_ids") or []]),
        manual_review_required=manual_review,
        extra={"is_hazard": plan_status == STATUS_INVALID,
               "blocked_items": blocked, "invalid_items": invalid,
               "manual_review_items": manual_items,
               "manual_review_breakdown": by_type,
               "manual_review_details": manual_entries,
               "hierarchy_issue_targets": list(hierarchy_targets or [])},
    )


def _is_workplace(entry: Mapping[str, Any]) -> bool:
    return str(entry.get("location_type") or "") in WORKPLACE_LOCATION_TYPES


def _verified_entry(
    entry: Mapping[str, Any],
    plan: Mapping[str, Any],
    rules: RuleSet,
    effective_window: Mapping[str, Any],
) -> dict[str, Any]:
    """核验通过的权威需风量；后续比对只能引用这里的值。"""
    return {
        "target_id": entry.get("target_id"),
        "target_type": entry.get("target_type"),
        "location_type": entry.get("location_type"),
        "target_name": entry.get("target_name"),
        "required_airflow_m3_min": entry.get("calculated_required_airflow_m3_min"),
        "planned_required_airflow_m3_min": entry.get("planned_required_airflow_m3_min"),
        "required_source": "verified_monthly_air_distribution_plan",
        "plan_id": plan.get("plan_id"),
        "plan_version": plan.get("plan_version"),
        "required_effective_window": dict(effective_window),
        "approved_at": plan.get("approved_at"),
        "required_reuse_key": entry.get("required_reuse_key") or "",
        "is_single_return_system_equivalent": entry.get("is_single_return_system_equivalent", False),
        "limit_floor_m3_min": entry.get("min_airflow_m3_min"),
        "limit_ceiling_m3_min": entry.get("max_airflow_m3_min"),
        "rule_version": rules.rule_version,
        "evidence_ids": entry.get("evidence_ids") or [],
    }


def _reuse_key(target_id: str, hierarchy: Mapping[str, Any]) -> str:
    equivalence = hierarchy.get("single_return_equivalence") or {}
    if not equivalence.get("equivalent"):
        return ""
    if target_id in (equivalence.get("return_system_id"), equivalence.get("mine_total_id")):
        return str(equivalence.get("reuse_key") or "")
    return ""


__all__ = ["verify_plan", "STATUS_VALID", "STATUS_INVALID", "STATUS_MANUAL"]