bid-doctor/evals/gold.py

"""标注集：读取、定位、校验（evals/gold/<ID>.json，规范见 evals/gold/GUIDELINES.md）。

标注用「页码 + 原文片段」定位，不用句子编号：切句规则一改编号就变，片段在同一页里照样找得到。
定位时把这一页的句子连起来找片段，片段覆盖到的句子都算这条标注的句子（切句变了也不影响）。

用法（在 server/ 下）：../.venv/bin/python ../evals/gold.py check S01 [S02 ...]
"""

from __future__ import annotations

import functools
import json
import pathlib
import re
import sys
from dataclasses import dataclass
from typing import Any

from bid_doctor.document.pipeline import text_layer
from bid_doctor.document.pipeline.text_layer import Sentence

ROOT = pathlib.Path(__file__).resolve().parent
GOLD = ROOT / "gold"
DOCS = ROOT / "out" / "docs"
CATS = ("substantive", "qualification", "scoring", "schedule", "tech_param")
BASES = ("star", "asterisk", "italic_underline", "explicit_invalid")
INFO_KEYS = (
    "name",
    "code",
    "purchaser",
    "agency",
    "procurement_method",
    "legal_regime",
    "budget_total",
    "ceiling_total",
    "eval_method",
    "accept_consortium",
    "accept_imported",
    "sme_only",
)
SHIFT = 3  # 标注页码找不到时，往前后各找这么多页
MARKUP = re.compile(r"</?(?:iu|u|i)>|〔[^〕]*〕|^\[p\d+s\d+\]\s*")


def norm(s: str) -> str:
    return text_layer.norm(MARKUP.sub("", s))


def sid(s: Sentence) -> str:
    return f"p{s.page_no}s{s.seq}"


@functools.cache
def sentences(doc_id: str) -> list[Sentence]:
    pages = text_layer.extract("D", (DOCS / doc_id / "normalized.pdf").read_bytes())
    return text_layer.body_sentences(pages)


@dataclass
class Unit:
    id: str
    cat: str
    sids: set[str]
    optional: bool
    raw: dict[str, Any]


def locate(doc_id: str, page: int, text: str, nth: int = 1) -> tuple[list[str], int]:
    """片段覆盖到的句子编号，以及片段在那一页出现的次数。先在标注的页码上找；找不到就到前后 SHIFT 页里按远近找
    （转换方式、字体变了，分页会移动几页，例如 S07 从「先转 DOCX」改成直接修正 DOC 字体表）。"""
    for p in sorted(range(page - SHIFT, page + SHIFT + 1), key=lambda x: abs(x - page)):
        hits, n = _locate_on(doc_id, p, text, nth)
        if hits:
            return hits, n
    return [], 0


def _locate_on(doc_id: str, page: int, text: str, nth: int = 1) -> tuple[list[str], int]:
    sents = [s for s in sentences(doc_id) if s.page_no == page]
    joined, spans = "", []
    for s in sents:
        t = norm(s.text)
        spans.append((len(joined), len(joined) + len(t), sid(s)))
        joined += t
    q = norm(text)
    if not q:
        return [], 0
    at = -1
    for _ in range(nth):  # 同一页上逐字相同的两段（两台设备的同一条参数）：n 指第几次出现
        at = joined.find(q, at + 1)
        if at < 0:
            return [], 0
    hits = [k for b, e, k in spans if b < at + len(q) and e > at]
    return hits, joined.count(q)


def load(doc_id: str) -> dict[str, Any]:
    return json.loads((GOLD / f"{doc_id}.json").read_text())


def units(doc_id: str) -> tuple[list[Unit], list[str]]:
    """标注单元（定位到句子编号）和定位时的问题。"""
    g = load(doc_id)
    out, problems = [], []
    seen: set[str] = set()
    for it in g.get("items", []):
        uid = str(it.get("id"))
        if uid in seen:
            problems.append(f"{uid}：编号重复")
        seen.add(uid)
        if it.get("cat") not in CATS:
            problems.append(f"{uid}：cat 不对（{it.get('cat')}）")
        if it.get("cat") == "substantive" and it.get("basis") not in BASES:
            problems.append(f"{uid}：substantive 要写 basis（{it.get('basis')}）")
        sids: set[str] = set()
        for a in it.get("anchors") or []:
            hits, n = locate(doc_id, int(a.get("p", 0)), str(a.get("t", "")), int(a.get("n", 1)))
            if not hits:
                problems.append(f"{uid}：第 {a.get('p')} 页找不到「{a.get('t')}」")
            elif n > 1 and len(norm(str(a.get("t", "")))) < 12 and "n" not in a:
                problems.append(f"{uid}：「{a.get('t')}」在第 {a.get('p')} 页出现 {n} 次，片段再长一点")
            sids.update(hits)
        if not it.get("anchors"):
            problems.append(f"{uid}：没有 anchors")
        out.append(Unit(uid, str(it.get("cat")), sids, bool(it.get("optional")), it))
    info = g.get("project_info") or {}
    missing = [k for k in INFO_KEYS if k not in info]
    if missing:
        problems.append(f"project_info 缺字段：{missing}")
    return out, problems


def main() -> None:
    if len(sys.argv) < 3 or sys.argv[1] != "check":
        print(__doc__)
        return
    for doc_id in sys.argv[2:]:
        us, problems = units(doc_id)
        counts: dict[str, int] = {}
        for u in us:
            key = (
                u.cat
                + ("/" + str(u.raw.get("basis")) if u.cat == "substantive" else "")
                + ("（可选）" if u.optional else "")
            )
            counts[key] = counts.get(key, 0) + 1
        print(f"{doc_id}：{len(us)} 条，" + "，".join(f"{k} {n}" for k, n in sorted(counts.items())))
        print(f"  技术参数抽样页：{load(doc_id).get('tech_param_pages')}")
        for p in problems:
            print("  ✗", p)
        if not problems:
            print("  ✓ 全部定位成功")


if __name__ == "__main__":
    main()