"""原始样例筛查、关联边界、效率量纲、文档原件及协议全链路。"""
import copy
from dataclasses import replace
from datetime import timedelta
import hashlib
import json
from pathlib import Path
from types import SimpleNamespace

import httpx
import pytest
from PIL import Image
from pydantic import ValidationError

from algorithms.main_fan.analysis import screen
from algorithms.main_fan.models import DocumentFacts, ScreeningInput
from agents import agent_factory, agent_registry
from agents.workspace import session_root
from db import business_store, chat_store, main_fan_store
from extensions import registry as extensions
from extensions.main_fan.documents import read_document, fingerprint
from extensions.main_fan import service, data_source
from extensions.main_fan.schemas import MainFanInput
from extensions.main_fan.tools import analyze_main_fan
from extensions.registry import ExtensionRegistry, GatewayBinding
from extensions.shaanxi.client import GatewayClient, GatewayStopped
from extensions.shaanxi.settings import GatewaySettings
from mock_gateway.main_fan import MockState, create_app, task_input, load_sample, INTERNAL_TOKEN, MANAGE_TOKEN
from runtime.business_tasks import BusinessTaskController
from runtime.engine import ExecutionResult
from runtime.environment import RuntimeEnvironment
from runtime.execution import execute
from runtime.business_models import RemoteFile
from extensions.main_fan.documents import rebind_sources, model_extract
from db import db_conn


def codes(result):
    return {f.code for f in result.findings}


def synthetic():
    return ScreeningInput.model_validate(task_input("synthetic"))


def test_raw_file_mock_never_loads_transcribed_facts(tmp_path, monkeypatch):
    def forbidden():
        raise AssertionError("原件模拟模式不得读取人工夹具")
    monkeypatch.setattr('mock_gateway.main_fan.load_sample', forbidden)
    (tmp_path/'通风机性能测定报告.pdf').write_bytes(b'synthetic raw PDF bytes')
    (tmp_path/'snapshot.png').write_bytes(b'synthetic raw image bytes')
    (tmp_path/'ignored.json').write_text('{"reports":[]}')
    state=MockState.from_raw_files(tmp_path)
    files=state.file_index()
    assert {f['filename'] for f in files} == {'通风机性能测定报告.pdf','snapshot.png'}
    assert state.sample['devices']==[] and state.sample['points']==[]
    assert 'document_facts' not in state.sample
    for file in files:
        assert file['version']==hashlib.sha256((tmp_path/file['filename']).read_bytes()).hexdigest()[:16]


def test_raw_file_mock_rejects_changed_original(tmp_path):
    source=tmp_path/'通风机性能测定报告.pdf';source.write_bytes(b'first original')
    state=MockState.from_raw_files(tmp_path)
    source.write_bytes(b'changed original')
    with pytest.raises(ValueError,match='刷新文件索引'):
        state.file_index()


def test_actual_sample_never_joins_foreign_mine_or_invents_flow():
    sample = load_sample()
    data = task_input("original")
    data.pop("use_gateway")
    data.update(devices=sample["devices"], points=sample["points"], reports=sample["document_facts"]["reports"])
    result = screen(ScreeningInput.model_validate(data))
    assert {"report_mine_mismatch", "stale_point", "device_missing_time", "zero_velocity"} <= codes(result)
    assert result.associations[0].state == "candidate"
    assert not result.associations[0].eligible_for_calculation
    assert not result.calculations
    assert "running_zero_velocity" not in codes(result)
    assert any("单点风速" in m for m in result.missing_materials)


def test_actual_report_reproduces_efficiency_but_only_as_history():
    data = task_input("report")
    data.pop("use_gateway")
    data["reports"] = load_sample()["document_facts"]["reports"]
    result = screen(ScreeningInput.model_validate(data))
    assert {"report_fan_mismatch", "report_recommended_date_passed", "historical_report"} <= codes(result)
    assert len(result.calculations) == 7
    assert all(c.scope == "historical_report" for c in result.calculations)
    assert result.calculations[4].efficiency_pct == pytest.approx(64.7675, abs=.0001)
    assert "report_formula_inconsistent" not in codes(result)
    assert result.engineering_status == "review_required"


@pytest.mark.parametrize("name,state", [("主扇通风_PLC1_二号风机风速", "unmatched"),
    ("主扇PLC1风速", "unmatched"), ("地面主通一级风速", "unmatched"),
    ("一号风机风速", "candidate"), ("1#风机风速", "candidate"),
    ("1号风机2号风机共用风速", "ambiguous"), ("1号风速测点", "unmatched")])
def test_name_matching_is_only_a_candidate_and_never_uses_plc_or_stage(name, state):
    data = synthetic()
    data.confirmed_bindings = []
    data.points = [data.points[0]]
    data.points[0].name, data.points[0].device_name = name, ""
    result = screen(data)
    assert result.associations[0].state == state
    assert not result.calculations


def test_confirmed_synchronized_quantities_compute_with_correct_units():
    data = synthetic()
    data.points[0].value, data.points[0].unit = 3600, "m3/min"
    data.points[1].value, data.points[1].unit = .5, "kPa"
    data.points[2].value, data.points[2].unit = 50000, "W"
    result = screen(data)
    assert result.calculations[0].efficiency_pct == 60
    assert result.calculations[0].scope == "current_confirmed"
    assert all(a.state == "confirmed" for a in result.associations)


@pytest.mark.parametrize("change,expected", [("stale", "stale_point"), ("future", "future_point"),
    ("time", "missing_time"), ("status", "point_status"), ("unit", "unknown_unit"),
    ("negative", "negative_value"), ("duplicate", "duplicate_point"), ("unsync", "unsynchronized"),
    ("conflict", "binding_conflict"), ("zero_power", "zero_shaft_power")])
def test_invalid_measurements_never_participate_in_efficiency(change, expected):
    data = synthetic()
    point = data.points[0]
    if change == "stale": point.observed_at -= timedelta(hours=1)
    if change == "future": point.observed_at += timedelta(hours=1)
    if change == "time": point.observed_at = None
    if change == "status": point.status = "断线"
    if change == "unit": point.unit = "m/s"
    if change == "negative": point.value = -1
    if change == "duplicate": data.points.append(point.model_copy())
    if change == "unsync": point.observed_at -= timedelta(seconds=240)
    if change == "conflict": point.name = "2号风机风量"
    if change == "zero_power": data.points[2].value = 0
    result = screen(data)
    assert expected in codes(result)
    assert not result.calculations


@pytest.mark.parametrize("replacement", ["input_power", "pressure"])
def test_electric_power_and_unspecified_pressure_cannot_replace_required_quantities(replacement):
    data = synthetic()
    data.points[2 if replacement == "input_power" else 1].metric = replacement
    assert not screen(data).calculations


def test_cross_mine_name_conflict_is_rejected_even_when_code_is_same():
    data = synthetic()
    data.points[0].mine_name = "另一矿井"
    result = screen(data)
    assert result.associations[0].state == "excluded"
    assert not result.calculations


def test_nonfinite_values_and_blank_identity_are_rejected():
    data = task_input("synthetic")
    data["points"][0]["value"] = float("nan")
    with pytest.raises(ValidationError): MainFanInput.model_validate(data)
    with pytest.raises(ValidationError): MainFanInput(mine_code=" ", mine_name="矿井")


@pytest.mark.parametrize("device_seconds,conflict", [(0, True), (-690, False), (240, False)])
def test_running_zero_velocity_requires_confirmed_binding_and_valid_synchronous_state(device_seconds, conflict):
    data = task_input("synthetic")
    data["points"] = [{**data["points"][0], "metric": "velocity", "unit": "m/s", "value": 0}]
    data["devices"] = [{"device_code": data["fan_id"], "name": data["fan_name"],
        "mine_code": data["mine_code"], "state": "running", "observed_at": data["evaluation_time"]}]
    checked = ScreeningInput.model_validate(data)
    checked.devices[0].observed_at += timedelta(seconds=device_seconds)
    assert ("running_zero_velocity" in codes(screen(checked))) == conflict


@pytest.mark.asyncio
async def test_vision_model_isolated_callbacks_and_separate_usage(monkeypatch, tmp_path):
    monkeypatch.setenv("MAIN_FAN_VISION_MODEL", "test-vision")
    monkeypatch.setattr("billing.guard.precheck", lambda *a: {})
    monkeypatch.setattr("billing.pricing.compute_cost", lambda *a: (0, "CNY"))
    seen = []
    class Vision:
        async def ainvoke(self, messages, config):
            from langchain_core.messages import AIMessage
            seen.append((messages, config))
            return AIMessage(content=json.dumps({"reports": [{"report_id": "R", "mine_name": "矿",
                "fan_name": "1号风机", "evidence": {"page": 1, "excerpt": "受检单位/报告编号"}}]}),
                usage_metadata={"input_tokens": 120, "output_tokens": 40, "total_tokens": 160})
    monkeypatch.setattr("agents.model_factory.build_model", lambda **kw: Vision())
    (tmp_path / "page.png").write_bytes(b"synthetic-image-bytes")
    result = await model_extract([{"page": 1, "text": "", "image": "page.png"}], tmp_path,
                                 user_name="vision-u", session_id="vision-s")
    assert result.reports[0].report_id == "R"
    assert seen[0][1] == {"callbacks": []}
    rows = db_conn.query("SELECT * FROM billing_usage WHERE user_name = ?", ("vision-u",))
    assert len(rows) == 1 and rows[0]["model"] == "test-vision" and rows[0]["kind"] == "main_fan_ocr"
    assert rows[0]["input_tokens"] == 120 and rows[0]["output_tokens"] == 40


def test_report_formula_disagreement_is_recorded_without_rewriting_original():
    data = task_input("report"); data.pop("use_gateway")
    data["reports"] = copy.deepcopy(load_sample()["document_facts"]["reports"][:1])
    data["reports"][0]["points"][0]["reported_efficiency_pct"] = 20
    result = screen(ScreeningInput.model_validate(data))
    assert "report_formula_inconsistent" in codes(result)
    assert result.calculations[0].reported_efficiency_pct == 20


@pytest.mark.asyncio
async def test_canonical_result_restores_model_edits_and_enforces_session_ownership():
    result = await service.analyze(task_input("synthetic"), "u", "s")
    edited = {**result, "result": "模型虚构的合格结论", "alarm_level": "红"}
    canonical = service.validate_result(edited, "u", "s")
    assert canonical["result"] == result["result"]
    assert canonical["alarm_level"] is None
    with pytest.raises(ValueError, match="当前会话"):
        service.validate_result(result, "other", "s")
    with pytest.raises(ValueError, match="analysis_id"):
        service.validate_result({**result, "analysis_id": ""}, "u", "s")


@pytest.mark.asyncio
async def test_queue_tool_uses_server_input_instead_of_model_invented_bindings():
    trusted = task_input("synthetic"); trusted["confirmed_bindings"] = []
    runtime = SimpleNamespace(config={"configurable": {"business_agent": "main_fan",
        "workspace_user": "tool-u", "workspace_session": "tool-s", "business_task_input": trusted}})
    result = await analyze_main_fan(task_input("synthetic"), runtime)
    result = service.validate_result(result,'tool-u','tool-s')
    assert not result["analysis"]["calculations"]
    assert all(a["state"] == "candidate" for a in result["analysis"]["associations"])


@pytest.mark.asyncio
async def test_real_langgraph_writes_executes_and_submits_agent_code(monkeypatch):
    from test_business_agents import ToolModel
    from sandbox_agent_helpers import sandbox_messages
    from runtime.python_sandbox import capability
    if not capability()['available']:
        pytest.skip('当前平台未安装执行沙箱')
    model = ToolModel(messages=sandbox_messages('fan-e2e', task_input('synthetic')))
    monkeypatch.setattr(agent_factory.model_factory, 'get_model', lambda: model)
    agent_registry.invalidate()
    async with RuntimeEnvironment(prewarm=False, billing_retry=False):
        session = chat_store.create_session('fan-e2e')
        result = await execute(agent_name='main_fan', user_name='fan-e2e', session_id=session['id'],
            message='请根据业务Skill自行解析、计算并生成报告。',
            graph_config={'configurable': {'business_task_input': task_input('synthetic')}})
    assert result.status == 'success', result.error
    assert result.structured_output['analysis'] is None
    assert result.structured_output['assessment']['agent_result']['details']['efficiency_pct'] == 60
    assert result.structured_output['assessment']['decision']['result_code'] == 'UNDETERMINED'
    assert '# 自编报告' in result.output and '60.0%' in result.output
    assert result.structured_output['assessment']['exploration']['python_runs']
    agent_registry.invalidate()


@pytest.mark.asyncio
async def test_scanned_pdf_renders_in_isolation_and_cache_is_hash_bound(tmp_path):
    source = tmp_path / "scan.pdf"
    Image.new("RGB", (120, 160), "white").save(source, "PDF")
    called = []
    async def extractor(pages, folder):
        called.append(pages)
        assert (folder / pages[0]["image"]).is_file()
        return {"reports": [{"report_id": "test-report", "mine_name": "测试矿", "fan_name": "1号风机",
                              "evidence": {"page": 1, "excerpt": "测试样例"}}]}
    value = await read_document(source, "/workspace/a.pdf", extractor=extractor)
    assert called and value.reports[0].evidence.sha256 == fingerprint(source)
    second = await read_document(source, "/workspace/b.pdf", extractor=extractor)
    assert len(called) == 1
    assert second.reports[0].evidence.source == "/workspace/b.pdf"
    with source.open("ab") as stream: stream.write(b"\nchanged")
    await read_document(source, "/workspace/c.pdf", extractor=extractor)
    assert len(called) == 2


@pytest.mark.asyncio
async def test_document_extraction_rejects_invented_page(tmp_path):
    source = tmp_path / "scan.pdf"
    Image.new("RGB", (120, 160), "white").save(source, "PDF")
    async def extractor(*args):
        return {"reports": [{"report_id": "R", "mine_name": "矿", "fan_name": "1号风机", "evidence": {"page": 99}}]}
    with pytest.raises(ValueError, match="原页"):
        await read_document(source, "/workspace/a.pdf", extractor=extractor)
    assert main_fan_store.document(fingerprint(source)) is None


@pytest.mark.asyncio
async def test_no_vision_model_returns_explicit_missing_materials(tmp_path):
    root = session_root("no-ocr", "s")
    local = root / "workspace" / "uploads" / "scan.pdf"; local.parent.mkdir(parents=True, exist_ok=True)
    Image.new("RGB", (120, 160), "white").save(local, "PDF")
    result = await service.analyze({"mine_code": "m", "mine_name": "矿", "fan_name": "1号风机",
                                  "materials": [{"path": "/workspace/uploads/scan.pdf"}]}, "no-ocr", "s")
    assert result["status"] == "needs_materials"
    assert "恢复文档识别服务" in result["result"]
    assert "MAIN_FAN_VISION_MODEL" in result["analysis"]["findings"][-1]["message"]


def test_workspace_path_and_symlink_cannot_escape(tmp_path):
    root = tmp_path / "session"; (root / "workspace").mkdir(parents=True)
    outside = tmp_path / "private.pdf"; outside.write_bytes(b"private")
    (root / "workspace" / "escape.pdf").symlink_to(outside)
    for value in (str(outside), "/workspace/../private.pdf", "/workspace/escape.pdf"):
        with pytest.raises(ValueError): service.workspace_file(root, value)


@pytest.fixture
def mock_state(tmp_path):
    state = MockState(tmp_path)
    source = tmp_path / "performance.pdf"
    source.write_bytes(b"%PDF-test-file-bytes")
    state.sample["assets"]["performance_pdf"] = {"relative_path": source.name,
                        "sha256": hashlib.sha256(source.read_bytes()).hexdigest()}
    return state


@pytest.fixture
def gateway_catalog(monkeypatch):
    base = extensions.load_extensions()
    catalog = ExtensionRegistry()
    catalog.tools, catalog.agents, catalog.business_agents = dict(base.tools), dict(base.agents), dict(base.business_agents)
    catalog.business_agents["main_fan"] = replace(base.business("main_fan"), gateway=GatewayBinding("mock-main-fan", "mock-fan-screening"))
    monkeypatch.setattr(extensions, "_loaded", catalog)
    return catalog


@pytest.mark.asyncio
async def test_gateway_json_file_tokens_upload_push_and_manage(mock_state, gateway_catalog, tmp_path):
    app = create_app(mock_state)
    transport = httpx.ASGITransport(app=app)
    client = GatewayClient(GatewaySettings("http://mock", INTERNAL_TOKEN, "http://mock"), transport=transport)
    spec = gateway_catalog.business("main_fan")
    try:
        await client.register(spec)
        value = await client.proxy(spec.gateway.agent_code, "/mock-platform/main-fan/points", params={"mineCode": "mock-ci"})
        assert len(data_source.items_of(value)) == 1
        target = tmp_path / "download.pdf"
        metadata = await client.download(spec.gateway.agent_code, RemoteFile.model_validate(mock_state.file_index()[0]), target)
        assert metadata["sha256"] == mock_state.sample["assets"]["performance_pdf"]["sha256"]
        direct = next(c for c in mock_state.calls if c["operation"] == "direct_file")
        assert not direct["gateway_identity_header_present"]
        uploaded = await client.upload(spec.gateway.agent_code, target)
        assert uploaded in mock_state.uploads
        payload = {"agentCode": spec.gateway.agent_code, "taskId": "t1", "mineCode": "mock-ci", "mineName": "测试矿",
                   "result": "初步筛查", "checkTypeCode": "mock-fan-screening", "clues": []}
        assert await client.push(spec.gateway.agent_code, payload) == await client.push(spec.gateway.agent_code, payload)
        async with httpx.AsyncClient(transport=transport, base_url="http://mock") as http:
            assert (await http.get("/mock-platform/files/performance_pdf", headers=client.headers(spec.gateway.agent_code))).status_code == 401
            response = await http.get("/openapi/agent/manage", params={"agentCode": spec.gateway.agent_code, "action": "stop", "token": MANAGE_TOKEN})
            assert response.json()["data"]["onlineStatus"] == "offline"
        await client.register(spec)  # 重连注册不能解除远程停止。
        with pytest.raises(GatewayStopped):
            await client.token(spec.gateway.agent_code)
    finally:
        await client.close()


@pytest.mark.asyncio
async def test_durable_job_fetches_from_gateway_and_pushes_to_separate_platform(mock_state, gateway_catalog, monkeypatch, tmp_path):
    from extensions.result_platform.client import ResultPlatformClient
    from extensions.result_platform.settings import ResultPlatformSettings
    from mock_result_platform.app import LOCAL_TOKEN, ResultStore, create_app as create_result_app
    for key, value in {"MAIN_FAN_DEVICES_PATH": "/mock-platform/main-fan/devices",
                       "MAIN_FAN_POINTS_PATH": "/mock-platform/main-fan/points",
                       "MAIN_FAN_REPORTS_PATH": "/mock-platform/main-fan/reports"}.items():
        monkeypatch.setenv(key, value)
    transport = httpx.ASGITransport(app=create_app(mock_state))
    gateway = GatewayClient(GatewaySettings("http://mock", INTERNAL_TOKEN, "http://mock"), transport=transport)
    received = ResultStore(tmp_path / "results.sqlite")
    publisher = ResultPlatformClient(ResultPlatformSettings("http://separate-platform/api/judgments", token=LOCAL_TOKEN),
        transport=httpx.ASGITransport(app=create_result_app(received)))
    try:
        # 此单元测试把文件解析替换为样例事实；实际原 PDF 在离线演示按真实哈希验证。
        async def reader(local, virtual):
            facts = DocumentFacts.model_validate(mock_state.sample["document_facts"])
            return rebind_sources(facts, virtual, fingerprint(local))
        async def executor(**kwargs):
            data = kwargs["graph_config"]["configurable"]["business_task_input"]
            value = await service.analyze(data, kwargs["user_name"], kwargs["session_id"], gateway=gateway, document_reader=reader,
                job_id=kwargs["graph_config"]["configurable"]["business_task_id"])
            value["result"] = "模型改写的错误结论"  # 结果验证器必须恢复原件。
            return ExecutionResult("test-run", kwargs["session_id"], "success", "", "", {}, [], [], value)
        controller = BusinessTaskController(gateway, publisher=publisher, executor=executor)
        job, _ = business_store.enqueue("main_fan", "u", task_input("original"), [], publish=True)
        await controller.run_job(business_store.claim("main_fan"))
        finished = business_store.get(job["id"])
        assert finished["status"] == "completed"
        assert "模型改写" not in finished["result_json"]
        await controller.publish_job(finished)
        pushed = business_store.get(job["id"])
        assert pushed["push_status"] == "pushed"
        assert len(mock_state.uploads) == 0  # 异矿报告未参与研判，不作为本矿井证据上传。
        payload = received.rows()[0]["payload"]
        assert '陕西众源' not in payload['judgment']['reason']
        assert '523.57' not in payload['judgment']['reason']
        assert "alarmLevel" not in payload
        assert payload['judgment']['evidenceLink'] == ''
        assert not mock_state.receipts
        assert not any(call['operation'] in {'register','push','upload'} for call in mock_state.calls)
    finally:
        await gateway.close()
        await publisher.close()