"""风量研判引擎的离线单测：规则库、表达式、复算、层级、核验、比对、校核、去重、证据链。 基准样例覆盖需求 §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("本月中旬") == {}