vent_ai_platform/tests/test_document_processing.py

from app.services.document_processing import (
    _build_structured_summary,
    _compact_analysis_summary,
    extract_dust_dispersion_free_silica_plan,
    extract_safety_monitoring_ledger,
    normalize_extraction_result,
    resolve_effective_parse_mode,
)


def test_compact_analysis_summary_keeps_numbered_results() -> None:
    summary = (
        "1. 有效期：正常；2. 双机能力：一致；3. 单位资质：需核验；"
        "4. 授权签字人：合规；5. 检验仪器：有效；6. 特性曲线：完整；"
        "7. 证书证号：齐全；8. 风量风压：差异较小。"
    )
    compacted = _compact_analysis_summary(summary)
    assert "1. 有效期：正常" in compacted
    assert "8. 风量风压：差异较小" in compacted
    assert len(compacted.splitlines()) == 8


def test_legacy_word_file_upgrades_to_precision_mode() -> None:
    assert resolve_effective_parse_mode("lightweight", ".doc") == "precision"
    assert resolve_effective_parse_mode("lightweight", ".docx") == "lightweight"


def test_compact_summary_does_not_split_decimal_values() -> None:
    summary = "1. 主要路线：883.2Pa，误差2.58%；2. 次要路线：866.42Pa，误差0.63%。"
    compacted = _compact_analysis_summary(summary)
    assert "883.2Pa" in compacted
    assert "866.42Pa" in compacted


def test_resistance_summary_uses_structured_results() -> None:
    summary = _build_structured_summary(
        "mine_ventilation_resistance_report",
        {
            "main_route_measured_resistance": "883.2 Pa",
            "main_route_relative_error": "2.58",
            "secondary_route_measured_resistance": "866.42 Pa",
            "secondary_route_relative_error": "0.63",
            "equivalent_orifice": "6.77",
            "ventilation_difficulty": "容易",
            "effective_airflow_rate": "94.6",
            "effective_airflow_rate_pass": "合格",
            "overall_assessment": "满足要求",
        },
    )
    assert "主要路线：883.2 Pa" in summary
    assert "次要路线：866.42 Pa" in summary


def test_effective_airflow_rate_above_85_is_qualified() -> None:
    summary = _build_structured_summary(
        "airflow_measurement_decadal_report",
        {
            "measurement_date": "2025-05-16",
            "mine_name": "上湾煤矿",
            "effective_airflow_rate": "93.93",
        },
    )
    assert "有效风量率 93.93%，合格" in summary


def test_safety_monitoring_ledger_deterministic_parser(tmp_path) -> None:
    from openpyxl import Workbook

    path = tmp_path / "安全监控系统台账202507.xlsx"
    workbook = Workbook()
    total = workbook.active
    total.title = "总台账"
    total.append(["编码", "安装地点", "设备名称", "安装位置", "单位", "报警值", "断电值", "复电值", "断电范围"])
    total.append([1, "主井变电所", "综合分站", "主井变电所", "---", "---", "---", "---", "---"])
    total.append([1, "主井变电所", "甲烷", "原煤仓", "%", ">=1.5%", ">=1.5%", "<1.5%", "原煤仓电源"])
    cover = workbook.create_sheet("台账封面")
    cover["A1"] = "上湾矿安全监测系统设备台账"
    cover["A2"] = "上湾煤矿通风队"
    workbook.save(path)
    workbook.close()

    result = extract_safety_monitoring_ledger(str(path))
    assert result is not None
    fields = {item["field_key"]: item["normalized_value"] for item in result["fields"]}
    assert fields["substation_count"] == 1
    assert fields["sensor_count"] == 1
    assert fields["gas_sensor_count"] == 1


def test_dust_dispersion_free_silica_plan_parser(tmp_path) -> None:
    from openpyxl import Workbook

    path = tmp_path / "粉尘分散度游离二氧化硅监测计划.xlsx"
    workbook = Workbook()
    sheet = workbook.active
    sheet.title = "粉尘检测计划"
    sheet.append(["附件：2025年采掘作业场所粉尘分散度、游离二氧化硅含量监测计划表"])
    sheet.append(
        [
            "序号",
            "矿井名称",
            "现场采样时间",
            "检测地点",
            "检测项目",
            "频次/周期",
            "联系人及联系方式",
            "",
            "备注",
            "计划检测时间",
            "实际检测日期",
        ]
    )
    sheet.append(
        [
            1,
            "上湾煤矿",
            "1-12月",
            "12煤、22煤采掘工作面",
            "总粉尘\n粉尘分散度\n游离二氧化硅",
            "1次/半年",
            "李玉福",
            "18047378769",
            "每个煤层至少选择一个采掘工作面",
            "",
            "",
        ]
    )
    sheet.append(["", "", "", "", "", "", "现场检测联系人", "王杰\n18047382145"])
    workbook.save(path)
    workbook.close()

    result = extract_dust_dispersion_free_silica_plan(str(path))
    assert result is not None
    fields = {item["field_key"]: item["normalized_value"] for item in result["fields"]}
    assert fields["monitoring_year"] == "2025"
    assert fields["mine_count"] == 1
    assert fields["plan_record_count"] == 1
    assert fields["monitoring_items"] == ["总粉尘", "粉尘分散度", "游离二氧化硅"]
    assert fields["onsite_contacts"][0]["contact_name"] == "王杰"
    assert fields["onsite_contacts"][0]["phone"] == "18047382145"


def test_normalized_field_uses_template_chinese_name() -> None:
    result = normalize_extraction_result(
        {
            "fields": [
                {
                    "field_key": "mine_count",
                    "field_name": "mine_count",
                    "field_value": 13,
                    "normalized_value": 13,
                }
            ]
        },
        [
            {
                "key": "mine_count",
                "name": "覆盖矿井数量",
                "type": "number",
                "group_key": "plan_summary",
                "group_name": "监测计划汇总",
                "group_order": 2,
                "field_order": 0,
            }
        ],
        model_status="test",
        parsed_markdown="",
        template_code="dust_dispersion_free_silica_monitoring_ledger",
    )
    assert result["fields"][0]["field_name"] == "覆盖矿井数量"