# -*- coding: utf-8 -*- """票 09 终验收：A4 客户管理 Bruno 文档联调就绪核验。 核验项（对 D:\\code\\crm-api-docs\\A4 客户管理 全量 .bru，排除 folder.bru）： V1 文件数与结构对账：磁盘文件 ↔ canonical-endpoint-response-mapping（117）+ 票 10-N3 补充（1）双向零缺零余，对账钥匙行一致 V2 「非真实返回」占位全仓 = 0 V3 四件套结构：meta / 请求块 / docs{ / 请求参数节 / 响应 data 结构节 / 响应示例节（二进制端点按票 05 口径豁免 data 结构与示例节） V4 嵌套 DTO 展开：`List<非原始类型>` 字段表行后必须跟子表（field[]. 前缀行）；其余泛型/对象行软性报告 V5 说明质量（票 06 口径）：字段表行说明列非空且无断行残片（<3 格行） V6 UTF-8 无 BOM 输出：控制台摘要 + verify-integration-ready.json（本目录）。 """ import json import os import re import sys DOCS = r"D:\code\crm-api-docs\A4 客户管理" MAPPING = r"D:\code\crm-backend-matt\.scratch\a4-doc-fields\canonical-endpoint-response-mapping.md" HERE = os.path.dirname(os.path.abspath(__file__)) OUT_JSON = os.path.join(HERE, "verify-integration-ready.json") # 票 10-N3 新增端点（mapping 建于 a4-doc-fields 轮，彼时尚无关联项目端点） SUPPLEMENT = { "客户详情/关联项目/关联项目页签.bru": "GET /api/customer/project/page", } PRIMITIVES = { "String", "Long", "Integer", "Boolean", "Double", "Float", "BigDecimal", "Object", "LocalDate", "LocalDateTime", "LocalTime", "Date", "MultipartFile", } ROW_RE = re.compile(r"^\|(.+)$") LIST_RE = re.compile(r"List<\s*([A-Z]\w*)") GENERIC_RE = re.compile(r"<([A-Z]\w*)") def iter_bru(): for root, dirs, files in os.walk(DOCS): for fn in sorted(files): if fn.endswith(".bru") and fn != "folder.bru": p = os.path.join(root, fn) rel = os.path.relpath(p, DOCS).replace("\\", "/") yield rel, p def read_file(p): raw = open(p, "rb").read() bom = raw[:3] == b"\xef\xbb\xbf" return bom, raw.decode("utf-8-sig", errors="replace") def parse_bru(txt): """返回 dict：meta/req/docs 各块有无 + docs 文本（去 2 空格缩进）+ 对账钥匙。""" meta = re.search(r"^meta\s*\{", txt, re.M) is not None req = re.search(r"^(get|post|put|delete)\s*\{", txt, re.M | re.I) is not None m = re.search(r"^docs\s*\{", txt, re.M) docs = "" key = None if m: lines = [] for ln in txt[m.end():].splitlines(): if ln == "}": # 顶格 } = docs 块结束（JSON 花括号均在缩进内） break lines.append(ln[2:] if ln.startswith(" ") else ln) docs = "\n".join(lines) km = re.search(r"`(GET|POST|PUT|DELETE) (\S+)`", docs) if km: key = km.group(1) + " " + km.group(2) return {"meta": meta, "req": req, "has_docs": m is not None, "docs": docs, "key": key} def load_mapping(): """canonical-endpoint-response-mapping.md → {relpath: (method_path, status)}""" out = {} with open(MAPPING, "r", encoding="utf-8") as f: for ln in f: m = re.match(r"^\|\s*`([^`]+\.bru)`\s*\|\s*`(GET|POST|PUT|DELETE) (\S+)`\s*\|\s*([^|]+)\|", ln) if m: out[m.group(1)] = (m.group(2) + " " + m.group(3), m.group(4).strip()) return out def table_rows(docs): """docs 内所有字段表行（去缩进后以 | 开头），跳过表头分隔行。返回 [(idx, cells)]""" rows = [] for i, ln in enumerate(docs.splitlines()): s = ln.strip() if not s.startswith("|"): continue cells = [c.strip() for c in s.strip("|").split("|")] if all(re.fullmatch(r"-{3,}", c) for c in cells if c): continue rows.append((i, cells)) return rows def check_nested(docs, fname, hard_fail, soft): """List<非原始> 行须有 field[]. 子行；其余泛型/对象行软报告。""" rows = table_rows(docs) for idx, (i, cells) in enumerate(rows): if len(cells) < 2: continue field = cells[0].replace("`", "").strip() ftype = cells[1].replace("`", "").strip() if not re.fullmatch(r"[A-Za-z_<>\[\], .\[\]]+", ftype): continue lm = LIST_RE.search(ftype) if lm: if lm.group(1) in PRIMITIVES: continue prefix = field + "[]." found = False prev_j = i for j, cs in rows[idx + 1:]: if j - prev_j > 2: break # 中间断行/空行/标题 → 同表结束 prev_j = j f0 = cs[0].replace("`", "").strip() if f0.startswith(prefix) or f0.startswith(field + "."): found = True break if f0[:1].isupper() and "." not in f0: break # 新的顶层字段开始，同表结束 if not found: hard_fail.append(f"{fname}: `{field}` 行类型 {ftype} 后无 {prefix} 子行") elif GENERIC_RE.search(ftype) and ftype not in PRIMITIVES: # Map / BatchResult 等非 List 泛型：软报告 prefix_exists = any( cs[0].replace("`", "").strip().startswith(field + ("[]." if "List<" in ftype else ".")) for _, cs in rows[idx + 1:] if cs[0].replace("`", "").strip().startswith(field) ) if not prefix_exists: soft.append(f"{fname}: `{field}` 非List泛型 {ftype} 无点路径子行（软报告）") def main(): mapping = load_mapping() expected = dict(mapping) for k, v in SUPPLEMENT.items(): expected[k] = (v, "票10-N3 新增") binary = {k for k, (_, st) in expected.items() if "二进制" in st} files = {} v2_hits, v3_fails, v5_fails, v6_fails = [], [], [], [] v4_hard, v4_soft = [], [] for rel, p in iter_bru(): bom, txt = read_file(p) files[rel] = p if bom: v6_fails.append(rel) if "非真实返回" in txt: v2_hits.append(rel) info = parse_bru(txt) docs = info["docs"] # V3 四件套 miss = [] if not info["meta"]: miss.append("meta块") if not info["req"]: miss.append("请求块") if not info["has_docs"]: miss.append("docs块") else: if "## 请求参数" not in docs: miss.append("请求参数节") if rel not in binary: if "## 响应 data 结构" not in docs: miss.append("响应data结构节") if "## 响应示例" not in docs: miss.append("响应示例节") if miss: v3_fails.append(f"{rel}: 缺 {' / '.join(miss)}") if docs: # V4 嵌套展开 check_nested(docs, rel, v4_hard, v4_soft) # V5 说明质量 for _, cells in table_rows(docs): if len(cells) < 3: v5_fails.append(f"{rel}: 断行残片行 `| {'|'.join(cells)}`") continue desc = cells[-1] if not desc or desc == "-" or len(desc) < 2: if cells[0].strip() in ("参数", "字段"): # 表头行 continue v5_fails.append(f"{rel}: `{cells[0]}` 说明列为空") # V1 结构对账 missing = sorted(set(expected) - set(files)) extra = sorted(set(files) - set(expected)) key_mismatch = [] for rel, p in sorted(files.items()): if rel not in expected: continue info = parse_bru(read_file(p)[1]) if info["key"] != expected[rel][0]: key_mismatch.append(f"{rel}: 钥匙 `{info['key']}` ≠ 期望 `{expected[rel][0]}`") ok = lambda b: "PASS" if b else "FAIL" print("=" * 72) print("V1 结构对账 ", ok(not missing and not extra and not key_mismatch), f"（磁盘 {len(files)} 文件 ↔ 期望 {len(expected)} = mapping {len(mapping)} + 票10补充 {len(SUPPLEMENT)}）") for x in missing: print(" 缺文件:", x) for x in extra: print(" 多文件:", x) for x in key_mismatch: print(" 钥匙不符:", x) print("V2 非真实返回=0 ", ok(not v2_hits), f"（命中 {len(v2_hits)}）", v2_hits[:5]) print("V3 四件套结构 ", ok(not v3_fails), f"（缺节 {len(v3_fails)}）") for x in v3_fails[:15]: print(" ", x) print("V4 嵌套DTO展开 ", ok(not v4_hard), f"（硬失败 {len(v4_hard)}，软报告 {len(v4_soft)}）") for x in v4_hard[:15]: print(" ", x) for x in v4_soft[:10]: print(" [软]", x) print("V5 说明质量 ", ok(not v5_fails), f"（问题行 {len(v5_fails)}）") for x in v5_fails[:15]: print(" ", x) print("V6 UTF-8 无 BOM ", ok(not v6_fails), f"（BOM {len(v6_fails)}）", v6_fails[:5]) print("=" * 72) all_pass = not (missing or extra or key_mismatch or v2_hits or v3_fails or v4_hard or v5_fails or v6_fails) print("总体判定:", "ALL_PASS —— 联调就绪核验通过" if all_pass else "存在 FAIL 项，见上") with open(OUT_JSON, "w", encoding="utf-8") as f: json.dump({ "total_files": len(files), "expected": len(expected), "missing": missing, "extra": extra, "key_mismatch": key_mismatch, "v2_hits": v2_hits, "v3_fails": v3_fails, "v4_hard": v4_hard, "v4_soft": v4_soft, "v5_fails": v5_fails, "v6_fails": v6_fails, "all_pass": all_pass, }, f, ensure_ascii=False, indent=2) print("明细 JSON:", OUT_JSON) sys.exit(0 if all_pass else 1) if __name__ == "__main__": main()