fix(server): 按 S01–S10 评测修复读标抽取(F1–F8、口径 Q1–Q4)

- 合并时待确认条目不再借正式条目进清单,没通过的类别记在 pending_categories
- 结构识别:章(按章跳过不相干的部分)、表后「上述……否则无效」说明、★ 节标题
- INVALID 补写法;输出用到上限 92% 按截断处理,拆到底时采用已有输出
- 格式范本正文、整表概括句不收;项目信息多看评标方法所在分块
- WPS 旧版 DOC 直接修正字体表字符集;表格按单元格切句(s3-v4)
- 口径:★ 节继承实质性、★ 定义为计分项时降为重要、「必须/应」声明记 keyword 规则、
  政府采购资格条件按 87 号令第六十三条带无效类别
- 提示词 substantive v5,qualification / schedule / tech_param v4

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
YANG JIANKUAN
2026-11-03 23:19:30 +08:00
parent f99bb8b358
commit f98fd51128
17 changed files with 912 additions and 62 deletions

View File

@@ -3,6 +3,10 @@ from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Protocol
from bid_doctor.core.config import get_settings
NEAR_LIMIT = 0.92
@dataclass(frozen=True, slots=True)
class Completion:
@@ -12,10 +16,17 @@ class Completion:
tokens_in: int | None = None
tokens_out: int | None = None
output_mode: str = "json_mode"
# 输出不完整:被截断(finish_reason=length)或不是合法 JSON。调用方要拆小输入重跑,不能当成「没有结果」
# 输出不完整:被截断(finish_reason=length)、用到了上限的 92% 以上,或不是合法 JSON。
# 调用方要拆小输入重跑,不能当成「没有结果」
incomplete: bool = False
def near_limit(tokens_out: int | None) -> bool:
"""输出用到了上限的 92% 以上:模型常在快到上限时自己把 JSON 收尾,后面的条目整段不输出,
finish_reason 却是正常结束(评测 §3.5:S03 一块 7757 / 8192 token,后两页参数全丢)。按截断处理。"""
return tokens_out is not None and tokens_out >= NEAR_LIMIT * get_settings().ai_max_output_tokens
class TransientError(Exception):
"""限流、5xx、网络抖动:可以重试;重试后仍失败按基础设施错误处理(退回额度)。"""

View File

@@ -11,7 +11,7 @@ import json
import httpx
from bid_doctor.ai.providers.base import Completion, TransientError
from bid_doctor.ai.providers.base import Completion, TransientError, near_limit
from bid_doctor.core.config import get_settings
@@ -26,6 +26,7 @@ class OpenAICompatProvider:
"model": model,
"temperature": 0,
"response_format": {"type": "json_object"},
"max_tokens": s.ai_max_output_tokens,
"messages": [{"role": "system", "content": system}, {"role": "user", "content": user}],
}
try:
@@ -44,7 +45,7 @@ class OpenAICompatProvider:
choice = payload["choices"][0]
raw = choice["message"]["content"] or ""
usage = payload.get("usage") or {}
incomplete = choice.get("finish_reason") == "length"
incomplete = choice.get("finish_reason") == "length" or near_limit(usage.get("completion_tokens"))
try:
data = json.loads(raw)
except json.JSONDecodeError:

View File

@@ -21,7 +21,12 @@ RETRIES = 2
class OutputIncomplete(Exception):
"""模型输出被截断或不是合法 JSON:调用方把输入拆小重跑(extract.run),不能当成「没有结果」。"""
"""模型输出被截断、接近上限或不是合法 JSON:调用方把输入拆小重跑(extract.run),不能当成「没有结果」。
data 是已经解析出来的输出(接近上限时 JSON 往往是完整的),拆到底还不完整时调用方用它。"""
def __init__(self, task: str, data: dict[str, Any] | None = None) -> None:
super().__init__(task)
self.data: dict[str, Any] = data or {}
def provider() -> Provider:
@@ -72,5 +77,5 @@ async def run_json(
raise error
assert result is not None
if result.incomplete:
raise OutputIncomplete(task)
raise OutputIncomplete(task, result.data)
return result.data

View File

@@ -26,7 +26,7 @@ from zoneinfo import ZoneInfo
from bid_doctor.ai import service as ai
from bid_doctor.core.db import TenantContext
from bid_doctor.document.pipeline import structure
from bid_doctor.document.pipeline import structure, text_layer
from bid_doctor.document.pipeline.text_layer import Sentence, norm
CHUNK_CHARS = 6000
@@ -39,13 +39,19 @@ CN_TZ = ZoneInfo("Asia/Shanghai")
INVALID = re.compile(
r"投标无效|响应无效|无效投标|无效响应|作无效处理|按无效|做无效|视为无效|非实质性响应|未实质性响应"
r"|(?:投标|响应)文件无效|不实质性响应" # 「否则投标文件无效」(S05)「视为不实质性响应」(S07)
r"|(?:投标|响应)((?:投标|响应))无效" # 「投标(响应)无效」:范本里两种叫法并写
r"|不能通过资格审查|不通过资格审查|资格审查不合格" # 资格审查不通过就不进入评审,等同于响应无效
r"|(?:没有|未作|未做)实质性响应"
r"|否决其投标|否决投标|不予受理|予以拒绝|拒绝其|拒收"
# 「其投标将被拒绝」「拒绝接收截止时间后上传的投标文件」(评测 §3.2)
r"|被拒绝|拒绝接收|拒绝接受|无资格参加|不得参加"
r"|视为[^,。;]{0,10}串通" # 87 号令第三十七条:视为串通投标的,投标无效
)
# 按上级标题补漏时只认投标本身无效的说法;「不予受理」「予以拒绝」也用于质疑、投诉(国动样本 28.5 节「无效质疑」)
BID_INVALID_HEADING = re.compile(r"投标无效|响应无效|无效投标|无效响应|作无效处理|未实质性响应|否决投标|否决其投标")
BID_INVALID_HEADING = re.compile(
r"投标无效|响应无效|无效投标|无效响应|作无效处理|未实质性响应|否决投标|否决其投标|视为[^,。;]{0,10}串通"
)
RULE_NOTE = re.compile(r"(★|*|\*|斜体|下划线).{0,30}(实质性|无效|否决)|(实质性).{0,20}(★|*|\*|斜体|下划线)")
# 在描述标注方式本身的句子:规则说明、偏离表表头这类,不是条款(国动样本 p67 的表头碎片)
MARKER_TALK = re.compile(r"斜体.{0,8}下划线|下划线.{0,8}斜体|[★**]\s*号|加\s*[★**]")
@@ -57,13 +63,35 @@ CN_NUMBER = re.compile(r"([零〇一二两三四五六七八九十百千]+)(?=[
CN_DIGIT = {"零": 0, "〇": 0, "一": 1, "二": 2, "两": 2, "三": 3, "四": 4, "五": 5, "六": 6, "七": 7, "八": 8, "九": 9}
CN_UNIT = {"十": 10, "百": 100, "千": 1000}
END_PUNCT = "。;;,,、::!!??"
OUR_MARKUP = re.compile(r"</?(?:iu|u|i)>|〔(?:续下页|上级:[^〕]*)〕")
OUR_MARKUP = re.compile(r"</?(?:iu|u|i)>|〔(?:续下页|上级:[^〕]*|表后说明:[^〕]*)〕")
# ★ 的含义被文件定义成别的(「加★为核心技术参数」),而且没有任何地方说 ★ 是实质性要求或不满足就无效(口径 Q2)
STAR_DEFINED = re.compile(r"(?:加|标|打|带)\s*[“\"]?★[”\"]?\s*号?\s*(?:的|为|是|表示)+\s*([^,。;]{0,12})")
STAR_IMPORTANT = "star:important" # 放进 Doc.declared:这份文件的 ★ 只是「重要」,不是实质性
# 「必须」「应」字样的要求都是实质性要求(S07 第 18 页):只在文件层面提示,不逐句抽(口径 Q3)
KEYWORD_NOTE = re.compile(r"[“\"](?:必须|应|应当|应(应当))[”\"]")
SID = re.compile(r"^p\d+s\d+$")
# 招标文件里待投标人填写的空白:「____元」「××」「( )」「待填」——不是取值
PLACEHOLDER = re.compile(r"_{2,}|_{2,}|×{2,}|[xX]{2,}|\*{2,}|(\s*)|\(\s*\)|待填|TBD", re.IGNORECASE)
# 没通过校验的原因码(记在清单项 ai_values.failures 里,解析结果按原因计数)。每一条对应提示词里的一条规则,
# 见 prompts/README.md「提示词规则和服务端校验」
# 「本项目采用综合评分法。」;不认须知里的条件句「采用综合评分法的项目,未中标的投标人可……」
# 第六章格式里的声明函、承诺函范本正文(「我单位具有良好的商业信誉……」):提示词禁止了,模型照样抽(评测第二轮)
TEMPLATE = re.compile(r"^(?:[一二三四五六七八九十\d]{1,3}[、..]\s*)?(?:我单位|我公司|本公司|本单位|我方|我们|本人|兹)")
# 概括整张表、整个列表的后果句(「有任何一项不符合……的,投标无效」):只说明后果,条款是表里的各项
BLANKET = re.compile(
r"任何一项|任一项|^(?:(\d{1,2})|\(\d{1,2}\)|\d{1,2}[.、.]|总?备注[::]|注[::]|说明[::])?\s*(?:上述|以上|下列)"
)
EVAL_METHOD = re.compile(r"(?:采用|方法为|方法:|■)\s*(?:综合评分法|最低评标价法|最低价法|性价比法)(?![的评])")
# 抽取任务按章跳过不相干的部分(评测 §3.3:资格任务把技术参数当成资格条件)。章认不出时不跳过。
# 资格任务不能跳过格式章:江苏范本常把资格证明材料清单只写在第六章「投标文件内容与格式」里(S05)
SKIP_CHAPTERS = {
"extract.qualification": {"contract", "requirements", "award", "complaint"},
"extract.schedule": {"contract", "requirements", "formats", "award"},
"extract.tech_param": {"contract", "formats", "award", "complaint"},
"extract.scoring": {"contract", "formats", "award", "complaint"},
}
FAILURES = {
"evidence_missing": "没有给证据",
"sid_format": "句子编号格式不对",
@@ -107,6 +135,7 @@ class MarkingNote:
markers: set[str]
sentence: Sentence
declared_count: int | None
meaning: str = "substantive" # 「重要」:★ 被文件定义成计分项(口径 Q2)
@dataclass(slots=True)
@@ -133,6 +162,10 @@ class Doc:
continuations: dict[str, Sentence]
notes: list[MarkingNote]
declared: set[str]
chapters: dict[str, str] = field(default_factory=dict) # 句子编号 → 章的类型(structure.chapters)
trailing: dict[str, Sentence] = field(default_factory=dict) # 句子编号 → 表后往回指的说明
star_scope: dict[str, Sentence] = field(default_factory=dict) # 句子编号 → 带 ★ 的节标题(口径 Q1)
regime: str = "unknown"
@property
def note_keys(self) -> set[str]:
@@ -140,10 +173,12 @@ class Doc:
def marking_notes(sentences: list[Sentence]) -> list[MarkingNote]:
"""「标注规则说明」:一句里可能同时声明了几种标注(如「斜体且有下划线部分或加*号条款为实质性要求」)。"""
"""「标注规则说明」:一句里可能同时声明了几种标注(如「斜体且有下划线部分或加*号条款为实质性要求」)。
★ 被文件定义成计分项、没有任何地方说它是实质性要求时,补一条「重要」的说明(口径 Q2)。"""
notes: list[MarkingNote] = []
for s in sentences:
if not RULE_NOTE.search(s.text):
# 「必须」「应」字样的声明不一定同时提到 ★ 或斜体(顾问审读第 12 条)
if not (RULE_NOTE.search(s.text) or (KEYWORD_NOTE.search(s.text) and "实质性" in s.text)):
continue
markers = set()
if "斜体" in s.text and "下划线" in s.text:
@@ -152,9 +187,57 @@ def marking_notes(sentences: list[Sentence]) -> list[MarkingNote]:
markers.add("star")
if re.search(r"[**]", s.text):
markers.add("asterisk")
if KEYWORD_NOTE.search(s.text):
markers.add("keyword")
m = DECLARED.search(s.text)
notes.append(MarkingNote(markers=markers, sentence=s, declared_count=int(m.group(1)) if m else None))
return notes[:10]
notes = notes[:10]
if not any("star" in n.markers for n in notes) and (d := star_scored(sentences)) is not None:
notes.append(MarkingNote(markers={"star"}, sentence=d, declared_count=None, meaning="important"))
return notes
STAR_CONSEQUENCE = re.compile(r"废标|负偏离|不得偏离|未响应|不响应|不予通过|实质|否决|拒绝")
SCORED = re.compile(r"扣\s*\d|减\s*\d|扣分|得分|加分|计分|评分")
def star_scored(sentences: list[Sentence]) -> Sentence | None:
"""★ 只是计分项的定义句(口径 Q2,S02「备注:加★为核心技术参数」),没有就返回 None。
降级的代价是把致命问题显示成「重要」,所以条件从严(顾问审读第 1 条):
1. 定义句说 ★ 是什么(「核心技术参数」),这句本身不带任何后果写法;
2. 有计分证据:这个说法出现在写着扣分、得分的句子里(S02 p29「每有一项核心技术参数不满足……减 4 分」);
3. 全文没有一句带 ★ 的话写着无效、废标、负偏离、未响应、实质性这类后果。
"""
if any("★" in s.text and (INVALID.search(s.text) or STAR_CONSEQUENCE.search(s.text)) for s in sentences):
return None
for s in sentences:
m = STAR_DEFINED.search(s.text)
if not m or INVALID.search(s.text) or STAR_CONSEQUENCE.search(s.text) or re.search(r"必须|关键", s.text):
continue
term = re.sub(r"^[的为是]+", "", m.group(1)).strip()
if len(term) >= 4 and any(term in x.text and SCORED.search(x.text) for x in sentences if x is not s):
return s
return None
def build_doc(sentences: list[Sentence]) -> Doc:
"""一份文档给各抽取任务共用的结构信息(解析流水线和离线评测都用它)。"""
notes = marking_notes(sentences)
declared = {m for n in notes if n.meaning == "substantive" for m in n.markers}
if any(n.meaning == "important" and "star" in n.markers for n in notes):
declared.add(STAR_IMPORTANT)
return Doc(
sentences=sentences,
parents=structure.parents(sentences),
continuations=text_layer.continuations(sentences),
notes=notes,
declared=declared,
chapters=structure.chapters(sentences),
trailing=structure.trailing_notes(sentences),
star_scope=structure.star_sections(sentences) if STAR_IMPORTANT not in declared else {},
regime=legal_regime_by_rule(sentences),
)
def is_marker_talk(s: Sentence) -> bool:
@@ -170,7 +253,7 @@ def is_marked(s: Sentence, declared: set[str]) -> str | None:
if is_marker_talk(s):
return None
if "★" in s.text:
return "star"
return None if STAR_IMPORTANT in declared else "star"
if "asterisk" in declared and ASTERISK.match(s.text.strip()):
return "asterisk"
iu = sum(e - b for b, e, f in s.fmt_runs if "i" in f and "u" in f)
@@ -179,11 +262,17 @@ def is_marked(s: Sentence, declared: set[str]) -> str | None:
return None
def importance_of(sents: list[Sentence], declared: set[str]) -> tuple[str, str | None]:
"""重要性按原文标注确定,不信模型:★ / 声明过的标注 → 实质性;▲ → 重要。"""
if any(is_marked(s, declared) for s in sents):
def inherited_star(s: Sentence, doc: Doc) -> bool:
"""在带 ★ 的节标题下面(「三、★商务要求」下的各句):整节都是实质性要求(口径 Q1)。"""
return s.key in doc.star_scope and not is_marker_talk(s)
def importance_of(sents: list[Sentence], doc: Doc) -> tuple[str, str | None]:
"""重要性按原文标注确定,不信模型:★ / 声明过的标注 / ★ 节里的句子 → 实质性;▲ → 重要。
文件把 ★ 定义成计分项时(口径 Q2),★ 和 ▲ 一样是重要。"""
if any(is_marked(s, doc.declared) or inherited_star(s, doc) for s in sents):
return "substantive", "marker"
if any("▲" in s.text for s in sents):
if any("▲" in s.text or "★" in s.text for s in sents):
return "important", "marker"
return "normal", None
@@ -254,10 +343,19 @@ def _render(chunk: list[Sentence], doc: Doc, marks: bool = False) -> str:
line = f"[{_sid(s)}] {s.marked if marks else s.text}"
if parent := doc.parents.get(s.key):
line += f"〔上级:{parent.text[:60]}〕"
if (note := doc.trailing.get(s.key)) is not None and marks: # 只给找实质性条款的任务(提示词 v5 认这个附注)
line += f"〔表后说明:{note.text[:60]}〕"
if s.key in doc.continuations:
line += "〔续下页〕" # 这句在页尾被截断,下一行是它在下一页的接续部分
lines.append(line)
notes = "\n".join(f"- {n.sentence.text[:200]}" for n in doc.notes) or "(未找到标注规则说明)"
notes = (
"\n".join(
f"- {n.sentence.text[:200]}"
+ ("(本文件的 ★ 只是计分项,不是实质性标注)" if n.meaning == "important" else "")
for n in doc.notes
)
or "(未找到标注规则说明)"
)
return "## 标注规则说明\n" + notes + "\n\n## 待审分块\n" + "\n".join(lines)
@@ -373,9 +471,13 @@ def check_basis(basis: str, sents: list[Sentence], chunk: list[Sentence], doc: D
"""实质性条款:模型声称的依据,用确定性信号核对(S4:不加这一道,采纳条目只有 36.7% 属实)。"""
if any(is_marker_talk(s) for s in sents) and all(is_marker_talk(s) or s.key in doc.note_keys for s in sents):
return False
if basis == "star" and any(inherited_star(s, doc) for s in sents):
return True
if basis in ("star", "asterisk", "italic_underline"):
return any(is_marked(s, doc.declared) == basis for s in sents)
if basis == "explicit_invalid":
if all(doc.chapters.get(s.key) == "contract" for s in sents):
return False # 合同条款里的「拒绝接受」「不得」是履约约定,不是投标无效(顾问审读第 8 条)
idx = {s.key: i for i, s in enumerate(chunk)}
near = [
chunk[j]
@@ -385,6 +487,7 @@ def check_basis(basis: str, sents: list[Sentence], chunk: list[Sentence], doc: D
if 0 <= j < len(chunk)
]
near += [p for s in sents for p in structure.ancestors(s.key, doc.parents)] # 上级条款写明了无效情形
near += [doc.trailing[s.key] for s in sents if s.key in doc.trailing] # 表后的说明写明了无效
return any(INVALID.search(s.text) for s in near)
return False
@@ -457,10 +560,10 @@ def _item(
passed = [*passed, "basis"]
accepted = ok and value_ok
sents = _extend(sents, doc)
importance, imp_basis = importance_of(sents, doc.declared)
importance, imp_basis = importance_of(sents, doc)
if "basis" not in kw and importance == "substantive":
# 按哪种标注认定为实质性:清单项据此挂到对应的标注规则上
kw["basis"] = next((m for x in sents if (m := is_marked(x, doc.declared))), None)
# 按哪种标注认定为实质性:清单项据此挂到对应的标注规则上;★ 节里的句子挂 ★
kw["basis"] = next((m for x in sents if (m := is_marked(x, doc.declared))), "star")
if importance == "substantive" and "invalidity" not in categories:
categories = [*categories, "invalidity"] # ★ 条款同时进「无效 / 废标」类(FR-M2-05)
return Extracted(
@@ -497,7 +600,8 @@ def _substantive(
basis=basis,
spec={"invalidity": {"source": "substantive_marker" if basis != "explicit_invalid" else "tender_text"}},
)
e.importance = "substantive"
if not (basis == "star" and STAR_IMPORTANT in doc.declared): # ★ 只是计分项的文件:按 importance_of 的结果
e.importance = "substantive"
e.importance_basis = e.importance_basis or ("tender_text" if basis == "explicit_invalid" else "marker")
return e
@@ -507,7 +611,7 @@ def _qualification(
) -> Extracted | None:
del chunk
kind = it.get("kind") if it.get("kind") in ("general", "specific", "credit") else "specific"
return _item(
e = _item(
"qualification",
["qualification"],
it.get("summary"),
@@ -521,6 +625,14 @@ def _qualification(
evidence_required=it.get("evidence_required"),
lot_no=it.get("lot_no"),
)
if doc.regime == "gov_procurement":
# 口径 Q4:政府采购项目不具备资格要求的投标无效(87 号令第六十三条;磋商同理,资格审查不通过即无效响应)。
# 这条依据来自法规,不是招标文件的标注,重要性依据记 statute
e.categories = [*dict.fromkeys([*e.categories, "invalidity"])]
e.spec = {**e.spec, "invalidity": {"source": "statute"}}
if e.importance != "substantive":
e.importance, e.importance_basis = "substantive", "statute"
return e
def _scoring(
@@ -615,11 +727,11 @@ def _tech_param(
TASKS = [
Task("extract.substantive", 4, _substantive, marks=True),
Task("extract.qualification", 3, _qualification),
Task("extract.substantive", 5, _substantive, marks=True),
Task("extract.qualification", 4, _qualification),
Task("extract.scoring", 3, _scoring),
Task("extract.schedule", 3, _schedule),
Task("extract.tech_param", 3, _tech_param),
Task("extract.schedule", 4, _schedule),
Task("extract.tech_param", 4, _tech_param),
]
INFO_TASK = ("extract.project_info", 2)
INFO_KEYS = (
@@ -654,8 +766,16 @@ async def run(
) -> tuple[list[Extracted], ProjectInfo, dict[str, Any]]:
"""返回 (清单项, 项目基本信息, 统计)。统计里有对半重跑的次数、拆到底仍不完整的分块数和各原因码的条数。"""
parts = chunks(doc.sentences, doc.continuations)
task_parts = {t.name: task_chunks(t.name, doc) for t in TASKS}
info_parts = parts[:INFO_CHUNKS]
# 评标方法写在评审一章(「本项目采用综合评分法」),常常不在文件开头几块里(评测 §3.6)。
# 按写法找,不靠章识别(S02 的评审章标题不是单独一句)
seen_keys = {x.key for c in info_parts for x in c}
hit = next((c for c in parts[INFO_CHUNKS:] if any(EVAL_METHOD.search(s.text) for s in c)), None)
if hit is not None and not any(EVAL_METHOD.search(s.text) for s in doc.sentences if s.key in seen_keys):
info_parts = [*info_parts, hit]
gate = asyncio.Semaphore(CONCURRENCY)
total = len(parts) * len(TASKS) + min(INFO_CHUNKS, len(parts))
total = sum(len(v) for v in task_parts.values()) + len(info_parts)
done = 0
stats: Counter[str] = Counter()
@@ -674,10 +794,11 @@ async def run(
user=_render(chunk, doc, marks),
)
return [(chunk, data)]
except ai.OutputIncomplete:
except ai.OutputIncomplete as e:
if len(chunk) < 2 or depth >= SPLIT_DEPTH:
# 拆到底还不完整:输出接近上限但 JSON 是完整的,就用它(顾问审读第 7 条),并记下来
stats["output_incomplete"] += 1
return [(chunk, {})]
return [(chunk, e.data)]
stats["chunks_split"] += 1
mid = len(chunk) // 2
while mid < len(chunk) - 1 and chunk[mid - 1].key in doc.continuations:
@@ -695,10 +816,13 @@ async def run(
await progress(done / total)
return got
jobs = [(t, call(t.name, t.prompt_version, c, t.marks)) for t in TASKS for c in parts]
info_jobs = [call(*INFO_TASK, c) for c in parts[:INFO_CHUNKS]]
jobs = [(t, call(t.name, t.prompt_version, c, t.marks)) for t in TASKS for c in task_parts[t.name]]
info_jobs = [call(*INFO_TASK, c) for c in info_parts]
results = await asyncio.gather(*(j for _, j in jobs), *info_jobs)
leads = {p.key for p in doc.parents.values() if p.text.strip().endswith((":", ":")) or len(norm(p.text)) <= 20}
leads |= {n.key for n in doc.trailing.values()}
leads |= {s.key for s in doc.sentences if BLANKET.search(s.text.strip()) and INVALID.search(s.text)}
out: list[Extracted] = []
for (task, _), got in zip(jobs, results[: len(jobs)], strict=True):
for chunk, data in got:
@@ -711,6 +835,10 @@ async def run(
sents = without_headings(sents, doc)
if sents and all(s.key in doc.note_keys or is_marker_talk(s) for s in sents):
continue # 只是「标注规则说明」或表头,不是条款(提示词要求不输出,这里再兜一道)
if task.name == "extract.substantive" and sents and all(s.key in leads for s in sents):
continue # 只引了列表的引导句、表后的概括说明:它们只说明后果,条款是下面列出的各项(评测 §3.1)
if task.name != "extract.substantive" and sents and all(TEMPLATE.match(s.text.strip()) for s in sents):
continue # 只引了格式范本正文(声明函、承诺函):那是投标人要填的格式,不是要求
if (e := task.build(it, sents, passed, ok, chunk, doc)) is not None:
e.failures = [*dict.fromkeys([*fails, *e.failures])] if e.zone == "pending" else []
out += split_listed(e, doc)
@@ -726,6 +854,13 @@ async def run(
return items, info, dict(stats)
def task_chunks(task: str, doc: Doc) -> list[list[Sentence]]:
"""这个任务要看的分块:跳过不相干的章(SKIP_CHAPTERS)。跳过之后什么都不剩(章认错了)就看全文。"""
skip = SKIP_CHAPTERS.get(task, set())
kept = [s for s in doc.sentences if doc.chapters.get(s.key) not in skip]
return chunks(kept or doc.sentences, doc.continuations)
def without_headings(sents: list[Sentence], doc: Doc) -> list[Sentence]:
"""证据里同时有上级标题和它下面的条目时,标题只是上下文,不算这条要求的句子。
否则以标题为首句的几条结果会被合并成一大条(国动样本 25.3 节)。依据校验照样认上级标题(doc.parents)。"""
@@ -788,8 +923,14 @@ def merge_items(items: list[Extracted]) -> list[Extracted]:
group.sort(key=lambda e: (e.zone != "formal", order.index(e.task)))
head = group[0]
for e in group[1:]:
if e.task == head.task and e.zone == "pending":
continue # 同一任务在两个分块里重复抽到:留通过校验的那条
if e.zone == "pending" and head.zone == "formal":
# 没通过校验的结果不能借别的条目进正式清单:以前别的任务的待确认条目会把类别、重要性并进正式条目,
# 没写无效后果的技术参数因此成了「实质性」(评测 §3.1,S07 一份 93 条)。
# 也不能丢:多出来的类别记成「待确认的类别」,由人确认(顾问审读第 3 条)
head.ai_values = {**e.ai_values, **head.ai_values}
extra = [c for c in e.categories if c not in head.categories]
add_pending_categories(head, extra, e.failures or ["basis_unmatched"])
continue
head.categories = [*dict.fromkeys([*head.categories, *e.categories])]
if rank[e.importance] > rank[head.importance]:
head.importance, head.importance_basis = e.importance, e.importance_basis
@@ -815,10 +956,25 @@ def merge_items(items: list[Extracted]) -> list[Extracted]:
return merged + unique
def add_pending_categories(e: Extracted, categories: list[str], failures: list[str]) -> None:
"""正式条目上「待确认的类别」:例如资格条件已经确认,但「不满足就投标无效」没有通过校验。
记在 ai_values.pending_categories({类别: 原因码}),不改条目本身的类别和区。"""
if not categories:
return
pending = dict(e.ai_values.get("pending_categories") or {})
for c in categories:
pending[c] = sorted({*pending.get(c, []), *failures})
e.ai_values = {**e.ai_values, "pending_categories": pending}
def fallback(items: list[Extracted], doc: Doc) -> list[Extracted]:
"""规则兜底:标注句没被正式条目覆盖 → 生成一条待确认(TV1:标注类召回 ≥ 99%)。跨页接续的两句算一条。"""
covered = {s.key for e in items if e.zone == "formal" for s in e.sentences}
seen = {s.key for e in items for s in e.sentences}
"""规则兜底:标注句没被带「无效」类别的正式条目覆盖 → 生成一条待确认(TV1:标注类召回 ≥ 99%)。
跨页接续的两句算一条。这句已经是别的正式条目(资格、参数……)的首句时,不另生成条目(同一首句只能有一条),
在那条上记「待确认的类别」。"""
covered = {s.key for e in items if e.zone == "formal" and "invalidity" in e.categories for s in e.sentences}
heads = {e.sentences[0].key: e for e in items if e.zone == "formal" and e.sentences}
seen = {s.key for e in items if e.zone != "formal" for s in e.sentences}
seen |= {k for k, e in heads.items() if "invalidity" in (e.ai_values.get("pending_categories") or {})}
out: list[Extracted] = []
skip: set[str] = set()
for s in doc.sentences:
@@ -831,6 +987,7 @@ def fallback(items: list[Extracted], doc: Doc) -> list[Extracted]:
listed = numbered and any(
BID_INVALID_HEADING.search(p.text) and "质疑" not in p.text for p in structure.ancestors(s.key, doc.parents)
)
listed = listed or s.key in doc.trailing # 表格下方「上述……否则无效」覆盖的行(评测 §3.2)
if not marker and listed:
marker = "explicit_invalid" # 列在「无效情形」标题下的条目:上级标题就是依据
# 标注规则说明本身不是条款;但列在「无效情形」下的那一条(如国动 25.3.6「不符合……实质性要求的(斜体且有
@@ -842,6 +999,9 @@ def fallback(items: list[Extracted], doc: Doc) -> list[Extracted]:
skip.update(x.key for x in sents)
if s.key in seen: # 模型抽到了但没通过校验:已经在待确认区,不再重复生成
continue
if s.key in heads:
add_pending_categories(heads[s.key], ["invalidity"], ["marker_uncovered"])
continue
out.append(
Extracted(
task="rule",
@@ -928,7 +1088,8 @@ def _project_info(results: Parts) -> ProjectInfo:
BOOL_SUBJECT = {"accept_consortium": "联合体", "accept_imported": "进口", "sme_only": "中小"}
NEGATIVE = re.compile(r"不接受|不允许|不得|不可|不能|不面向|拒绝|否")
# 「非专门面向中小企业」「不专门面向」(评测 §3.6:以前 8/10 份核对不通过)
NEGATIVE = re.compile(r"不接受|不允许|不得|不可|不能|不面向|不专门|非专门|拒绝|否")
POSITIVE = re.compile(r"接受|允许|可以|面向|是")

View File

@@ -64,7 +64,9 @@ async def merge(
project_id=doc.project_id,
document_id=doc.id,
marker=marker,
meaning="substantive",
# 「必须」「应」字样:文件层面的提示,不逐句抽成清单项(口径 Q3)
marker_text="必须、应(应当)" if marker == "keyword" else None,
meaning=n.meaning,
declared_count=n.declared_count if marker == "star" else None,
citation_id=c.id,
origin="rule",

View File

@@ -46,6 +46,8 @@ class Settings(BaseSettings):
ai_model: str = ""
# 按能力选模型(T1 场景路由):{"reasoning": "deepseek-reasoner"};没配的能力用 ai_model
ai_models: dict[str, str] = Field(default_factory=dict)
# 每次调用的输出上限(token)。输出用到 92% 以上就按截断处理、拆小重跑(ai/providers/base.near_limit)
ai_max_output_tokens: int = 8192
upload_part_size: int = 8 * 1024 * 1024
download_url_ttl_s: int = 300

View File

@@ -3,7 +3,7 @@
TODO(S1):
- 转换镜像要带按宋体行高调整过的字体包(BD Song / BD Hei)和 fontconfig 替换规则,现在用的是 Gotenberg 默认字体,
分页偏差会偏大(S1:默认配置 +17%~+37%);
- WPS 生成的旧版 DOC 要先转 DOCX、修正字体字符集再转 PDF,否则正文可能是空白(S1 §1.3)。现在直接转。
- WPS 生成的旧版 DOC 的字体字符集已在 DOC 字体表里直接修正(fix_wps_doc_charset,评测 F7),不再需要先转 DOCX。
"""
from __future__ import annotations
@@ -45,11 +45,36 @@ def fix_wps_charset(docx: bytes) -> bytes:
return out.getvalue()
# 旧版 DOC 字体表(SttbfFfn)里的一条字体记录(FFN):名字是 UTF-16 字符串,从记录开头第 40 字节起;
# 第 0 字节是记录长度减 1,第 4 字节是字符集(chs)。名字后面以两个 0 字节结束
FFN_NAME = re.compile(rb"(?:[\x00-\xff][\x4e-\x9f]){2,12}\x00\x00|(?:[A-Za-z ]\x00){4,20}\x00\x00")
FFN_NAME_AT = 40
FFN_CHS_AT = 4
def fix_wps_doc_charset(doc: bytes) -> bytes:
"""旧版 DOC 的同一个问题(S07:正文全变成「第第第第」):在字体表里找中文字体名,往前定位到记录开头,
记录长度对得上、字符集是 SYMBOL(02) 的改成 GB2312(86)。直接改字节,不用先转 DOCX(Gotenberg 只能转出 PDF)。"""
buf = bytearray(doc)
for m in FFN_NAME.finditer(doc):
name = doc[m.start() : m.end() - 2].decode("utf-16-le", "ignore")
start = m.start() - FFN_NAME_AT
if not (re.search(r"[一-鿿]", name) or name in CJK_LATIN) or start < 0 or buf[start + FFN_CHS_AT] != 0x02:
continue
# 名字后面可能还跟着备用名(也是 UTF-16),记录比名字长偶数个字节;长度对不上的是正文里碰巧像字体名的字节
extra = buf[start] + 1 - (m.end() - start)
if 0 <= extra <= 64 and extra % 2 == 0:
buf[start + FFN_CHS_AT] = 0x86
return bytes(buf)
async def to_pdf(data: bytes, fmt: str) -> bytes:
if fmt == "pdf":
return data
if fmt == "docx":
data = fix_wps_charset(data)
if fmt == "doc":
data = fix_wps_doc_charset(data)
async with httpx.AsyncClient(base_url=get_settings().gotenberg_url, timeout=TIMEOUT) as c:
r = await c.post("/forms/libreoffice/convert", files={"files": (f"input.{fmt}", data)})
if r.status_code == 200:

View File

@@ -28,7 +28,7 @@ from bid_doctor.core.errors import AppError
from bid_doctor.core.ids import uuid7
from bid_doctor.core.storage import storage
from bid_doctor.document.models import DocumentPage, ParseJob, SourceDocument
from bid_doctor.document.pipeline import convert, structure, text_layer
from bid_doctor.document.pipeline import convert, text_layer
from bid_doctor.document.service import doc_prefix
from bid_doctor.notification.models import Notification
from bid_doctor.project import extracted as project_extracted
@@ -46,7 +46,7 @@ PROGRESS = {
"merge": 95,
}
# text_layer v2:页眉页脚带单独成句,句子编号和 v1 不同(同一份规范化 PDF 在 v1、v2 下的编号不能混用)
PROFILE = {"converter": "gotenberg-8", "fonts": "gotenberg-default", "text_layer": "v3", "sentence_split": "s3-v3"}
PROFILE = {"converter": "gotenberg-8", "fonts": "gotenberg-default", "text_layer": "v4", "sentence_split": "s3-v4"}
class Cancelled(Exception):
@@ -168,14 +168,8 @@ async def _stages(ctx: TenantContext, job_id: UUID) -> dict[str, Any]:
await _stage(ctx, job_id, "structure")
body = text_layer.body_sentences(pages)
notes = extract.marking_notes(body)
doc_info = extract.Doc(
sentences=body,
parents=structure.parents(body),
continuations=text_layer.continuations(body),
notes=notes,
declared={m for n in notes for m in n.markers},
)
doc_info = extract.build_doc(body)
notes = doc_info.notes
await _stage(ctx, job_id, "extract")
async with tenant_tx(ctx) as s:

View File

@@ -9,7 +9,12 @@
3. 中文编号的小节:「三、符合条件(必须提供,否则投标无效)」下面的 1. 2. 3.……(S02 第 20 页);
4. 以冒号结尾的编号条目:「1.报价符合性审查:」下面不带编号的句子,到下一个编号为止。
上级还可以有上级(第 4 种的上级常常是第 3 种),用 ancestors() 取整条链。
TODO(结构识别):章、节层级,表格结构,分包识别。
另外三种结构(S01–S10 评测,docs/architecture/extract-eval.md §3):
- chapters():每句属于哪一章(公告、须知、合同、采购需求、评审、格式……),抽取任务据此跳过不相干的章;
- trailing_notes():表格、列表下方往回指的说明(「上述资格证明文件 1~6 项必须全部提供,否则投标无效」);
- star_sections():★ 标在节标题上(「三、★商务要求」),整节都是实质性要求(口径 Q1)。
TODO(结构识别):表格结构,分包识别。
"""
from __future__ import annotations
@@ -23,6 +28,7 @@ LIST_ITEM = re.compile(r"^((\d{1,2})|\(\d{1,2}\)|\d{1,2}[.、.](?!\d)|[①
CHAPTER = re.compile(r"^第[一二三四五六七八九十百\d]+[章节]")
CN_SECTION = re.compile(r"^(?:[一二三四五六七八九十]{1,3}、|([一二三四五六七八九十]{1,3}))")
LEAD_MAX = 30 # 冒号引出的列表,最多往后认这么多句
CN_NUM = "零一二三四五六七八九十"
def _list_no(text: str) -> int | None:
@@ -96,3 +102,165 @@ def ancestors(key: str, parents: dict[str, Sentence], depth: int = 3) -> list[Se
out.append(p)
key = p.key
return out
# ---------------------------------------------------------------- 章
# 章标题:「第四章采购需求」,也可能前面粘着页眉(S03「常州市政府采购项目公开招标文件第三章资格审查」)
CHAPTER_TITLE = re.compile(r"第([一二三四五六七八九十]{1,3})章\s*(\S{2,16})")
# 章的类型按标题关键词判断,先判断的优先(「合同授予」不是合同,「主要条款及格式」算合同)
CHAPTER_KINDS = (
("award", ("授予",)),
("complaint", ("质疑", "投诉")),
("notice", ("公告", "邀请")),
("instructions", ("须知", "资料表")),
("qualification", ("资格审查",)),
("evaluation", ("评标", "评审", "评分")),
("contract", ("合同", "条款")),
("requirements", ("需求", "技术规格", "技术要求")),
("formats", ("格式", "组成")),
)
def cn_int(s: str) -> int:
"""「三」→ 3,「十二」→ 12(只处理章节编号用到的范围)。"""
if "十" not in s:
return CN_NUM.index(s) if len(s) == 1 else 0
head, _, tail = s.partition("十")
return (CN_NUM.index(head) if head else 1) * 10 + (CN_NUM.index(tail) if tail else 0)
def chapters(sentences: list[Sentence]) -> dict[str, str]:
"""{句子编号: 章的类型}。认不出章的句子不在里面(调用方按「不跳过」处理)。
只认正文里的章标题:句子短(不超过 40 字)、不是目录(目录行带省略号,或者前后紧挨着另一个章标题,中间只隔页码)、
不是引用(「见第一章《投标邀请》」「(8)第三章评标标准中……」)、章号比上一章大。
认出来的章不可靠时(不到 3 章,或最后一章占了全文一半以上)返回空:按章跳过的任务就看全文,宁可多看不漏看。
"""
parents_of = parents(sentences)
texts = [s.text.strip() for s in sentences]
cands: list[tuple[int, int, str]] = [] # (句子位置, 章号, 类型)
for i, s in enumerate(sentences):
text = texts[i]
m = CHAPTER_TITLE.search(text)
if not m or len(text) > 40 or "…" in text or "...." in text or "见" in text[: m.start()] or "《" in text:
continue
if LIST_ITEM.match(text) or DOTTED.match(text) or s.key in parents_of or re.match(r"[第\d]", m.group(2)):
continue # 编号条目里提到某一章、引用「第二章第26.1.2条」,不是章标题
before = [t for t in texts[max(0, i - 3) : i] if not t.isdigit()][-1:]
after = [t for t in texts[i + 1 : i + 4] if not t.isdigit()][:1]
if any(CHAPTER_TITLE.match(t) and len(t) <= 40 for t in before + after):
continue # 目录:几个章标题挨着(中间可能隔着页码)
kind = next((k for k, words in CHAPTER_KINDS if any(w in m.group(2) for w in words)), "other")
cands.append((i, cn_int(m.group(1)), kind))
# 章号要严格递增:取最长的递增序列,正文里偶尔提到「第八章」的句子就被排除了(S04 p14)。
# 一样长时取靠后的(目录在前、正文在后)
best: list[list[tuple[int, int, str]]] = []
for c in cands:
prev = max((b for b in best if b[-1][1] < c[1]), key=lambda b: (len(b), b[0][0]), default=[])
best.append([*prev, c])
heads = [(i, kind) for i, _, kind in max(best, key=lambda b: (len(b), b[0][0]), default=[])]
spans = list(zip(heads, [*heads[1:], (len(sentences), "")], strict=False))
# 目录那一串胜出时,正文会全被算成最后一章:最后一章超过全文一半就不可靠(采购需求章本身可以很长,S06)
if len(heads) < 3 or len(sentences) - heads[-1][0] > 0.5 * len(sentences):
return {}
out: dict[str, str] = {}
for (start, kind), (end, _) in spans:
for s in sentences[start:end]:
out[s.key] = kind
return out
# ---------------------------------------------------------------- 表后往回指的说明
TRAILING_NOTE = re.compile(
r"^(?:(\d{1,2})|\(\d{1,2}\)|\d{1,2}[.、.]|总?备注[::]|注[::]|说明[::])?\s*(?:上述|以上)"
)
CONSEQUENCE = re.compile(r"无效|不实质性响应|未实质性响应|不合格|否决|不予受理|拒绝")
TRAILING_MAX = 40 # 往回最多认这么多句
CN_TOP = re.compile(r"^[一二三四五六七八九十]{1,3}、")
TRAILING_ROW_MAX = 60
# 说明的是标注方式,或者说的是「不导致无效」:不是往回指的无效说明(顾问审读第 5 条)
NOT_CONSEQUENCE = re.compile(r"★|*|\*|斜体|下划线|非实质|不导致|不属于|不作为|不视为")
def trailing_notes(sentences: list[Sentence]) -> dict[str, Sentence]:
"""表格、列表下方往回指的说明:{被说明的句子编号: 说明那一句}。
「(1)上述资格证明文件 1~6 项必须全部提供……否则投标文件无效」写在资格证明表的下面(S05 p54),
表里的每一行本身不写「无效」,上级条款也没有;只往上找上级的话认不出来(评测 §3.2,33 条漏抽)。
范围:从说明往回,到章、「一、」这样的中文编号小节、十进制编号条款为止(「附件十一、……」这样的附件标题
算在里面),最多 TRAILING_MAX 句、不超过前两页。
"""
out: dict[str, Sentence] = {}
for i, s in enumerate(sentences):
text = s.text.strip()
if not (TRAILING_NOTE.match(text) and CONSEQUENCE.search(text)) or NOT_CONSEQUENCE.search(text):
continue
for x in reversed(sentences[max(0, i - TRAILING_MAX) : i]):
xt = x.text.strip()
if x.page_no < s.page_no - 2 or CHAPTER.match(xt) or CN_TOP.match(xt) or DOTTED.match(xt):
break
if TRAILING_NOTE.match(xt) and CONSEQUENCE.search(xt):
continue # 同一组说明里的另一条
if xt.endswith((":", ":")) and len(xt) <= 8:
continue # 「说明:」这类引导句
if not (LIST_ITEM.match(xt) or DOTTED.match(xt) or len(xt) <= TRAILING_ROW_MAX):
continue # 只覆盖编号条目和表格行,不覆盖夹在中间的解释段落(S05 p54「专门面向中小企业,是指……」)
out.setdefault(x.key, s)
if xt.startswith("附件"):
break
return out
# ---------------------------------------------------------------- ★ 节
# 编号 + ★ 的节标题:「三、★商务要求」「★(一)技术要求」「8★售后服务(质保期)」
STAR_HEADING = re.compile(
r"^(?P<s1>★)?\s*(?:(?P<cn>[一二三四五六七八九十]{1,3})、|((?P<paren>[一二三四五六七八九十]{1,3}))|(?P<num>\d{1,2})(?![\d..])[、]?)"
r"\s*(?P<s2>★)?"
)
# 节标题去掉编号和 ★ 后的文字:短、不带标点和数字(「★5、采用2*15W功放芯片,声音清晰;」是参数行,不是节标题)
STAR_TITLE = re.compile(r"^[^,,;;::。\d]{2,12}(?:([^)]{1,8}))?$")
STAR_SCOPE_MAX = 800
def _heading(text: str) -> tuple[str, int] | None:
"""节编号的写法和序号:("cn", 3)("paren", 1)("num", 8)。"""
m = STAR_HEADING.match(text)
if not m or not any(m.group(k) for k in ("cn", "paren", "num")):
return None
for family in ("cn", "paren", "num"):
if m.group(family):
v = m.group(family)
return family, int(v) if family == "num" else cn_int(v)
return None
def star_sections(sentences: list[Sentence]) -> dict[str, Sentence]:
"""★ 标在节标题上时,节里的每一句:{句子编号: 节标题}(标题本身不在里面,它自己带 ★)。
节标题:带编号、带 ★、不超过 20 字、不以句号结尾。节到下一个同级编号(三 → 四、(一)→(二)、8 → 9)、
更高一级的标题或下一章为止。
"""
out: dict[str, Sentence] = {}
for i, s in enumerate(sentences):
text = s.text.strip()
h = _heading(text)
m = STAR_HEADING.match(text)
if h is None or m is None or not (m.group("s1") or m.group("s2")) or not STAR_TITLE.match(text[m.end() :]):
continue
family, no = h
higher = {"cn": (), "paren": ("cn",), "num": ("cn", "paren")}[family]
for x in sentences[i + 1 : i + 1 + STAR_SCOPE_MAX]:
xt = x.text.strip()
if CHAPTER.match(xt) or (CHAPTER_TITLE.search(xt[:40]) and len(xt) <= 40):
break
xh = _heading(xt)
if xh and (xh[0] in higher or (xh[0] == family and xh[1] == no + 1)):
break
out.setdefault(x.key, s)
return out

View File

@@ -1,11 +1,14 @@
"""文字层:从规范化 PDF 取逐字坐标和格式 → 切句 → 每句生成 quads 和稳定编号(架构 §3.5,技术验证 S2、S3)。
切句规则:
1. 同一行里字间空隙大于 1.5 个字号时断开(表格单元格、制表位),得到「行片段」;
2. 句子在 。!?; 处结束;行片段没写到右边距附近(段落末行)时也结束;
3. 句子不跨页。编号 <文档>:p<页>:s<页内序号>;规范化 PDF 不再重新生成,所以编号稳定。
1. 有框线的表格按单元格取字:一格读完再读下一格(同一行从左到右),每格的结束就是句子的结束。
以前按整页横向一行取字,一行里几格各自换行时,各格的第 1 行拼在一起、第 2 行又拼在一起,
竖排窄列成了一字一句(S02 评分表「项/目/实/施」,评测 F8);
2. 表格以外,同一行里字间空隙大于 1.5 个字号时断开(制表位),得到「行片段」;
3. 句子在 。!?; 处结束;行片段没写到右边距附近(段落末行)时也结束,单元格里按这一格的右边界算;
4. 句子不跨页。编号 <文档>:p<页>:s<页内序号>;规范化 PDF 不再重新生成,所以编号稳定。
格式:斜体看字体名或字形倾斜;下划线看字下方的横线;加粗看字体名(S2:斜体且下划线按字符 100% 一致)。
TODO(S3 改进):表格按单元格切句;页眉页脚带单独识别(现在都算正文)。
页眉页脚带里的页码和重复行单独成句(location = header_footer)。
"""
from __future__ import annotations
@@ -140,7 +143,45 @@ def _flags(c: dict[str, Any], hlines: list[tuple[float, float, float]]) -> str:
return ("b" if "Bold" in name else "") + ("i" if italic else "") + ("u" if under else "")
def _fragments(chars: list[dict[str, Any]]) -> list[list[dict[str, Any]]]:
Cell = tuple[float, float, float, float] # 单元格 (x0, top, x1, bottom)
@dataclass(slots=True)
class FragInfo:
right: float | None = None # 单元格的右边界:判断段落末行用它,不用页面右边距
cell_end: bool = False # 单元格的最后一个片段:句子在这里结束
def _cell_of(c: dict[str, Any], cells: list[Cell]) -> int | None:
x, y = (c["x0"] + c["x1"]) / 2, (c["top"] + c["bottom"]) / 2
return next((i for i, (x0, top, x1, bottom) in enumerate(cells) if x0 <= x <= x1 and top <= y <= bottom), None)
def _fragments(
chars: list[dict[str, Any]], cells: list[Cell] | None = None
) -> tuple[list[list[dict[str, Any]]], list[FragInfo]]:
"""行片段,按阅读顺序:表格外的文字一行一行,表格里一格一格(格子按上边、左边排序)。"""
cells = cells or []
by_cell: dict[int | None, list[dict[str, Any]]] = {}
for c in chars:
by_cell.setdefault(_cell_of(c, cells), []).append(c)
blocks: list[tuple[tuple[float, float], list[list[dict[str, Any]]], FragInfo | None]] = []
for frag in _line_fragments(by_cell.pop(None, [])):
blocks.append(((min(c["top"] for c in frag), frag[0]["x0"]), [frag], None))
for i, cell_chars in by_cell.items():
assert i is not None
x0, top, x1, _ = cells[i]
blocks.append(((top, x0), _line_fragments(cell_chars), FragInfo(right=x1)))
frags: list[list[dict[str, Any]]] = []
infos: list[FragInfo] = []
for _, fs, info in sorted(blocks, key=lambda b: b[0]):
for j, f in enumerate(fs):
frags.append(f)
infos.append(FragInfo(right=info.right, cell_end=j == len(fs) - 1) if info else FragInfo())
return frags, infos
def _line_fragments(chars: list[dict[str, Any]]) -> list[list[dict[str, Any]]]:
chars = sorted(chars, key=lambda c: (round(c["bottom"]), c["x0"]))
lines: list[list[dict[str, Any]]] = []
cur: list[dict[str, Any]] = []
@@ -206,7 +247,7 @@ def _underlines(page: Any) -> list[tuple[float, float, float]]:
return [(x0, x1, y) for x0, x1, y in hl if not (joined(x0, y) and joined(x1, y))]
def _page(doc_id: str, page: Any, frags: list[list[dict[str, Any]]], repeats: set[str]) -> Page:
def _page(doc_id: str, page: Any, frags: list[list[dict[str, Any]]], infos: list[FragInfo], repeats: set[str]) -> Page:
"""页眉页脚带里的页码和重复行单独成句(location = header_footer),不并进正文句子(S3 的改进点 2)。"""
hlines = _underlines(page)
chars = [c for c in page.chars if str(c["text"]).strip()]
@@ -237,9 +278,10 @@ def _page(doc_id: str, page: Any, frags: list[list[dict[str, Any]]], repeats: se
cur.append((fi, c))
if c["text"] in END:
ended = True
if cur and not ended and frag[-1]["x1"] < right - 2 * frag[-1]["size"]:
limit = infos[fi].right or right
if cur and (infos[fi].cell_end or (not ended and frag[-1]["x1"] < limit - 2 * frag[-1]["size"])):
groups.append(("body", cur))
cur = []
cur, ended = [], False
if cur:
groups.append(("body", cur))
open_tail = not ended # 页面写完了句子还没结束
@@ -282,9 +324,14 @@ def _page(doc_id: str, page: Any, frags: list[list[dict[str, Any]]], repeats: se
def extract(doc_id: str, pdf: bytes) -> list[Page]:
with pdfplumber.open(io.BytesIO(pdf)) as doc:
pages = [(p, _fragments([c for c in p.chars if str(c["text"]).strip()])) for p in doc.pages]
repeats = _band_repeats(pages)
return [_page(doc_id, p, frags, repeats) for p, frags in pages]
pages = [(p, *_fragments([c for c in p.chars if str(c["text"]).strip()], _cells(p))) for p in doc.pages]
repeats = _band_repeats([(p, frags) for p, frags, _ in pages])
return [_page(doc_id, p, frags, infos, repeats) for p, frags, infos in pages]
def _cells(page: Any) -> list[Cell]:
"""有框线的表格的单元格(pdfplumber 按横竖线找表格)。"""
return [(c[0], c[1], c[2], c[3]) for t in page.find_tables() for c in t.cells if c]
CLOSERS = set("”’))》」』】")

View File

@@ -186,7 +186,10 @@ async def test_all_categories_and_project_info(user: User) -> None:
for clause in ("25.3.1", "25.3.2"): # 本身没写「无效」,靠上级标题认定
it = item(clause)
assert it["zone"] == "formal" and "invalidity" in it["categories"], it
assert item("资格要求")["categories"] == ["qualification"]
# 口径 Q4:政府采购项目不具备资格要求的投标无效(87 号令第六十三条),资格条件带「无效」类别,依据是法规
qual = item("资格要求")
assert qual["categories"] == ["qualification", "invalidity"]
assert (qual["importance"], qual["importance_basis"]) == ("substantive", "statute")
assert item("技术方案")["spec"]["scoring"]["points"] == "20"
assert item("系统支持")["spec"]["tech_param"] == {"param": "系统支持", "op": ">=", "value": "200", "unit": "个"}
deadline = item("响应文件递交截止时间")

View File

@@ -402,3 +402,155 @@ def test_project_info_status() -> None:
assert info.status["budget_total"]["codes"] == ["ambiguous"]
assert info.status["code"]["codes"] == ["not_found_with_evidence"]
assert info.status["sme_only"]["codes"] == ["missing_answer"]
# ---------------------------------------------------------------- S01–S10 评测后的修正(extract-eval §3、§5)
def lines(*texts: str) -> list[Sentence]:
return [sent(t, seq=i + 1) for i, t in enumerate(texts)]
def test_chapters_skip_toc_and_references() -> None:
from bid_doctor.document.pipeline.structure import chapters
ss = lines(
"第一章招标公告", # 目录:章标题挨着
"第二章投标人须知",
"投标邀请",
"第一章招标公告",
"公告正文。",
"资格要求见第四章《采购需求》。", # 引用,不是章标题
"第二章投标人须知",
"须知正文。",
"第八章投标文件格式要求另行说明", # 正文里偶尔提到后面的章:章号要严格递增,取最长的序列
"须知正文。",
"第三章采购需求",
"参数。",
"第四章投标文件格式",
)
ch = chapters(ss)
assert [ch.get(s.key) for s in ss] == [
None, None, None, "notice", "notice", "notice", "instructions", "instructions", "instructions",
"instructions", "requirements", "requirements", "formats",
] # fmt: skip
def test_trailing_note_covers_rows_above() -> None:
from bid_doctor.document.pipeline.structure import trailing_notes
ss = lines(
"一、资格证明文件",
"营业执照或相关部门登记证明",
"财务状况报告",
"说明:",
"(1)上述资格证明文件1~2项必须全部按要求提供,否则投标文件无效。",
)
notes = trailing_notes(ss)
assert set(notes) == {ss[1].key, ss[2].key} # 到「一、」为止;「说明:」引导句不算
d = doc(sentences=ss)
d.trailing = notes
assert check_basis("explicit_invalid", [ss[1]], [ss[1]], d)
def test_star_section_heading() -> None:
from bid_doctor.checklist.extract import importance_of
from bid_doctor.document.pipeline.structure import star_sections
ss = lines(
"三、★商务要求", "1.交货期", "自合同签定之日起60工作日内安装调试完成。", "四、其他说明", "本项目不接受分包。"
)
scope = star_sections(ss)
assert set(scope) == {ss[1].key, ss[2].key} # 到下一个同级编号「四、」为止
d = doc(sentences=ss)
d.star_scope = scope
assert importance_of([ss[2]], d) == ("substantive", "marker")
assert check_basis("star", [ss[2]], [ss[2]], d)
assert importance_of([ss[4]], d) == ("normal", None)
def test_star_defined_as_scored_is_important() -> None:
"""文件把 ★ 定义成「核心技术参数」、不满足只扣分(S02):★ 是「重要」,不是实质性(口径 Q2)。"""
from bid_doctor.checklist.extract import build_doc, importance_of
star, define = sent("★设备采用国产自主可控CPU芯片。", seq=1), sent("备注:加★为核心技术参数;", seq=2)
scored = sent("每有一项核心技术参数不满足的减4分。", seq=3)
d = build_doc([star, define, scored])
assert is_marked(star, d.declared) is None
assert importance_of([star], d) == ("important", "marker")
assert [(n.markers, n.meaning) for n in d.notes] == [({"star"}, "important")]
# 没有计分证据:维持实质性(降级的代价是把致命显示成重要)
assert is_marked(star, build_doc([star, define]).declared) == "star"
# 有一处说 ★ 是实质性要求、或者带 ★ 的句子写着后果,就按实质性(顾问审读第 1 条的反例)
for extra in ("本文件中打“★”内容为实质性要求。", "打★号的参数负偏离的,视为未响应。"):
assert is_marked(star, build_doc([star, define, scored, sent(extra, seq=4)]).declared) == "star"
def test_heuristics_reject_lookalikes() -> None:
"""顾问审读第 2、5、6 条:目录隔着页码、参数行带 ★ 编号、否定写法的表后说明。"""
from bid_doctor.document.pipeline.structure import chapters, star_sections, trailing_notes
toc = lines("第一章招标公告", "3", "第二章投标人须知", "10", "第三章采购需求", "20")
assert chapters(toc) == {} # 只有目录:认不出可靠的章,就不跳过
ss = lines("★5、采用2*15W功放芯片,声音清晰;", "符合参数要求的产品彩页复印件。", "6、输出功率≥30W。")
assert star_sections(ss) == {} # 参数行不是节标题
ss = lines("三、商务要求(★为实质性要求)", "1.交货期30天。")
assert star_sections(ss) == {} # ★ 不紧挨编号:是图例,不是节标题
ss = lines("一、参数表", "内存不低于32GB", "以上参数为非实质性要求,不满足不导致投标无效。")
assert trailing_notes(ss) == {}
def test_pending_item_does_not_upgrade_formal_item() -> None:
"""待确认的条目不能借同一句的正式条目进正式清单(评测 §3.1:S07 普通参数被标成实质性)。"""
s = sent("支持SM2、SM3、SM4国密算法。")
tech = Extracted(task="tech_param", categories=["tech_param"], summary="国密算法", sentences=[s], zone="formal")
sub = Extracted(
task="substantive", categories=["invalidity"], summary="国密算法", sentences=[s], zone="pending",
importance="substantive", failures=["basis_unmatched"],
) # fmt: skip
(merged,) = merge_items([tech, sub])
assert (merged.categories, merged.importance, merged.zone) == (["tech_param"], "normal", "formal")
def test_near_output_limit_counts_as_incomplete() -> None:
from bid_doctor.ai.providers.base import near_limit
assert near_limit(7757) and near_limit(8192) # S03:7757 / 8192 自己收了尾
assert not near_limit(5000) and not near_limit(None)
def test_wps_doc_font_charset() -> None:
from bid_doctor.document.pipeline.convert import fix_wps_doc_charset
name = "宋体".encode("utf-16-le") + b"\x00\x00"
ffn = bytes([40 + len(name) - 1, 0, 0x90, 0x01, 0x02, 0]) + bytes(34) + name # chs = 02(SYMBOL)
fixed = fix_wps_doc_charset(b"\xd0\xcf" + ffn + b"\x00" * 8)
assert fixed[2 + 4] == 0x86
assert fix_wps_doc_charset(b"\xd0\xcf" + ffn.replace(b"\x02", b"\x00", 1)) == b"\xd0\xcf" + ffn.replace(
b"\x02", b"\x00", 1
) # 字符集本来没错:不动
def test_pending_category_kept_on_formal_item() -> None:
"""待确认的「无效」类别不并进正式条目,但也不丢:记在正式条目上(顾问审读第 3 条)。"""
s = sent("25.3.1投标人未按要求提供资格证明文件的。")
qual = Extracted(
task="qualification", categories=["qualification"], summary="资格证明", sentences=[s], zone="formal"
)
sub = Extracted(
task="substantive", categories=["invalidity"], summary="资格证明", sentences=[s], zone="pending",
importance="substantive", failures=["quote_not_original"],
) # fmt: skip
(merged,) = merge_items([qual, sub])
assert merged.categories == ["qualification"]
assert merged.ai_values["pending_categories"] == {"invalidity": ["quote_not_original"]}
def test_template_and_blanket_patterns() -> None:
from bid_doctor.checklist.extract import BLANKET, TEMPLATE
assert TEMPLATE.match("二、我单位具有良好的商业信誉和健全的财务会计制度。")
assert not TEMPLATE.match("投标人须具有良好的商业信誉和健全的财务会计制度。")
assert BLANKET.search("3投标人《资格证明文件》有任何一项不符合《资格审查要求》的,其投标无效。")
assert BLANKET.search("(1)上述资格证明文件1~6项必须全部提供,否则投标无效。")
assert not BLANKET.search("投标文件未按上述要求签章的,投标无效。") # 这是一条无效情形,不是概括句