#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ OCR完整性检查 / OCR Integrity Check ================================== 物理依赖链第一环：OCR → 建表 检查OCR后的.md文件是否有未补全的乱码/占位符。 如果有，拒绝生成checkpoint，建表脚本将因此中止。 用法： python3 ocr-integrity-check.py <工作目录> 检查模式： 1. 公司名乱码：连续3+英文字母出现在中文语境中（如"HAT"代替公司名） 2. 地址残缺：【…】、〔…〕、"了 B}"等占位符模式 3. 关键数字乱码：字母代替数字（如"5 1 te"代替"号1幢"） 4. OCR标记残留：OCR模糊、OCR无法识别、待补全等 通过 → 生成 step1.verified checkpoint（含签名） 失败 → 列出所有需要vision补全的字段，不生成checkpoint 退出码：0=通过，1=有问题需补全 🔴 安全机制：checkpoint文件包含HMAC签名，防止手动创建 """ import sys, os, re, json, glob, hmac, hashlib from datetime import datetime # checkpoint签名密钥（固定值，脚本内置） CHECKPOINT_SECRET = b'ocr-integrity-check-v1-2026-06-29' def generate_checkpoint_signature(workdir, md_files, timestamp): """生成checkpoint签名""" # 签名内容：工作目录+文件列表+时间戳 sign_content = f"{workdir}|{','.join(sorted(md_files))}|{timestamp}" sig = hmac.new(CHECKPOINT_SECRET, sign_content.encode('utf-8'), hashlib.sha256).hexdigest() return sig[:16] # 取前16位 def check_ocr_file(md_path): """检查单个.md文件的OCR完整性""" issues = [] with open(md_path, 'r', encoding='utf-8') as f: content = f.read() lines = content.split('\n') # 模式1：公司名乱码 - 连续3+大写字母或混合大小写英文（中文语境中） # 例如："HAT CP RRA)" 应该是公司名 company_garble = re.finditer(r'[A-Z]{3,}(?:\s+[A-Z]{2,}){0,2}', content) for m in company_garble: # 排除常见的合理英文（如OCR、PDF、API等） if m.group() not in ['OCR', 'PDF', 'API', 'URL', 'HTTP', 'HTTPS', 'JSON', 'XML', 'LOGO', 'EMS', 'WPS', 'PAGE', 'REMARK', 'NOTE', 'TODO', 'FIXME', 'HACK', 'XXX', 'EOF']: # 排除统一社会信用代码中的字母部分（前后有数字的字母序列） pre = content[max(0, m.start()-2):m.start()] post = content[m.end():min(len(content), m.end()+2)] if (pre and pre[-1:].isdigit()) or (post and post[:1].isdigit()): continue # 信用代码中的字母，跳过 line_num = content[:m.start()].count('\n') + 1 context_start = max(0, m.start() - 20) context_end = min(len(content), m.end() + 20) context = content[context_start:context_end].replace('\n', ' ') issues.append({ 'type': '公司名乱码', 'line': line_num, 'pattern': m.group(), 'context': f"...{context}...", 'fix': '用vision看原图，补全公司全称' }) # 模式2：地址残缺占位符 placeholder_patterns = [ (r'【\.{1,5}】', '中文省略号占位符'), (r'【…】', '中文省略号占位符'), (r'〔\.{1,5}〕', '方括号省略号占位符'), (r'了\s*[A-Za-z0-9]}', '地址残缺模式（如"了 B}"）'), ] for pat, desc in placeholder_patterns: for m in re.finditer(pat, content): line_num = content[:m.start()].count('\n') + 1 context_start = max(0, m.start() - 20) context_end = min(len(content), m.end() + 20) context = content[context_start:context_end].replace('\n', ' ') issues.append({ 'type': desc, 'line': line_num, 'pattern': m.group(), 'context': f"...{context}...", 'fix': '用vision看原图，补全完整地址' }) # 模式3：OCR标记残留 ocr_markers = [ r'OCR模糊', r'OCR无法识别', r'OCR乱码', r'OCR识别失败', r'待补全', r'〔待核PDF〕', r'〔待核〕', ] for marker in ocr_markers: for m in re.finditer(marker, content): line_num = content[:m.start()].count('\n') + 1 context_start = max(0, m.start() - 15) context_end = min(len(content), m.end() + 15) context = content[context_start:context_end].replace('\n', ' ') issues.append({ 'type': 'OCR标记残留', 'line': line_num, 'pattern': m.group(), 'context': f"...{context}...", 'fix': '用vision看原图补全，删除标记' }) # 模式4：关键字段中的字母代替数字 # 例如："5 1 te 202" 应该是 "号1幢202" # 检测：中文+空格+单个字母+空格+数字 的模式 digit_garble = re.finditer(r'[\u4e00-\u9fff]\s+[a-zA-Z]\s+\d{2,4}', content) for m in digit_garble: line_num = content[:m.start()].count('\n') + 1 context_start = max(0, m.start() - 10) context_end = min(len(content), m.end() + 10) context = content[context_start:context_end].replace('\n', ' ') issues.append({ 'type': '数字乱码（字母代替）', 'line': line_num, 'pattern': m.group(), 'context': f"...{context}...", 'fix': '用vision看原图，确认正确数字' }) # 模式5：严重乱码段落（连续3+英文无义词，如"peMet iy ecsapae"、"Fak"开头） # 表明该区域OCR完全失败，不可用于模版比对 garble_patterns = re.finditer(r'(?:[a-zA-Z]{3,}\s+){2,}[a-zA-Z]{2,}', content) for m in garble_patterns: # 排除已知英文短语 text = m.group().strip() if any(kw in text.lower() for kw in ['page', 'total', 'remark', 'note']): continue line_num = content[:m.start()].count('\n') + 1 context_start = max(0, m.start() - 15) context_end = min(len(content), m.end() + 15) context = content[context_start:context_end].replace('\n', ' ') issues.append({ 'type': '严重乱码段落（OCR完全失败）', 'line': line_num, 'pattern': text[:40], 'context': f"...{context[:60]}...", 'fix': '该区域OCR不可读，必须vision核实原文后才能写入比对报告' }) return issues def main(): if len(sys.argv) < 2: print("用法: python3 ocr-integrity-check.py <工作目录>") print("例: python3 ocr-integrity-check.py /tmp/人民中路") sys.exit(1) workdir = sys.argv[1] if not os.path.isdir(workdir): print(f"❌ 工作目录不存在: {workdir}") sys.exit(1) # 查找所有OCR产出的.md文件 md_files = glob.glob(os.path.join(workdir, '*.md')) md_files += glob.glob(os.path.join(workdir, 'md', '*.md')) md_files += glob.glob(os.path.join(workdir, 'ocr', '*.md')) if not md_files: print(f"❌ 未找到OCR产出的.md文件: {workdir}") sys.exit(1) print(f"检查 {len(md_files)} 个OCR文件的完整性...\n") all_issues = [] for md_file in md_files: fname = os.path.basename(md_file) issues = check_ocr_file(md_file) if issues: all_issues.append((fname, issues)) checkpoint_path = os.path.join(workdir, 'step1.verified') if all_issues: print("❌ OCR完整性检查失败\n") for fname, issues in all_issues: print(f" {fname}: {len(issues)} 个问题") for i, issue in enumerate(issues[:5], 1): # 只显示前5个 print(f" {i}. [{issue['type']}] 第{issue['line']}行") print(f" {issue['context']}") print(f" → {issue['fix']}") if len(issues) > 5: print(f" ... 还有 {len(issues) - 5} 个问题") print(f"\n⛔ 未生成checkpoint: {checkpoint_path}") print(" 建表脚本将拒绝执行，直到所有问题用vision补全。") # 写入问题清单供后续处理 issues_log = os.path.join(workdir, 'step1.issues.json') with open(issues_log, 'w', encoding='utf-8') as f: json.dump({ 'timestamp': datetime.now().isoformat(), 'issues': {fname: issues for fname, issues in all_issues} }, f, ensure_ascii=False, indent=2) print(f" 问题清单已保存: {issues_log}") sys.exit(1) else: print("✅ OCR完整性检查通过\n") for md_file in md_files: print(f" ✓ {os.path.basename(md_file)}") # 生成checkpoint（含签名） timestamp = datetime.now().isoformat() signature = generate_checkpoint_signature(workdir, md_files, timestamp) with open(checkpoint_path, 'w', encoding='utf-8') as f: f.write(f"OCR完整性检查通过\n") f.write(f"时间: {timestamp}\n") f.write(f"文件数: {len(md_files)}\n") for md_file in md_files: f.write(f" - {os.path.basename(md_file)}\n") f.write(f"签名: {signature}\n") print(f"\n✓ checkpoint已生成: {checkpoint_path}") print(f" 签名: {signature}") sys.exit(0) if __name__ == '__main__': main()