"""查标第 1 步（m3-plan §5）：上传投标文件 → 检查 → 处理问题 → 复查后状态沿用；以及检查的计费（红线 9）。

假模型按字面重合判断有没有响应，结论没有参考价值；这里验证的是链路、证据、状态沿用和计费。
"""

from __future__ import annotations

from typing import Any
from uuid import UUID

import httpx
from sqlalchemy import text

from tests.conftest import FIXTURES, User, balance, project_with_tender, register, run_parse, upload


async def upload_bid(u: User, project_id: str, fixture: str, replaces: str | None = None) -> dict[str, Any]:
    done = await upload(u, fixture, (FIXTURES / fixture).read_bytes())
    assert done.status_code == 200, done.text
    body: dict[str, Any] = {"upload_id": done.json()["id"], "role": "bid_volume", "volume_kind": "technical"}
    if replaces:
        body["replaces_document_id"] = replaces
    r = await u.post(f"/projects/{project_id}/documents", json=body)
    assert r.status_code == 201, r.text
    return r.json()


async def run_check(u: User, project_id: str, doc_ids: list[str], expected: int = 0) -> dict[str, Any]:
    """发起检查，并在测试进程里直接跑流水线（不经过 worker）。"""
    from bid_doctor.check import runner

    r = await u.post(f"/projects/{project_id}/check-runs", json={"document_ids": doc_ids, "expected_charge": expected})
    assert r.status_code == 201, r.text
    await runner.run(UUID(u.tenant_id), UUID(r.json()["id"]))
    return (await u.get(f"/check-runs/{r.json()['id']}")).json()


async def checked_project(u: User) -> tuple[str, dict[str, Any], dict[str, Any]]:
    project_id, tender_id = await project_with_tender(u)
    assert (await run_parse(u, tender_id))["status"] == "succeeded"
    bid = await upload_bid(u, project_id, "bid-mini-v1.pdf")
    run = await run_check(u, project_id, [bid["id"]])
    assert run["status"] == "succeeded", run
    return project_id, bid, run


async def findings(u: User, run_id: str) -> list[dict[str, Any]]:
    return (await u.get(f"/check-runs/{run_id}/findings")).json()["items"]


async def use_pro_plan(u: User) -> None:
    from bid_doctor.core.db import TenantContext, tenant_tx

    ctx = TenantContext(tenant_id=UUID(u.tenant_id), user_id=UUID(str(u.me["user"]["id"])))
    async with tenant_tx(ctx) as s:
        await s.execute(text("UPDATE subscription SET plan_code = 'pro' WHERE tenant_id = :t"), {"t": u.tenant_id})


async def test_check_flow(user: User) -> None:
    project_id, tender_id = await project_with_tender(user)
    await run_parse(user, tender_id)
    preview = (await user.get(f"/projects/{project_id}/check-preview")).json()
    assert preview["blockers"] == ["no_bid_documents"] and not preview["can_start"]

    bid = await upload_bid(user, project_id, "bid-mini-v1.pdf")
    assert bid["label"] == "技术分册" and bid["volume_kind"] == "technical" and bid["is_latest_version"]
    preview = (await user.get(f"/projects/{project_id}/check-preview")).json()
    assert preview["can_start"] and preview["billing"] == "free_plan" and preview["charge"] == 0
    assert preview["checklist_confirmed"] is False and [d["id"] for d in preview["bid_documents"]] == [bid["id"]]

    run = await run_check(user, project_id, [bid["id"]])
    assert run["status"] == "succeeded" and run["is_latest"] and run["checklist_confirmed"] is False
    assert run["counts"] == {"fatal": 1, "important": 0, "hint": 0, "open_fatal": 1, "resolved_this_run": 0, "new": 1}
    assert "清单未确认，检查结果以当前清单为准" in run["coverage"]["not_checked"]
    assert run["coverage"]["items"]["checked"] == 3

    (f,) = await findings(user, run["id"])
    assert f["type"] == "unresponded" and f["severity"] == "fatal" and f["presence"] == "new"
    assert "BFD" in f["title"] and f["detector"] == "ai" and f["checklist_item"]["importance"] == "substantive"
    assert f["tender_citations"][0]["quote"].startswith("★交换机") and f["bid_citations"] == []
    assert f["search_scope"]["documents"] == ["技术分册"]
    assert "建议" not in f["explanation"]  # 只写事实和出处，不写整改建议

    # 「已响应」也留结论和出处（C9）
    results = (await user.get(f"/check-runs/{run['id']}/item-results")).json()["items"]
    responded = [r for r in results if r["verdict"] == "responded"]
    assert len(results) == 3 and len(responded) == 2
    for r in responded:
        (c,) = r["bid_citations"]
        assert r["evidence_status"] == "verified" and c["document_label"] == "技术分册" and c["quads"]

    project = (await user.get(f"/projects/{project_id}")).json()
    assert project["status"] == "checked" and project["readiness"] == {
        "status": "checked",
        "open_fatal_count": 1,
        "ready": False,
    }
    notes = (await user.get("/notifications")).json()["items"]
    assert notes[0]["kind"] == "check_succeeded"


async def test_resolve_and_reopen(user: User) -> None:
    project_id, _, run = await checked_project(user)
    (f,) = await findings(user, run["id"])
    r = await user.post(f"/findings/{f['id']}/resolve", json={"status": "confirmed_ok"})
    assert r.status_code == 422 and r.json()["code"] == "note_required"  # 致命问题确认无误也要写说明（U4）
    r = await user.post(f"/findings/{f['id']}/resolve", json={"status": "ignored", "note": "  "})
    assert r.status_code == 422 and r.json()["code"] == "note_required"

    r = await user.post(f"/findings/{f['id']}/resolve", json={"status": "confirmed_ok", "note": "附件 5 第 3 页有说明"})
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["finding"]["status"] == "confirmed_ok" and body["finding"]["resolved_by"]["name"] == "张工"
    assert body["project"] == {"status": "checked", "open_fatal_count": 0, "ready": True}

    from bid_doctor.core.db import TenantContext, tenant_tx

    ctx = TenantContext(tenant_id=UUID(user.tenant_id), user_id=UUID(str(user.me["user"]["id"])))
    async with tenant_tx(ctx) as s:
        fb = (await s.execute(text("SELECT object_type, kind, source, note FROM feedback"))).all()
    assert [tuple(x) for x in fb] == [("finding", "false_positive", "fatal_confirm", "附件 5 第 3 页有说明")]

    r = await user.post(f"/findings/{f['id']}/reopen")
    assert r.json()["finding"]["status"] == "pending" and r.json()["project"]["open_fatal_count"] == 1

    r = await user.post(
        f"/check-runs/{run['id']}/findings/bulk-resolve",
        json={"finding_ids": [f["id"]], "status": "ignored", "note": "x"},
    )
    assert r.status_code == 422 and r.json()["code"] == "bulk_only_hint" and r.json()["finding_ids"] == [f["id"]]
    assert (await user.get(f"/projects/{project_id}")).json()["readiness"]["open_fatal_count"] == 1


async def test_recheck_carries_status_and_resolves(user: User) -> None:
    project_id, bid, first = await checked_project(user)
    (f,) = await findings(user, first["id"])
    await user.post(f"/findings/{f['id']}/resolve", json={"status": "ignored", "note": "BFD 写在附件里"})

    # 同一份投标文件再查：问题还在，沿用上次的处理状态
    second = await run_check(user, project_id, [bid["id"]])
    (g,) = await findings(user, second["id"])
    assert g["presence"] == "carried" and g["status"] == "ignored" and g["carried_status"] == "ignored"
    assert g["resolution_note"] == "BFD 写在附件里" and second["counts"]["open_fatal"] == 0

    # 历史批次只读
    r = await user.post(f"/findings/{f['id']}/reopen")
    assert r.status_code == 409 and r.json()["code"] == "run_superseded"

    # 上次「已修改」又查出来：回到待处理
    await user.post(f"/findings/{g['id']}/resolve", json={"status": "modified"})
    third = await run_check(user, project_id, [bid["id"]])
    (h,) = await findings(user, third["id"])
    assert h["presence"] == "carried" and h["status"] == "pending" and h["carried_status"] == "modified"

    # 换上补了 BFD 的新版本：问题本次已解决
    v2 = await upload_bid(user, project_id, "bid-mini-v2.pdf", replaces=bid["id"])
    assert v2["label"] == "技术分册 v2" and v2["lineage_id"] == bid["lineage_id"]
    assert not (await user.get(f"/documents/{bid['id']}")).json()["is_latest_version"]
    preview = (await user.get(f"/projects/{project_id}/check-preview")).json()
    assert [d["id"] for d in preview["bid_documents"]] == [v2["id"]]
    fourth = await run_check(user, project_id, [v2["id"]])
    (k,) = await findings(user, fourth["id"])
    assert k["presence"] == "resolved" and fourth["counts"]["resolved_this_run"] == 1
    assert fourth["counts"]["fatal"] == 0 and fourth["counts"]["open_fatal"] == 0
    r = await user.post(f"/findings/{k['id']}/resolve", json={"status": "ignored", "note": "x"})
    assert r.status_code == 409 and r.json()["code"] == "finding_resolved_this_run"
    project = (await user.get(f"/projects/{project_id}")).json()
    assert project["readiness"]["ready"] is True and project["latest_check_run_id"] == fourth["id"]


async def test_pro_plan_charges_then_waives_recheck(app_client: httpx.AsyncClient) -> None:
    u = await register(app_client)
    await use_pro_plan(u)
    project_id, tender_id = await project_with_tender(u)
    await run_parse(u, tender_id)
    bid = await upload_bid(u, project_id, "bid-mini-v1.pdf")
    assert await balance(u) == 2

    preview = (await u.get(f"/projects/{project_id}/check-preview")).json()
    assert preview["billing"] == "charged" and preview["charge"] == 1 and preview["waive_until"] is None
    r = await u.post(f"/projects/{project_id}/check-runs", json={"document_ids": [bid["id"]], "expected_charge": 0})
    assert r.status_code == 409 and r.json()["code"] == "billing_changed" and r.json()["preview"]["charge"] == 1
    assert await balance(u) == 2  # 扣点说明和实际不一致：拒绝，不扣点（API4）

    first = await run_check(u, project_id, [bid["id"]], expected=1)
    assert first["billing"] == "charged" and await balance(u) == 1

    preview = (await u.get(f"/projects/{project_id}/check-preview")).json()
    assert preview["billing"] == "waived_recheck" and preview["charge"] == 0 and preview["waive_until"]
    second = await run_check(u, project_id, [bid["id"]], expected=0)
    assert second["billing"] == "waived_recheck" and await balance(u) == 1

    from bid_doctor.core.db import TenantContext, tenant_tx

    ctx = TenantContext(tenant_id=UUID(u.tenant_id), user_id=UUID(str(u.me["user"]["id"])))
    async with tenant_tx(ctx) as s:
        rows = (
            await s.execute(
                text("SELECT kind, amount, reason FROM quota_entry WHERE ref_type = 'check_run' ORDER BY created_at")
            )
        ).all()
    assert [tuple(x) for x in rows] == [("hold", -1, None), ("settle", 0, None), ("waive", 0, "recheck_24h")]


async def test_unreadable_bid_fails_and_refunds(app_client: httpx.AsyncClient) -> None:
    """投标文件读不出文字（扫描件）：按输入问题失败，退回额度，不生成问题。"""
    u = await register(app_client)
    await use_pro_plan(u)
    project_id, tender_id = await project_with_tender(u)
    await run_parse(u, tender_id)
    bid = await upload_bid(u, project_id, "blank.pdf")
    run = await run_check(u, project_id, [bid["id"]], expected=1)
    assert run["status"] == "failed" and run["error"]["code"] == "bid_text_missing" and run["error"]["refunded"]
    assert await balance(u) == 2 and await findings(u, run["id"]) == []
    project = (await u.get(f"/projects/{project_id}")).json()
    assert project["status"] == "drafting" and project["latest_check_run_id"] is None


async def test_duplicate_check_is_rejected(user: User) -> None:
    project_id, tender_id = await project_with_tender(user)
    await run_parse(user, tender_id)
    bid = await upload_bid(user, project_id, "bid-mini-v1.pdf")
    body = {"document_ids": [bid["id"]], "expected_charge": 0}
    assert (await user.post(f"/projects/{project_id}/check-runs", json=body)).status_code == 201
    r = await user.post(f"/projects/{project_id}/check-runs", json=body)
    assert r.status_code == 409 and r.json()["code"] == "job_already_running" and r.json()["job_id"]
    preview = (await user.get(f"/projects/{project_id}/check-preview")).json()
    assert "job_already_running" in preview["blockers"]


async def test_platform_encrypted_bid_is_rejected(user: User) -> None:
    done = await upload(user, "投标文件.kedt", b"PK\x03\x04not-really-a-zip")
    assert done.status_code == 422 and done.json()["code"] == "bid_file_platform_encrypted"
    assert "苏采云" in done.json()["detail"]


async def test_other_tenant_cannot_see_check(app_client: httpx.AsyncClient) -> None:
    a = await register(app_client, "甲公司")
    b = await register(app_client, "乙公司")
    project_id, _, run = await checked_project(a)
    (f,) = await findings(a, run["id"])
    for path in (
        f"/projects/{project_id}/check-preview",
        f"/projects/{project_id}/check-runs",
        f"/check-runs/{run['id']}",
        f"/check-runs/{run['id']}/item-results",
    ):
        r = await b.get(path)
        assert r.status_code == 404, (path, r.status_code, r.text)
    assert (await b.get(f"/check-runs/{run['id']}/findings")).status_code == 404
    assert (await b.post(f"/findings/{f['id']}/resolve", json={"status": "modified"})).status_code == 404
    assert (await b.post(f"/findings/{f['id']}/reopen")).status_code == 404
    assert (await b.post(f"/check-runs/{run['id']}/cancel")).status_code == 404


async def test_response_modes_are_stored_editable_and_used(user: User) -> None:
    """响应方式（C12）：解析时存进清单；用户改成「默认接受」后，查标对没复述的一般要求不再报问题。"""
    project_id, tender_id = await project_with_tender(user)
    await run_parse(user, tender_id)
    modes = {m["category"]: m for m in (await user.get(f"/projects/{project_id}/checklist")).json()["response_modes"]}
    assert set(modes) == {"technical", "commercial"}
    assert modes["technical"]["mode"] == "explicit" and modes["technical"]["status"] == "unknown"

    r = await user.put(f"/projects/{project_id}/checklist/response-modes/technical", json={"mode": "implicit"})
    assert r.status_code == 200, r.text
    tech = next(m for m in r.json()["response_modes"] if m["category"] == "technical")
    assert tech["origin"] == "human" and tech["mode"] == "implicit" and tech["set_by"]["name"] == "张工"

    bid = await upload_bid(user, project_id, "bid-mini-v1.pdf")
    run = await run_check(user, project_id, [bid["id"]])
    # 漏掉的 BFD 是 ★ 条款：默认接受下不报致命，改为提示级「请人工核对」
    (f,) = await findings(user, run["id"])
    assert f["type"] == "manual_review" and f["severity"] == "hint"
    assert run["counts"]["fatal"] == 0


async def test_rule_findings_are_written_with_citations_and_carried(user: User) -> None:
    """规则检查（报价形式、形式要件）：问题挂规则编号和投标原文，复查时按 match_key 沿用处理状态。"""
    project_id, tender_id = await project_with_tender(user)
    await run_parse(user, tender_id)
    bid = await upload_bid(user, project_id, "bid-mini-forms.pdf")
    run = await run_check(user, project_id, [bid["id"]])
    assert run["status"] == "succeeded", run
    rule = {f["rule"]["code"]: f for f in await findings(user, run["id"]) if f["detector"] == "rule"}
    assert set(rule) == {
        "PRICE.CAPITAL_MISMATCH",
        "PRICE.MULTIPLE_QUOTES",
        "FORM.BLANK_FIELD",
        "FORM.CONDITION_ADDED",
    }
    cap = rule["PRICE.CAPITAL_MISMATCH"]
    assert cap["type"] == "price_form" and cap["severity"] == "important" and cap["checklist_item"] is None
    assert cap["rule"]["description"] == "大写金额和小写金额核对"
    (c,) = cap["bid_citations"]  # 同一页的几句合成一条出处，各句分别画框
    assert "壹拾贰万捌仟元整" in c["quote"] and "1,280,000.00" in c["quote"] and len(c["quads"]) >= 2
    assert c["document_label"] == "技术分册"
    # 「让利后总价」和投标总价不同：多半另报了一个价，但不是「备选方案」那种明确的两个报价，按重要
    assert rule["PRICE.MULTIPLE_QUOTES"]["severity"] == "important" and rule["FORM.BLANK_FIELD"]["severity"] == "hint"
    assert any("报价形式" in x for x in run["coverage"]["checked"])
    assert run["coverage"]["rule_set_version"] == "2026.11-r2"

    await user.post(f"/findings/{cap['id']}/resolve", json={"status": "ignored", "note": "大写栏已手工更正"})
    again = await run_check(user, project_id, [bid["id"]])
    cap2 = next(
        f for f in await findings(user, again["id"]) if (f["rule"] or {}).get("code") == "PRICE.CAPITAL_MISMATCH"
    )
    assert cap2["presence"] == "carried" and cap2["status"] == "ignored"


async def test_manual_checks_are_kept_per_project(user: User) -> None:
    """需人工核对：来自清单（按主题合并），勾选按项目存、写勾选人和时间，复查后还在（决策 P2）。"""
    project_id, _, _ = await checked_project(user)
    for title in ("须用CA数字证书加密投标文件", "开标后30分钟内在线解密投标文件", "本次采购不收取投标保证金"):
        r = await user.post(
            f"/projects/{project_id}/checklist/items", json={"categories": ["schedule"], "title": title}
        )
        assert r.status_code == 201, r.text
    items = (await user.get(f"/projects/{project_id}/manual-checks")).json()["items"]
    assert [(i["check_key"], i["checked"]) for i in items] == [
        ("rule:MANUAL.DECRYPT", False),
        ("rule:MANUAL.SIGN_ENCRYPT", False),
    ]
    assert items[1]["requirement_text"] == "须用CA数字证书加密投标文件"

    r = await user.put(
        f"/projects/{project_id}/manual-checks",
        json={"check_key": "rule:MANUAL.DECRYPT", "checked": True, "note": "已演练"},
    )
    assert r.status_code == 200, r.text
    m = r.json()
    assert m["checked"] and m["note"] == "已演练" and m["checked_by"]["name"] and m["checked_at"]
    bad = await user.put(
        f"/projects/{project_id}/manual-checks", json={"check_key": "rule:MANUAL.NOPE", "checked": True}
    )
    assert bad.status_code == 422

    bid_id = (await user.get(f"/projects/{project_id}/check-preview")).json()["bid_documents"][0]["id"]
    again = await run_check(user, project_id, [bid_id])
    assert any("列入「需人工核对」2 条" in x for x in again["coverage"]["not_checked"])
    kept = (await user.get(f"/projects/{project_id}/manual-checks")).json()["items"]
    assert [i["checked"] for i in kept] == [True, False]


async def test_recheck_reuses_unchanged_model_results(user: User) -> None:
    """复查复用（C2）：同一份投标文件再查，送给模型的内容没变，模型那一层沿用上次的结论，不再调模型。"""
    from bid_doctor.ai.models import AiCall
    from bid_doctor.core.db import TenantContext, tenant_tx

    project_id, bid, first = await checked_project(user)
    second = await run_check(user, project_id, [bid["id"]])
    assert second["counts"]["fatal"] == first["counts"]["fatal"]
    reused = [x for x in second["coverage"]["checked"] if "沿用上次的结论" in x]
    assert reused, second["coverage"]["checked"]

    ctx = TenantContext(tenant_id=UUID(user.tenant_id), user_id=UUID(str(user.me["user"]["id"])))
    async with tenant_tx(ctx) as s:
        from sqlalchemy import func, select

        n = (await s.execute(select(func.count()).where(AiCall.job_id == UUID(second["id"])))).scalar_one()
    assert n == 0
    results = (await user.get(f"/check-runs/{second['id']}/item-results")).json()["items"]
    assert all(r["bid_citations"] for r in results if r["verdict"] == "responded")  # 沿用的结论证据也找回来了