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

"""实际风量与需风量比对（工具 6 的确定性内核）。

同一套比较逻辑服务矿井总进风/总回风、回风系统与工作面，只靠规则库里的
threshold_policy 区分对象与阈值（需求 §2.3 工具 6 要求 2）。

三条硬约束：
1. 只有 plan_verification_status = valid 才允许把计划需风量当正式基准；
2. 比对前先校验时间对齐，对不上不得判定；
3. 缺实测值或缺有效需风值时只输出数据不足，不得判正常（§2.6 第 13 条）。
"""
from __future__ import annotations

from typing import Any, Mapping

from algorithms.airflow_review.contracts import (
    ALIGN_ALIGNED,
    as_window,
    ALIGN_FAILED,
    ALIGN_MANUAL,
    ALIGN_PARTIAL,
    COMPARE_INSUFFICIENT,
    COMPARE_MAJOR,
    COMPARE_NORMAL,
    COMPARE_NOT_COMPARABLE,
    FINDING_DATA_INSUFFICIENT,
    FINDING_TIME_ALIGNMENT,
    LOC_OTHER,
    LOC_STANDBY,
    LOC_WORKPLACE_COAL,
    LOC_WORKPLACE_HEADING,
    SEV_ATTENTION,
    SEV_MAJOR,
    SEV_MANUAL,
    TARGET_MINE_INTAKE,
    TARGET_MINE_RETURN,
    TARGET_RETURN_SYSTEM,
    TARGET_WORKPLACE,
    parse_time,
    ratio as compute_ratio,
    window_contains,
    window_overlaps,
)
from algorithms.airflow_review.engine.findings import FindingFactory
from algorithms.airflow_review.rules.loader import RuleSet

# location_type 到比对阈值策略的映射：阈值策略按目标类型区分，不重复建比较逻辑
_POLICY_KEY_BY_LOCATION = {
    LOC_WORKPLACE_COAL: TARGET_WORKPLACE,
    LOC_WORKPLACE_HEADING: TARGET_WORKPLACE,
    LOC_STANDBY: TARGET_WORKPLACE,
    LOC_OTHER: TARGET_RETURN_SYSTEM,
}


def policy_key(target: Mapping[str, Any]) -> str:
    """决定该对象用哪条阈值策略：目标类型优先，工作面按地点类型细分。"""
    target_type = str(target.get("target_type") or "")
    if target_type in (TARGET_MINE_INTAKE, TARGET_MINE_RETURN):
        return TARGET_MINE_INTAKE if target_type == TARGET_MINE_INTAKE else TARGET_MINE_RETURN
    if target_type == TARGET_RETURN_SYSTEM:
        return TARGET_RETURN_SYSTEM
    if target_type == TARGET_WORKPLACE:
        # 地点类型缺失/无法判定时按工作面策略（0.75）走，**不能回落到回风系统的 1.0**：
        # 那等于把工作面按 100% 判定，用错了规则；具体地点类型只影响是否 75% 红线。
        return _POLICY_KEY_BY_LOCATION.get(str(target.get("location_type") or ""), TARGET_WORKPLACE)
    return TARGET_RETURN_SYSTEM


def compare_one(
    target: Mapping[str, Any],
    rules: RuleSet,
    factory: FindingFactory,
    *,
    mine_id: str = "",
) -> dict[str, Any]:
    """比对单个对象，返回比对记录与（必要时）问题记录。"""
    key = policy_key(target)
    policy = rules.compare_policy(key)
    threshold_ratio = float(target.get("threshold_policy", {}).get("threshold_ratio", policy["threshold_ratio"]))
    actual = _number(target.get("actual_airflow_m3_min"))
    required = _number(target.get("required_airflow_m3_min"))
    plan_status = str(target.get("plan_verification_status") or "")
    # 模型会把时间窗写成字符串（实测传过 "2024-12-01 ~ 2024-12-31"），
    # 归一化后仍不可解析才按"时间无法对齐"转人工复核——不能直接 dict() 崩掉整轮。
    actual_window = as_window(target.get("actual_window"))
    required_window = as_window(target.get("required_effective_window"))
    alignment, alignment_reason = align_windows(actual_window, required_window)

    record: dict[str, Any] = {
        "comparison_id": f"C{target.get('target_id')}",
        "mine_id": mine_id,
        "target_id": target.get("target_id"),
        "target_type": target.get("target_type"),
        "location_type": target.get("location_type"),
        "target_name": target.get("target_name"),
        "actual_value_m3_min": actual,
        "actual_source": target.get("actual_source"),
        "actual_window": actual_window,
        "required_value_m3_min": required,
        "required_source": target.get("required_source"),
        "required_reuse_key": target.get("required_reuse_key", ""),
        "required_plan_version": target.get("plan_version", ""),
        "required_effective_window": required_window,
        "time_alignment_status": alignment,
        "time_alignment_reason": alignment_reason,
        "threshold_ratio": threshold_ratio,
        "policy_key": key,
        "difference_m3_min": None,
        "shortfall_m3_min": None,
        "ratio": None,
        "comparison_status": COMPARE_NOT_COMPARABLE,
        "plan_verification_status": plan_status,
        "severity": "",
        "finding": None,
        "evidence_ids": list(target.get("evidence_ids") or []),
    }

    if plan_status != "valid":
        record["comparison_status"] = COMPARE_NOT_COMPARABLE
        record["finding"] = factory.make(
            finding_type=FINDING_DATA_INSUFFICIENT,
            conclusion=f"{target.get('target_name') or target.get('target_id')} 未取得权威需风量，暂不比对："
                       f"配风计划核验状态为 {plan_status or '未知'}",
            target_id=str(target.get("target_id") or ""),
            target_type=str(target.get("target_type") or ""),
            target_name=str(target.get("target_name") or ""),
            severity=SEV_MANUAL,
            mine_id=mine_id,
            plan_version=str(target.get("plan_version") or ""),
            plan_verification_status=plan_status,
            required_effective_window=required_window,
            rule_version=rules.rule_version,
            evidence_ids=record["evidence_ids"],
            manual_review_required=True,
        )
        return record

    if actual is None or required is None:
        record["comparison_status"] = COMPARE_INSUFFICIENT
        missing = "有效实测风量" if actual is None else "有效需风量"
        record["finding"] = factory.make(
            finding_type=FINDING_DATA_INSUFFICIENT,
            conclusion=f"{target.get('target_name') or target.get('target_id')} 缺少{missing}，无法比对",
            target_id=str(target.get("target_id") or ""),
            target_type=str(target.get("target_type") or ""),
            target_name=str(target.get("target_name") or ""),
            severity=SEV_MANUAL,
            mine_id=mine_id,
            actual_value=actual,
            required_value=required,
            plan_version=str(target.get("plan_version") or ""),
            plan_verification_status=plan_status,
            actual_window=actual_window,
            required_effective_window=required_window,
            rule_version=rules.rule_version,
            evidence_ids=record["evidence_ids"],
            manual_review_required=True,
        )
        return record

    if alignment in (ALIGN_FAILED, ALIGN_MANUAL):
        # 时间窗缺失或解析不出，与"窗口无交集"一样不能定性：不知道实测是什么时候的，
        # 拿它跟计划窗口比毫无意义（§2.8 第 6 条：实测时间无法匹配有效窗口必须转人工复核）。
        record["comparison_status"] = COMPARE_INSUFFICIENT
        record["finding"] = factory.make(
            finding_type=FINDING_TIME_ALIGNMENT,
            conclusion=f"{target.get('target_name') or target.get('target_id')} 实测时间与计划生效窗口"
                       f"无法对齐（{alignment_reason}），不判定风量是否满足",
            target_id=str(target.get("target_id") or ""),
            target_type=str(target.get("target_type") or ""),
            target_name=str(target.get("target_name") or ""),
            severity=SEV_MANUAL,
            mine_id=mine_id,
            actual_value=actual,
            required_value=required,
            plan_version=str(target.get("plan_version") or ""),
            plan_verification_status=plan_status,
            actual_window=actual_window,
            required_effective_window=required_window,
            time_alignment_status=alignment,
            rule_version=rules.rule_version,
            evidence_ids=record["evidence_ids"],
            manual_review_required=True,
        )
        return record

    ratio_value = compute_ratio(actual, required)
    record["ratio"] = ratio_value
    record["difference_m3_min"] = round(actual - required, 4)
    record["shortfall_m3_min"] = round(required - actual, 4) if actual < required else 0.0

    if ratio_value is None:
        record["comparison_status"] = COMPARE_INSUFFICIENT
        return record

    if ratio_value < threshold_ratio:
        severity = str(policy.get("below_threshold_severity") or SEV_MAJOR)
        status = COMPARE_MAJOR
        percent = round(ratio_value * 100.0, 2)
        threshold_percent = round(threshold_ratio * 100.0, 2)
        if key == TARGET_WORKPLACE:
            conclusion = (f"{target.get('target_name')} 实际供风量为需风量的 {percent}%，"
                          f"低于 {threshold_percent}% 红线，缺 {record['shortfall_m3_min']} m³/min")
        elif key == TARGET_RETURN_SYSTEM:
            conclusion = (f"{target.get('target_name')} 实际风量为需风量的 {percent}%，"
                          f"低于需风量，缺 {record['shortfall_m3_min']} m³/min")
        else:
            conclusion = (f"{target.get('target_name')} 实际风量为需风量的 {percent}%，"
                          f"低于需风量，缺 {record['shortfall_m3_min']} m³/min")
    elif ratio_value < 1.0:
        severity = str(policy.get("between_threshold_and_required_severity") or SEV_ATTENTION)
        status = COMPARE_INSUFFICIENT
        conclusion = (f"{target.get('target_name')} 实际风量为需风量的 {round(ratio_value * 100.0, 2)}%，"
                      f"低于需风量但高于 {round(threshold_ratio * 100.0, 2)}% 红线，需关注")
    else:
        record["comparison_status"] = COMPARE_NORMAL
        record["severity"] = ""
        return record

    record["comparison_status"] = status
    record["severity"] = severity
    record["finding"] = factory.make(
        finding_type=str(policy.get("finding_type")),
        conclusion=conclusion,
        target_id=str(target.get("target_id") or ""),
        target_type=str(target.get("target_type") or ""),
        target_name=str(target.get("target_name") or ""),
        severity=severity,
        mine_id=mine_id,
        actual_value=actual,
        required_value=required,
        difference=record["difference_m3_min"],
        ratio=ratio_value,
        plan_id=str(target.get("plan_id") or ""),
        plan_version=str(target.get("plan_version") or ""),
        plan_verification_status=plan_status,
        actual_window=actual_window,
        required_effective_window=required_window,
        time_alignment_status=alignment,
        rule_version=rules.rule_version,
        evidence_ids=record["evidence_ids"],
        extra={
            "is_abnormal": True,
            "is_major_hazard": status == COMPARE_MAJOR and key == TARGET_WORKPLACE,
            "shortfall_m3_min": record["shortfall_m3_min"],
            "actual_required_ratio": ratio_value,
            "threshold_ratio": threshold_ratio,
            "required_reuse_key": record["required_reuse_key"],
            "disposal_note": "请核实是否已停止作业、撤人、断电并处理" if status == COMPARE_MAJOR else "",
        },
    )
    return record


def compare_targets(
    targets: list[Mapping[str, Any]],
    rules: RuleSet,
    *,
    mine_id: str = "",
    factory: FindingFactory | None = None,
) -> dict[str, Any]:
    """批量比对；一次工具调用处理全部对象，避免逐个调用往返回模型。"""
    finder = factory or FindingFactory()
    comparisons: list[dict[str, Any]] = []
    findings: list[dict[str, Any]] = []
    for target in targets:
        record = compare_one(target, rules, finder, mine_id=mine_id)
        comparisons.append(record)
        if record.get("finding"):
            findings.append(record["finding"])
    return {
        "mine_id": mine_id,
        "comparisons": comparisons,
        "findings": findings,
        "rule_version": rules.rule_version,
        "summary": _summarize(comparisons),
    }


def align_windows(
    actual_window: Mapping[str, Any] | None,
    required_window: Mapping[str, Any] | None,
) -> tuple[str, str]:
    """时间对齐判定：实测窗口落在计划生效窗口内为 aligned。

    实测时间窗长于计划窗口（跨月旬报之类）只算 partial，需要人工确认按哪个
    版本取值；完全不相交判 failed。任一侧缺时间一律 manual_review。
    """
    if not actual_window or not required_window:
        return ALIGN_MANUAL, "缺少实测窗口或计划生效窗口"
    if parse_time(actual_window.get("start")) is None or parse_time(actual_window.get("end")) is None:
        return ALIGN_MANUAL, "实测窗口时间无法解析"
    if parse_time(required_window.get("start")) is None or parse_time(required_window.get("end")) is None:
        return ALIGN_MANUAL, "计划生效窗口时间无法解析"
    if window_contains(required_window, actual_window):
        return ALIGN_ALIGNED, "实测窗口落在计划生效窗口内"
    if not window_overlaps(actual_window, required_window):
        return ALIGN_FAILED, "实测窗口与计划生效窗口无交集"
    return ALIGN_PARTIAL, "实测窗口与计划生效窗口部分重叠，需确认取值口径"


def _summarize(comparisons: list[Mapping[str, Any]]) -> dict[str, Any]:
    counts: dict[str, int] = {}
    for record in comparisons:
        status = str(record.get("comparison_status") or "")
        counts[status] = counts.get(status, 0) + 1
    return {
        "total": len(comparisons),
        "by_status": counts,
        "abnormal_count": sum(1 for record in comparisons
                              if record.get("comparison_status") in (COMPARE_MAJOR, COMPARE_INSUFFICIENT)),
        "major_risk_count": counts.get(COMPARE_MAJOR, 0),
        "not_comparable_count": counts.get(COMPARE_NOT_COMPARABLE, 0),
    }


def _number(value: Any) -> float | None:
    if isinstance(value, bool):
        return None
    if isinstance(value, (int, float)):
        return float(value)
    if isinstance(value, str):
        try:
            return float(value.strip())
        except ValueError:
            return None
    return None