vent_agent_2.0/tests/test_main_fan_reporting.py

"""面向读者的报告仍须严格保留设备边界、数据口径和实际问题。"""
from datetime import date, datetime

import pytest

from algorithms.main_fan.analysis import screen
from algorithms.main_fan.models import Evidence, Finding, PerformancePoint, PerformanceReport, ScreeningInput
from extensions.main_fan.reporting import render_result


def report_input():
    report = PerformanceReport(report_id="TEST-REPORT-1", mine_name="报告测试矿", fan_name="1号通风机",
        model="TEST-MODEL", serial_number="TEST-SERIAL-1", inspected_on=date(2022, 3, 18),
        recommended_next_on=date(2025, 3, 17), original_conclusion="该次检测合格",
        basis=["测试报告原载检测依据"], evidence=Evidence(source="/workspace/test.pdf", page=3, method="synthetic"))
    for frequency in (30, 40, 50):
        for index in range(7):
            q, pressure, power = 40+index+frequency-30, 900-40*index+(frequency-30)*20, 70+(frequency-30)*2
            row = PerformancePoint(label=f"测点{index+1}", frequency_hz=frequency, flow_m3_s=q,
                static_pressure_pa=pressure, shaft_power_kw=power, input_power_kw=power+5,
                reported_efficiency_pct=round(q*pressure/(10*power), 2),
                evidence=Evidence(source="/workspace/test.pdf", page=8, method="synthetic"))
            report.points.append(row)
            factor = 50/frequency
            report.points.append(row.model_copy(update={"condition": "converted", "flow_m3_s": q*factor,
                "static_pressure_pa": pressure*factor**2, "shaft_power_kw": power*factor**3}))
            if frequency == 30 and index == 3:
                operating = row.model_copy(update={"label": "运行工况"})
                report.points.append(operating)
                report.points.append(report.points[-2].model_copy(update={"label": "运行工况"}))
                report.curves = [operating.model_copy(update={"label": "性能曲线", "static_pressure_pa": pressure-20,
                    "reported_efficiency_pct": round(q*(pressure-20)/(10*power), 2)})]
    foreign = report.model_copy(deep=True, update={"report_id": "OTHER-FAN", "fan_name": "2号通风机", "serial_number": "OTHER-SERIAL"})
    foreign.points[0].flow_m3_s = 9999
    return ScreeningInput(mine_code="test", mine_name="报告测试矿", fan_id="fan-1", fan_name="1号通风机",
        evaluation_time=datetime(2026, 10, 14), reports=[report, foreign])


def test_overall_report_aggregates_real_ranges_and_keeps_foreign_fan_out():
    data = report_input()
    analysis = screen(data)
    body = render_result(data, analysis)
    assert 1500 <= len(body) <= 3500
    assert "| 30 Hz | 40.00～46.00 | 660.00～900.00 |" in body
    assert len([line for line in body.splitlines() if line.startswith("| ")]) == 5
    assert "22行现场实测记录、22行折算记录" in body
    assert "检查了44行历史性能记录" in body
    assert "OTHER-SERIAL" not in body and "9999" not in body
    assert "1份未匹配" in body
    assert "测点1" not in body and "PDF页" not in body
    assert "需复核" not in body and "初步筛查" not in body and "仅复核" not in body
    assert "TEST-SERIAL-1" in body and "2022-03-18" in body and "2025-03-17" in body
    assert "**研判对象：** 报告测试矿" in body
    assert "**报告运行工况。** 数据表" in body
    assert "折算记录" in body and "未取得本矿井的实时监测记录" in body
    assert analysis.engineering_status == "review_required"


def test_table_curve_difference_is_explained_without_rewriting_original_values():
    data = report_input()
    analysis = screen(data)
    body = render_result(data, analysis)
    assert "数据表静压780.00 Pa，图注静压760.00 Pa，相差20.00 Pa" in body
    assert "表列效率47.91%，图注效率46.69%，相差1.22个百分点" in body
    assert data.reports[0].curves[0].static_pressure_pa == 760
    assert next(r for r in data.reports[0].points if r.label == "运行工况").static_pressure_pa == 780
    assert "未取平均或相互覆盖" in body


def test_extraction_notes_do_not_become_page_by_page_business_problems():
    data = report_input()
    analysis = screen(data)
    analysis.findings.extend(Finding(code="document_extraction", severity="review",
        message=f"PDF页{page}：封面或声明页，未提供性能数据，需复核。") for page in range(1, 27))
    analysis.missing_materials.append("核验文档识别结果及原件")
    body = render_result(data, analysis)
    assert "PDF页" not in body and "声明页" not in body and "需复核" not in body
    assert "表格与曲线参数差异" in body and "实时运行依据缺项" in body
    assert len(analysis.findings) >= 26


def test_future_reports_and_missing_metadata_do_not_gain_invented_dates_or_performance():
    data = report_input()
    data.reports = [data.reports[0]]
    data.reports[0].inspected_on = date(2030, 1, 1)
    body = render_result(data, screen(data))
    assert "检测日期晚于本次研判日期" in body
    assert "TEST-MODEL" not in body and "780.00" not in body
    data.reports[0].inspected_on = None
    body = render_result(data, screen(data))
    assert "未载明日期" in body
    assert "2030-01-01" not in body and "2022-03-18" not in body


def test_current_valid_and_invalid_data_lead_to_different_conclusions():
    from mock_gateway.main_fan import task_input
    data = ScreeningInput.model_validate({k:v for k,v in task_input("synthetic").items() if k in ScreeningInput.model_fields})
    body = render_result(data, screen(data))
    assert "计算静压效率为60.00%" in body
    assert "实时运行依据缺项" not in body
    data.points[0].observed_at = None
    body = render_result(data, screen(data))
    assert "计算静压效率为60.00%" not in body
    assert "缺少采集时间（1项）" in body
    assert "补齐本机同步监测" in body


def test_invalid_efficiency_stays_a_specific_problem_not_a_pass_claim():
    data = report_input()
    data.reports[0].points[0].reported_efficiency_pct = 20
    body = render_result(data, screen(data))
    assert "1行存在公式结果超出合理范围或与表列效率差异较大的情况" in body
    assert "性能数值一致性" in body
    assert "未出现超过" not in body
    assert data.reports[0].points[0].reported_efficiency_pct == 20


def test_new_detection_report_does_not_inherit_old_report_update_issue():
    data = report_input()
    latest = data.reports[0].model_copy(deep=True, update={"report_id": "LATEST-REPORT",
        "inspected_on": date(2026, 5, 1), "recommended_next_on": date(2029, 4, 30)})
    data.reports.append(latest)
    body = render_result(data, screen(data))
    assert "主要依据为2026-05-01" in body and "LATEST-REPORT" in body
    assert "当前资料未反映该日期之后的检测结果" not in body
    assert "检测资料更新。" not in body


@pytest.mark.asyncio
async def test_persisted_report_keeps_full_audit_and_restores_model_edits():
    from extensions.main_fan import service
    data = report_input().model_dump(mode="json")
    result = await service.analyze({**data,'paired_assessment':False}, "report-user", "report-session")
    assert result["result_format"] == "report"
    assert len(result["analysis"]["calculations"]) == 44
    assert "需复核" not in result["result"] + result["result_title"]
    assert any("需复核" in f["message"] for f in result["analysis"]["findings"])
    edited = {**result, "result": "当前设备合格", "result_format": None, "sources": []}
    assert service.validate_result(edited, "report-user", "report-session") == result


@pytest.mark.asyncio
async def test_only_canonical_report_is_streamed_and_saved(pricing_ready):
    from langchain_core.messages import AIMessageChunk
    from types import SimpleNamespace
    from db import chat_store
    from extensions.main_fan import service
    from runtime.execution import admit, build_execution
    session = chat_store.create_session("report-user")
    canonical = await service.analyze(report_input().model_dump(mode="json"), "report-user", session["id"])

    class Agent:
        async def aget_state(self, config):
            return SimpleNamespace(tasks=[])

        async def astream(self, *args, **kwargs):
            yield {"type": "messages", "data": (AIMessageChunk(content="模型错误地说当前设备合格", usage_metadata={
                "input_tokens": 8, "output_tokens": 5, "total_tokens": 13}), {})}
            yield {"type": "updates", "data": {"model": {"structured_response": {
                **canonical, "result": "模型篡改的报告", "result_format": None}}}}

    admission = await admit("report-user", session["id"])
    try:
        execution = await build_execution(admission, agent_name="main_fan", agent=Agent(),
            session_id=session["id"], thread_id=session["id"], user_name="report-user", message="审核测试风机",
            config={"configurable": {"thread_id": session["id"]}})
        events = [event async for event in execution.events()]
    finally:
        admission.release()
    expected = f"# {canonical['result_title']}\n\n{canonical['result']}"
    assert execution.result().output == expected
    assert "".join(e["content"] for e in events if e["type"] == "token") == expected
    assert chat_store.get_messages(session["id"])[-1]["content"] == expected
    assert "需要补充：" not in expected and "模型" not in expected


@pytest.mark.asyncio
async def test_file_sources_and_tool_receipt_keep_report_out_of_orchestrator_payload(monkeypatch):
    from types import SimpleNamespace
    from algorithms.main_fan.models import DocumentFacts
    from agents.workspace import session_root
    from extensions.main_fan import service
    from extensions.main_fan.tools import analyze_main_fan
    from extensions.main_fan.schemas import MainFanResult
    from runtime.source_tracker import SourceTracker
    user, session = "report-source", "s"
    path = session_root(user, session) / "workspace" / "uploads" / "test.pdf"
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(b"synthetic PDF unit-test input")

    async def reader(local, virtual):
        return DocumentFacts(reports=report_input().reports, extraction_audit={"mode": "synthetic-test"})

    data = report_input().model_dump(mode="json")
    data["reports"] = []
    data["materials"] = [{"path": "/workspace/uploads/test.pdf", "filename": "测试性能报告.pdf"}]
    canonical = await service.analyze(data, user, session, document_reader=reader)

    async def existing(*args, **kwargs):
        return canonical

    monkeypatch.setattr(service, "analyze", existing)
    runtime = SimpleNamespace(config={"configurable": {"workspace_user": user, "workspace_session": session}})
    receipt = await analyze_main_fan(data, runtime)
    assert receipt["analysis"] is None and receipt["clues"] == [] and receipt["missing_materials"] == []
    assert len(receipt["result"]) < 100 and receipt["analysis_id"] == canonical["analysis_id"]
    assert MainFanResult.model_validate(receipt).sources == canonical["sources"]
    sources = SourceTracker().note_tool_result("analyze_main_fan", receipt)
    assert sources[0]["type"] == "file" and sources[0]["title"] == "测试性能报告.pdf"
    assert service.validate_result(receipt, user, session) == canonical