"""查标的纯函数:检索、投标侧证据校验、问题级别(m3-plan §2)。""" from __future__ import annotations from uuid import uuid4 import pytest from bid_doctor.check import findings, match from bid_doctor.check.retrieval import Index, tokens from bid_doctor.document.pipeline.text_layer import Sentence def sent(seq: int, text: str, page: int = 1) -> Sentence: return Sentence(key=f"doc:p{page}:s{seq}", page_no=page, seq=seq, text=text, quads=[[0, 0, 1, 1]]) def test_tokens_keep_number_unit_and_model_names() -> None: t = tokens("服务器内存不低于32GB,支持10GE和硬件BFD") assert "32gb" in t and "10ge" in t and "bfd" in t and "内存" in t def test_bm25_ranks_the_responding_sentence_first() -> None: index = Index( [ sent(1, "我公司提供7×24小时售后服务。"), sent(2, "服务器内存配置为64GB,并提供产品彩页。"), sent(3, "交换机支持硬件BFD,最小发送间隔3.3毫秒。"), ] ) (top, *_) = index.search("交换机支持硬件BFD 最小报文发送间隔不大于50毫秒", 3) assert top.sentence.seq == 3 assert index.search("完全无关的词语组合", 3) == [] def bid() -> tuple[match.BidDoc, dict[str, tuple[match.BidDoc, Sentence]]]: s = sent(9, "服务器内存不低于32GB的要求:我方服务器内存配置为64GB。") d = match.BidDoc(uuid4(), 1, "技术分册", [s]) return d, {"d1p1s9": (d, s)} def test_responded_needs_verified_evidence() -> None: """模型说已响应,但证据没通过:不能判已响应,记「未找到响应」并带原因码(C3)。""" _, allowed = bid() ok = match.judge( {"verdict": "responded", "evidence": [{"sid": "d1p1s9", "quote": "我方服务器内存配置为64GB。"}]}, allowed, [] ) assert ok.verdict == "responded" and ok.evidence_status == "verified" and len(ok.evidence) == 1 no_ev = match.judge({"verdict": "responded", "evidence": []}, allowed, []) assert (no_ev.verdict, no_ev.evidence_status, no_ev.reason_code) == ("not_found", "unverified", "evidence_missing") rewritten = match.judge( {"verdict": "responded", "evidence": [{"sid": "d1p1s9", "quote": "内存为64GB,满足要求。"}]}, allowed, [] ) assert rewritten.reason_code == "quote_not_original" and rewritten.detail["claimed_pages"] == [1] other = match.judge({"verdict": "responded", "evidence": [{"sid": "d1p2s1", "quote": "x" * 10}]}, allowed, []) assert other.reason_code == "sid_unknown" missing = match.judge(None, allowed, ["技术分册"]) assert (missing.verdict, missing.reason_code) == ("undetermined", "missing_answer") def test_severity_comes_from_the_item_not_the_model() -> None: assert findings.severity({"importance": "substantive", "categories": ["tech_param", "invalidity"]}) == "fatal" assert findings.severity({"importance": "normal", "categories": ["qualification"]}) == "fatal" assert findings.severity({"importance": "important", "categories": ["tech_param"]}) == "important" assert findings.severity({"importance": "normal", "categories": ["scoring"]}) == "important" assert findings.severity({"importance": "normal", "categories": ["tech_param"]}) == "hint" def test_no_value_is_a_hint_and_unverified_is_explained() -> None: item = {"title": "服务器内存", "importance": "substantive", "categories": ["tech_param"]} nv = findings.spec(match.Result("no_value", "model", "verified"), item) assert nv is not None and nv.severity == "hint" and nv.title.startswith("未给出具体值") uv = findings.spec( match.Result("not_found", "model", "unverified", "quote_not_original", detail={"claimed_pages": [3]}), item ) assert uv is not None and uv.severity == "fatal" and "第 3 页" in uv.explanation and "没能核对" in uv.explanation assert findings.spec(match.Result("responded", "model", "verified"), item) is None def test_response_mode_by_category() -> None: from bid_doctor.checklist import response_mode sents = [ sent(1, "技术规格偏离表"), sent(2, "投标人应将所有偏离的条目列入上表,未列入上表的视作投标人完全接受。"), sent(3, "商务条款偏离表"), sent(4, "投标人须对商务要求逐条作出响应。"), ] m = response_mode.detect(sents) assert (m["technical"].mode, m["technical"].status) == ("implicit", "found") assert (m["commercial"].mode, m["commercial"].status) == ("explicit", "found") both = response_mode.detect( [sent(1, "须对技术要求逐项作出响应。"), sent(2, "否则将认为投标人接受招标文件的要求。")] ) assert (both["technical"].mode, both["technical"].status) == ("explicit", "conflict") # 写法冲突按逐条(C12) none = response_mode.detect([sent(1, "本项目为货物采购。")]) assert (none["commercial"].mode, none["commercial"].status) == ("explicit", "unknown") def test_applicability_skips_items_that_need_no_response() -> None: from bid_doctor.check import applicability from bid_doctor.checklist import response_mode modes = response_mode.detect([sent(1, "未列入偏离表的技术要求视作投标人完全接受。")]) decide = applicability.decide assert decide({"categories": ["invalidity"], "importance_basis": "tender_text"}, modes).skip == "condition" assert decide({"categories": ["invalidity"], "importance_basis": "marker"}, modes).match # ★ 条款是要求 assert decide({"categories": ["schedule"]}, modes).skip == "schedule" assert decide({"categories": ["scoring"]}, modes).skip == "scoring" assert decide({"categories": ["scoring"], "evidence_required": "业绩合同"}, modes).match assert decide({"categories": ["scoring"], "title": "业绩(2 分):提供近三年类似项目合同复印件"}, modes).match assert ( decide({"categories": ["scoring"], "title": "提供检测报告,中标后签订合同前提供原件"}, modes).skip == "scoring" ) assert decide({"categories": ["tech_param"], "title": "支持硬件BFD,提供官网截图证明"}, modes).evidence assert decide({"categories": ["qualification"], "title": "未被列入信用中国失信名单"}, modes).skip == "no_proof" d = decide({"categories": ["tech_param"], "title": "内存"}, modes) assert d.match and d.mode == "implicit" def test_implicit_not_found_is_accepted_but_substantive_gets_a_hint() -> None: from uuid import uuid4 from bid_doctor.check import applicability a, b = uuid4(), uuid4() results = {a: match.Result("not_found", "model"), b: match.Result("not_found", "model")} items = {a: {"importance": "normal"}, b: {"importance": "substantive", "title": "★双电源"}} decisions = {a: applicability.Decision(True, mode="implicit"), b: applicability.Decision(True, mode="implicit")} match.apply_modes(results, items, decisions) assert results[a].verdict == "implicit_accepted" and findings.spec(results[a], items[a]) is None spec = findings.spec(results[b], items[b]) assert results[b].verdict == "implicit_review" and spec is not None and spec.severity == "hint" def test_presence_names_and_find() -> None: from bid_doctor.check import presence assert presence.names("须提供加盖公章的《法定代表人授权委托书》,否则投标无效") == ["法定代表人授权委托书"] assert presence.names("投标人须提交投标函及开标一览表") == ["投标函"] heading = sent(3, "4.法定代表人授权委托书") body = sent(4, "本授权委托书声明:我单位授权……(这是正文里顺带提到的,很长" + "," * 60 + ")") assert presence.find("法定代表人授权委托书", [body, heading]) is heading assert presence.find("营业执照", [heading]) is None def test_scanned_fallback_turns_missing_evidence_into_a_hint() -> None: from uuid import uuid4 from bid_doctor.check import applicability a, b = uuid4(), uuid4() results = {a: match.Result("not_found", "model"), b: match.Result("not_found", "model")} items = { a: {"categories": ["qualification"], "title": "营业执照"}, b: {"categories": ["tech_param"], "title": "内存"}, } decisions = {a: applicability.Decision(True, evidence=True), b: applicability.Decision(True)} docs = [match.BidDoc(uuid4(), 1, "资格证明册", [], [12, 13, 14, 20])] match.scanned_fallback(results, items, decisions, docs) spec = findings.spec(results[a], items[a]) assert ( results[a].verdict == "scanned" and spec is not None and spec.severity == "hint" and "资格证明册第 12–14、20 页 是扫描件(共 4 页)" in spec.explanation ) assert match.page_ranges([5, 3, 4, 9]) == "3–5、9" 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("投标文件递交截止") # 写明截止时间的做代表 def test_direct_read_turns_negative_deviation_cell_into_finding() -> None: """偏离栏直读:证据句所在的表格行里偏离格写「偏离」,模型判了已响应也改判负偏离(check-eval §8)。""" from bid_doctor.check import deviation def c(seq: int, text: str, x0: float, y: float) -> Sentence: s = Sentence(key=f"b:p1:s{seq}", page_no=1, seq=seq, text=text, quads=[[x0 + 2, y + 4, x0 + 90, y + 16]]) s.cell = [x0, y, x0 + 100, y + 30] return s req, resp, dev = c(1, "支持链路聚合", 100, 50), c(2, "满足,支持链路聚合", 200, 50), c(3, "偏离", 300, 50) req2, resp2, dev2 = c(4, "支持堆叠", 100, 80), c(5, "满足,支持堆叠", 200, 80), c(6, "无偏离", 300, 80) prose = Sentence( key="b:p1:s7", page_no=1, seq=7, text="如不满足上述要求,按负偏离处理。", quads=[[74, 120, 500, 132]] ) d = match.BidDoc(uuid4(), 1, "技术分册", [req, resp, dev, req2, resp2, dev2, prose]) results = { uuid4(): match.Result("responded", "ai", "verified", evidence=[(d, resp)]), uuid4(): match.Result("responded", "ai", "verified", evidence=[(d, resp2)]), uuid4(): match.Result("responded", "ai", "verified", evidence=[(d, prose)]), } assert deviation.direct_read(results, [d]) == 1 first, second, third = results.values() assert first.verdict == "negative_deviation" and [s.text for _, s in first.evidence] == [ "满足,支持链路聚合", "偏离", ] assert first.detail["model_verdict"] == "responded" assert second.verdict == third.verdict == "responded" def test_mismatch_read_only_when_response_cell_is_unrelated() -> None: """响应对象错位(D09):证据只是照抄的要求格、同一行响应格写的是别的内容 → 未找到响应。 指引、相关内容、只写满足的不算。""" from bid_doctor.check import deviation def c(seq: int, text: str, x0: float, y: float) -> Sentence: s = Sentence(key=f"b:p1:s{seq}", page_no=1, seq=seq, text=text, quads=[[x0 + 2, y + 4, x0 + 90, y + 16]]) s.cell = [x0, y, x0 + 100, y + 30] return s rows = [ ("支持云管理、VLAN隔离、标准交换三种工作模式", "显示器升降行程:150mm"), ("支持端口汇聚、端口流量统计", "满足,支持端口汇聚和端口流量统计"), ("批量数据处理在非工作时段内完成", "满足,详见技术方案第12页"), ("系统支持国产操作系统", "满足"), # 几个肯定写法连着写(国动真实投标文件,check-eval §14):仍是肯定回答 ("支持按区域、距离和服务类型筛选并排序", "完全满足,无偏离。"), ("后台支持批量导入并提示错误行", "符合,无负偏离;"), ("支持离线缓存最近浏览的十条记录", "完全满足,无偏离;页面截图详见附件三"), ("短信验证码有效期不超过五分钟", "完全满足,显示器升降行程:150mm"), ] sents, results, items = [], {}, {} for n, (req, resp) in enumerate(rows): a, b, dev = ( c(n * 3 + 1, req, 100, 50 + n * 40), c(n * 3 + 2, resp, 200, 50 + n * 40), c(n * 3 + 3, "无偏离", 300, 50 + n * 40), ) sents += [a, b, dev] k = uuid4() items[k] = {"title": req, "requirement_text": req} results[k] = (a, b) d = match.BidDoc(uuid4(), 1, "技术分册", sents) res = {k: match.Result("responded", "ai", "verified", evidence=[(d, a)]) for k, (a, _) in results.items()} assert deviation.mismatch_read(res, items, [d]) == 2 verdicts = [r.verdict for r in res.values()] assert verdicts == [ "not_found", "responded", "responded", "responded", "responded", "responded", "responded", "not_found", ] first = next(iter(res.values())) assert first.detail["mismatch"] == "显示器升降行程:150mm" and first.reason_code == "response_mismatch" async def test_over_budget_items_are_undetermined_without_model_calls() -> None: """硬预算(C2):超出预算的批次不调模型,条目记「超出预算」,覆盖范围另外写明。""" from bid_doctor.core.db import TenantContext d, _ = bid() items = { uuid4(): { "title": "服务器内存不低于32GB", "requirement_text": "服务器内存不低于32GB", "categories": ["tech_param"], } } async def progress(_: float) -> None: return None ctx = TenantContext(tenant_id=uuid4(), user_id=uuid4()) results, stats = await match.run(ctx, uuid4(), items, [d], progress, budget=10) (r,) = results.values() assert (r.verdict, r.reason_code, stats.calls, stats.over_budget) == ("undetermined", "over_budget", 0, 1) def test_award_time_wording_is_not_award_stage_evidence() -> None: """「以合同签订时间为准」是业绩的时间口径,不是「签订合同时才交」:要附合同的评分项照常查(check-eval §9)。""" from bid_doctor.check import applicability scoring = { "categories": ["scoring"], "title": "项目业绩(10 分)", "requirement_text": "提供2023年以来类似项目业绩,以合同签订时间为准,需提供合同复印件。", } assert applicability.decide(scoring, {}).match later = {**scoring, "requirement_text": "签订合同时提供业绩合同原件备查。"} assert applicability.decide(later, {}).skip == "scoring" def test_response_mode_ignores_declaration_boilerplate_and_negative_consequence() -> None: """响应方式识别(check-eval §11):声明函「逐一、准确填写产品名称……」不算要求逐条响应; 「否则视为无效响应」说的是不响应的后果,不算默认接受。""" from bid_doctor.checklist import response_mode implicit = [ sent(1, "供应商需在《声明函》中逐一、准确填写产品名称、厂名、生产厂址三项信息。"), sent(2, "对招标文件中的所有商务、技术要求,除本表所列明的所有偏离外,均视作供应商已对之理解和响应。"), ] modes = response_mode.detect(implicit) assert (modes["technical"].mode, modes["technical"].status) == ("implicit", "found") explicit = [ sent(1, "供应商需对磋商文件中的技术要求与商务要求逐项作出响应或偏离说明。"), sent(2, "供应商须对技术要求中的实质性要求做出明确响应,否则视为无效响应。"), ] modes = response_mode.detect(explicit) assert (modes["technical"].mode, modes["technical"].status) == ("explicit", "found") def test_direct_read_uses_header_column_and_flags_unclear_values() -> None: """偏离栏按表头认出来:写的不是明确的「无偏离 / 满足」,改判负偏离并标「请核对」。 级别最高到重要(check-eval §11)。""" from bid_doctor.check import deviation def c(seq: int, text: str, x0: float, y: float) -> Sentence: s = Sentence(key=f"b:p1:s{seq}", page_no=1, seq=seq, text=text, quads=[[x0 + 2, y + 4, x0 + 90, y + 16]]) s.cell = [x0, y, x0 + 100, y + 30] return s head = [c(1, "序号", 0, 20), c(2, "招标要求", 100, 20), c(3, "投标响应", 200, 20), c(4, "偏离", 300, 20)] row1 = [c(5, "1", 0, 50), c(6, "支持双电源", 100, 50), c(7, "满足,支持双电源", 200, 50), c(8, "接近要求", 300, 50)] row2 = [ c(9, "2", 0, 80), c(10, "支持热插拔", 100, 80), c(11, "满足,支持热插拔", 200, 80), c(12, "无偏离", 300, 80), ] d = match.BidDoc(uuid4(), 1, "技术分册", [*head, *row1, *row2]) a, b = uuid4(), uuid4() results = { a: match.Result("responded", "ai", "verified", evidence=[(d, row1[2])]), b: match.Result("responded", "ai", "verified", evidence=[(d, row2[2])]), } assert deviation.direct_read(results, [d]) == 1 assert results[a].verdict == "negative_deviation" and results[a].detail["deviation_unclear"] is True assert results[b].verdict == "responded" spec = findings.spec(results[a], {"title": "支持双电源", "importance": "substantive", "categories": ["tech_param"]}) assert spec is not None and spec.severity == "important" and spec.title.startswith("偏离栏需核对") def test_scanned_page_with_caption_counts_as_empty() -> None: """扫描件常带一行文字说明(「合同扫描件第 2 页(共 15 页)」):按图片占比和字数一起判断(check-eval §14)。""" from bid_doctor.document.pipeline.text_layer import Page def page(chars: int, cover: float) -> str: return Page(1, 595, 842, 0, chars, [], cover).text_source assert page(0, 0.0) == "empty" assert page(26, 0.28) == "empty" # 「2.1 营业执照」+ 执照扫描图 assert page(97, 0.53) == "empty" # 一段说明 + 整页合同扫描 assert page(30, 0.0) == "text" # 只有一行字的正文页(没有图) assert page(495, 0.25) == "text" # 带架构图的正文页 assert page(179, 0.45) == "text" # 图多、字也不少 assert ( Page.from_json( {"page_no": 1, "width_pt": 1, "height_pt": 1, "rotation": 0, "char_count": 30, "sentences": []} ).text_source == "text" ) # v4 文字层没有图片占比 def test_applicability_skips_conditions_the_bidder_does_not_claim() -> None: """不需要这家投标人回答的条目(check-eval §14):写明没有的、总括性的、中标后才做的、 投标文件没声称适用的条件性要求(联合体、价格扣除)。声称了(投标文件里有对应文件)就照常查。""" from bid_doctor.check import applicability def it(title: str, cats: list[str], req: str = "") -> dict[str, object]: return {"title": title, "categories": cats, "requirement_text": req or title} items = [ it("本项目第2包无特定资格要求", ["qualification"]), it("提交特定资格要求的证明材料", ["qualification"], "特定资格要求的证明材料(如有,必须提供)"), it("投标人应满足招标文件规定的资格条件", ["qualification"]), it("中标后十日内提交详细实施计划", ["tech_param"]), it("联合体投标的,须提交联合体协议书", ["qualification"]), it("残疾人福利性单位价格扣除", ["scoring"], "提供《残疾人福利性单位声明函》,给予价格扣除。"), it("提供营业执照副本复印件", ["qualification"]), ] ctx = applicability.context(items, ["营业执照副本复印件", "我公司不属于残疾人福利性单位。"]) got = [applicability.decide(i, {}, ctx).skip for i in items] assert got == ["none_required", "none_required", "umbrella", "after_award", "not_claimed", None, None] # 没有「无特定资格要求」时特定资格证明照查;不知道投标文件写了什么(ctx 为空)时条件性要求照查 assert applicability.decide(items[1], {}, applicability.Context(frozenset())).skip is None assert applicability.decide(items[4], {}).skip is None consortium = applicability.context(items, ["联合体协议书(附后)"]) assert applicability.decide(items[4], {}, consortium).skip is None async def test_citation_from_another_item_in_the_same_batch_is_verified(monkeypatch: pytest.MonkeyPatch) -> None: """片段每批只送一次,模型引用了同一批别的条目的候选句:引文核对得上就算(check-eval §14); 没送过的编号照样是 sid_unknown。""" from bid_doctor.core.db import TenantContext net = sent(1, "系统部署于互联网区与电子政务外网区,两区之间只经前置交换。", page=7) mem = sent(2, "服务器内存配置为64GB,满足不低于32GB的要求。", page=9) d = match.BidDoc(uuid4(), 1, "投标文件", [net, mem]) a, b = uuid4(), uuid4() items = { a: {"title": "系统部署在电子政务外网区", "categories": ["tech_param"]}, b: {"title": "服务器内存不低于32GB", "categories": ["tech_param"]}, } # 前提:每条只取 1 个候选句时,b 的候选句里没有 net(net 只是 a 的候选句) assert [h.sentence for h in Index([net, mem]).search(match.query_text(items[b]), 1)] == [mem] async def fake(ctx: object, **kw: object) -> dict[str, object]: ev = {"sid": "d1p7s1", "quote": "系统部署于互联网区与电子政务外网区"} return { "results": [ {"item": "I1", "verdict": "responded", "evidence": [ev]}, {"item": "I2", "verdict": "responded", "evidence": [ev]}, # 引的是 a 的候选句 ] } monkeypatch.setattr(match, "CANDIDATES", 1) monkeypatch.setattr(match.ai, "run_json", fake) async def progress(_: float) -> None: return None ctx = TenantContext(tenant_id=uuid4(), user_id=uuid4()) results, _ = await match.run(ctx, uuid4(), items, [d], progress) assert results[a].verdict == "responded" and results[b].verdict == "responded" assert results[b].evidence[0][1] is net unknown = match.judge({"verdict": "responded", "evidence": [{"sid": "d1p8s1", "quote": "x" * 10}]}, {}, []) assert unknown.reason_code == "sid_unknown" def test_outline_requirement_is_searched_item_by_item() -> None: """「方案应明确 A、B、C……」:列出的每一项再各检索一次(check-eval §14)。""" item = {"requirement_text": "运维方案应包括巡检频次、备份策略、故障上报流程以及应急演练安排,并确保措施可落地。"} assert match.outline_terms(item) == ["巡检频次", "备份策略", "故障上报流程", "应急演练安排"] assert match.outline_terms({"requirement_text": "方案应明确工期。"}) == [] assert match.outline_terms({"requirement_text": "支持巡检、备份和应急演练。"}) == [] def test_reuse_skips_conclusions_whose_citation_failed() -> None: """出处没过校验的结论不沿用(复查时重新问模型);正常的结论按证据位置找回来。""" from bid_doctor.check import reuse d, allowed = bid() (_, s) = allowed["d1p1s9"] ok = match.Result("responded", "model", "verified", evidence=[(d, s)]) assert reuse.restore(reuse.snapshot(ok), [d], "r1") is not None failed = match.judge({"verdict": "responded", "evidence": [{"sid": "d1p9s9", "quote": "x" * 10}]}, allowed, []) assert reuse.restore(reuse.snapshot(failed), [d], "r1") is None