"""Preparation substages describe observed work without echoing source text."""

from __future__ import annotations

import os
from types import SimpleNamespace

os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")

import pytest
from PySide6.QtCore import QEvent, QTimer
from PySide6.QtWidgets import QApplication

from doctor_workstation.core.errors import ApiBusinessError
from doctor_workstation.ui.dialogs import ai_consult as module


@pytest.fixture(scope="module")
def application():
    return QApplication.instance() or QApplication([])


@pytest.fixture
def dialog(application, monkeypatch):
    monkeypatch.setattr(module, "QThreadPool", SimpleNamespace(
        globalInstance=lambda: SimpleNamespace(start=lambda _: None),
    ))
    instance = module.AiConsultDialog(object())
    monkeypatch.setattr(instance, "_load_workspace", lambda _: None)
    instance.open_for(diagnosis_id=501, patient_id=1501)
    yield instance
    for timer in instance.findChildren(QTimer):
        timer.stop()
    instance.close()
    instance.deleteLater()
    application.sendPostedEvents(None, QEvent.Type.DeferredDelete)
    application.processEvents()


def emit(dialog, payload):
    dialog._stream_event(dialog._generation, dialog._stream_generation, payload)


def test_queued_request_at_52_seconds_does_not_claim_preparation(dialog, monkeypatch):
    now = [100.0]
    monkeypatch.setattr(module, "monotonic", lambda: now[0])
    dialog._ask("合成完整资料核对")
    worker, bubble = dialog._stream_worker, dialog._stream_bubble
    assert worker.prompt == "合成完整资料核对" and worker.diagnosis_id == 501
    assert worker.run_started_at is None
    now[0] = 152.0
    dialog._stream_elapsed_timer.timeout.emit()
    assert bubble._raw_payload == "正在连接 AI 服务，等待资料处理进度…"
    assert "52 秒" in bubble.stamp.text()
    assert "正在整理" not in bubble.body.toPlainText()
    assert dialog._asking and dialog._stream_timeout_timer.isActive()

    emit(dialog, {"event": "start", "message": "禁止显示的患者临床信息"})
    assert bubble._raw_payload == module.AI_STREAM_WAITING_TEXT
    assert module.AI_STREAM_WAITING_TEXT in bubble.transfer_status.text()
    emit(dialog, {"event": "progress", "phase": "preparing"})
    assert bubble._raw_payload == module.AI_STREAM_PREPARING_TEXT


@pytest.mark.parametrize("status,expected", [
    ("aggregating", "正在汇总本次授权范围内的患者资料…"),
    ("aggregated", "患者资料已汇总，正在检查提交内容…"),
    ("ready", "资料正文已整理，正在准备提交…"),
    (None, module.AI_STREAM_PREPARING_TEXT),
    ("禁止显示的患者临床信息", module.AI_STREAM_PREPARING_TEXT),
    (["禁止显示的患者临床信息"], module.AI_STREAM_PREPARING_TEXT),
])
def test_known_preparation_states_use_safe_local_copy(dialog, status, expected):
    dialog._ask("合成资料核对")
    emit(dialog, {
        "event": "progress", "phase": "preparing", "status": status,
        "message": "禁止显示的患者临床信息 https://example.test/private.pdf",
    })
    bubble = dialog._stream_bubble
    assert bubble._raw_payload == expected
    assert bubble.transfer_status.text().splitlines()[0] == expected
    assert "禁止显示" not in bubble.body.toPlainText() + bubble.transfer_status.text()
    assert "example.test" not in bubble.body.toPlainText() + bubble.transfer_status.text()
    assert "资料已提交" not in bubble._raw_payload
    assert dialog._asking and not dialog._stream_completed


@pytest.mark.parametrize("completed,total", [(0, 3), (1, 3), (3, 3)])
def test_compacting_counts_same_model_history_work_and_keeps_all_13_receipts(
    dialog, completed, total,
):
    dialog._ask("合成完整病历核对")
    manifest = [
        {"key": key, "label": "禁止显示的患者临床信息", "status": "prepared"}
        for key, _ in module._REQUIRED_TRANSFER_SECTIONS
    ]
    emit(dialog, {
        "event": "progress", "phase": "preparing", "status": "aggregated",
        "required_sections": manifest, "source_summary": {"diagnosis_count": 23},
    })
    emit(dialog, {
        "event": "progress", "phase": "preparing", "status": "compacting",
        "completed": completed, "total": total, "message": "禁止显示的患者临床信息",
    })
    bubble = dialog._stream_bubble
    assert "当前 AI 模型整理历史资料" in bubble._raw_payload
    assert f"已完成 {completed}/{total}" in bubble._raw_payload
    assert "完整病历尚未提交综合分析" in bubble._raw_payload
    receipt = bubble.transfer_status.text()
    assert all(f"{label}：已整理，待提交" in receipt
               for _, label in module._REQUIRED_TRANSFER_SECTIONS)
    assert len(manifest) == 13
    assert dialog._stream_meta["required_sections"] == manifest
    assert "诊单 23 条" in receipt and "禁止显示" not in receipt
    assert "资料正文已提交" not in receipt
    assert dialog._asking and dialog._stream_first_delta_at is None


@pytest.mark.parametrize("completed,total", [
    (None, 3), (0, None), (-1, 3), (4, 3), (0, 0), (-1, -2),
    (True, 3), (1, False), (1.0, 3), (1, 3.0), ("1", 3), (1, "3"),
    (float("nan"), 3), (1, float("inf")),
    ({"private": "禁止显示的患者临床信息"}, 3),
    (1, ["禁止显示的患者临床信息"]),
])
def test_invalid_compacting_counts_are_not_rendered(dialog, completed, total):
    dialog._ask("合成资料核对")
    emit(dialog, {
        "event": "progress", "phase": "preparing", "status": "compacting",
        "completed": completed, "total": total, "message": "禁止显示的患者临床信息",
    })
    bubble = dialog._stream_bubble
    assert bubble._raw_payload == (
        "正在由当前 AI 模型整理历史资料，完整病历尚未提交综合分析…"
    )
    assert "已完成" not in bubble.transfer_status.text()
    assert "禁止显示" not in bubble.body.toPlainText() + bubble.transfer_status.text()


def test_late_preparation_progress_never_replaces_received_answer(dialog):
    dialog._ask("合成资料核对")
    emit(dialog, {"event": "delta", "text": "合成正文"})
    for status in ("aggregating", "aggregated", "compacting", "ready"):
        emit(dialog, {
            "event": "progress", "phase": "preparing", "status": status,
            "completed": 1, "total": 2, "message": "禁止显示的患者临床信息",
        })
        assert dialog._stream_bubble._raw_payload == "合成正文"
    emit(dialog, {"event": "delta", "text": "未刷新尾段"})
    emit(dialog, {"event": "done"})
    assert dialog._stream_bubble._raw_payload == "合成正文未刷新尾段"
    assert dialog.send_button.isEnabled() and not dialog._asking


@pytest.mark.parametrize("ending", ["done", "error", "eof", "timeout", "cancelled"])
def test_compacting_exits_on_terminal_path_and_ignores_late_preparation(dialog, ending):
    dialog._ask("合成资料核对")
    emit(dialog, {
        "event": "progress", "phase": "preparing", "status": "compacting",
        "completed": 1, "total": 2,
    })
    generation, stream = dialog._generation, dialog._stream_generation
    worker, bubble = dialog._stream_worker, dialog._stream_bubble
    if ending == "done":
        emit(dialog, {"event": "done", "answer": "合成最终正文"})
        assert bubble._raw_payload == "合成最终正文"
    elif ending == "error":
        dialog._stream_failed(generation, stream, RuntimeError("合成失败"))
        assert "本次请求失败" in bubble.transfer_status.text()
    elif ending == "eof":
        dialog._stream_finished(generation, stream, worker)
        assert bubble._raw_payload == module.AI_STREAM_SILENT_TEXT
    elif ending == "timeout":
        dialog._stream_timed_out()
        assert module.AI_STREAM_TIMEOUT_TEXT in bubble.transfer_status.text()
        assert worker.is_cancelled()
    else:
        emit(dialog, {"event": "progress", "phase": "cancelled", "cancel_requested": True})
        dialog._stream_failed(generation, stream, ApiBusinessError(
            "已取消", code=409, data={"code": "AI_ASSISTANT_CANCELLED"},
        ))
        assert "取消" in bubble.transfer_status.text()
    assert not dialog._asking and dialog.send_button.isEnabled()
    assert not dialog._stream_timeout_timer.isActive()
    assert not dialog._stream_elapsed_timer.isActive()
    before = bubble._raw_payload, bubble.transfer_status.text(), bubble.stamp.text()
    dialog._stream_event(generation, stream, {
        "event": "progress", "phase": "preparing", "status": "compacting",
        "completed": 2, "total": 2,
    })
    assert (bubble._raw_payload, bubble.transfer_status.text(), bubble.stamp.text()) == before