bid-doctor/evals/check/run_check.py

"""离线查标评测：拿读标评测的清单（evals/out/runs/<运行>/<招标文件>/items.json）对一份虚构投标文件跑响应匹配。

用法（在 server/ 下，模型配置读根目录 .env）：
  ../.venv/bin/python ../evals/check/run_check.py <投标文件编号> [--checklist fix3] [--name <运行名>]
输入：evals/out/check/<投标文件编号>/v*.pdf（render.py 渲染的）、manifest.json 里的 tender。
输出：evals/out/check/<投标文件编号>/runs/<运行名>/results.json（每条清单项的结论、投标侧证据、生成的问题）。
同一模型 + 同一提示词版本 + 同一输入只调用一次模型（和读标评测共用 evals/out/cache/），只改规则时重放不花钱。
"""

from __future__ import annotations

import argparse
import asyncio
import hashlib
import json
import pathlib
import re
import sys
from typing import Any
from uuid import UUID, uuid4

from bid_doctor.ai import prompts
from bid_doctor.ai import service as ai_service
from bid_doctor.ai.providers.base import near_limit
from bid_doctor.ai.service import OutputIncomplete, provider
from bid_doctor.check import findings, match
from bid_doctor.core.config import get_settings
from bid_doctor.core.db import TenantContext
from bid_doctor.document.pipeline import text_layer

EVALS = pathlib.Path(__file__).resolve().parents[1]
sys.path.insert(0, str(EVALS))
import gold  # noqa: E402

OUT = EVALS / "out" / "check"
CACHE = EVALS / "out" / "cache"
usage = {"calls": 0, "cached": 0, "tokens_in": 0, "tokens_out": 0}


async def cached_run_json(
    ctx: Any, *, job_type: str, job_id: Any, task: str, prompt_id: str, version: int, user: str
) -> dict[str, Any]:
    del ctx, job_type, job_id
    p = prompts.get(prompt_id, version)
    s = get_settings()
    model = s.ai_models.get(p.capability) or s.ai_model
    key = hashlib.sha256(f"{model}\n{prompt_id}\n{version}\n{p.body}\n{user}".encode()).hexdigest()[:32]
    f = CACHE / f"{key}.json"
    if f.exists():
        data = json.loads(f.read_text())
        usage["cached"] += 1
    else:
        r = await provider().complete_json(p.body, user, p.capability, task)
        data = {
            "data": r.data,
            "model": r.model,
            "tokens_in": r.tokens_in,
            "tokens_out": r.tokens_out,
            "incomplete": r.incomplete,
        }
        f.write_text(json.dumps(data, ensure_ascii=False))
    usage["calls"] += 1
    usage["tokens_in"] += data.get("tokens_in") or 0
    usage["tokens_out"] += data.get("tokens_out") or 0
    if data.get("incomplete") or near_limit(data.get("tokens_out")):
        raise OutputIncomplete(task, data["data"])
    return data["data"]


ITEM = re.compile(r"^\[(I\d+)\] ")
CANDS = re.compile(r"^候选句：(.*)$")
FRAG = re.compile(r"^\[(d\d+p\d+s\d+)\]（[^）]*）(.*)$")


async def baseline_run_json(
    ctx: Any, *, job_type: str, job_id: Any, task: str, prompt_id: str, version: int, user: str
) -> dict[str, Any]:
    """只检索基线（GUIDELINES §7）：不调模型，检索第一名就当作已响应。"""
    del ctx, job_type, job_id, task, prompt_id, version
    frags = {m.group(1): m.group(2) for line in user.splitlines() if (m := FRAG.match(line.strip()))}
    results, key = [], None
    for line in user.splitlines():
        if m := ITEM.match(line.strip()):
            key = m.group(1)
        elif (m := CANDS.match(line.strip())) and key:
            first = m.group(1).split("、")[0]
            results.append(
                {"item": key, "verdict": "responded", "evidence": [{"sid": first, "quote": frags.get(first, "")}]}
            )
    return {"results": results}


def checklist(tender: str, run: str) -> dict[UUID, dict[str, Any]]:
    """读标评测输出的正式清单 → 查标的清单项（和检查快照的 payload 同样的字段）。"""
    rows = json.loads((EVALS / "out" / "runs" / run / tender / "items.json").read_text())
    by_sid = {gold.sid(s): s for s in gold.sentences(tender)}
    out: dict[UUID, dict[str, Any]] = {}
    for i, r in enumerate(rows):
        if r.get("zone") != "formal":
            continue
        text = "".join(by_sid[k].text for k in r.get("sentences") or [] if k in by_sid)
        raw = (r.get("raw") or {}).get("tech_param") or {}
        spec = {"tech_param": {k: raw[k] for k in ("param", "op", "value", "unit") if raw.get(k)}} if raw else {}
        out[UUID(int=i + 1)] = {
            "title": r.get("title"),
            "requirement_text": text or None,
            "categories": r.get("categories") or [],
            "importance": r.get("importance"),
            "importance_basis": r.get("importance_basis"),
            "spec": spec,
            "sentences": r.get("sentences") or [],
        }
    return out


async def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("bid")
    ap.add_argument("--checklist", default="fix3")
    ap.add_argument("--name", default="base")
    ap.add_argument("--baseline", action="store_true", help="只检索、不调模型的基线")
    ap.add_argument("--v1", action="store_true", help="第 1 步的做法：提示词 v1、不带同一行的单元格（对比基线）")
    args = ap.parse_args()
    d = OUT / args.bid
    manifest = json.loads((d / "manifest.json").read_text())
    items = checklist(manifest["tender"], args.checklist)
    docs = []
    for n, pdf in enumerate(sorted(d.glob("v*.pdf")), start=1):
        pages = text_layer.extract(f"{args.bid}-{pdf.stem}", pdf.read_bytes())
        docs.append(match.BidDoc(uuid4(), n, pdf.stem, text_layer.body_sentences(pages)))
    ai_service.run_json = baseline_run_json if args.baseline else cached_run_json  # type: ignore[assignment]
    if args.v1:
        match.PROMPT = ("check.match", 1)
        match.ROW_MATES = 0

    async def progress(_: float) -> None:
        return None

    ctx = TenantContext(tenant_id=UUID(int=0), user_id=UUID(int=0))
    results, stats = await match.run(ctx, uuid4(), items, docs, progress)
    rows = []
    for item_id, item in items.items():
        res = results[item_id]
        spec = findings.spec(res, item)
        rows.append(
            {
                "title": item["title"],
                "sentences": item["sentences"],
                "categories": item["categories"],
                "importance": item["importance"],
                "verdict": res.verdict,
                "layer": res.layer,
                "evidence_status": res.evidence_status,
                "reason_code": res.reason_code,
                "evidence": [{"volume": doc.label, "p": s.page_no, "text": s.text} for doc, s in res.evidence],
                "detail": res.detail,
                "finding": {"type": spec.type, "severity": spec.severity, "title": spec.title} if spec else None,
            }
        )
    out = d / "runs" / args.name
    out.mkdir(parents=True, exist_ok=True)
    summary = {"items": len(rows), "usage": usage, "verify_failures": stats.verify_failures, "splits": stats.splits}
    (out / "results.json").write_text(json.dumps({"summary": summary, "items": rows}, ensure_ascii=False, indent=1))
    findings_n = sum(r["finding"] is not None for r in rows)
    print(
        f"{args.bid}：清单 {len(rows)} 条，问题 {findings_n} 个，模型调用 {usage['calls']} 次（缓存 {usage['cached']}），"
        f"token {usage['tokens_in']} / {usage['tokens_out']}"
    )


if __name__ == "__main__":
    asyncio.run(main())