vent_agent_2.0/tests/test_airflow_engine.py

"""风量研判引擎的离线单测：规则库、表达式、复算、层级、核验、比对、校核、去重、证据链。

基准样例覆盖需求 §2.11.7 要求的九类场景：正常、计划计算错误、层级归属错误、单一回风
系统、总风量不足、回风系统风量不足、工作面低于 75%、数据冲突、时间无法对齐。
"""
from __future__ import annotations

import pytest

from algorithms.airflow_review.contracts import to_m3_min
from algorithms.airflow_review.engine.compare import compare_targets
from algorithms.airflow_review.engine.dedupe import dedupe_findings
from algorithms.airflow_review.engine.expression import ExpressionError, evaluate, missing_params
from algorithms.airflow_review.engine.findings import summarize
from algorithms.airflow_review.engine.hierarchy import build_hierarchy
from algorithms.airflow_review.engine.recalc import recalc_all, recalc_plan_item, within_tolerance
from algorithms.airflow_review.engine.reconcile import reconcile_sources
from algorithms.airflow_review.engine.trace import trace_evidence
from algorithms.airflow_review.engine.verify import verify_plan
from algorithms.airflow_review.rules.loader import (
    RuleError,
    available_versions,
    classify_location,
    derive_params,
    load_rules,
)

PLAN_WINDOW = {"start": "2026-05-01T00:00:00+08:00", "end": "2026-05-31T23:59:59+08:00"}
MAY_POINT = {"start": "2026-05-15T09:00:00+08:00", "end": "2026-05-15T09:00:00+08:00"}


@pytest.fixture(scope="module")
def rules():
    return load_rules()


# ---- 规则库与表达式 -------------------------------------------------------

def test_rule_library_is_versioned_and_loaded(rules):
    assert rules.rule_version in available_versions()
    assert rules.thresholds()["compare"]["workplace"]["threshold_ratio"] == 0.75
    assert rules.compare_policy("mine_total_intake")["finding_type"] == "total_airflow_shortage"
    assert rules.tolerance()["absolute_m3_min"] == 1.0


def test_unknown_rule_version_is_rejected():
    with pytest.raises(RuleError, match="规则版本不可用"):
        load_rules("AIR-RULE-1900.1")


def test_expression_evaluator_is_whitelisted():
    assert evaluate("60 * 0.7 * 1.5 * 11.3 * 1.2 * 1.5", {}) == pytest.approx(1281.42, rel=1e-6)
    assert evaluate("sum(values) * factor", {"values": [1, 2, 3], "factor": 1.2}) == pytest.approx(7.2)
    with pytest.raises(ExpressionError):
        evaluate("__import__('os').system('echo x')", {})
    with pytest.raises(ExpressionError):
        evaluate("a ** 99", {"a": 2})
    assert missing_params("a + b + c", {"a": 1}) == ["b", "c"]


def test_table_lookup_interpolates_and_clamps(rules):
    value, row = rules.lookup_table("v_cfi_by_temp", "max_temperature", 24.5, ["min_speed", "max_speed"])
    assert 1.5 < value < 1.8
    assert row["max_temp"] == 26
    height, _ = rules.lookup_table("k_ch_by_height", "cutting_height", 3.4, "k")
    assert height == 1.2
    eta, _ = rules.lookup_table("leakage_eta_by_distance", "distance", 1500, "max_eta")
    assert eta == 2


def test_derive_params_records_provenance(rules):
    formula = rules.formulas("采煤工作面")[0]
    params, missing, trace = derive_params(formula, {"max_temperature": 22, "cutting_height": 2.3,
                                                     "face_length": 180, "max_support_width": 4.8,
                                                     "min_support_width": 4.2}, rules)
    assert not missing
    assert params["k_chi"] == 1.1
    assert any(entry.get("table") == "k_ch_by_height" for entry in trace)


def test_unit_conversion_and_unknown_unit():
    value, warning = to_m3_min(1.5, "m3/s")
    assert value == 90.0 and warning is None
    value, warning = to_m3_min("120", "m3/h")
    assert value == 2.0 and warning is None
    value, warning = to_m3_min(100, "方/分钟")
    assert value is None and "未识别" in warning


def test_location_classification_and_unknown_falls_through():
    assert classify_location("22210 综采工作面")[0] == "采煤工作面"
    assert classify_location("12 上305 辅运顺槽工作面局扇安设点")[0] == "掘进工作面"
    assert classify_location("一水平消防材料库")[0] == "硐室"
    assert classify_location("二水平北边界巷")[0] == "其他巷道"
    location_type, reason = classify_location("某个没见过的东西")
    assert location_type == "" and "未命中" in reason


# ---- 复算 -----------------------------------------------------------------

def test_recalc_uses_section_area_variant(rules):
    """真实计划里 S_cfi 已含 70% 有效断面系数，变体必须命中，不能重复相乘。"""
    item = _workplace_item()
    result = recalc_plan_item(item, rules)
    assert result["calculated_required_airflow_m3_min"] == pytest.approx(1220.4)
    assert result["governing_method_id"] == "WF-CFI-01"
    assert result["methods"][0]["variant"] == {"section_area_basis": "有效断面积"}
    reproduced = result["stated_reproduction"]
    assert reproduced["source"] == "rule_library"
    assert reproduced["recomputed_m3_min"] == pytest.approx(1220.4)


def test_recalc_marks_insufficient_when_params_missing(rules):
    item = {"target_id": "L9", "target_type": "workplace", "location_type": "采煤工作面",
            "target_name": "缺参数面", "planned_required_airflow_m3_min": 900, "planned_parameters": {}}
    result = recalc_plan_item(item, rules)
    assert result["calculation_status"] == "insufficient"
    assert result["calculated_required_airflow_m3_min"] is None
    assert any(method["status"] == "insufficient" for method in result["methods"])


def test_tolerance_absorbs_rounding_but_not_errors():
    within, difference, limit = within_tolerance(1221, 1220.4, {"absolute_m3_min": 1.0, "relative_percent": 0.5})
    assert within and difference == pytest.approx(-0.6) and limit == pytest.approx(6.105)
    within, _, _ = within_tolerance(900, 850, {"absolute_m3_min": 1.0, "relative_percent": 0.5})
    assert not within


def test_aggregation_uses_component_planned_values(rules):
    items = [_workplace_item(), _return_system_item()]
    results = recalc_all(items, rules)
    aggregate = results["R1"]
    assert aggregate["aggregation_basis"] == "planned_components"
    assert aggregate["calculated_required_airflow_m3_min"] == pytest.approx(1465.2)


# ---- 层级 -----------------------------------------------------------------

def test_hierarchy_detects_orphan_duplicate_and_cycle():
    items = [
        {"target_id": "L1", "target_type": "workplace", "target_name": "A", "parent_return_system_id": "R1",
         "planned_required_airflow_m3_min": 100},
        {"target_id": "L2", "target_type": "workplace", "target_name": "B",
         "planned_required_airflow_m3_min": 100},
        {"target_id": "R1", "target_type": "return_air_system", "target_name": "系统一",
         "component_workplace_ids": ["L1", "L1"], "parent_target_id": "MINE"},
        {"target_id": "MINE", "target_type": "mine_total", "target_name": "总",
         "component_return_system_ids": ["R1"]},
    ]
    result = build_hierarchy(items)
    codes = {issue["code"] for issue in result["issues"]}
    assert "orphan_target" in codes
    assert "duplicate_in_aggregation" in codes


def test_hierarchy_detects_cycle():
    items = [
        {"target_id": "MINE", "target_type": "mine_total", "target_name": "总",
         "component_return_system_ids": ["R1"], "parent_target_id": "R2"},
        {"target_id": "R1", "target_type": "return_air_system", "target_name": "系统",
         "component_workplace_ids": ["L1"], "parent_target_id": "MINE"},
        {"target_id": "R2", "target_type": "return_air_system", "target_name": "系统2",
         "component_workplace_ids": ["L1"], "parent_target_id": "R1"},
        {"target_id": "L1", "target_type": "workplace", "target_name": "A", "parent_return_system_id": "R1",
         "planned_required_airflow_m3_min": 100},
    ]
    result = build_hierarchy(items)
    assert any(issue["code"] == "cyclic_reference" for issue in result["issues"])


def test_single_return_system_equivalence_and_reuse_key(rules):
    items = [_workplace_item(), _return_system_item()]
    items[1]["component_workplace_ids"] = ["L1"]
    items[1]["planned_required_airflow_m3_min"] = 1465.2
    items.append({"target_id": "MINE_TOTAL", "target_type": "mine_total", "target_name": "矿井总需风量",
                  "component_return_system_ids": ["R1"], "planned_required_airflow_m3_min": 1465.2})
    result = verify_plan({"plan_id": "P", "plan_version": "V1", "effective_window": PLAN_WINDOW,
                          "items": items}, rules)
    equivalence = result["hierarchy"]["single_return_equivalence"]
    assert equivalence["applicable"] and equivalence["equivalent"]
    reuse_keys = {item["target_id"]: item["required_reuse_key"] for item in result["verified_items"]}
    assert reuse_keys["R1"] == reuse_keys["MINE_TOTAL"] != ""


# ---- 基准样例：九类场景 ---------------------------------------------------

def test_baseline_normal_plan_and_comparison(rules):
    result = verify_plan(_normal_plan(), rules)
    assert result["plan_calculation_status"] == "valid"
    assert result["plan_authoritative"] is True
    assert not result["blocked_items"]

    comparison = compare_targets([{
        "target_id": "MINE_TOTAL_INTAKE", "target_type": "mine_total_intake", "target_name": "矿井总进风量",
        "actual_airflow_m3_min": 7300, "required_airflow_m3_min": 6800,
        "plan_verification_status": "valid", "actual_window": MAY_POINT,
        "required_effective_window": PLAN_WINDOW,
    }], rules)
    record = comparison["comparisons"][0]
    assert record["comparison_status"] == "normal"
    assert not comparison["findings"]


def test_baseline_plan_calculation_error_blocks_ancestors(rules):
    plan = _normal_plan()
    plan["items"][0]["planned_required_airflow_m3_min"] = 900  # 复算值 1220.4，超出容差
    result = verify_plan(plan, rules)
    assert result["plan_calculation_status"] == "invalid"
    assert result["plan_authoritative"] is False
    assert "L1" in result["blocked_items"]
    assert "R1" in result["blocked_items"]          # 错误沿层级向上传播
    assert "MINE_TOTAL" in result["blocked_items"]
    finding = next(item for item in result["findings"] if item["target_id"] == "L1")
    assert finding["finding_type"] == "plan_calculation_mismatch"
    assert finding["is_hazard"] is True and finding["severity"] == "major"


def test_baseline_hierarchy_ownership_error(rules):
    plan = _normal_plan()
    plan["items"][0].pop("parent_return_system_id")
    result = verify_plan(plan, rules)
    issues = {issue["code"] for issue in result["hierarchy"]["issues"]}
    assert "orphan_target" in issues
    assert result["plan_calculation_status"] == "manual_review"


def test_baseline_total_airflow_shortage(rules):
    comparison = compare_targets([{
        "target_id": "MINE_TOTAL_RETURN", "target_type": "mine_total_return", "target_name": "矿井总回风量",
        "actual_airflow_m3_min": 6000, "required_airflow_m3_min": 6800,
        "plan_verification_status": "valid", "actual_window": MAY_POINT,
        "required_effective_window": PLAN_WINDOW,
    }], rules)
    finding = comparison["findings"][0]
    assert finding["finding_type"] == "total_airflow_shortage"
    assert finding["severity"] == "major"
    assert finding["shortfall_m3_min"] == pytest.approx(800)


def test_baseline_return_system_shortage(rules):
    comparison = compare_targets([{
        "target_id": "R1", "target_type": "return_air_system", "target_name": "南翼回风系统",
        "actual_airflow_m3_min": 3000, "required_airflow_m3_min": 3200,
        "plan_verification_status": "valid", "actual_window": MAY_POINT,
        "required_effective_window": PLAN_WINDOW,
    }], rules)
    finding = comparison["findings"][0]
    assert finding["finding_type"] == "return_system_airflow_shortage"
    assert finding["severity"] == "major"


def test_baseline_workplace_below_75_percent(rules):
    comparison = compare_targets([{
        "target_id": "L1", "target_type": "workplace", "location_type": "采煤工作面",
        "target_name": "1301采煤工作面",
        "actual_airflow_m3_min": 640, "required_airflow_m3_min": 900,
        "plan_verification_status": "valid", "actual_window": MAY_POINT,
        "required_effective_window": PLAN_WINDOW,
    }], rules)
    finding = comparison["findings"][0]
    assert finding["finding_type"] == "workplace_airflow_below_75_percent"
    assert finding["is_major_hazard"] is True
    assert finding["ratio"] == pytest.approx(0.7111)
    assert finding["disposal_note"]


def test_baseline_workplace_between_75_and_100_is_attention(rules):
    comparison = compare_targets([{
        "target_id": "L1", "target_type": "workplace", "location_type": "掘进工作面",
        "target_name": "12202辅运顺槽", "actual_airflow_m3_min": 800, "required_airflow_m3_min": 900,
        "plan_verification_status": "valid", "actual_window": MAY_POINT,
        "required_effective_window": PLAN_WINDOW,
    }], rules)
    finding = comparison["findings"][0]
    assert finding["severity"] == "attention"
    assert finding["is_major_hazard"] is False


def test_baseline_time_not_aligned_is_manual_review(rules):
    comparison = compare_targets([{
        "target_id": "MINE_TOTAL_INTAKE", "target_type": "mine_total_intake", "target_name": "矿井总进风量",
        "actual_airflow_m3_min": 35181, "required_airflow_m3_min": 32497.2,
        "plan_verification_status": "valid",
        "actual_window": {"start": "2026-05-15T09:00:00+08:00", "end": "2026-05-15T09:00:00+08:00"},
        "required_effective_window": {"start": "2024-12-01T00:00:00+08:00", "end": "2024-12-31T23:59:59+08:00"},
    }], rules)
    record = comparison["comparisons"][0]
    assert record["time_alignment_status"] == "failed"
    assert record["comparison_status"] == "insufficient"
    assert comparison["findings"][0]["finding_type"] == "time_alignment_failed"


def test_baseline_plan_not_verified_is_not_comparable(rules):
    comparison = compare_targets([{
        "target_id": "L1", "target_type": "workplace", "target_name": "1301采煤工作面",
        "actual_airflow_m3_min": 640, "required_airflow_m3_min": 900,
        "plan_verification_status": "invalid", "actual_window": MAY_POINT,
        "required_effective_window": PLAN_WINDOW,
    }], rules)
    record = comparison["comparisons"][0]
    assert record["comparison_status"] == "not_comparable"
    assert "未取得权威需风量" in record["finding"]["conclusion"]


def test_baseline_missing_actual_is_insufficient_not_normal(rules):
    comparison = compare_targets([{
        "target_id": "L1", "target_type": "workplace", "target_name": "1301采煤工作面",
        "actual_airflow_m3_min": None, "required_airflow_m3_min": 900,
        "plan_verification_status": "valid", "actual_window": MAY_POINT,
        "required_effective_window": PLAN_WINDOW,
    }], rules)
    record = comparison["comparisons"][0]
    assert record["comparison_status"] == "insufficient"
    assert comparison["findings"][0]["manual_review_required"] is True


def test_baseline_data_conflict_is_reported_without_averaging(rules):
    result = reconcile_sources({
        "target_id": "L1", "target_type": "workplace", "target_name": "1301采煤工作面",
        "comparison_window": MAY_POINT,
        "sources": [
            {"source_type": "measurement_report", "airflow_m3_min": 860,
             "observed_at": "2026-05-15T09:00:00+08:00"},
            {"source_type": "monitoring", "airflow_m3_min": 600,
             "observed_at": "2026-05-15T09:05:00+08:00"},
        ],
    }, rules)
    assert result["consistency"] == "conflict"
    assert result["requires_manual_review"] is True
    assert result["finding"]["finding_type"] == "data_conflict"
    assert result["recommended_airflow_m3_min"] == 860  # 采用高优先级来源，不做平均


def test_reconcile_accepts_consistent_sources_and_flags_time_offset(rules):
    ok = reconcile_sources({
        "target_id": "L1", "target_type": "workplace", "target_name": "面",
        "comparison_window": MAY_POINT,
        "sources": [
            {"source_type": "measurement_report", "airflow_m3_min": 860, "observed_at": "2026-05-15T09:00:00+08:00"},
            {"source_type": "monitoring", "airflow_m3_min": 820, "observed_at": "2026-05-15T09:30:00+08:00"},
        ],
    }, rules)
    assert ok["consistency"] == "acceptable"
    assert ok["deviation_percent"] == pytest.approx(4.878, rel=1e-3)
    assert ok["recommended_source"] == "measurement_report"

    offset = reconcile_sources({
        "target_id": "L1", "target_type": "workplace", "target_name": "面",
        "comparison_window": MAY_POINT,
        "sources": [
            {"source_type": "measurement_report", "airflow_m3_min": 860, "observed_at": "2026-05-15T09:00:00+08:00"},
            {"source_type": "monitoring", "airflow_m3_min": 860, "observed_at": "2026-05-15T15:00:00+08:00"},
        ],
    }, rules)
    assert offset["time_alignment_status"] == "failed"
    assert offset["finding"]["finding_type"] == "time_alignment_failed"


# ---- 去重与证据链 ---------------------------------------------------------

def test_dedupe_merges_same_target_and_dedups_reuse(rules):
    comparison = compare_targets([
        {"target_id": "MINE_TOTAL", "target_type": "mine_total_intake", "target_name": "矿井总进风量",
         "actual_airflow_m3_min": 6000, "required_airflow_m3_min": 6800, "plan_verification_status": "valid",
         "actual_window": MAY_POINT, "required_effective_window": PLAN_WINDOW},
        {"target_id": "R1", "target_type": "return_air_system", "target_name": "唯一回风系统",
         "actual_airflow_m3_min": 6000, "required_airflow_m3_min": 6800, "plan_verification_status": "valid",
         "actual_window": MAY_POINT, "required_effective_window": PLAN_WINDOW,
         "required_reuse_key": "reuse:MINE_TOTAL"},
        {"target_id": "MINE_TOTAL", "target_type": "mine_total_return", "target_name": "矿井总回风量",
         "actual_airflow_m3_min": 6100, "required_airflow_m3_min": 6800, "plan_verification_status": "valid",
         "actual_window": MAY_POINT, "required_effective_window": PLAN_WINDOW,
         "required_reuse_key": "reuse:MINE_TOTAL"},
    ], rules)
    deduped = dedupe_findings(comparison["findings"])
    assert deduped["merged_count"] == 1                      # 同一复用基准不重复计数
    assert "reuse:MINE_TOTAL" in deduped["reuse_groups"]
    merged = dedupe_findings(comparison["findings"] + [dict(comparison["findings"][0],
                                                            finding_id="F999")])
    assert merged["merged_count"] >= 1


def test_trace_evidence_reports_missing_ids():
    result = trace_evidence(
        evidence={"e001": {"evidence_id": "e001", "source_file": "报表.xlsx", "source_type": "measurement_report",
                           "normalized_value": 860, "locator": {"sheet": "测风报表", "cell": "N20"},
                           "unit": "m3/min", "confidence": 0.98}},
        evidence_ids=["e001", "e404"], finding_id="F001",
    )
    assert result["is_complete"] is False
    assert result["missing_evidence_ids"] == ["e404"]
    assert result["manual_review_required"] is True
    assert result["evidence_chain"][0]["cell"] == "N20"


def test_summary_counts_by_severity_and_type(rules):
    comparison = compare_targets([
        {"target_id": "L1", "target_type": "workplace", "location_type": "采煤工作面",
         "target_name": "面", "actual_airflow_m3_min": 500, "required_airflow_m3_min": 900,
         "plan_verification_status": "valid", "actual_window": MAY_POINT,
         "required_effective_window": PLAN_WINDOW},
    ], rules)
    stats = summarize(comparison["findings"])
    assert stats["major_count"] == 1
    assert stats["by_finding_type"]["workplace_airflow_below_75_percent"] == 1


# ---- 构造辅助 -------------------------------------------------------------

def _workplace_item() -> dict:
    """布尔台矿 22210 综采工作面的真实计划参数（计划值 1221，复算 1220.4）。"""
    return {
        "target_id": "L1", "target_type": "workplace", "location_type": "采煤工作面",
        "target_name": "22210 综采工作面", "parent_return_system_id": "R1",
        "planned_required_airflow_m3_min": 1221, "planned_formula": "WF-CFI-01",
        "mining_method": "综采",
        "planned_parameters": {"v_cfi": 1.0, "S_cfi": 11.3, "k_chi": 1.2, "k_cli": 1.5,
                               "section_area_basis": "有效断面积",
                               "q_cgi": 3.4, "k_cgi": 1.6, "q_cci": 0.5, "k_cci": 1.5,
                               # 验算项需要控顶距与采高，缺了就判"依据不足"（引擎的严格性
                               # 是刻意的：没有这些参数就无法完成最小/最大风速验算）
                               "l_cbi": 5.0, "l_csi": 4.4, "h_cfi": 3.0},
        "stated_calculation": {"formula_id": "WF-CFI-01", "stated_result_m3_min": 1221,
                               "substituted_values": {"v_cfi": 1.0, "S_cfi": 11.3, "k_chi": 1.2,
                                                      "k_cli": 1.5, "section_area_basis": "有效断面积"}},
        "evidence_ids": ["e101"],
    }


def _return_system_item() -> dict:
    return {"target_id": "R1", "target_type": "return_air_system", "target_name": "南翼回风系统",
            "parent_target_id": "MINE_TOTAL", "component_workplace_ids": ["L1", "L2"],
            "planned_required_airflow_m3_min": 1465.2,
            "planned_parameters": {"reserve_coefficient": 1.2},
            "evidence_ids": ["e102"]}


def _normal_plan() -> dict:
    return {
        "plan_id": "PLAN-TEST", "plan_version": "V1", "mine_id": "M1",
        "approved_at": "2026-04-28", "effective_window": PLAN_WINDOW,
        "items": [
            _workplace_item(),
            {"target_id": "L2", "target_type": "workplace", "location_type": "掘进工作面",
             "target_name": "12202 辅运顺槽", "parent_return_system_id": "R1",
             "planned_required_airflow_m3_min": 367.3, "planned_formula": "WF-HFI-04",
             "roadway_type": "煤巷",
             "planned_parameters": {"q_hgi": 0.83, "k_hgi": 1.6, "S_hfi": 17.82,
                                    "S_hdi": 17.82, "fan_intake_sum": 100,
                                    "outlet_airflow": 90, "distance": 800}},
            {"target_id": "R1", "target_type": "return_air_system", "target_name": "南翼回风系统",
             "parent_target_id": "MINE_TOTAL", "component_workplace_ids": ["L1", "L2"],
             "planned_required_airflow_m3_min": 1905.24,
             "planned_parameters": {"reserve_coefficient": 1.2}},
            {"target_id": "MINE_TOTAL", "target_type": "mine_total", "target_name": "矿井总需风量",
             "component_return_system_ids": ["R1"], "planned_required_airflow_m3_min": 1905.24},
        ],
    }


# ---- 参数口径的变体选择与自洽核验 -----------------------------------------

def test_variant_inferred_from_plan_stated_result(rules):
    """计划未标注参数口径时，按计划写出的结果反推变体，不能把正确计划算成 854.28。"""
    item = _workplace_item()
    item["planned_parameters"]["section_area_basis"] = "平均有效净断面积"   # 模型曾写出的同义说法
    result = recalc_plan_item(item, rules)
    assert result["calculated_required_airflow_m3_min"] == pytest.approx(1220.4)
    method = result["methods"][0]
    assert method["variant_selection"] == "inferred_from_stated_calculation"
    assert method["variant"] == {"section_area_basis": "有效断面积"}
    assert any("反推" in warning for warning in result["warnings"])


def test_variant_declared_is_not_overridden(rules):
    """口径标记写对时按声明走，不做反推。"""
    item = _workplace_item()
    result = recalc_plan_item(item, rules)
    assert result["methods"][0]["variant_selection"] == "declared"
    assert not result["warnings"]


def test_self_consistency_accepts_any_allowed_convention(rules):
    """自洽核验：计划写出的数值在某个允许口径下能复现即算自洽。"""
    item = _workplace_item()
    item["planned_parameters"]["section_area_basis"] = "平均有效净断面积"
    item["stated_calculation"]["expression"] = "60x70%x1.0x11.3x1.2x1.5"
    stated = recalc_plan_item(item, rules)["stated_reproduction"]
    assert stated["status"] == "ok"
    assert stated["recomputed_m3_min"] == pytest.approx(1220.4)
    # 命中即返回，但试算过的表达式都留在计算链里，人能看出是哪个口径对上的
    assert stated["candidates"][0]["expression"] == "60 * v_cfi * S_cfi * k_chi * k_cli"
    assert stated["candidates"][0]["variant"] == {"section_area_basis": "有效断面积"}


def test_self_consistency_flags_when_no_convention_reproduces(rules):
    """任何口径都复现不出来，才是真的算错（这里让计划值与基础口径一致，只留自洽性问题）。"""
    item = _workplace_item()
    item["planned_parameters"]["section_area_basis"] = "平均有效净断面积"
    item["planned_required_airflow_m3_min"] = 854.28              # 与基础口径复算值一致
    item["stated_calculation"]["stated_result_m3_min"] = 2000     # 但与代入值无关
    stated = recalc_plan_item(item, rules)["stated_reproduction"]
    assert stated["status"] == "inconsistent"
    assert "任何口径" in stated["reason"]
    plan = {"plan_id": "P", "plan_version": "V1", "effective_window": PLAN_WINDOW, "items": [item]}
    result = verify_plan(plan, rules)
    assert result["items"][0]["verification_status"] == "invalid"
    assert "不自洽" in result["items"][0]["verification_reason"]


def test_parameter_markers_expose_allowed_values(rules):
    """口径标记的允许取值必须来自规则库，供提取模板约束模型写法。"""
    markers = rules.parameter_markers()
    assert set(markers["section_area_basis"]["allowed"]) == {"有效断面积", "净断面积"}
    assert "WF-CFI-01" in markers["section_area_basis"]["affects"]
    assert markers["roadway_type"]["allowed"]
    assert "煤巷" in markers["roadway_type"]["allowed"]
    assert markers["room_type"]["allowed"]


def test_workplace_without_location_type_uses_workplace_policy(rules):
    """地点类型缺失/判不出时按工作面策略（0.75）走，不能回落到回风系统的 1.0——
    那等于把工作面按 100% 判定，用错了规则。"""
    comparison = compare_targets([{
        "target_id": "L1", "target_type": "workplace", "target_name": "某采面",
        "actual_airflow_m3_min": 800, "required_airflow_m3_min": 900,
        "plan_verification_status": "valid", "actual_window": MAY_POINT,
        "required_effective_window": PLAN_WINDOW,
    }], rules)
    record = comparison["comparisons"][0]
    assert record["policy_key"] == "workplace"
    assert record["threshold_ratio"] == 0.75
    # 800/900 = 88.9%，未破 75% 红线 → 关注项而不是重大风险
    assert record["comparison_status"] == "insufficient"
    assert comparison["findings"][0]["severity"] == "attention"


def test_unaligned_or_unparsable_time_stops_auto_determination(rules):
    """实测窗口解析不出时不得定性——不知道实测是什么时候的，比了也没意义。"""
    comparison = compare_targets([{
        "target_id": "L1", "target_type": "workplace", "location_type": "采煤工作面",
        "target_name": "某采面", "actual_airflow_m3_min": 800, "required_airflow_m3_min": 900,
        "plan_verification_status": "valid",
        "actual_window": "本月中旬",   # 认不出 → 归一化后为空
        "required_effective_window": PLAN_WINDOW,
    }], rules)
    record = comparison["comparisons"][0]
    assert record["time_alignment_status"] == "manual_review"
    assert record["comparison_status"] == "insufficient"
    finding = comparison["findings"][0]
    assert finding["finding_type"] == "time_alignment_failed"
    assert finding["manual_review_required"] is True
    assert "不判定风量是否满足" in finding["conclusion"]


def test_tolerant_time_parsing_for_non_iso_forms():
    """材料里常见的中文/斜杠写法要能认出来，否则会被误判成"时间无法对齐"。"""
    from algorithms.airflow_review.contracts import as_window, parse_time

    assert parse_time("2026年05月29日") is not None
    assert parse_time("2024年12月1日") is not None
    assert parse_time("2024/12/01") is not None
    assert parse_time("2024-12-01 9:00") is not None
    assert parse_time("不是时间") is None
    window = as_window("自2024年12月1日至2024年12月31日")
    assert parse_time(window["start"]) < parse_time(window["end"])
    assert window["normalized_from"] == "string"
    assert as_window("本月中旬") == {}