#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 交付前闸门 / Delivery Gate ========================= 每次交付南通新东方梳理表前，必须跑这个脚本。 不通过 = 不发文件。不通过 = 回去补 workflow。 用法： python3 delivery-gate.py <工作目录> <校区名> 例： python3 delivery-gate.py /tmp/金飞达/金飞达-梳理-MJ-20260627.xlsx /tmp/金飞达 金飞达 检查项： G1. 模版比对 subagent 是否执行过（工作目录是否有 模版比对-*.md） G2. K/L 分工自检是否通过 G3. 格式自检是否通过（标题酒红底、段标题红底、列头正确） G4. 整体风险分析段是否存在 G5. 数据行数是否合理（至少 文件数 行） G6. 文件是否已上传 Nextcloud G7. OCR占位符残留检测（【…】/〔待核PDF〕/OCR模糊 等，任一残留=禁止交付） """ import sys, os, subprocess, re import openpyxl def fail(msg): print(f"❌ {msg}") return False def ok(msg): print(f"✅ {msg}") return True def check_g1(workdir): """G1: 模版比对 subagent 是否执行过""" import glob files = glob.glob(os.path.join(workdir, "模版比对-*.md")) if not files: return fail("G1 模版比对：未找到模版比对输出文件。Step2 动作B 必须 delegate_task subagent 做模版比对。") f = files[0] size = os.path.getsize(f) if size < 1000: return fail(f"G1 模版比对：{os.path.basename(f)} 仅 {size} 字节，内容过短，可能未完整执行。") return ok(f"G1 模版比对：{os.path.basename(f)} ({size:,} 字节)") def check_g2(xlsx_path): """G2: K/L 分工自检""" # 直接用 kl-separation-check.py script = os.path.expanduser("~/.hermes/skills/legal/contract-portfolio-analysis/scripts/kl-separation-check.py") r = subprocess.run(["python3", script, xlsx_path], capture_output=True, text=True) if r.returncode != 0: return fail(f"G2 K/L自检：未通过\n{r.stdout}") return ok("G2 K/L自检：通过") def check_g3(xlsx_path): """G3: 格式自检""" try: wb = openpyxl.load_workbook(xlsx_path) ws = wb[wb.sheetnames[0]] # 找到第一个列头行（含"序号"的） hdr_row = None for r in range(1, 30): if ws.cell(r, 1).value == '序号': hdr_row = r break if not hdr_row: return fail("G3 格式：未找到列头行") checks = [ ("标题酒红底", ws.cell(1,1).fill.start_color.rgb == 'FF8B1A2B'), ("段标题(K列头)", ws.cell(hdr_row, 11).value == '法律风险（站乙方立场）'), ("段标题(L列头)", ws.cell(hdr_row, 12).value == '与07标准模版差异'), ] for name, passed in checks: if not passed: return fail(f"G3 格式：{name} 不正确") return ok("G3 格式自检：通过") except Exception as e: return fail(f"G3 格式：打开失败 - {e}") def check_g4(xlsx_path): """G4: 整体风险分析段""" try: wb = openpyxl.load_workbook(xlsx_path) ws = wb[wb.sheetnames[0]] for r in range(ws.max_row, 1, -1): v = ws.cell(r, 1).value if v and '整体风险分析' in str(v): return ok(f"G4 整体风险分析段：存在（行{r}）") return fail("G4 整体风险分析段：未找到。每校区必须包含「整体风险分析与建议」段。") except Exception as e: return fail(f"G4 整体风险分析段：打开失败 - {e}") def check_g5(xlsx_path, min_rows=3): """G5: 数据行数""" try: wb = openpyxl.load_workbook(xlsx_path) ws = wb[wb.sheetnames[0]] data_rows = 0 for r in range(5, ws.max_row + 1): v = ws.cell(r, 1).value if v and re.match(r'^\d+$', str(v).strip()): data_rows += 1 if data_rows < min_rows: return fail(f"G5 数据行数：仅 {data_rows} 行，需 ≥ {min_rows}") return ok(f"G5 数据行数：{data_rows} 行") except Exception as e: return fail(f"G5 数据行数：打开失败 - {e}") def check_g6(xlsx_path, campus): """G6: 文件是否已上传 Nextcloud""" fname = os.path.basename(xlsx_path) nc_path = f"/var/www/html/data/admin/files/小Maggie协作区/南通新东方/履约期内非集采合同-综办/房租物业合同/{campus}/{fname}" r = subprocess.run( ["docker", "exec", "nextcloud-nextcloud-1", "test", "-f", nc_path], capture_output=True ) if r.returncode == 0: return ok(f"G6 Nextcloud：已上传 {campus}/{fname}") else: return fail(f"G6 Nextcloud：未上传到 {campus}/{fname}。请先 docker cp + files:scan。") def check_g7(xlsx_path): """G7: OCR占位符残留检测（【…】、〔待核PDF〕、OCR模糊 等）""" patterns = [ r'【\.{1,5}】', # 【…】、【...】 r'【…】', # 中文省略号 r'〔待核PDF〕', # 待核标记 r'〔待核〕', r'OCR模糊', # OCR模糊标记 r'OCR无法识别', r'OCR乱码', r'待补全', ] try: wb = openpyxl.load_workbook(xlsx_path, data_only=True) hits = [] for ws in wb.worksheets: for r in range(1, ws.max_row + 1): for c in range(1, ws.max_column + 1): v = ws.cell(r, c).value if not v: continue sv = str(v) for pat in patterns: if re.search(pat, sv): col_letter = openpyxl.utils.get_column_letter(c) m = re.search(pat, sv) start = max(0, m.start() - 10) end = min(len(sv), m.end() + 10) ctx = sv[start:end].replace('\n', ' ') hits.append(f" [{ws.title}] {col_letter}{r}: ...{ctx}...") if hits: msg = f"G7 OCR占位符残留：发现 {len(hits)} 处未补全的OCR标记\n" + "\n".join(hits[:5]) if len(hits) > 5: msg += f"\n ...（共 {len(hits)} 处，仅显示前5处）" msg += "\n 铁律：OCR读不出的必须用vision看原图补全，不能用占位符交付。" return fail(msg) return ok("G7 OCR占位符：无残留（全表扫描通过）") except Exception as e: return fail(f"G7 OCR占位符：检查失败 - {e}") def check_g8(workdir): """G8: OCR完整性checkpoint（物理依赖链第一环）""" checkpoint = os.path.join(workdir, 'step1.verified') if not os.path.exists(checkpoint): return fail(f"G8 OCR依赖链：checkpoint不存在\n 路径: {checkpoint}\n 说明: Step1 OCR完成后必须运行 ocr-integrity-check.py 生成checkpoint\n 动作: python3 scripts/ocr-integrity-check.py {workdir}") # 读取checkpoint内容确认 with open(checkpoint, 'r', encoding='utf-8') as f: content = f.read() if 'OCR完整性检查通过' not in content: return fail(f"G8 OCR依赖链：checkpoint内容异常\n {checkpoint} 未包含'OCR完整性检查通过'") return ok(f"G8 OCR依赖链：checkpoint存在且有效") def check_g9(workdir): """G9: 模版比对checkpoint（物理依赖链第二环）""" checkpoint = os.path.join(workdir, 'step2b.verified') if not os.path.exists(checkpoint): return fail(f"G9 模版比对依赖链：checkpoint不存在\n 路径: {checkpoint}\n 说明: Step2 动作B必须delegate subagent做模版比对，然后运行验证脚本\n 动作: python3 scripts/template-diff-verify.py {workdir}") # 读取checkpoint内容确认 with open(checkpoint, 'r', encoding='utf-8') as f: content = f.read() if '模版比对验证通过' not in content: return fail(f"G9 模版比对依赖链：checkpoint内容异常\n {checkpoint} 未包含'模版比对验证通过'") return ok(f"G9 模版比对依赖链：checkpoint存在且有效") def main(): if len(sys.argv) < 4: print("用法: python3 delivery-gate.py <工作目录> <校区名>") print("例: python3 delivery-gate.py /tmp/金飞达/金飞达-梳理-MJ-20260627.xlsx /tmp/金飞达 金飞达") sys.exit(1) xlsx = sys.argv[1] workdir = sys.argv[2] campus = sys.argv[3] if not os.path.exists(xlsx): print(f"❌ 文件不存在: {xlsx}") sys.exit(1) bar = "=" * 60 print(bar) print(f" 交付闸门 · {campus}") print(bar) print() results = [ check_g1(workdir), check_g2(xlsx), check_g3(xlsx), check_g4(xlsx), check_g5(xlsx, min_rows=3), check_g6(xlsx, campus), check_g7(xlsx), check_g8(workdir), check_g9(workdir), ] print() print(bar) passed = sum(1 for r in results if r) total = len(results) if all(results): print(f" ✅ 全部通过 ({passed}/{total}) —— 可以交付") print(bar) sys.exit(0) else: print(f" ❌ {total - passed}/{total} 项未通过 —— 禁止交付，回去补 workflow") print(bar) sys.exit(1) if __name__ == "__main__": main()