"""资质匹配（FR-M3-05，qualification-plan §3）：证书在投标截止日有没有过期、等级够不够、有没有覆盖，不调模型（QL1）。 - 证书类别词表：IT 集成常见的企业资质，别名含国标等同写法；等级按类别比较（数字越大越高，或「一级」最高）； - 产品证书（节能认证、CCC、商用密码产品）和人员证书不进资质匹配：厂商、人员的证书不是用户的资质（红线 5）； - 证据两个来源：投标文字层（资质情况表那一行、写着有效期的那一句）按清单项定级；只有资质库记录的降一级（QL2）； - 投标文件和资质库冲突时以投标文件为准（QL4）；资质库有记录但缺这一类才报「未覆盖」，空库不报（QL3）； - 一条清单项里每类证书出一个问题（过期和等级不足分开报）；它比扫描件兜底的「请人工核对」更具体，有资质问题时代替那一条； - 说明只写事实，不写整改建议；认不出类别、日期归属不清的不判断，写进覆盖范围。 """ from __future__ import annotations import re from dataclasses import dataclass, field from datetime import date from typing import Any from uuid import UUID from bid_doctor.check.findings import severity as item_severity from bid_doctor.check.match import BidDoc, Result from bid_doctor.check.rules import Hit, dates_in from bid_doctor.document.pipeline import layout from bid_doctor.document.pipeline.text_layer import Sentence, norm from bid_doctor.qualification.service import Record LABEL = "资质核对（证书有效期、等级、有没有覆盖）" @dataclass(frozen=True, slots=True) class Family: key: str label: str pattern: re.Pattern[str] # 在 norm() 后转大写的文本上匹配 levels: str | None = None # asc：数字越大越高；cn：「一级」最高；jia：甲 > 乙 > 丙 prefix: str | None = None # 等级写法的前缀（CS3、CMMI3） subtypes: tuple[str, ...] = () # 服务类别（CCRC 的安全集成、风险评估……） def _f(key: str, label: str, pattern: str, **kw: Any) -> Family: return Family(key, label, re.compile(pattern), **kw) FAMILIES = ( _f("iso9001", "ISO 9001 质量管理体系认证", r"ISO(?:/IEC)?/?9001|GB/?T19001|(? str: return norm(s).upper() def families_in(text: str) -> list[Family]: t = text_of(text) return [f for f in FAMILIES if f.pattern.search(t)] def family_of_record(r: Record) -> Family | None: if r.kind == "software_copyright": return BY_KEY["softcopy"] found = [f for f in families_in(r.name) if f.key != "softcopy"] return found[0] if found else None def level_rank(fam: Family, text: str) -> tuple[int, str] | None: """文字里写的等级 → (名次, 原文)，名次越大越高；认不出返回 None。""" t = text_of(text) if fam.levels == "asc": m = re.search(rf"(?:{fam.prefix})\s*-?\s*([1-5])(?![0-9])", t) or re.search(r"([一二三四五壹贰叁肆伍1-5])级", t) if m: g = m.group(1) return (int(g) if g.isdigit() else CN_NUM[g]), m.group(0) elif fam.levels == "cn": if m := re.search(r"([特一二三四壹贰叁肆1-4])级", t): g = m.group(1) return 10 - (int(g) if g.isdigit() else CN_NUM[g]), m.group(0) elif fam.levels == "jia": if m := re.search(r"([甲乙丙])级", t): return JIA[m.group(1)], m.group(0) return None def segment(fam: Family, text: str) -> str: """要求里属于这一类证书的那一段：从它的名称起，到下一类证书的名称为止（「ITSS 三级得 1 分；CS3 及以上得 2 分」 里，ITSS 的等级不能读成 CS 的）。没有别的证书时就是全文。""" t = text_of(text) m = fam.pattern.search(t) if not m: return t others = [o.start() for f in FAMILIES if f.key != fam.key for o in f.pattern.finditer(t) if o.start() > m.start()] return t[m.start() : min(others, default=len(t))] def subtype_of(fam: Family, text: str) -> str | None: t = text_of(text) return next((x for x in fam.subtypes if x in t), None) @dataclass(slots=True) class BidFact: """投标文字层上读到的一张证书：在哪一句（和同一行的格子），有效期、等级。""" evidence: list[tuple[BidDoc, Sentence]] valid_to: date | None = None long_term: bool = False level: tuple[int, str] | None = None @dataclass(slots=True) class DocIndex: """一份投标文件的版面索引，每次检查建一次：句子 → 所在行、每页的行、规范化后的文字。""" doc: BidDoc rows: dict[int, list[Sentence]] page_rows: dict[int, list[list[Sentence]]] texts: list[tuple[Sentence, str]] @classmethod def build(cls, doc: BidDoc) -> DocIndex: page_rows = {no: layout.rows(page) for no, page in layout.pages(doc.sentences).items()} rows = {id(c): row for page in page_rows.values() for row in page for c in row} return cls(doc, rows, page_rows, [(s, text_of(s.text)) for s in doc.sentences]) def row_level(fam: Family, row: list[Sentence], row_text: str) -> tuple[int, str] | None: """一行里的等级：先看短的格子（等级那一栏），证书编号里常带等级字样（「CS3-3201-…」），整行读会读错。""" short = [lv for x in row if len(norm(x.text)) <= LEVEL_CELL_MAX and (lv := level_rank(fam, x.text))] return short[0] if short else level_rank(fam, row_text) def bid_facts(fam: Family, docs: list[DocIndex], requirement: str) -> list[BidFact]: """投标文字层上写着这一类证书的句子：取同一行（表格）或同一句里的有效期、等级。 只认这一行 / 这一句里只有这一类证书的；照抄要求的句子不算。""" req = text_of(requirement) out: list[BidFact] = [] for ix in docs: seen: set[int] = set() for s, t in ix.texts: if not fam.pattern.search(t) or REQUIREMENT_LIKE.search(t): continue if req and (req in t or (t in req and len(t) >= COPY_RATIO * len(req))): continue # 整句就是要求原文（证书名称只是要求的一部分时不算） row = ix.rows.get(id(s)) or [s] if id(row[0]) in seen: continue seen.add(id(row[0])) row_text = CELL_SEP.join(x.text for x in row) if {f.key for f in families_in(row_text)} - {fam.key, "softcopy"}: continue # 一行里有两类证书，日期归属不清 evidence = [(ix.doc, x) for x in row if x is s or dates_in(x.text) or level_rank(fam, x.text)] fact = BidFact(evidence=evidence, level=row_level(fam, row, row_text)) labeled = VALIDITY_LABEL.search(row_text) or _header_has_validity(ix.page_rows.get(s.page_no, []), row) ds = dates_in(row_text) if LONG_TERM.search(row_text) and labeled: fact.long_term = True elif ds and labeled: fact.valid_to = max(ds) # 起止日期都在这一行时，晚的那个是「有效期至」 out.append(fact) return out def _header_has_validity(rows: list[list[Sentence]], row: list[Sentence]) -> bool: """表头（同一页往上几行）里有没有「有效期」这一栏。""" try: i = next(k for k, r in enumerate(rows) if r and row and r[0] is row[0]) except StopIteration: return False return any(VALIDITY_LABEL.search(CELL_SEP.join(x.text for x in r)) for r in rows[max(0, i - HEADER_LOOKBACK) : i]) @dataclass(slots=True) class Coverage: library: int = 0 checked: int = 0 by_library: int = 0 unrecognized: int = 0 excluded: int = 0 no_deadline: bool = False def to_api(self) -> dict[str, Any]: lines: list[str] = [] if self.library == 0: lines.append("资质库还没有录入资质，有效期和等级没有按资质库核对（只看投标文件文字部分）") if self.no_deadline: lines.append("项目信息里没有投标截止时间，证书有效期没有比对") if self.unrecognized: lines.append(f"{self.unrecognized} 条资质要求没认出是哪类证书，有效期和等级没有核对") lines.append("证书主体和投标人是否一致：还没有上线") checked = f"{LABEL}：{self.checked} 条资质要求" if self.by_library: checked += f"，其中 {self.by_library} 条按资质库记录判断" if self.excluded: checked += f"；产品证书、人员证书 {self.excluded} 条不在资质匹配范围内" return { "checked": checked, "not_checked": lines, "detail": { "library": self.library, "checked": self.checked, "by_library": self.by_library, "unrecognized": self.unrecognized, "excluded": self.excluded, }, } @dataclass(slots=True) class Issue: kind: str # expired / level_low / not_covered source: str # bid_text / library family: Family text: str severity: str evidence: list[tuple[BidDoc, Sentence]] = field(default_factory=list[tuple[BidDoc, Sentence]]) detail: dict[str, Any] = field(default_factory=dict[str, Any]) # 投标文件里有（已响应）、可能在扫描件里、或评分项没逐条找响应（None）：资质库的过期、等级照常提示； # 文字版投标文件确实没找到的，保留「未找到响应」，资质库的情况只补进说明 SEEN = ("responded", "scanned", None) CODES = {"expired": "QUAL.EXPIRED", "level_low": "QUAL.LEVEL_LOW", "not_covered": "QUAL.NOT_COVERED"} TITLES = {"expired": "证书在投标截止日已过期", "level_low": "证书等级低于要求", "not_covered": "资质库里没有这类资质"} ORDER = {"expired": 0, "level_low": 1, "not_covered": 2} def _d(x: date) -> str: return f"{x.year}年{x.month}月{x.day}日" def _lower(sev: str) -> str: return "important" if sev == "fatal" else sev def _best_valid(records: list[Record]) -> Record | None: """同一类有几条（旧证书和换发后的新证书）时取有效期最晚的；长期有效最好。""" return max(records, key=lambda r: date.max if r.valid_to is None else r.valid_to) if records else None def _rec(r: Record) -> str: no = f"，编号 {r.cert_no}" if r.cert_no else "" to = f"有效期至 {_d(r.valid_to)}" if r.valid_to else "长期有效" return f"《{r.name}》（{to}{no}）" def check_item( item: dict[str, Any], result: Result | None, fams: list[Family], docs: list[DocIndex], library: list[Record], by_family: dict[str, list[Record]], deadline: date | None, cov: Coverage, ) -> tuple[list[Issue], list[str]]: """一条清单项：返回资质问题和给原有问题补的说明（资质库的情况）。""" requirement = " ".join(str(item.get(k) or "") for k in ("title", "requirement_text", "evidence_required")) full = item_severity(item) verdict = result.verdict if result else None issues: list[Issue] = [] notes: list[str] = [] used_library = False for fam in fams: facts = bid_facts(fam, docs, requirement) recs = by_family.get(fam.key, []) if fam.key == "softcopy": # 软著长期有效、没有等级：只给扫描件兜底补一句 if recs and verdict in ("scanned", "not_found"): notes.append(f"资质库里有 {len(recs)} 件软件著作权") continue best = _best_valid(recs) # 有效期：投标文字层读到日期就以它为准（QL4），读不到再看资质库 dated = [x for x in facts if x.valid_to or x.long_term] if deadline and dated: if not any(x.long_term or (x.valid_to and x.valid_to >= deadline) for x in dated): latest = max(x.valid_to for x in dated if x.valid_to) text = f"投标文件里写的有效期至 {_d(latest)}，早于投标截止日 {_d(deadline)}。" if best and (best.valid_to is None or best.valid_to >= deadline): text += f"资质库里有{_rec(best)}，投标文件附的可能是旧证书。" issues.append( Issue( "expired", "bid_text", fam, text, full, [e for x in dated for e in x.evidence], {"bid_valid_to": latest.isoformat(), **({"library": best.snapshot()} if best else {})}, ) ) elif deadline and best and best.valid_to and best.valid_to < deadline and verdict in SEEN: used_library = True text = f"按资质库记录，{_rec(best)}早于投标截止日 {_d(deadline)}。投标文件文字部分没有写有效期。" issues.append(Issue("expired", "library", fam, text, _lower(full), [], {"library": best.snapshot()})) elif deadline and best and best.valid_to and best.valid_to < deadline and verdict == "not_found": notes.append(f"资质库里的{_rec(best)}也早于投标截止日") # 等级：能比较时才比 need = level_rank(fam, segment(fam, requirement)) if fam.levels else None if need: got = [x.level for x in facts if x.level] if got: top = max(got) if top[0] < need[0]: text = f"要求「{need[1]}」，投标文件里写的是「{top[1]}」。" issues.append( Issue( "level_low", "bid_text", fam, text, full, [e for x in facts if x.level for e in x.evidence], {"required_level": need[1], "bid_level": top[1]}, ) ) elif recs and verdict in SEEN: ranked = [(level_rank(fam, r.level or ""), r) for r in recs] comparable = [(lv, r) for lv, r in ranked if lv] if comparable and len(comparable) == len(ranked) and max(lv[0] for lv, _ in comparable) < need[0]: used_library = True r = max(comparable, key=lambda x: x[0][0])[1] text = f"要求「{need[1]}」，按资质库记录，{_rec(r)}的等级是「{r.level}」。" issues.append( Issue( "level_low", "library", fam, text, _lower(full), [], {"required_level": need[1], "library": r.snapshot()}, ) ) # 服务类别（CCRC）：要求写了类别，资质库的记录名称里都没写 sub = subtype_of(fam, segment(fam, requirement)) if sub and recs and not any(sub in text_of(r.name) for r in recs) and not facts: used_library = True text = f"要求的是「{sub}」类，资质库里的{fam.label}记录没写服务类别，请核对是不是「{sub}」。" issues.append(Issue("not_covered", "library", fam, text, "hint", [], {"subtype": sub})) # 未覆盖：投标文字层没找到、资质库有记录但没有这一类（QL3） if library and not recs and not facts and verdict in ("scanned", "not_found"): if verdict == "scanned": used_library = True text = ( f"投标文件文字部分没有找到{fam.label}，资质库里也没有这类资质（资质库有 {len(library)} 条记录）。" ) issues.append(Issue("not_covered", "library", fam, text, "important", [], {})) else: notes.append(f"资质库里也没有{fam.label}") elif best and verdict == "scanned" and not issues: notes.append(f"资质库里有{_rec(best)}，请核对扫描件里附的是这一张") if used_library: cov.by_library += 1 return issues, notes def run( items: dict[UUID, dict[str, Any]], results: dict[UUID, Result], docs: list[BidDoc], library: list[Record], deadline: date | None, ) -> tuple[list[Hit], Coverage]: """资质匹配：对资格条件、评分项里认出证书类别的清单项逐条核对。有资质问题的条目，扫描件兜底那条不再单独出； 资质库的情况补进原有问题的说明（results[...].detail["qualification_note"]）。""" cov = Coverage(library=len(library), no_deadline=deadline is None) index = [DocIndex.build(d) for d in docs] by_family: dict[str, list[Record]] = {} for r in library: if fam := family_of_record(r): by_family.setdefault(fam.key, []).append(r) hits: list[Hit] = [] for item_id, item in items.items(): cats = set(item.get("categories") or []) if not cats & {"qualification", "scoring"}: continue requirement = " ".join(str(item.get(k) or "") for k in ("title", "requirement_text", "evidence_required")) t = text_of(requirement) fams = families_in(requirement) if EXCLUDE.search(t): if fams or CERT_WORDS.search(t): cov.excluded += 1 continue if not fams: if CERT_WORDS.search(t) and not NOT_CERT.search(t): cov.unrecognized += 1 continue cov.checked += 1 result = results.get(item_id) issues, notes = check_item(item, result, fams, index, library, by_family, deadline, cov) if result is not None and notes: result.detail = {**result.detail, "qualification_note": "；".join(notes) + "。"} if not issues: continue if result is not None and result.verdict == "scanned": result.detail = {**result.detail, "superseded_by": "QUAL"} for fam in dict.fromkeys(x.family.key for x in issues): # 一类证书一个问题：过期和等级不足是两件事 mine = sorted( (x for x in issues if x.family.key == fam), key=lambda x: (ORDER[x.kind], x.source != "bid_text") ) top = mine[0] hits.append( Hit( code=CODES[top.kind], type="qualification", severity=min((x.severity for x in mine), key=["fatal", "important", "hint"].index), title=f"{TITLES[top.kind]}：{top.family.label}", explanation=" ".join(x.text for x in mine), evidence=[e for x in mine for e in x.evidence], target=f"{item_id}:{fam}", item=item_id, detail={ "kind": top.kind, "source": top.source, "family": fam, "deadline": deadline.isoformat() if deadline else None, "importance_basis": item.get("importance_basis"), "issues": [{"kind": x.kind, "source": x.source, **x.detail} for x in mine], }, tender_refs=[UUID(c) for c in item.get("citation_ids") or []], ) ) return hits, cov