"""检查批次与问题处理（FR-M3-02、07、08，api-design §4.1、§4.5）。

- 计费（data-model §4.3）：免费版不记流水；专业版 24 小时内复查免扣（记一行 0）；其余在解析额度上预扣 1 次。
  预览和发起共用同一个计费函数；前端带来的 expected_charge 和服务端当下算的不一致时拒绝，不扣点（API4）。
- 「预扣 → 建批次 → 入队」在同一个事务里（架构 §3.4）。
- 只有最近一次检查的问题可以处理；致命问题「确认无误」也要写说明，说明同时进误报反馈（U4）。
"""

from __future__ import annotations

from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from typing import Any
from uuid import UUID

from fastapi.encoders import jsonable_encoder
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from bid_doctor.audit import service as audit
from bid_doctor.billing import service as billing
from bid_doctor.check import manual, rules
from bid_doctor.check.models import CheckItemResult, CheckRun, CheckRunDocument, Finding, FindingCitation, ManualCheck
from bid_doctor.checklist import edit as checklist_edit
from bid_doctor.checklist import service as checklist
from bid_doctor.core import queue
from bid_doctor.core.db import TenantContext
from bid_doctor.core.errors import AppError
from bid_doctor.core.ids import uuid7
from bid_doctor.document import citations
from bid_doctor.document import service as documents
from bid_doctor.document.models import SourceDocument
from bid_doctor.document.pipeline.text_layer import norm
from bid_doctor.identity import service as identity
from bid_doctor.ops import service as feedback
from bid_doctor.project import service as projects
from bid_doctor.project.models import Project

CHECK_TASK = "check_project"
RULE_SET_VERSION = "2026.11-r2"  # 规则集有版本，检查批次记下当时用的版本（架构 §3.7）；r2：加分项 / 单价上限
WAIVE_WINDOW = timedelta(hours=24)
ERROR_MESSAGES = {
    "bid_text_missing": "投标文件里几乎读不出文字（可能都是扫描件），扫描件识别还没有上线",
    "file_corrupted": "投标文件无法转换，可能已加密或损坏",
    "plan_limit_exceeded": "投标文件页数超出套餐上限",
    "checklist_empty": "正式清单还没有条目",
    "converter_unavailable": "文档转换服务暂时不可用",
    "model_unavailable": "模型服务暂时不可用",
    "internal_error": "系统出错了",
}


@dataclass(frozen=True, slots=True)
class Billing:
    mode: str  # charged / waived_recheck / free_plan
    charge: int
    waive_until: datetime | None


async def billing_mode(s: AsyncSession, ctx: TenantContext, p: Project) -> Billing:
    _, plan = await billing.current_plan(s, ctx)
    if plan.code == "free":
        return Billing("free_plan", 0, None)  # 免费版只限 1 个项目，项目内检查不限次数（01 §4）
    last = p.last_charged_check_at
    if last is not None and datetime.now(UTC) - last < WAIVE_WINDOW:
        return Billing("waived_recheck", 0, last + WAIVE_WINDOW)  # 从上一次扣了额度的检查算起，不重新计时（P1）
    return Billing("charged", 1, None)


async def _running(s: AsyncSession, project_id: UUID) -> CheckRun | None:
    return (
        await s.execute(
            select(CheckRun).where(CheckRun.project_id == project_id, CheckRun.status.in_(("queued", "running")))
        )
    ).scalar_one_or_none()


async def _tender_ready(s: AsyncSession, project_id: UUID) -> bool:
    n = (
        await s.execute(
            select(func.count()).where(
                SourceDocument.project_id == project_id,
                SourceDocument.role == "tender",
                SourceDocument.status == "ready",
            )
        )
    ).scalar_one()
    return n > 0


async def preview(s: AsyncSession, ctx: TenantContext, project_id: UUID) -> dict[str, Any]:
    p = await projects.get(s, ctx, project_id)
    mode = await billing_mode(s, ctx, p)
    bids = await documents.latest_bid_volumes(s, project_id)
    blockers: list[str] = []
    if p.status in ("submitted", "archived"):
        blockers.append("project_read_only")
    if not await _tender_ready(s, project_id):
        blockers.append("no_parsed_tender")
    elif not await checklist.items(s, project_id, zone="formal"):
        blockers.append("checklist_empty")
    if not bids:
        blockers.append("no_bid_documents")
    if await _running(s, project_id):
        blockers.append("job_already_running")
    if mode.charge and await billing.balance(s, ctx, "parse") < 1:
        blockers.append("quota_exhausted")
    states = await checklist.statuses(s, [project_id])
    return {
        "can_start": not blockers,
        "blockers": blockers,
        "billing": mode.mode,
        "charge": mode.charge,
        "charge_unit": "parse",
        "waive_until": mode.waive_until,
        "checklist_confirmed": states.get(project_id) == "confirmed",
        "pending_item_count": (await checklist.pending_counts(s, [project_id])).get(project_id, 0),
        "bid_documents": [await documents.doc_to_api(s, d) for d in bids],
        "lot_ids": await projects.bidding_lot_ids(s, project_id),
    }


async def start(
    s: AsyncSession, ctx: TenantContext, project_id: UUID, document_ids: list[UUID], expected_charge: int
) -> CheckRun:
    p = await projects.get(s, ctx, project_id)
    projects.ensure_editable(p)
    if running := await _running(s, project_id):
        raise AppError("job_already_running", job_id=str(running.id))
    if not await _tender_ready(s, project_id):
        raise AppError("no_parsed_tender")
    docs = (
        (await s.execute(select(SourceDocument).where(SourceDocument.id.in_(document_ids)))).scalars().all()
        if document_ids
        else []
    )
    if not docs or len(docs) != len(set(document_ids)):
        raise AppError("no_bid_documents")
    if any(d.project_id != project_id or d.role != "bid_volume" for d in docs):
        raise AppError("not_found")
    mode = await billing_mode(s, ctx, p)
    if expected_charge != mode.charge:
        raise AppError("billing_changed", preview=jsonable_encoder(await preview(s, ctx, project_id)))
    version, confirmed = await checklist_edit.check_snapshot(s, ctx, project_id)
    run_no = (
        await s.execute(
            select(func.coalesce(func.max(CheckRun.run_no), 0) + 1).where(CheckRun.project_id == project_id)
        )
    ).scalar_one()
    run = CheckRun(
        id=uuid7(),
        tenant_id=ctx.tenant_id,
        project_id=project_id,
        run_no=run_no,
        checklist_version_id=version.id,
        checklist_confirmed=confirmed,
        rule_set_version=RULE_SET_VERSION,
        legal_regime=p.legal_regime,
        lot_ids=await projects.bidding_lot_ids(s, project_id),
        previous_run_id=p.latest_check_run_id,
        billing=mode.mode,
        created_by=ctx.user_id,
    )
    if mode.mode == "charged":
        run.quota_hold_id = await billing.hold(
            s, ctx, unit="parse", ref_type="check_run", ref_id=run.id, project_id=project_id
        )
    s.add(run)
    await s.flush()
    for d in docs:
        s.add(CheckRunDocument(tenant_id=ctx.tenant_id, check_run_id=run.id, document_id=d.id))
    if mode.mode == "waived_recheck":
        await billing.waive(
            s, ctx, unit="parse", ref_type="check_run", ref_id=run.id, project_id=project_id, reason="recheck_24h"
        )
    run.queue_job_id = await queue.enqueue(
        s, CHECK_TASK, {"tenant_id": str(ctx.tenant_id), "run_id": str(run.id)}, lock=f"project:{project_id}"
    )
    await audit.record(
        s,
        ctx,
        "check_run.start",
        "check_run",
        run.id,
        project_id=project_id,
        after={"run_no": run_no, "billing": mode.mode, "documents": [str(d.id) for d in docs]},
    )
    await s.flush()
    await s.refresh(run)
    return run


async def get_run(s: AsyncSession, run_id: UUID) -> CheckRun:
    run = (await s.execute(select(CheckRun).where(CheckRun.id == run_id))).scalar_one_or_none()
    if run is None:
        raise AppError("not_found")
    return run


async def request_cancel(s: AsyncSession, run_id: UUID) -> CheckRun:
    run = await get_run(s, run_id)
    if run.status not in ("queued", "running"):
        raise AppError("job_finished")
    run.cancel_requested_at = datetime.now(UTC)
    await s.flush()
    return run


async def list_runs(s: AsyncSession, project_id: UUID) -> list[CheckRun]:
    rows = await s.execute(select(CheckRun).where(CheckRun.project_id == project_id).order_by(CheckRun.run_no.desc()))
    return list(rows.scalars())


async def run_to_api(s: AsyncSession, ctx: TenantContext, run: CheckRun) -> dict[str, Any]:
    p = await projects.get(s, ctx, run.project_id)
    doc_ids = list(
        (await s.execute(select(CheckRunDocument.document_id).where(CheckRunDocument.check_run_id == run.id))).scalars()
    )
    docs = (await s.execute(select(SourceDocument).where(SourceDocument.id.in_(doc_ids)))).scalars().all()
    version_no = (await checklist_edit.version_numbers(s, [run.checklist_version_id])).get(run.checklist_version_id, 0)
    names = await identity.display_names(s, [run.created_by])
    error = None
    if run.status in ("failed", "cancelled") and run.error_code:
        error = {
            "code": run.error_code,
            "message": ERROR_MESSAGES.get(run.error_code, run.error_code),
            "refunded": run.billing == "charged",
        }
    return {
        "id": run.id,
        "run_no": run.run_no,
        "status": run.status,
        "stage": run.stage,
        "progress": run.progress,
        "cancel_requested": run.cancel_requested_at is not None,
        "error": error,
        "billing": run.billing,
        "is_latest": p.latest_check_run_id == run.id,
        "checklist_version_no": version_no,
        "checklist_confirmed": run.checklist_confirmed,
        "rule_set_version": run.rule_set_version,
        "legal_regime": run.legal_regime,
        "documents": [await documents.doc_to_api(s, d) for d in docs],
        "coverage": run.coverage,
        "counts": run.counts,
        "created_by": {"id": run.created_by, "name": names.get(run.created_by, "")},
        "created_at": run.created_at,
        "finished_at": run.finished_at,
    }


# ------------------------------------------------------------------ 问题


async def _items_of(s: AsyncSession, run: CheckRun) -> dict[UUID, dict[str, Any]]:
    return await checklist_edit.version_items(s, run.checklist_version_id)


async def findings(
    s: AsyncSession, run_id: UUID, *, severity: str | None, status: str | None, presence: str | None
) -> list[dict[str, Any]]:
    run = await get_run(s, run_id)
    q = select(Finding).where(Finding.check_run_id == run_id)
    if severity:
        q = q.where(Finding.severity == severity)
    if status:
        q = q.where(Finding.status == status)
    if presence:
        q = q.where(Finding.presence == presence)
    rows = list((await s.execute(q)).scalars())
    return await _to_api_many(s, run, rows)


ORDER = {"fatal": 0, "important": 1, "hint": 2}


async def _to_api_many(s: AsyncSession, run: CheckRun, rows: list[Finding]) -> list[dict[str, Any]]:
    """按级别、再按招标原文顺序（清单项的排序键）。"""
    items = await _items_of(s, run)
    links = (
        await s.execute(
            select(FindingCitation)
            .where(FindingCitation.finding_id.in_([f.id for f in rows]))
            .order_by(FindingCitation.position)
        )
    ).scalars()
    by_finding: dict[UUID, list[FindingCitation]] = {}
    for link in links:
        by_finding.setdefault(link.finding_id, []).append(link)
    cites = await citations.to_api_many(s, [link.citation_id for ls in by_finding.values() for link in ls])
    names = await identity.display_names(s, [f.resolved_by for f in rows if f.resolved_by])
    sort_keys = await checklist.sort_keys(s, [f.checklist_item_id for f in rows if f.checklist_item_id])
    # 重复的要求（决策 C5 修订）：每条照常单独判断，问题类型相同、要求原文完全相同的折叠显示
    groups: dict[tuple[str, str], list[UUID]] = {}
    for f in rows:
        item = items.get(f.checklist_item_id) if f.checklist_item_id else None
        text = norm(str((item or {}).get("requirement_text") or (item or {}).get("title") or ""))
        if item and f.presence != "resolved" and len(text) >= 6:
            groups.setdefault((f.type, text), []).append(f.id)
    same = {fid: [x for x in ids if x != fid] for ids in groups.values() if len(ids) > 1 for fid in ids}

    def api(f: Finding) -> dict[str, Any]:
        item = items.get(f.checklist_item_id) if f.checklist_item_id else None
        ls = by_finding.get(f.id, [])
        return {
            "id": f.id,
            "type": f.type,
            "severity": f.severity,
            "title": f.title,
            "explanation": f.explanation,
            "detector": f.detector,
            "checklist_item": {
                "id": f.checklist_item_id,
                "title": item.get("title"),
                "importance": item.get("importance"),
                "categories": item.get("categories"),
            }
            if item
            else None,
            "rule": {"code": f.rule_code, "description": rules.LABELS.get(f.rule_code, "")}
            if f.rule_code and not item
            else None,
            "detail": f.detail,
            "tender_citations": [cites[x.citation_id] for x in ls if x.side == "tender" and x.citation_id in cites],
            "bid_citations": [cites[x.citation_id] for x in ls if x.side == "bid" and x.citation_id in cites],
            "search_scope": f.search_scope,
            "duplicate_ids": same.get(f.id, []),
            "presence": f.presence,
            "carried_status": (f.detail or {}).get("carried_status"),
            "status": f.status,
            "resolution_note": f.resolution_note,
            "resolved_by": {"id": f.resolved_by, "name": names.get(f.resolved_by, "")} if f.resolved_by else None,
            "resolved_at": f.resolved_at,
        }

    rows.sort(key=lambda f: (ORDER[f.severity], sort_keys.get(f.checklist_item_id or UUID(int=0), "~"), f.title))
    return [api(f) for f in rows]


async def _get_finding(s: AsyncSession, finding_id: UUID) -> Finding:
    f = (await s.execute(select(Finding).where(Finding.id == finding_id))).scalar_one_or_none()
    if f is None:
        raise AppError("not_found")
    return f


async def _editable_run(s: AsyncSession, ctx: TenantContext, run_id: UUID) -> tuple[CheckRun, Project]:
    run = await get_run(s, run_id)
    p = await projects.get(s, ctx, run.project_id)
    projects.ensure_editable(p)
    if p.latest_check_run_id != run.id:
        raise AppError("run_superseded")
    return run, p


async def open_fatal(s: AsyncSession, run_id: UUID) -> int:
    return (
        await s.execute(
            select(func.count()).where(
                Finding.check_run_id == run_id,
                Finding.severity == "fatal",
                Finding.status == "pending",
                Finding.presence != "resolved",
            )
        )
    ).scalar_one()


def _require_note(severity: str, status: str, note: str | None) -> None:
    if (status == "ignored" or (severity == "fatal" and status == "confirmed_ok")) and not (note or "").strip():
        raise AppError("note_required")


async def _apply(s: AsyncSession, ctx: TenantContext, f: Finding, p: Project, status: str, note: str | None) -> None:
    before = {"status": f.status, "note": f.resolution_note}
    done = status != "pending"
    f.status = status
    f.resolution_note = ((note or "").strip() or None) if done else None
    f.resolved_by, f.resolved_at = (ctx.user_id, datetime.now(UTC)) if done else (None, None)
    if f.severity == "fatal" and status == "confirmed_ok":
        await feedback.record(
            s,
            ctx,
            project_id=p.id,
            object_type="finding",
            object_id=f.id,
            kind="false_positive",
            source="fatal_confirm",
            note=f.resolution_note,
        )
    await audit.record(
        s,
        ctx,
        "finding.resolve" if done else "finding.reopen",
        "finding",
        f.id,
        project_id=p.id,
        before=before,
        after={"status": f.status, "note": f.resolution_note},
    )


async def resolve(
    s: AsyncSession, ctx: TenantContext, finding_id: UUID, status: str, note: str | None
) -> dict[str, Any]:
    f = await _get_finding(s, finding_id)
    run, p = await _editable_run(s, ctx, f.check_run_id)
    if f.presence == "resolved":
        raise AppError("finding_resolved_this_run")
    _require_note(f.severity, status, note)
    await _apply(s, ctx, f, p, status, note)
    await s.flush()
    p.open_fatal_count = await open_fatal(s, run.id)
    await s.flush()
    (out,) = await _to_api_many(s, run, [f])
    return {"finding": out, "project": projects.readiness(p)}


async def reopen(s: AsyncSession, ctx: TenantContext, finding_id: UUID) -> dict[str, Any]:
    f = await _get_finding(s, finding_id)
    run, p = await _editable_run(s, ctx, f.check_run_id)
    if f.presence == "resolved":
        raise AppError("finding_resolved_this_run")
    await _apply(s, ctx, f, p, "pending", None)
    await s.flush()
    p.open_fatal_count = await open_fatal(s, run.id)
    await s.flush()
    (out,) = await _to_api_many(s, run, [f])
    return {"finding": out, "project": projects.readiness(p)}


async def bulk_resolve(
    s: AsyncSession, ctx: TenantContext, run_id: UUID, finding_ids: list[UUID], status: str, note: str | None
) -> dict[str, Any]:
    """只限「提示」级（FR-M3-07）；混进了别的级别就整批拒绝，返回不符合的问题编号。"""
    run, p = await _editable_run(s, ctx, run_id)
    rows = list((await s.execute(select(Finding).where(Finding.id.in_(finding_ids)))).scalars())
    if len(rows) != len(set(finding_ids)) or any(f.check_run_id != run.id for f in rows):
        raise AppError("not_found")
    bad = [str(f.id) for f in rows if f.severity != "hint" or f.presence == "resolved"]
    if bad:
        raise AppError("bulk_only_hint", finding_ids=bad)
    _require_note("hint", status, note)
    for f in rows:
        await _apply(s, ctx, f, p, status, note)
    await s.flush()
    p.open_fatal_count = await open_fatal(s, run.id)
    await s.flush()
    return {"items": await _to_api_many(s, run, rows), "project": projects.readiness(p)}


async def item_results(s: AsyncSession, run_id: UUID) -> list[dict[str, Any]]:
    """每条清单项的检查结论（C9）：「已响应」也有出处，可以复核、可以反馈判断有误。"""
    run = await get_run(s, run_id)
    items = await _items_of(s, run)
    rows = list((await s.execute(select(CheckItemResult).where(CheckItemResult.check_run_id == run_id))).scalars())
    cites = await citations.to_api_many(s, [c for r in rows for c in r.bid_citation_ids])
    return [
        {
            "checklist_item_id": r.checklist_item_id,
            "title": (items.get(r.checklist_item_id) or {}).get("title"),
            "verdict": r.verdict,
            "layer": r.layer,
            "evidence_status": r.evidence_status,
            "reason_code": r.reason_code,
            "bid_citations": [cites[c] for c in r.bid_citation_ids if c in cites],
            "detail": r.detail,
            "reused": r.reused_from_run_id is not None,
        }
        for r in rows
    ]


async def manual_checks(s: AsyncSession, ctx: TenantContext, project_id: UUID) -> list[dict[str, Any]]:
    """需人工核对的条目（按主题合并，来自当前正式清单）和勾选结果（按项目存，复查沿用，P2）。"""
    await projects.get(s, ctx, project_id)
    topics = manual.topics(await checklist.items(s, project_id, zone="formal"))
    rows = {
        r.check_key: r
        for r in (await s.execute(select(ManualCheck).where(ManualCheck.project_id == project_id))).scalars()
    }
    names = await identity.display_names(s, [r.checked_by for r in rows.values() if r.checked])
    out = []
    for t in topics:
        first = t.items[0]
        r = rows.get(t.check_key)
        done = bool(r and r.checked)
        out.append(
            {
                "check_key": t.check_key,
                "label": t.label,
                "requirement_text": first.get("title"),
                "citation": (first.get("citations") or [None])[0],
                "related_items": [{"id": i["id"], "title": i["title"]} for i in t.items],
                "checked": done,
                "note": r.note if r else None,
                "checked_by": {"id": r.checked_by, "name": names.get(r.checked_by, "")} if r and done else None,
                "checked_at": r.checked_at if r and done else None,
            }
        )
    return out


async def set_manual_check(
    s: AsyncSession, ctx: TenantContext, project_id: UUID, check_key: str, checked: bool, note: str | None
) -> dict[str, Any]:
    p = await projects.get(s, ctx, project_id)
    projects.ensure_editable(p)
    current = {x["check_key"]: x for x in await manual_checks(s, ctx, project_id)}
    if check_key not in current:
        raise AppError("validation_failed", errors=[{"field": "check_key", "code": "unknown", "message": "没有这一条"}])
    r = (
        await s.execute(
            select(ManualCheck).where(ManualCheck.project_id == project_id, ManualCheck.check_key == check_key)
        )
    ).scalar_one_or_none()
    before = {"checked": r.checked, "note": r.note} if r else None
    note = (note or "").strip() or None
    if r is None:
        r = ManualCheck(tenant_id=ctx.tenant_id, project_id=project_id, check_key=check_key, checked_by=ctx.user_id)
        s.add(r)
    r.checked, r.note, r.checked_by, r.checked_at = checked, note, ctx.user_id, datetime.now(UTC)
    await audit.record(
        s,
        ctx,
        "manual_check.set",
        "project",
        project_id,
        project_id=project_id,
        before=before,
        after={"check_key": check_key, "checked": checked, "note": note},
    )
    await s.flush()
    return next(x for x in await manual_checks(s, ctx, project_id) if x["check_key"] == check_key)