feat(check): 需人工核对(P2)和分项 / 单价上限(C7)
- 需人工核对:份数、密封、递交、CA 签章加密、解密、保证金这类文件里看不出来的要求, 从正式清单按主题合并、带招标原文出处;manual-checks 接口按项目存勾选、写勾选人和时间, 复查沿用;查标页加「需人工核对」一块,覆盖范围写明列了几条 - 「报价」类清单项:读标从采购需求的清单表规则读出分项 / 单价上限(checklist/price.py); 查标交给规则 PRICE.OVER_CEILING_ITEM 比对分项报价表,不交给模型;规则集 2026.11-r2 - 版面工具(分页、分行、归栏)挪到 document/pipeline/layout.py,读标和查标共用 - 评测:price_defects.py 从规矩投标文件派生只带报价问题的 4 份(P01、P02、P03) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -24,6 +24,12 @@ class ResolveIn(BaseModel):
|
||||
note: str | None = Field(default=None, max_length=500)
|
||||
|
||||
|
||||
class ManualCheckIn(BaseModel):
|
||||
check_key: str = Field(min_length=1, max_length=200)
|
||||
checked: bool
|
||||
note: str | None = Field(default=None, max_length=500)
|
||||
|
||||
|
||||
class BulkResolveIn(BaseModel):
|
||||
finding_ids: list[UUID] = Field(min_length=1, max_length=500)
|
||||
status: Literal["confirmed_ok", "ignored"]
|
||||
@@ -96,3 +102,15 @@ async def resolve_finding(finding_id: UUID, body: ResolveIn, ctx: Ctx) -> dict[s
|
||||
async def reopen_finding(finding_id: UUID, ctx: Ctx) -> dict[str, Any]:
|
||||
async with tenant_tx(ctx) as s:
|
||||
return await service.reopen(s, ctx, finding_id)
|
||||
|
||||
|
||||
@router.get("/projects/{project_id}/manual-checks")
|
||||
async def list_manual_checks(project_id: UUID, ctx: Ctx) -> dict[str, Any]:
|
||||
async with tenant_tx(ctx) as s:
|
||||
return {"items": await service.manual_checks(s, ctx, project_id)}
|
||||
|
||||
|
||||
@router.put("/projects/{project_id}/manual-checks")
|
||||
async def set_manual_check(project_id: UUID, body: ManualCheckIn, ctx: Ctx) -> dict[str, Any]:
|
||||
async with tenant_tx(ctx) as s:
|
||||
return await service.set_manual_check(s, ctx, project_id, body.check_key, body.checked, body.note)
|
||||
|
||||
@@ -26,6 +26,7 @@ SKIP_LABELS = {
|
||||
"schedule": "日期与递交要求",
|
||||
"scoring": "评分标准(没要求附证明材料)",
|
||||
"no_proof": "资格条件(写明无需提供证明,或由采购人查询)",
|
||||
"price": "报价类要求(分项、单价上限):由规则比对分项报价表",
|
||||
}
|
||||
|
||||
|
||||
@@ -43,6 +44,8 @@ def decide(item: dict[str, Any], modes: dict[str, Mode]) -> Decision:
|
||||
text = f"{item.get('title') or ''}{item.get('requirement_text') or ''}"
|
||||
if cats == {"invalidity"} and item.get("importance_basis") != "marker":
|
||||
return Decision(False, skip="condition") # 「……的,投标无效」这类情形;★ / 斜体下划线标注的条款是要求,照常找
|
||||
if cats == {"price"}:
|
||||
return Decision(False, skip="price")
|
||||
if "schedule" in cats and not cats & {"tech_param", "qualification", "scoring"}:
|
||||
return Decision(False, skip="schedule")
|
||||
evidence = bool(item.get("evidence_required") or EVIDENCE.search(text)) and not AWARD.search(text)
|
||||
|
||||
76
server/src/bid_doctor/check/manual.py
Normal file
76
server/src/bid_doctor/check/manual.py
Normal file
@@ -0,0 +1,76 @@
|
||||
"""需人工核对的条目(FR-M3-04、决策 P2):份数、密封、递交、CA 签章加密、解密、保证金——投标文件里看不出来,
|
||||
列出来请用户递交前自己核对;勾选按项目存在 manual_check,复查沿用,报告写勾选人和时间。
|
||||
|
||||
- 条目只从正式清单里来(带招标原文出处),不凭空生成提醒;技术参数、评分项、资格条件不算;
|
||||
- 同一主题在清单里常有好几条(公告、须知、无效情形各写一遍),按主题合并成一条,check_key = rule:MANUAL.<主题>,
|
||||
清单改了、条目编号变了,勾选也还在;
|
||||
- 招标文件写明「免收保证金」的,保证金这条不列。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
|
||||
TOPICS: tuple[tuple[str, str, re.Pattern[str]], ...] = (
|
||||
("DECRYPT", "开标时在规定时间内解密投标文件", re.compile(r"解密")),
|
||||
("SIGN_ENCRYPT", "用 CA 数字证书加盖电子签章、加密投标文件", re.compile(r"CA|电子签章|电子公章|数字证书|加密")),
|
||||
("PAPER", "纸质投标文件的份数、装订、密封", re.compile(r"正本|副本|份数|装订|密封|封装|封套|纸质|U盘|光盘")),
|
||||
("DEPOSIT", "投标保证金按要求缴纳", re.compile(r"保证金")),
|
||||
(
|
||||
"SUBMIT",
|
||||
"在投标截止时间前递交(上传)投标文件",
|
||||
re.compile(
|
||||
r"(上传|递交|提交|送达).{0,12}(投标|响应|磋商)文件|(投标|响应|磋商)文件.{0,12}(上传|递交|提交|送达)|递交截止"
|
||||
),
|
||||
),
|
||||
)
|
||||
SKIP_CATEGORIES = {"tech_param", "scoring", "qualification"}
|
||||
NOT_NEEDED = re.compile(r"免收|免交|不收取|无需|无须|不需|不交纳")
|
||||
NOT_MANUAL = re.compile(r"相似|雷同|同一单位|同一账户|《") # 围串标情形;点名要交的文件(由「文件要在」检查)
|
||||
PREFERRED = {"SIGN_ENCRYPT": re.compile(r"CA|加密")} # 同一主题里拿哪条做代表
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class Topic:
|
||||
key: str
|
||||
label: str
|
||||
items: list[dict[str, Any]] = field(default_factory=list[dict[str, Any]])
|
||||
|
||||
@property
|
||||
def check_key(self) -> str:
|
||||
return f"rule:MANUAL.{self.key}"
|
||||
|
||||
|
||||
def topic_of(item: dict[str, Any]) -> str | None:
|
||||
cats = set(item.get("categories") or [])
|
||||
title = str(item.get("title") or "")
|
||||
if cats & SKIP_CATEGORIES or NOT_MANUAL.search(title):
|
||||
return None
|
||||
for key, _, pattern in TOPICS:
|
||||
if pattern.search(title):
|
||||
return None if key == "DEPOSIT" and NOT_NEEDED.search(title) else key
|
||||
return None
|
||||
|
||||
|
||||
def topics(items: list[dict[str, Any]]) -> list[Topic]:
|
||||
"""按主题合并的需人工核对条目,按 TOPICS 的顺序;每个主题里写明截止时间的、写着主题核心词的条目排前面(做代表)。"""
|
||||
found: dict[str, Topic] = {}
|
||||
for item in items:
|
||||
key = topic_of(item)
|
||||
if key is None:
|
||||
continue
|
||||
label = next(lbl for k, lbl, _ in TOPICS if k == key)
|
||||
found.setdefault(key, Topic(key, label)).items.append(item)
|
||||
order = [k for k, _, _ in TOPICS]
|
||||
out = sorted(found.values(), key=lambda t: order.index(t.key))
|
||||
for t in out:
|
||||
pref = PREFERRED.get(t.key)
|
||||
t.items.sort(
|
||||
key=lambda i: (
|
||||
not ((i.get("spec") or {}).get("schedule") or {}).get("at"),
|
||||
bool(pref) and not pref.search(str(i.get("title") or "")),
|
||||
)
|
||||
)
|
||||
return out
|
||||
@@ -22,11 +22,16 @@ from typing import Any
|
||||
from uuid import UUID
|
||||
|
||||
from bid_doctor.check.match import BidDoc
|
||||
from bid_doctor.document.pipeline import layout
|
||||
from bid_doctor.document.pipeline.layout import column as _col
|
||||
from bid_doctor.document.pipeline.layout import rows as rows_of
|
||||
from bid_doctor.document.pipeline.layout import span as _span
|
||||
from bid_doctor.document.pipeline.text_layer import Sentence, norm
|
||||
|
||||
LABELS = {
|
||||
"PRICE.OVER_CEILING_TOTAL": "报价和最高限价比对",
|
||||
"PRICE.OVER_CEILING_LOT": "报价和所投采购包的最高限价比对",
|
||||
"PRICE.OVER_CEILING_ITEM": "分项报价和分项 / 单价上限比对",
|
||||
"PRICE.CAPITAL_MISMATCH": "大写金额和小写金额核对",
|
||||
"PRICE.SUM_MISMATCH": "分项报价合计核对",
|
||||
"PRICE.TOTAL_INCONSISTENT": "投标总价前后一致",
|
||||
@@ -39,7 +44,7 @@ LABELS = {
|
||||
"FORM.RESPONSE_NEGATIVE": "响应表里的未响应、负偏离",
|
||||
}
|
||||
RULES_NOT_YET = (
|
||||
"分项报价表和采购需求清单逐项对照、分项 / 单价上限、格式文件固定文字被改、"
|
||||
"分项报价表和采购需求清单逐项对照、格式文件固定文字被改、"
|
||||
"授权代表和签字人一致、声明函填报、签章有无、低价说明、跨页的响应表"
|
||||
)
|
||||
|
||||
@@ -58,9 +63,33 @@ class Facts:
|
||||
ceiling_note: str | None = None # 没有可比的限价时的原因(写进覆盖范围)
|
||||
validity_days: int | None = None
|
||||
validity_item: UUID | None = None
|
||||
caps: list[ItemCap] = field(default_factory=list["ItemCap"]) # 所投采购包的分项 / 单价上限(「报价」类清单项)
|
||||
refs: dict[str, list[UUID]] = field(default_factory=dict[str, list[UUID]])
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ItemCap:
|
||||
"""分项上限(checklist/price.py 抽出的「报价」类清单项):per = unit 按单价比,item 按分项总价比。"""
|
||||
|
||||
name: str
|
||||
cap: Decimal
|
||||
per: str
|
||||
item: UUID | None = None
|
||||
|
||||
|
||||
def caps_of(items: dict[UUID, dict[str, Any]]) -> list[ItemCap]:
|
||||
out = []
|
||||
for item_id, p in items.items():
|
||||
spec = (p.get("spec") or {}).get("price") or {}
|
||||
if spec.get("kind") in ("unit_cap", "item_cap") and spec.get("name") and spec.get("cap"):
|
||||
try:
|
||||
cap = Decimal(str(spec["cap"]))
|
||||
except InvalidOperation:
|
||||
continue
|
||||
out.append(ItemCap(str(spec["name"]), cap, "unit" if spec["kind"] == "unit_cap" else "item", item_id))
|
||||
return out
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class Hit:
|
||||
code: str
|
||||
@@ -164,55 +193,9 @@ def yuan(v: Decimal) -> str:
|
||||
return f"{v:,.2f} 元"
|
||||
|
||||
|
||||
# ---------- 版面:页、行、单元格 ----------
|
||||
def _box(s: Sentence) -> tuple[float, float, float, float]:
|
||||
return (
|
||||
min(q[0] for q in s.quads),
|
||||
min(q[1] for q in s.quads),
|
||||
max(q[2] for q in s.quads),
|
||||
max(q[3] for q in s.quads),
|
||||
)
|
||||
|
||||
|
||||
# ---------- 版面:页、行、单元格(document/pipeline/layout.py) ----------
|
||||
def pages_of(doc: BidDoc) -> dict[int, list[Sentence]]:
|
||||
out: dict[int, list[Sentence]] = {}
|
||||
for s in doc.sentences:
|
||||
out.setdefault(s.page_no, []).append(s)
|
||||
return out
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class _Row:
|
||||
top: float
|
||||
bottom: float
|
||||
in_cell: bool
|
||||
items: list[Sentence]
|
||||
|
||||
|
||||
def rows_of(page: list[Sentence]) -> list[list[Sentence]]:
|
||||
"""把一页的句子分成行,行内从左到右。有框线的表格按单元格的上沿分行:同一行的格子上沿相同,
|
||||
垂直居中的短格、折成几句的长格、纵向合并的格(只算进它的第一行)都不会串行;其余按上下范围重叠。"""
|
||||
rows: list[_Row] = []
|
||||
|
||||
def top_of(s: Sentence) -> float:
|
||||
return s.cell[1] if s.cell else _box(s)[1]
|
||||
|
||||
for s in sorted((x for x in page if x.quads), key=top_of):
|
||||
if s.cell:
|
||||
r = next((r for r in rows if r.in_cell and abs(r.top - s.cell[1]) < 2), None)
|
||||
if r:
|
||||
r.items.append(s)
|
||||
else:
|
||||
rows.append(_Row(s.cell[1], s.cell[1], True, [s]))
|
||||
continue
|
||||
_, top, _, bottom = _box(s)
|
||||
last = rows[-1] if rows else None
|
||||
if last and not last.in_cell and top < last.bottom - 1:
|
||||
last.bottom = max(last.bottom, bottom)
|
||||
last.items.append(s)
|
||||
else:
|
||||
rows.append(_Row(top, bottom, False, [s]))
|
||||
return [sorted(r.items, key=lambda x: _box(x)[0]) for r in sorted(rows, key=lambda r: r.top)]
|
||||
return layout.pages(doc.sentences)
|
||||
|
||||
|
||||
def row_index(doc: BidDoc) -> dict[int, list[Sentence]]:
|
||||
@@ -220,25 +203,6 @@ def row_index(doc: BidDoc) -> dict[int, list[Sentence]]:
|
||||
return {id(c): row for page in pages_of(doc).values() for row in rows_of(page) for c in row}
|
||||
|
||||
|
||||
def _col(cell: Sentence, cols: dict[str, tuple[float, float]]) -> str | None:
|
||||
"""单元格落在哪一栏:和表头单元格左右范围重叠最多的那一栏;都不重叠(窄数字居中、表头字靠左)取中心最近的。
|
||||
cols 要包含表头的每一格,否则别的栏的数字会被归到最近的已知栏。"""
|
||||
x0, x1 = _span(cell)
|
||||
best, score = None, 0.0
|
||||
for name, (c0, c1) in cols.items():
|
||||
overlap = min(x1, c1) - max(x0, c0)
|
||||
if overlap > score:
|
||||
best, score = name, overlap
|
||||
if best is None and cols:
|
||||
mid = (x0 + x1) / 2
|
||||
best = min(cols, key=lambda n: abs((cols[n][0] + cols[n][1]) / 2 - mid))
|
||||
return best
|
||||
|
||||
|
||||
def _span(c: Sentence) -> tuple[float, float]:
|
||||
return (c.cell[0], c.cell[2]) if c.cell else (_box(c)[0], _box(c)[2])
|
||||
|
||||
|
||||
# ---------- 分项报价表、开标一览表 ----------
|
||||
@dataclass(slots=True)
|
||||
class ItemRow:
|
||||
@@ -507,6 +471,66 @@ def check_prices(
|
||||
)
|
||||
|
||||
|
||||
PAREN = re.compile(r"[((][^))]*[))]")
|
||||
|
||||
|
||||
def _match_cap(name: str, caps: list[ItemCap]) -> ItemCap | None:
|
||||
"""分项名称对上哪条上限。投标文件的分项名称通常照抄采购需求,所以要求很严:完全相同(或去掉括注后相同)优先;
|
||||
否则字对重合度不低于 0.8,或一方包含另一方且长度比不低于 0.8。「服务器」不会对上「服务器机柜」。"""
|
||||
n = norm(name)
|
||||
for c in caps:
|
||||
if norm(c.name) == n or PAREN.sub("", norm(c.name)) == PAREN.sub("", n):
|
||||
return c
|
||||
best, score = None, 0.0
|
||||
for c in caps:
|
||||
k = norm(c.name)
|
||||
x, y = _bigrams(k), _bigrams(n)
|
||||
dice = 2 * len(x & y) / (len(x) + len(y)) if x and y else 0.0
|
||||
contained = (k in n or n in k) and min(len(k), len(n)) / max(len(k), len(n)) >= 0.8
|
||||
s = max(dice, 0.8 if contained else 0.0)
|
||||
if s >= 0.8 and s > score:
|
||||
best, score = c, s
|
||||
return best
|
||||
|
||||
|
||||
def check_item_caps(facts: Facts, tables: list[PriceTable], hits: list[Hit], notes: list[str]) -> None:
|
||||
"""分项 / 单价上限(第三层限价,FR-M3-06):分项报价表里同名分项的单价(或分项总价)超过招标文件写的上限。"""
|
||||
if not facts.caps:
|
||||
return
|
||||
if not tables:
|
||||
notes.append(f"招标文件写了分项 / 单价上限 {len(facts.caps)} 条,投标文件里没认出分项报价表,没有比对")
|
||||
return
|
||||
matched = 0
|
||||
for t in tables:
|
||||
for r in t.rows:
|
||||
name = next((c.text for c in r.cells if not money(c.text, plain=True) and len(norm(c.text)) >= 2), "")
|
||||
cap = _match_cap(name, facts.caps) if name else None
|
||||
value = None if cap is None else r.unit_price if cap.per == "unit" else r.line
|
||||
if cap is None or value is None or r.ambiguous:
|
||||
continue
|
||||
matched += 1
|
||||
if value <= cap.cap:
|
||||
continue
|
||||
what = "单价" if cap.per == "unit" else "分项总价"
|
||||
hits.append(
|
||||
Hit(
|
||||
"PRICE.OVER_CEILING_ITEM",
|
||||
"price_form",
|
||||
"fatal",
|
||||
f"「{name[:20]}」的{what}超过上限",
|
||||
f"分项报价表里「{name[:30]}」的{what}是 {yuan(value)},招标文件写的{what}上限是 {yuan(cap.cap)}。"
|
||||
"超过分项或单价上限通常会导致投标无效,请核对。",
|
||||
[(t.doc, c) for c in r.cells if money(c.text, plain=True) or c.text == name],
|
||||
target(name),
|
||||
basis=f"cap:{cap.item}",
|
||||
item=cap.item,
|
||||
detail={"value": str(value), "cap": str(cap.cap), "per": cap.per},
|
||||
)
|
||||
)
|
||||
if matched < len(facts.caps):
|
||||
notes.append(f"分项 / 单价上限 {len(facts.caps)} 条,在分项报价表里对上名称的 {matched} 条,其余没有比对")
|
||||
|
||||
|
||||
LOWER_MARK = re.compile(r"小写|[¥¥]|报价|总价|合计|金额|价款")
|
||||
|
||||
|
||||
@@ -1065,6 +1089,7 @@ def run(facts: Facts, docs: list[BidDoc], tender_sentences: list[Sentence]) -> t
|
||||
hits: list[Hit] = []
|
||||
notes: list[str] = []
|
||||
check_prices(facts, docs, tables, totals, hits, notes)
|
||||
check_item_caps(facts, tables, hits, notes)
|
||||
check_quotes(docs, tender, tables, totals, hits)
|
||||
check_foreign(facts, docs, tender, hits, notes)
|
||||
check_blanks(docs, hits, notes)
|
||||
|
||||
@@ -22,7 +22,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from bid_doctor.ai.models import AiCall
|
||||
from bid_doctor.ai.providers.base import TransientError
|
||||
from bid_doctor.billing import service as billing
|
||||
from bid_doctor.check import applicability, findings, match, rules
|
||||
from bid_doctor.check import applicability, findings, manual, match, rules
|
||||
from bid_doctor.check.models import CheckItemResult, CheckRun, CheckRunDocument, Finding, FindingCitation
|
||||
from bid_doctor.check.service import RULE_SET_VERSION, open_fatal
|
||||
from bid_doctor.checklist import edit as checklist_edit
|
||||
@@ -207,6 +207,13 @@ async def _stages(ctx: TenantContext, run_id: UUID) -> None:
|
||||
for d, x in res.evidence
|
||||
}
|
||||
hits = rules.drop_covered(hits, covered)
|
||||
n_manual = len(manual.topics(list(snapshot.values())))
|
||||
unseen = "份数、密封、递交、CA 签章加密、解密、保证金等文件里看不出来的要求"
|
||||
rule_cov["not_checked"].append(
|
||||
f"{unseen}:列入「需人工核对」{n_manual} 条,请递交前自己核对"
|
||||
if n_manual
|
||||
else f"{unseen}:清单里没有,没有列「需人工核对」"
|
||||
)
|
||||
await _stage(ctx, run_id, "verify") # 前两道校验在匹配时逐条做完;这里只是进度节点
|
||||
await _stage(ctx, run_id, "carry_over")
|
||||
async with tenant_tx(ctx) as s:
|
||||
@@ -262,6 +269,13 @@ async def _facts(s: AsyncSession, ctx: TenantContext, r: CheckRun, snapshot: dic
|
||||
days, item = rules.required_validity(snapshot)
|
||||
if item is not None:
|
||||
refs["validity"] = [UUID(c) for c in snapshot[item].get("citation_ids") or []]
|
||||
bidding_ids = {x.id for x in bidding}
|
||||
in_scope = {
|
||||
i: x for i, x in snapshot.items() if not x.get("lot_id") or not bidding_ids or UUID(x["lot_id"]) in bidding_ids
|
||||
}
|
||||
caps = rules.caps_of(in_scope)
|
||||
for c in caps:
|
||||
refs[f"cap:{c.item}"] = [UUID(x) for x in snapshot[c.item].get("citation_ids") or []] if c.item else []
|
||||
refs["deadline"] = refs.pop("deadline_at", [])
|
||||
refs["ceiling"] = refs.pop("ceiling_total", [])
|
||||
return rules.Facts(
|
||||
@@ -275,6 +289,7 @@ async def _facts(s: AsyncSession, ctx: TenantContext, r: CheckRun, snapshot: dic
|
||||
ceiling_note=note,
|
||||
validity_days=days,
|
||||
validity_item=item,
|
||||
caps=caps,
|
||||
refs=refs,
|
||||
)
|
||||
|
||||
|
||||
@@ -19,8 +19,8 @@ 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 rules
|
||||
from bid_doctor.check.models import CheckItemResult, CheckRun, CheckRunDocument, Finding, FindingCitation
|
||||
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
|
||||
@@ -36,9 +36,7 @@ from bid_doctor.project import service as projects
|
||||
from bid_doctor.project.models import Project
|
||||
|
||||
CHECK_TASK = "check_project"
|
||||
RULE_SET_VERSION = (
|
||||
"2026.11-r1" # 规则集有版本,检查批次记下当时用的版本(架构 §3.7);r1:报价形式、形式要件(rules.py)
|
||||
)
|
||||
RULE_SET_VERSION = "2026.11-r2" # 规则集有版本,检查批次记下当时用的版本(架构 §3.7);r2:加分项 / 单价上限
|
||||
WAIVE_WINDOW = timedelta(hours=24)
|
||||
ERROR_MESSAGES = {
|
||||
"bid_text_missing": "投标文件里几乎读不出文字(可能都是扫描件),扫描件识别还没有上线",
|
||||
@@ -461,3 +459,66 @@ async def item_results(s: AsyncSession, run_id: UUID) -> list[dict[str, Any]]:
|
||||
}
|
||||
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)
|
||||
|
||||
@@ -25,6 +25,7 @@ from uuid import UUID
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
from bid_doctor.ai import service as ai
|
||||
from bid_doctor.checklist import price
|
||||
from bid_doctor.core.db import TenantContext
|
||||
from bid_doctor.document.pipeline import structure, text_layer
|
||||
from bid_doctor.document.pipeline.text_layer import Sentence, norm
|
||||
@@ -845,6 +846,7 @@ async def run(
|
||||
info = _project_info([pair for got in results[len(jobs) :] for pair in got])
|
||||
items = merge_items(out)
|
||||
items += fallback(items, doc)
|
||||
items += price_items(items, doc)
|
||||
for e in items:
|
||||
if e.failures:
|
||||
e.ai_values = {**e.ai_values, "failures": e.failures}
|
||||
@@ -967,6 +969,28 @@ def add_pending_categories(e: Extracted, categories: list[str], failures: list[s
|
||||
e.ai_values = {**e.ai_values, "pending_categories": pending}
|
||||
|
||||
|
||||
def price_items(items: list[Extracted], doc: Doc) -> list[Extracted]:
|
||||
"""「报价」类清单项(C7):分项限价、单价上限,规则从清单表里读出,不调模型(checklist/price.py)。"""
|
||||
used = {e.sentences[0].key for e in items if e.sentences}
|
||||
out = []
|
||||
for c in price.caps(doc.sentences):
|
||||
if c.sentences[0].key in used:
|
||||
continue
|
||||
out.append(
|
||||
Extracted(
|
||||
task="rule",
|
||||
categories=["price"],
|
||||
summary=c.title,
|
||||
sentences=c.sentences,
|
||||
zone="formal",
|
||||
checks_passed=["id", "quote"],
|
||||
spec=c.spec(),
|
||||
lot_no=c.lot_no,
|
||||
)
|
||||
)
|
||||
return out
|
||||
|
||||
|
||||
def fallback(items: list[Extracted], doc: Doc) -> list[Extracted]:
|
||||
"""规则兜底:标注句没被带「无效」类别的正式条目覆盖 → 生成一条待确认(TV1:标注类召回 ≥ 99%)。
|
||||
跨页接续的两句算一条。这句已经是别的正式条目(资格、参数……)的首句时,不另生成条目(同一首句只能有一条),
|
||||
|
||||
129
server/src/bid_doctor/checklist/price.py
Normal file
129
server/src/bid_doctor/checklist/price.py
Normal file
@@ -0,0 +1,129 @@
|
||||
"""「报价」类清单项:分项限价、单价上限(决策 C7,FR-M3-06 的第三层限价)。规则抽取,不调模型。
|
||||
|
||||
招标文件里分项上限几乎都写在采购需求的清单表里:表头有「名称」一栏和「单价上限 / 分项限价 / 最高限价」一栏,
|
||||
表前一句写着是哪个采购包(「采购包2:……」)。按表头归栏逐行读出名称、数量、上限,每行一条清单项,
|
||||
出处是表格那一行。总价、采购包两层限价在项目基本信息里,不在这里。
|
||||
|
||||
- 只认表格(版面工具按单元格分行);散文里的「单价不得超过 × 元」写法先不认,用户可以在清单里手工新增;
|
||||
- 表头写「万元」的按万元折算;采购包一级的表(名称栏写「采购包 / 标段」)不算分项;
|
||||
- 数字只来自招标文件原文(红线 6)。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from decimal import Decimal, InvalidOperation
|
||||
|
||||
from bid_doctor.document.pipeline import layout
|
||||
from bid_doctor.document.pipeline.text_layer import Sentence, norm
|
||||
|
||||
CAP_HEAD = re.compile(r"(单价上限|单价最高限价|最高单价|分项限价|分项最高限价|限价|最高限价|单价控制价|控制价)")
|
||||
NAME_HEAD = re.compile(r"名称|内容|品目|标的")
|
||||
QTY_HEAD = re.compile(r"^数量")
|
||||
LOT_HEAD = re.compile(r"采购包|标段|包号")
|
||||
LOT = re.compile(r"(?:采购包|标段|包)\s*([0-9一二三四五六七八九十]+)")
|
||||
NUMBER = re.compile(r"\d{1,3}(?:,\d{3})+(?:\.\d+)?|\d+(?:\.\d+)?")
|
||||
CN_DIGIT = {c: str(i) for i, c in enumerate("零一二三四五六七八九")}
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class Cap:
|
||||
"""一条分项上限:名称、上限(元)、按单价还是按分项总价、数量、所属采购包、出处句子。"""
|
||||
|
||||
name: str
|
||||
cap: Decimal
|
||||
per: str # unit:单价上限;item:分项总价上限
|
||||
qty: Decimal | None
|
||||
lot_no: str | None
|
||||
sentences: list[Sentence]
|
||||
|
||||
@property
|
||||
def title(self) -> str:
|
||||
what = "单价" if self.per == "unit" else "分项总价"
|
||||
return f"{self.name[:40]}:{what}不超过 {self.cap:,.2f} 元"
|
||||
|
||||
def spec(self) -> dict[str, object]:
|
||||
return {
|
||||
"price": {
|
||||
"kind": "unit_cap" if self.per == "unit" else "item_cap",
|
||||
"name": self.name,
|
||||
"cap": f"{self.cap:.2f}",
|
||||
**({"qty": str(self.qty)} if self.qty is not None else {}),
|
||||
**({"lot_no": self.lot_no} if self.lot_no else {}),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
def _number(text: str) -> Decimal | None:
|
||||
m = NUMBER.search(norm(text))
|
||||
if not m:
|
||||
return None
|
||||
try:
|
||||
return Decimal(m.group(0).replace(",", ""))
|
||||
except InvalidOperation:
|
||||
return None
|
||||
|
||||
|
||||
def _lot(text: str) -> str | None:
|
||||
m = LOT.search(text)
|
||||
if not m:
|
||||
return None
|
||||
v = m.group(1)
|
||||
return "".join(CN_DIGIT.get(c, c) for c in v) if not v.isdigit() else v
|
||||
|
||||
|
||||
def caps(sentences: list[Sentence]) -> list[Cap]:
|
||||
out: list[Cap] = []
|
||||
lot: str | None = None
|
||||
cur: tuple[dict[str, tuple[float, float]], int, str] | None = None # 栏、单位倍数、单价 / 分项
|
||||
for _, page in sorted(layout.pages(sentences).items()):
|
||||
for row in layout.rows(page):
|
||||
text = norm("".join(c.text for c in row))
|
||||
if not any(c.cell for c in row):
|
||||
lot = _lot(text) or lot # 表前「采购包2:……」
|
||||
if cur is not None and len(text) > 30:
|
||||
cur = None # 表格结束(正文)
|
||||
continue
|
||||
head = _head(row)
|
||||
if head is not None:
|
||||
cur = head
|
||||
continue
|
||||
if cur is None:
|
||||
continue
|
||||
cols, scale, per = cur
|
||||
# 名称栏里取不是纯数字的那一格(序号表头没认成单元格时,序号会被归到名称栏)
|
||||
name = next(
|
||||
(c.text for c in row if layout.column(c, cols) == "name" and not NUMBER.fullmatch(norm(c.text))), ""
|
||||
)
|
||||
cap = next((v for c in row if layout.column(c, cols) == "cap" and (v := _number(c.text))), None)
|
||||
qty = next((v for c in row if layout.column(c, cols) == "qty" and (v := _number(c.text))), None)
|
||||
if not name or cap is None:
|
||||
if any(layout.column(c, cols) == "cap" for c in row):
|
||||
cur = None
|
||||
continue # 表头折到下一页的残行、合计行
|
||||
if LOT.fullmatch(norm(name)) or re.search(r"合计|总计|小计", name):
|
||||
continue
|
||||
first = next(c for c in row if c.text == name) # 名称格做首句:清单按首句对齐,序号格容易撞
|
||||
out.append(Cap(norm(name), cap * scale, per, qty, lot, [first, *(c for c in row if c is not first)]))
|
||||
return out
|
||||
|
||||
|
||||
def _head(row: list[Sentence]) -> tuple[dict[str, tuple[float, float]], int, str] | None:
|
||||
cols: dict[str, tuple[float, float]] = {}
|
||||
scale, per = 1, "item"
|
||||
for n, c in enumerate(row):
|
||||
t = norm(c.text)
|
||||
if CAP_HEAD.search(t) and "cap" not in cols:
|
||||
cols["cap"] = layout.span(c)
|
||||
scale = 10_000 if "万元" in t else 1
|
||||
per = "unit" if "单价" in t else "item"
|
||||
elif NAME_HEAD.search(t) and "name" not in cols and not LOT_HEAD.search(t):
|
||||
cols["name"] = layout.span(c)
|
||||
elif QTY_HEAD.search(t) and "qty" not in cols:
|
||||
cols["qty"] = layout.span(c)
|
||||
else:
|
||||
cols[f"c{n}"] = layout.span(c)
|
||||
if not {"cap", "name"} <= cols.keys() or any(LOT_HEAD.search(norm(c.text)) for c in row):
|
||||
return None
|
||||
return cols, scale, per
|
||||
80
server/src/bid_doctor/document/pipeline/layout.py
Normal file
80
server/src/bid_doctor/document/pipeline/layout.py
Normal file
@@ -0,0 +1,80 @@
|
||||
"""版面工具:按坐标把文字层的句子分成页、行、栏(读标认表格、查标认投标文件的表格都用)。
|
||||
|
||||
有框线的表格按单元格的上沿分行(文字层给句子带了单元格框):同一行的格子上沿相同,垂直居中的短格、
|
||||
折成几句的长格、纵向合并的格(只算进它的第一行)都不会串行;其余按句子上下范围重叠分行。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
from bid_doctor.document.pipeline.text_layer import Sentence
|
||||
|
||||
|
||||
def box(s: Sentence) -> tuple[float, float, float, float]:
|
||||
return (
|
||||
min(q[0] for q in s.quads),
|
||||
min(q[1] for q in s.quads),
|
||||
max(q[2] for q in s.quads),
|
||||
max(q[3] for q in s.quads),
|
||||
)
|
||||
|
||||
|
||||
def span(s: Sentence) -> tuple[float, float]:
|
||||
"""左右范围:在表格里取单元格的,否则取文字的。"""
|
||||
return (s.cell[0], s.cell[2]) if s.cell else (box(s)[0], box(s)[2])
|
||||
|
||||
|
||||
def pages(sentences: list[Sentence]) -> dict[int, list[Sentence]]:
|
||||
out: dict[int, list[Sentence]] = {}
|
||||
for s in sentences:
|
||||
out.setdefault(s.page_no, []).append(s)
|
||||
return out
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class _Row:
|
||||
top: float
|
||||
bottom: float
|
||||
in_cell: bool
|
||||
items: list[Sentence]
|
||||
|
||||
|
||||
def rows(page: list[Sentence]) -> list[list[Sentence]]:
|
||||
"""一页的句子分成行,行内从左到右。"""
|
||||
out: list[_Row] = []
|
||||
|
||||
def top_of(s: Sentence) -> float:
|
||||
return s.cell[1] if s.cell else box(s)[1]
|
||||
|
||||
for s in sorted((x for x in page if x.quads), key=top_of):
|
||||
if s.cell:
|
||||
r = next((r for r in out if r.in_cell and abs(r.top - s.cell[1]) < 2), None)
|
||||
if r:
|
||||
r.items.append(s)
|
||||
else:
|
||||
out.append(_Row(s.cell[1], s.cell[1], True, [s]))
|
||||
continue
|
||||
_, top, _, bottom = box(s)
|
||||
last = out[-1] if out else None
|
||||
if last and not last.in_cell and top < last.bottom - 1:
|
||||
last.bottom = max(last.bottom, bottom)
|
||||
last.items.append(s)
|
||||
else:
|
||||
out.append(_Row(top, bottom, False, [s]))
|
||||
return [sorted(r.items, key=lambda x: box(x)[0]) for r in sorted(out, key=lambda r: r.top)]
|
||||
|
||||
|
||||
def column(cell: Sentence, cols: dict[str, tuple[float, float]]) -> str | None:
|
||||
"""单元格落在哪一栏:和表头单元格左右范围重叠最多的那一栏;都不重叠(窄数字居中、表头字靠左)取中心最近的。
|
||||
cols 要包含表头的每一格,否则别的栏的数字会被归到最近的已知栏。"""
|
||||
x0, x1 = span(cell)
|
||||
best, score = None, 0.0
|
||||
for name, (c0, c1) in cols.items():
|
||||
overlap = min(x1, c1) - max(x0, c0)
|
||||
if overlap > score:
|
||||
best, score = name, overlap
|
||||
if best is None and cols:
|
||||
mid = (x0 + x1) / 2
|
||||
best = min(cols, key=lambda n: abs((cols[n][0] + cols[n][1]) / 2 - mid))
|
||||
return best
|
||||
@@ -297,7 +297,7 @@ async def test_rule_findings_are_written_with_citations_and_carried(user: User)
|
||||
# 「让利后总价」和投标总价不同:多半另报了一个价,但不是「备选方案」那种明确的两个报价,按重要
|
||||
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-r1"
|
||||
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"]])
|
||||
@@ -305,3 +305,37 @@ async def test_rule_findings_are_written_with_citations_and_carried(user: User)
|
||||
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]
|
||||
|
||||
@@ -166,3 +166,28 @@ def test_scanned_fallback_turns_missing_evidence_into_a_hint() -> None:
|
||||
and "2 页扫描件" in spec.explanation
|
||||
)
|
||||
assert results[b].verdict == "not_found" # 技术参数不是证明材料,照常报未找到
|
||||
|
||||
|
||||
def test_manual_topics_merge_by_topic_and_skip_irrelevant() -> None:
|
||||
"""需人工核对(FR-M3-04):同一主题合并成一条;技术参数、免收保证金、围串标情形、点名要交的文件不算。"""
|
||||
from bid_doctor.check import manual
|
||||
|
||||
items = [
|
||||
{
|
||||
"title": "投标文件递交截止 2027年3月5日 9:30",
|
||||
"categories": ["schedule"],
|
||||
"spec": {"schedule": {"at": "2027-03-05T09:30"}},
|
||||
},
|
||||
{"title": "电子投标文件须在截止时间前上传", "categories": ["schedule"]},
|
||||
{"title": "未在规定时间内解密投标文件的投标无效", "categories": ["invalidity"]},
|
||||
{"title": "须提供加盖电子签章的《开标一览表》", "categories": ["format_doc"]},
|
||||
{"title": "须用CA数字证书加密投标文件", "categories": ["schedule"]},
|
||||
{"title": "本项目免收投标保证金", "categories": ["schedule"]},
|
||||
{"title": "数据库支持国密加密存储", "categories": ["tech_param"]},
|
||||
{"title": "不同投标人的投标文件装订明显相似的投标无效", "categories": ["invalidity"]},
|
||||
{"title": "纸质投标文件正本一份、副本四份,密封递交", "categories": ["schedule"]},
|
||||
]
|
||||
got = {t.key: [i["title"] for i in t.items] for t in manual.topics(items)}
|
||||
assert list(got) == ["DECRYPT", "SIGN_ENCRYPT", "PAPER", "SUBMIT"]
|
||||
assert got["SIGN_ENCRYPT"] == ["须用CA数字证书加密投标文件"]
|
||||
assert got["SUBMIT"][0].startswith("投标文件递交截止") # 写明截止时间的做代表
|
||||
|
||||
93
server/tests/test_price_caps.py
Normal file
93
server/tests/test_price_caps.py
Normal file
@@ -0,0 +1,93 @@
|
||||
"""「报价」类清单项(C7):从招标文件的清单表读分项 / 单价上限,查标时和分项报价表比对。内容虚构。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from decimal import Decimal
|
||||
from uuid import uuid4
|
||||
|
||||
from bid_doctor.check import applicability, match, rules
|
||||
from bid_doctor.checklist import price
|
||||
from bid_doctor.document.pipeline.text_layer import Sentence
|
||||
|
||||
COLS = [(80, 110), (110, 300), (300, 340), (340, 380), (380, 470)]
|
||||
|
||||
|
||||
def row(seq: int, y: float, texts: list[str], page: int = 1) -> list[Sentence]:
|
||||
out = []
|
||||
for n, (t, (x0, x1)) in enumerate(zip(texts, COLS, strict=False)):
|
||||
s = Sentence(
|
||||
key=f"t:p{page}:s{seq + n}", page_no=page, seq=seq + n, text=t, quads=[[x0 + 2, y + 4, x1 - 2, y + 16]]
|
||||
)
|
||||
s.cell = [x0, y, x1, y + 20]
|
||||
out.append(s)
|
||||
return out
|
||||
|
||||
|
||||
def line(seq: int, y: float, text: str, page: int = 1) -> Sentence:
|
||||
return Sentence(key=f"t:p{page}:s{seq}", page_no=page, seq=seq, text=text, quads=[[74, y, 500, y + 12]])
|
||||
|
||||
|
||||
def tender() -> list[Sentence]:
|
||||
return [
|
||||
line(1, 60, "采购包2:数据中心扩容"),
|
||||
*row(10, 80, ["序号", "分项名称", "数量", "单位", "单价上限(万元)"]),
|
||||
*row(20, 100, ["1", "服务器", "4", "台", "12.5"]),
|
||||
*row(30, 120, ["2", "服务器机柜", "2", "个", "0.8"]),
|
||||
line(40, 160, "采购包一览:各包的最高限价见下表,投标报价超过所投采购包最高限价的投标无效。"),
|
||||
*row(50, 180, ["序号", "采购包名称", "数量", "单位", "最高限价(元)"]), # 采购包一级:不是分项
|
||||
*row(60, 200, ["1", "采购包2", "1", "项", "600000"]),
|
||||
]
|
||||
|
||||
|
||||
def test_caps_from_tender_table() -> None:
|
||||
caps = price.caps(tender())
|
||||
assert [(c.lot_no, c.name, c.cap, c.per, c.qty) for c in caps] == [
|
||||
("2", "服务器", Decimal("125000.0"), "unit", Decimal(4)),
|
||||
("2", "服务器机柜", Decimal("8000.0"), "unit", Decimal(2)),
|
||||
]
|
||||
assert caps[0].title == "服务器:单价不超过 125,000.00 元" and caps[0].sentences[0].text == "服务器"
|
||||
assert caps[0].spec()["price"] == {
|
||||
"kind": "unit_cap",
|
||||
"name": "服务器",
|
||||
"cap": "125000.00",
|
||||
"qty": "4",
|
||||
"lot_no": "2",
|
||||
}
|
||||
|
||||
|
||||
def test_price_items_are_checked_by_rule_not_model() -> None:
|
||||
d = applicability.decide({"categories": ["price"], "title": "服务器:单价不超过 125,000.00 元"}, {})
|
||||
assert (d.match, d.skip) == (False, "price")
|
||||
|
||||
|
||||
def bid_row(seq: int, y: float, texts: list[str]) -> list[Sentence]:
|
||||
out = []
|
||||
for n, (t, (x0, x1)) in enumerate(
|
||||
zip(texts, [(80, 100), (100, 260), (260, 300), (300, 380), (380, 470)], strict=True)
|
||||
):
|
||||
s = Sentence(key=f"b:p1:s{seq + n}", page_no=1, seq=seq + n, text=t, quads=[[x0 + 2, y + 4, x1 - 2, y + 16]])
|
||||
s.cell = [x0, y, x1, y + 20]
|
||||
out.append(s)
|
||||
return out
|
||||
|
||||
|
||||
def test_item_over_cap_is_fatal_and_similar_names_do_not_match() -> None:
|
||||
bid = [
|
||||
*bid_row(1, 80, ["序号", "货物名称", "数量", "单价(元)", "合价(元)"]),
|
||||
*bid_row(10, 100, ["1", "服务器", "4", "126,000.00", "504,000.00"]),
|
||||
*bid_row(20, 120, ["2", "服务器机柜", "2", "7,900.00", "15,800.00"]),
|
||||
]
|
||||
item = uuid4()
|
||||
facts = rules.Facts(
|
||||
caps=[
|
||||
rules.ItemCap("服务器", Decimal(125_000), "unit", item),
|
||||
rules.ItemCap("服务器机柜", Decimal(8_000), "unit"),
|
||||
]
|
||||
)
|
||||
hits, cov = rules.run(facts, [match.BidDoc(uuid4(), 1, "报价分册", bid)], [])
|
||||
(h,) = [h for h in hits if h.code == "PRICE.OVER_CEILING_ITEM"]
|
||||
assert h.severity == "fatal" and h.item == item and h.basis == f"cap:{item}" and "服务器" in h.title
|
||||
assert "126,000.00 元" in h.explanation and "125,000.00 元" in h.explanation
|
||||
assert not any("其余没有比对" in x for x in cov["not_checked"]) # 两条上限都对上了名称
|
||||
|
||||
assert rules._match_cap("服务器", [rules.ItemCap("服务器机柜", Decimal(1), "unit")]) is None
|
||||
Reference in New Issue
Block a user