a07663cbdb
- build.py：从「招标文件档案」和读标清单生成规矩投标文件稿；技术参数按 op / value / unit 算满足的值， 分项报价按分项上限比例分摊，落款日期取截止日前 3 天，偏离表按格式表头决定列（含设备名称列）， 读标漏抽的技术参数也补上响应；routes 支持「+」组合和 prose 散文响应。 - mutate.py：按词库注入要求项问题（单点 / 连续块，带不带标志词）和规则问题，写响应清单； 符合性条件响应表的问题按表级记一个。 - score.py：按单元去重、类型族比对、读标漏抽单列、端到端召回、无法对照单列。 - run_check.py：加只检索不调模型的基线；render.py：扫描件按最后一次出现定位。 - vocab-dev.json：开发集注入词库（留出集的另行封存，不进 git）。 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
446 lines
20 KiB
Python
446 lines
20 KiB
Python
"""从「招标文件档案」和读标清单生成规矩投标文件的结构化稿 spec.json（evals/check/GUIDELINES.md §1、§5）。
|
||
|
||
cd server && ../.venv/bin/python ../evals/check/build.py <招标文件> <投标文件编号> --style A|B [--checklist fix3]
|
||
|
||
规矩靠构造保证：偏离表、响应表、报价表、证明材料由脚本生成；技术参数有 op / value / unit 的，按要求算一个满足的值。
|
||
LLM 只写散文（档案里的 prose）和 B / C 风格的改写（rewrite.json，可选）。
|
||
|
||
## 招标文件档案 evals/out/check/tenders/<ID>.json（照着招标原文整理，不进 git）
|
||
|
||
{
|
||
"tender": "S02",
|
||
"project": {"name": "…", "code": "…", "purchaser": "…", "agency": "…", "deadline": "2026-12-01",
|
||
"validity_days": 90, "ceiling_total": "1514100.00", "lots": [{"no": "1", "ceiling": "…"}]},
|
||
"modes": [{"category": "tech_param", "scope": "一般参数", "mode": "implicit", "when": null,
|
||
"basis": [{"p": 76, "t": "否则将认为投标人接受"}]}],
|
||
"forms": [ // 按格式章的顺序
|
||
{"key": "bid_letter", "title": "投标函", "kind": "text", "paragraphs": ["致：{agency}", "…{project_name}…"]},
|
||
{"key": "price_summary", "title": "开标一览表", "kind": "price_summary", "header": ["项目名称", "投标总价（元）", "…"]},
|
||
{"key": "price_items", "title": "分项价格表", "kind": "price_items", "header": ["序号", "名称", "…"],
|
||
"items": [{"name": "汇聚交换机", "qty": 3, "unit": "台"}]},
|
||
{"key": "tech_dev", "title": "偏离表", "kind": "response_table", "route": "tech_row",
|
||
"header": ["序号", "招标文件的要求", "投标文件中的内容", "偏离", "说明"], "only_deviations": true},
|
||
{"key": "compliance", "title": "采购需求符合性条件响应表", "kind": "fixed_rows", "header": ["…"],
|
||
"rows": [["1", "“(一)数量要求”的响应情况", "见招标文件第六章《采购需求》中相关要求", "完全响应", ""]]},
|
||
{"key": "auth", "title": "授权委托书", "kind": "text", "paragraphs": ["…"]},
|
||
{"key": "evidence", "title": "资格证明文件", "kind": "attachments"}
|
||
],
|
||
"attachments": [{"key": "a1", "name": "营业执照（副本）", "scan": true, "lines": ["…"]}],
|
||
"routes": {"i12": "tech_row", "i40": "commercial_row", "i55": "attach:a1", "i60": "form:compliance", "i70": "na"},
|
||
"prose": [{"title": "技术方案", "paragraphs": ["…"]}]
|
||
}
|
||
|
||
routes 的取值可以用「+」连起来（tech_row+attach:a3），「prose」表示靠技术方案散文响应（prose:<节标题> 指定哪一节）。
|
||
占位符：{bidder} {credit_code} {legal_rep} {agent} {project_name} {project_code} {purchaser} {agency} {date}
|
||
{validity_days} {total} {total_cn}。`routes` 的键是读标清单里正式条目的序号（items.json 的下标，前缀 i）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import datetime
|
||
import json
|
||
import pathlib
|
||
import re
|
||
import sys
|
||
from decimal import ROUND_HALF_UP, Decimal
|
||
from typing import Any
|
||
|
||
EVALS = pathlib.Path(__file__).resolve().parents[1]
|
||
sys.path.insert(0, str(EVALS))
|
||
import gold # noqa: E402
|
||
|
||
OUT = EVALS / "out" / "check"
|
||
BIDDER = {
|
||
"bidder": "江苏启衡信息技术有限公司（虚构）",
|
||
"credit_code": "91320100MA1XXXXX0Q",
|
||
"legal_rep": "周明远",
|
||
"agent": "陈立新",
|
||
"date": "2026年11月20日",
|
||
}
|
||
DIGITS = "零壹贰叁肆伍陆柒捌玖"
|
||
UNITS = ["", "拾", "佰", "仟"]
|
||
SECTIONS = ["", "万", "亿"]
|
||
|
||
|
||
def rmb_upper(amount: Decimal) -> str:
|
||
"""金额大写（元、角、分），用于开标一览表。"""
|
||
q = amount.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
|
||
yuan, cents = divmod(int(q * 100), 100)
|
||
if yuan == 0:
|
||
head = "零元"
|
||
else:
|
||
groups, out = [], ""
|
||
while yuan:
|
||
groups.append(yuan % 10000)
|
||
yuan //= 10000
|
||
for gi in range(len(groups) - 1, -1, -1):
|
||
g = groups[gi]
|
||
if g == 0:
|
||
if out and not out.endswith("零"):
|
||
out += "零"
|
||
continue
|
||
part, zero = "", False
|
||
for di in range(3, -1, -1):
|
||
d = (g // 10**di) % 10
|
||
if d == 0:
|
||
zero = bool(part)
|
||
continue
|
||
if zero:
|
||
part += "零"
|
||
zero = False
|
||
part += DIGITS[d] + UNITS[di]
|
||
if out and g < 1000 and not out.endswith("零"):
|
||
out += "零"
|
||
out += part + SECTIONS[gi]
|
||
head = out.rstrip("零") + "元"
|
||
jiao, fen = divmod(cents, 10)
|
||
if not cents:
|
||
return head + "整"
|
||
return head + (DIGITS[jiao] + "角" if jiao else "零") + (DIGITS[fen] + "分" if fen else "")
|
||
|
||
|
||
def fill(text: str, ctx: dict[str, str]) -> str:
|
||
return re.sub(r"\{(\w+)\}", lambda m: ctx.get(m.group(1), m.group(0)), text)
|
||
|
||
|
||
def comply(spec: dict[str, Any], positive: bool) -> tuple[str, str]:
|
||
"""按要求算一个满足的响应值。返回 (响应文字, 偏离)。positive 时给正偏离（T05）。"""
|
||
param, op, value, unit = (spec.get(k) or "" for k in ("param", "op", "value", "unit"))
|
||
try:
|
||
v = Decimal(str(value))
|
||
except Exception: # noqa: BLE001
|
||
v = None
|
||
if v is not None and op in (">=", "<="):
|
||
better = (v * 2 if op == ">=" else v / 2).normalize() if positive else v.normalize()
|
||
return f"{param}：{better:f}{unit}", "正偏离" if positive else "无偏离"
|
||
if v is not None and op == "=":
|
||
return f"{param}：{v.normalize():f}{unit}", "无偏离"
|
||
return f"支持{param}" if param else "满足", "无偏离"
|
||
|
||
|
||
def load_items(tender: str, run: str) -> dict[str, dict[str, Any]]:
|
||
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 = {}
|
||
for i, r in enumerate(rows):
|
||
if r.get("zone") != "formal":
|
||
continue
|
||
sents = [by_sid[k] for k in r.get("sentences") or [] if k in by_sid]
|
||
out[f"i{i}"] = {
|
||
**r,
|
||
"text": "".join(s.text for s in sents),
|
||
"req": {"p": sents[0].page_no, "t": gold.norm(sents[0].text)[:24]} if sents else None,
|
||
}
|
||
return out
|
||
|
||
|
||
def unit_of(tender: str) -> dict[str, str]:
|
||
"""清单条目 → 标注单元：读标清单条目的句子和标注单元的句子有重叠就对上（和读标评分同一口径）。"""
|
||
units, _ = gold.units(tender)
|
||
by_sent: dict[str, list[str]] = {}
|
||
for u in units:
|
||
for s in u.sids:
|
||
by_sent.setdefault(s, []).append(u.id)
|
||
return by_sent # 句子编号 → 单元编号
|
||
|
||
|
||
COLUMN_HINTS = [ # 表头文字 → 字段（按顺序匹配，先匹配到的为准）
|
||
("no", ("序号", "序")),
|
||
("deviation", ("偏离",)),
|
||
("note", ("说明", "备注")),
|
||
("answer", ("响应", "投标文件中的内容", "应答", "投标人", "实际")),
|
||
("requirement", ("要求", "条款", "参数", "指标")),
|
||
("device", ("名称", "设备", "货物", "服务")),
|
||
]
|
||
|
||
|
||
def columns_of(header: list[str]) -> list[str]:
|
||
"""按格式的表头决定每一列放什么（S04 的偏离表第 2 列是「项目名称」）。认不出的列留空。"""
|
||
out: list[str] = []
|
||
for h in header:
|
||
field = next((f for f, keys in COLUMN_HINTS if any(k in h for k in keys) and f not in out), "")
|
||
out.append(field)
|
||
if "answer" not in out and "" in out:
|
||
out[out.index("")] = "answer"
|
||
return out
|
||
|
||
|
||
def _row(k: int, r: dict[str, Any], columns: list[str]) -> dict[str, Any]:
|
||
value = {
|
||
"no": str(k),
|
||
"device": (r.get("tp") or {}).get("device") or "",
|
||
"requirement": r["requirement"],
|
||
"answer": r["answer"],
|
||
"deviation": r["deviation"],
|
||
"note": "",
|
||
}
|
||
return {
|
||
"cells": [value.get(c, "") for c in columns],
|
||
"anchor": columns.index("answer") if "answer" in columns else len(columns) - 1,
|
||
"responds": r["responds"],
|
||
"req": r["req"],
|
||
"iid": r["iid"],
|
||
"tp": r["tp"],
|
||
}
|
||
|
||
|
||
def build(tender: str, bid: str, style: str, run: str) -> None:
|
||
profile = json.loads((OUT / "tenders" / f"{tender}.json").read_text())
|
||
items = load_items(tender, run)
|
||
sent_units = unit_of(tender)
|
||
rewrite_path = OUT / bid / "rewrite.json"
|
||
rewrite = json.loads(rewrite_path.read_text()) if rewrite_path.exists() else {}
|
||
proj = profile["project"]
|
||
routes: dict[str, str] = profile.get("routes", {})
|
||
|
||
rows_by_route: dict[str, list[dict[str, Any]]] = {}
|
||
attach_responds: dict[str, list[str]] = {}
|
||
form_responds: dict[str, list[str]] = {}
|
||
for iid, it in items.items():
|
||
units = sorted({u for s in it.get("sentences") or [] for u in sent_units.get(s, [])})
|
||
# 一条要求可以走几个去处，用「+」连起来（如 tech_row+attach:a3：表里一行，另附证明材料）
|
||
parts = routes.get(iid, "na").split("+")
|
||
for part in parts:
|
||
if part.startswith("attach:"):
|
||
attach_responds.setdefault(part[7:], []).extend(units)
|
||
elif part.startswith("form:"):
|
||
form_responds.setdefault(part[5:], []).extend(units)
|
||
elif part.startswith("prose"):
|
||
form_responds.setdefault("prose:" + part[6:], []).extend(units)
|
||
route = next((x for x in parts if x in ("tech_row", "commercial_row")), None)
|
||
if route is None:
|
||
continue
|
||
raw = (it.get("raw") or {}).get("tech_param") or {}
|
||
positive = style == "A" and len(rows_by_route.get(route, [])) % 9 == 4 # 每 9 行一个正偏离（T05）
|
||
req_text = it["text"] or it["title"]
|
||
numeric = raw.get("op") in (">=", "<=", "=") and raw.get("value")
|
||
if numeric:
|
||
answer, dev = comply(raw, positive)
|
||
else: # 功能类要求：复述型（风格 A）照常写「满足，……」
|
||
answer, dev = "满足，" + re.sub(r"^[\s★▲*＊\d.、（）()]+", "", req_text)[:120], "无偏离"
|
||
rw = rewrite.get(iid) or {}
|
||
rows_by_route.setdefault(route, []).append(
|
||
{
|
||
"iid": iid,
|
||
"requirement": rw.get("requirement", req_text),
|
||
"answer": rw.get("answer", answer),
|
||
"deviation": dev,
|
||
"responds": units,
|
||
"req": it["req"],
|
||
"tp": raw,
|
||
}
|
||
)
|
||
covered = {u for v in rows_by_route.values() for r in v for u in r["responds"]}
|
||
covered |= {u for v in attach_responds.values() for u in v} | {u for v in form_responds.values() for u in v}
|
||
na_units = sorted(
|
||
{
|
||
u
|
||
for iid, it in items.items()
|
||
if routes.get(iid, "na") == "na"
|
||
for s in it.get("sentences") or []
|
||
for u in sent_units.get(s, [])
|
||
}
|
||
- covered
|
||
)
|
||
# 读标清单漏抽、标注集里有的技术参数：也按规矩响应，免得规矩的投标文件里「无意漏响应」
|
||
gold_units, _ = gold.units(tender)
|
||
by_sid = {gold.sid(x): x for x in gold.sentences(tender)}
|
||
for u in gold_units:
|
||
if u.cat != "tech_param" or u.id in covered or u.id in na_units or not u.sids:
|
||
continue
|
||
sents = sorted((by_sid[k] for k in u.sids if k in by_sid), key=lambda x: (x.page_no, x.seq))
|
||
if not sents:
|
||
continue
|
||
text = "".join(x.text for x in sents)
|
||
rows_by_route.setdefault("tech_row", []).append(
|
||
{
|
||
"iid": None,
|
||
"requirement": text,
|
||
"answer": "满足：" + text,
|
||
"deviation": "无偏离",
|
||
"responds": [u.id],
|
||
"req": {"p": sents[0].page_no, "t": gold.norm(sents[0].text)[:24]},
|
||
"tp": {},
|
||
}
|
||
)
|
||
|
||
if proj.get("deadline"): # 落款日期取投标截止日前 3 天（写死的日期可能晚于截止日，规矩的投标文件不能带 P08）
|
||
dl = datetime.date.fromisoformat(str(proj["deadline"])[:10]) - datetime.timedelta(days=3)
|
||
bidder_date = f"{dl.year}年{dl.month}月{dl.day}日"
|
||
else:
|
||
bidder_date = BIDDER["date"]
|
||
total = Decimal(str(proj.get("ceiling_total") or "1000000")) * Decimal("0.93")
|
||
total = total.quantize(Decimal("1"))
|
||
ctx = {
|
||
**BIDDER,
|
||
"date": bidder_date,
|
||
"project_name": proj.get("name", ""),
|
||
"project_code": proj.get("code", ""),
|
||
"purchaser": proj.get("purchaser", ""),
|
||
"agency": proj.get("agency", ""),
|
||
"validity_days": str(proj.get("validity_days") or 90),
|
||
"total": f"{total:,.2f}",
|
||
"total_cn": rmb_upper(total),
|
||
}
|
||
|
||
blocks: list[dict[str, Any]] = [{"type": "title", "text": f"{proj.get('name', '')}投标文件"}]
|
||
price_items = next((f.get("items") for f in profile["forms"] if f["kind"] == "price_items"), []) or []
|
||
weights = [Decimal(str(x.get("qty") or 1)) for x in price_items] or [Decimal(1)]
|
||
# 各分项的合价：有分项 / 单价上限的按上限的比例分摊（总价是限价的 93%，所以每项都在上限以内）；没有的按数量分摊
|
||
ceilings = [Decimal(str(x["ceiling"])) if x.get("ceiling") else None for x in price_items]
|
||
if price_items and all(c is not None for c in ceilings):
|
||
base = [
|
||
c * w if x.get("ceiling_is_unit") else c for c, w, x in zip(ceilings, weights, price_items, strict=False)
|
||
] # type: ignore[operator]
|
||
else:
|
||
base = list(weights)
|
||
lines = [(total * b / sum(base)).quantize(Decimal("0.01")) for b in base]
|
||
lines[-1] = total - sum(lines[:-1])
|
||
unit_prices = [(ln / w).quantize(Decimal("0.01")) for ln, w in zip(lines, weights, strict=False)]
|
||
for n, f in enumerate(profile["forms"], start=1):
|
||
head = {"type": "heading", "text": f"{n}．{f['title']}", "new_page": n > 1, "form": f["key"]}
|
||
kind = f["kind"]
|
||
if kind == "text":
|
||
blocks += [
|
||
head,
|
||
*(
|
||
{
|
||
"type": "para",
|
||
"text": fill(p, ctx),
|
||
"responds": form_responds.get(f["key"], []),
|
||
"form": f["key"],
|
||
}
|
||
for p in f["paragraphs"]
|
||
),
|
||
]
|
||
blocks.append({"type": "sign", "lines": [f"投标人：{ctx['bidder']}（电子签章）", f"日期：{ctx['date']}"]})
|
||
elif kind == "price_summary":
|
||
header = f.get("header") or ["项目名称", "投标总价（小写，元）", "投标总价（大写）"]
|
||
cells = [ctx["project_name"], ctx["total"], ctx["total_cn"]] + [""] * max(0, len(header) - 3)
|
||
blocks += [
|
||
head,
|
||
{
|
||
"type": "table",
|
||
"header": header,
|
||
"rows": [{"cells": cells[: len(header)], "anchor": 1}],
|
||
"price_summary": True,
|
||
},
|
||
]
|
||
elif kind == "price_items":
|
||
rows = []
|
||
for k, (x, up, ln) in enumerate(zip(price_items, unit_prices, lines, strict=False), start=1):
|
||
rows.append(
|
||
{
|
||
"cells": [
|
||
str(k),
|
||
x["name"],
|
||
f"{x.get('qty', 1)}",
|
||
x.get("unit", ""),
|
||
f"{up:,.2f}",
|
||
f"{ln:,.2f}",
|
||
],
|
||
"anchor": 1,
|
||
}
|
||
)
|
||
header = ["序号", "名称", "数量", "单位", "单价（元）", "总价（元）"]
|
||
blocks += [
|
||
head,
|
||
{"type": "table", "header": header, "rows": rows, "price_items": True},
|
||
{"type": "para", "text": f"合计：{ctx['total']}元（{ctx['total_cn']}）"},
|
||
]
|
||
elif kind == "response_table":
|
||
src = rows_by_route.get(f.get("route", "tech_row"), [])
|
||
header = f.get("header") or ["序号", "招标文件的要求", "投标文件中的内容", "偏离", "说明"]
|
||
columns = columns_of(header)
|
||
if f.get("only_deviations"):
|
||
devs = [r for r in src if r["deviation"] != "无偏离"]
|
||
rows = [_row(k, r, columns) for k, r in enumerate(devs, 1)]
|
||
empty = {"no": "1", "requirement": "—", "answer": "无偏离", "deviation": "无"}
|
||
rows = rows or [{"cells": [empty.get(c, "") for c in columns], "anchor": columns.index("answer")}]
|
||
pool = [
|
||
_row(0, r, columns) for r in src if r["deviation"] == "无偏离"
|
||
] # 没列出的（默认接受）：注入时可以挑来写成偏离
|
||
else:
|
||
rows = [_row(k, r, columns) for k, r in enumerate(src, 1)]
|
||
pool = []
|
||
blocks += [
|
||
head,
|
||
{
|
||
"type": "table",
|
||
"header": header,
|
||
"rows": rows,
|
||
"form": f["key"],
|
||
"route": f.get("route", "tech_row"),
|
||
"only_deviations": bool(f.get("only_deviations")),
|
||
"columns": columns,
|
||
"pool": pool,
|
||
},
|
||
]
|
||
elif kind == "fixed_rows":
|
||
rows = [
|
||
{"cells": [fill(c, ctx) for c in r], "responds": form_responds.get(f["key"], []), "anchor": len(r) - 2}
|
||
for r in f["rows"]
|
||
]
|
||
blocks += [head, {"type": "table", "header": f["header"], "rows": rows, "form": f["key"], "fixed": True}]
|
||
elif kind == "attachments":
|
||
blocks.append(head)
|
||
for a in profile.get("attachments", []):
|
||
b = {
|
||
"type": "scan" if a.get("scan") else "para",
|
||
"text": fill(a["name"], ctx),
|
||
"responds": sorted(set(attach_responds.get(a["key"], []))),
|
||
"attach": a["key"],
|
||
}
|
||
if a.get("scan"):
|
||
b["lines"] = [fill(x, ctx) for x in a.get("lines", [])]
|
||
else:
|
||
b["text"] = fill(a["name"] + "：" + "；".join(a.get("lines", [])), ctx)
|
||
blocks.append(b)
|
||
else:
|
||
raise ValueError(f"未知的表单类型：{kind}")
|
||
for sec in profile.get("prose", []):
|
||
blocks.append({"type": "heading", "text": sec["title"], "new_page": True})
|
||
units = sorted(set(form_responds.get("prose:" + sec["title"], []) + form_responds.get("prose:", [])))
|
||
blocks += [
|
||
{"type": "para", "text": fill(p, ctx), "responds": units if i == 0 else [], "form": "prose"}
|
||
for i, p in enumerate(sec["paragraphs"])
|
||
]
|
||
|
||
spec = {
|
||
"bid": bid,
|
||
"tender": tender,
|
||
"style": style,
|
||
"project_name": proj.get("name", ""),
|
||
"bidder": BIDDER,
|
||
"volumes": [{"id": "v1", "kind": "whole", "header": None, "blocks": blocks}],
|
||
"price": {
|
||
"total": f"{total:.2f}",
|
||
"items": [
|
||
{"name": x["name"], "qty": x.get("qty", 1), "unit_price": f"{p:.2f}"}
|
||
for x, p in zip(price_items, unit_prices, strict=False)
|
||
],
|
||
},
|
||
"routes_used": {r: len(v) for r, v in rows_by_route.items()},
|
||
"na_units": na_units,
|
||
"deadline": proj.get("deadline"),
|
||
"ceiling_total": proj.get("ceiling_total"),
|
||
"validity_days": proj.get("validity_days"),
|
||
}
|
||
d = OUT / bid
|
||
d.mkdir(parents=True, exist_ok=True)
|
||
(d / "spec.json").write_text(json.dumps(spec, ensure_ascii=False, indent=1))
|
||
print(
|
||
f"{bid}：{sum(len(v) for v in rows_by_route.values())} 行响应，{len(profile.get('attachments', []))} 份附件，总价 {ctx['total']}"
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("tender")
|
||
ap.add_argument("bid")
|
||
ap.add_argument("--style", default="A")
|
||
ap.add_argument("--checklist", default="fix3")
|
||
a = ap.parse_args()
|
||
build(a.tender, a.bid, a.style, a.checklist)
|