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

"""多源风量交叉校核（工具 5 的确定性内核）。

默认以正式测风报表为周期审查主依据、安全监控为连续校核依据；先校验时间与
统计口径，再算偏差。偏差超限或时间不对齐时**不得**给出单一确定结论，也不得
用平均值掩盖冲突（需求 §2.3 工具 5 要求 3、5）。
"""
from __future__ import annotations

from typing import Any, Mapping

from algorithms.airflow_review.contracts import (
    ALIGN_ALIGNED,
    as_mapping,
    as_window,
    ALIGN_FAILED,
    ALIGN_MANUAL,
    FINDING_DATA_CONFLICT,
    FINDING_DATA_INSUFFICIENT,
    FINDING_TIME_ALIGNMENT,
    SEV_ATTENTION,
    SEV_MANUAL,
    SOURCE_MEASUREMENT,
    SOURCE_MONITORING,
    parse_time,
)
from algorithms.airflow_review.engine.findings import FindingFactory
from algorithms.airflow_review.rules.loader import RuleSet

CONSISTENCY_ACCEPTABLE = "acceptable"
CONSISTENCY_CONFLICT = "conflict"
CONSISTENCY_SINGLE = "single_source"
CONSISTENCY_UNKNOWN = "unknown"


def reconcile_sources(
    request: Mapping[str, Any],
    rules: RuleSet,
    factory: FindingFactory | None = None,
    *,
    mine_id: str = "",
) -> dict[str, Any]:
    """对同一对象、同一时间窗的多个来源做对齐与偏差判定。"""
    finder = factory or FindingFactory()
    policy = rules.reconcile_policy()
    deviation_limit = float(policy.get("deviation_percent", 10.0))
    tolerance_minutes = float(policy.get("time_tolerance_minutes", 60))
    priority = list(policy.get("source_priority") or [SOURCE_MEASUREMENT, SOURCE_MONITORING])

    target_id = str(request.get("target_id") or "")
    target_type = str(request.get("target_type") or "")
    target_name = str(request.get("target_name") or target_id)
    window = as_window(request.get("comparison_window"))
    sources = [as_mapping(item) for item in (request.get("sources") or [])]
    sources = [item for item in sources if _number(item.get("airflow_m3_min")) is not None]
    sources = sorted(sources, key=lambda item: _priority_index(str(item.get("source_type") or ""), priority))

    result: dict[str, Any] = {
        "target_id": target_id,
        "target_type": target_type,
        "target_name": target_name,
        "period": request.get("period", ""),
        "comparison_window": window,
        "recommended_airflow_m3_min": None,
        "recommended_source": "",
        "deviation_percent": None,
        "consistency": CONSISTENCY_UNKNOWN,
        "time_alignment_status": ALIGN_MANUAL,
        "requires_manual_review": True,
        "sources": sources,
        "reason": "",
        "rule_version": rules.rule_version,
        "evidence_ids": list(request.get("evidence_ids") or []),
        "finding": None,
    }

    if not sources:
        result["reason"] = "该对象在本窗口内没有任何有效来源数据"
        result["finding"] = finder.make(
            finding_type=FINDING_DATA_INSUFFICIENT,
            conclusion=f"{target_name} 缺少有效风量数据，无法交叉校核",
            target_id=target_id, target_type=target_type, target_name=target_name,
            severity=SEV_MANUAL, mine_id=mine_id,
            actual_window=window, rule_version=rules.rule_version,
            evidence_ids=result["evidence_ids"], manual_review_required=True,
        )
        return result

    primary = sources[0]
    result["recommended_airflow_m3_min"] = _number(primary.get("airflow_m3_min"))
    result["recommended_source"] = str(primary.get("source_type") or "")

    if len(sources) == 1:
        result["consistency"] = CONSISTENCY_SINGLE
        result["time_alignment_status"] = ALIGN_ALIGNED
        result["requires_manual_review"] = False
        result["reason"] = (f"仅有 {result['recommended_source']} 单一来源，按来源优先级直接采用；"
                            f"缺少第二来源，无法交叉验证")
        return result

    secondary = sources[1]
    align_status, align_reason = _align_times(primary, secondary, tolerance_minutes, window)
    result["time_alignment_status"] = align_status
    if align_status == ALIGN_FAILED:
        result["consistency"] = CONSISTENCY_UNKNOWN
        result["reason"] = f"两来源时间不对齐（{align_reason}），不形成正式一致性结论"
        result["finding"] = finder.make(
            finding_type=FINDING_TIME_ALIGNMENT,
            conclusion=f"{target_name} 测风报表与安全监控数据时间不对齐（{align_reason}），需人工复核",
            target_id=target_id, target_type=target_type, target_name=target_name,
            severity=SEV_MANUAL, mine_id=mine_id,
            actual_value=_number(primary.get("airflow_m3_min")),
            required_value=_number(secondary.get("airflow_m3_min")),
            actual_window=window, time_alignment_status=ALIGN_FAILED,
            rule_version=rules.rule_version, evidence_ids=result["evidence_ids"],
            manual_review_required=True,
        )
        return result

    first, second = _number(primary.get("airflow_m3_min")), _number(secondary.get("airflow_m3_min"))
    deviation = _deviation_percent(first, second)
    result["deviation_percent"] = deviation
    if deviation <= deviation_limit:
        result["consistency"] = CONSISTENCY_ACCEPTABLE
        result["requires_manual_review"] = align_status != ALIGN_ALIGNED
        result["reason"] = (f"测风报表与安全监控数据处于允许偏差范围内（{deviation}% ≤ {deviation_limit}%），"
                            f"按来源优先级采用 {result['recommended_source']}")
        return result

    result["consistency"] = CONSISTENCY_CONFLICT
    result["reason"] = (f"两来源偏差 {deviation}% 超出允许阈值 {deviation_limit}%，"
                        f"不采用平均值掩盖冲突，转人工复核")
    result["finding"] = finder.make(
        finding_type=FINDING_DATA_CONFLICT,
        conclusion=(f"{target_name} 测风报表（{first} m³/min）与安全监控（{second} m³/min）"
                    f"偏差 {deviation}%，超过阈值 {deviation_limit}%，需人工复核确认采信值"),
        target_id=target_id, target_type=target_type, target_name=target_name,
        severity=SEV_ATTENTION, mine_id=mine_id,
        actual_value=first, required_value=second, ratio=deviation,
        actual_window=window, time_alignment_status=align_status,
        rule_version=rules.rule_version, evidence_ids=result["evidence_ids"],
        manual_review_required=True,
        extra={"deviation_limit_percent": deviation_limit,
               "candidate_values": [first, second],
               "disposal_note": "请核实传感器状态与测风时点后确认采信值"},
    )
    return result


def _align_times(
    primary: Mapping[str, Any],
    secondary: Mapping[str, Any],
    tolerance_minutes: float,
    window: Mapping[str, Any],
) -> tuple[str, str]:
    first = parse_time(primary.get("observed_at"))
    second = parse_time(secondary.get("observed_at"))
    if first is None or second is None:
        return ALIGN_MANUAL, "来源观测时间无法解析"
    delta_minutes = abs((first - second).total_seconds()) / 60.0
    if delta_minutes > tolerance_minutes:
        return ALIGN_FAILED, f"观测时点相差 {round(delta_minutes, 1)} 分钟，超过容差 {tolerance_minutes} 分钟"
    for moment in (first, second):
        if window and not _in_window(moment, window):
            return ALIGN_MANUAL, "观测时点不在声明的时间窗内"
    return ALIGN_ALIGNED, f"观测时点相差 {round(delta_minutes, 1)} 分钟，在容差内"


def _in_window(moment: Any, window: Mapping[str, Any]) -> bool:
    start, end = parse_time(window.get("start")), parse_time(window.get("end"))
    if start is None or end is None:
        return True
    return start <= moment <= end


def _deviation_percent(first: float | None, second: float | None) -> float:
    """偏差按两值中的较小者计算：偏保守，避免用大值稀释冲突。"""
    if not first or not second:
        return 100.0
    return round(abs(first - second) / min(first, second) * 100.0, 4)


def _priority_index(source_type: str, priority: list[str]) -> int:
    try:
        return priority.index(source_type)
    except ValueError:
        return len(priority)


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