#!/usr/bin/env python3 """ 大医网OCR/文本杂质清洗脚本 ----------------------------- 清洗7个核心库的JSON源文件中的OCR形近错字和格式杂质。 直接修改源文件（明确OCR校正，不改变语义）。 生成清洗报告到 04_分析报告/数据质量/OCR清洗报告.md """ import json import os import re from collections import defaultdict, OrderedDict # ============================================================ # 配置 # ============================================================ BASE_DIR = "/home/songyi/Documents/ai_agent_scraper_study/data/大医网" SOURCE_DIR = os.path.join(BASE_DIR, "01_来源数据") SUBDIRS = ["疾病", "方剂", "中药材", "针灸穴位", "症状", "术语", "药膳食疗"] REPORT_DIR = os.path.join(BASE_DIR, "04_分析报告", "数据质量") REPORT_FILE = os.path.join(REPORT_DIR, "OCR清洗报告.md") # ============================================================ # 清洗规则定义（按优先级） # 每条规则: (pattern, replacement, description, is_regex) # ============================================================ CLEAN_RULES = [ # --- OCR形近错字（注意负向零宽断言）--- (re.compile(r'(? 0: stats_counter[desc] += count file_stat[desc] += count total_changes += count text = new_text else: # 普通字符串替换（保留备用） count = text.count(pattern) if count > 0: text = text.replace(pattern, replacement) stats_counter[desc] += count file_stat[desc] += count total_changes += count return text, total_changes def walk_and_clean(data, stats_counter, file_stat): """递归遍历嵌套的dict/list结构，清洗所有字符串字段和键名""" total_changes = 0 if isinstance(data, dict): # 先处理键名：可能含有零宽字符等杂质 cleaned_keys = {} for key in list(data.keys()): if isinstance(key, str): new_key, changes = apply_clean_rules(key, stats_counter, file_stat) if changes > 0: cleaned_keys[key] = new_key total_changes += changes # 重建dict（如果键名有变化） if cleaned_keys: for old_key, new_key in cleaned_keys.items(): data[new_key] = data.pop(old_key) # 处理值 for key in list(data.keys()): val = data[key] if isinstance(val, str): new_val, changes = apply_clean_rules(val, stats_counter, file_stat) if changes > 0: data[key] = new_val total_changes += changes elif isinstance(val, (dict, list)): total_changes += walk_and_clean(val, stats_counter, file_stat) elif isinstance(data, list): for i in range(len(data)): val = data[i] if isinstance(val, str): new_val, changes = apply_clean_rules(val, stats_counter, file_stat) if changes > 0: data[i] = new_val total_changes += changes elif isinstance(val, (dict, list)): total_changes += walk_and_clean(val, stats_counter, file_stat) return total_changes def clean_file(filepath, stats_counter, file_changes_log): """ 清洗单个JSON文件： 1. json.load 读取 2. 递归遍历清洗 3. json.dump 写回 返回 (修改文件数:0或1, 修正总次数) """ try: with open(filepath, 'r', encoding='utf-8') as f: data = json.load(f) except (json.JSONDecodeError, UnicodeDecodeError) as e: print(f" [!] 跳过无法解析的JSON: {filepath} — {e}") return 0, 0 file_stat = defaultdict(int) changes = walk_and_clean(data, stats_counter, file_stat) if changes == 0: return 0, 0 # 记录此文件的修改详情 file_changes_log.append({ 'filepath': filepath, 'changes': changes, 'detail': {k: v for k, v in file_stat.items() if v > 0} }) # 写回 try: with open(filepath, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) return 1, changes except Exception as e: print(f" [!!] 写回失败: {filepath} — {e}") return 0, 0 def scan_for_residual(data): """扫描数据中是否仍有任何未清洗的匹配项""" residuals = defaultdict(int) def _scan(d): if isinstance(d, dict): for v in d.values(): _scan(v) elif isinstance(d, list): for v in d: _scan(v) elif isinstance(d, str): for pattern, replacement, desc, is_regex in CLEAN_RULES: if is_regex: matches = pattern.findall(d) if matches: residuals[desc] += len(matches) else: cnt = d.count(pattern) if cnt: residuals[desc] += cnt _scan(data) return residuals def validate_residuals(): """清洗后重新扫描所有JSON文件，验证零残留""" residual_counts = defaultdict(int) residual_files = defaultdict(list) for subdir in SUBDIRS: dirpath = os.path.join(SOURCE_DIR, subdir) if not os.path.isdir(dirpath): continue for fname in os.listdir(dirpath): if not fname.endswith('.json'): continue fpath = os.path.join(dirpath, fname) try: with open(fpath, 'r', encoding='utf-8') as f: data = json.load(f) except: continue res = scan_for_residual(data) if res: for desc, cnt in res.items(): residual_counts[desc] += cnt residual_files[desc].append(fname) return residual_counts, residual_files def get_dataset_name(subdir): """映射子目录名到中文数据集名""" mapping = { '疾病': '疾病库', '方剂': '方剂库', '中药材': '中药材库', '针灸穴位': '针灸穴位库', '症状': '症状库', '术语': '术语库', '药膳食疗': '药膳食疗库', } return mapping.get(subdir, subdir) def relative_path(filepath): """将绝对路径转为相对路径（用于报告）""" return filepath.replace(BASE_DIR, '.../大医网') def main(): print("=" * 70) print(" 大医网OCR/文本杂质清洗") print("=" * 70) # 全局统计 # stats_counter: { desc: total_count } stats_counter = defaultdict(int) # 按数据集的统计: { subdir: { 'files_cleaned': N, 'total_changes': N, 'detail': {desc: count} } } dataset_stats = {} # 所有修改的文件日志 all_file_changes_log = [] for subdir in SUBDIRS: dirpath = os.path.join(SOURCE_DIR, subdir) if not os.path.isdir(dirpath): print(f"\n [!] 目录不存在: {dirpath}") continue print(f"\n{'─'*50}") print(f" 处理: {subdir}/") print(f"{'─'*50}") json_files = sorted([f for f in os.listdir(dirpath) if f.endswith('.json')]) total_files = len(json_files) files_cleaned = 0 total_changes_subdir = 0 subdir_detail = defaultdict(int) for idx, fname in enumerate(json_files): fpath = os.path.join(dirpath, fname) file_changes_log_entry = [] cleaned_flag, changes = clean_file(fpath, stats_counter, file_changes_log_entry) files_cleaned += cleaned_flag total_changes_subdir += changes if file_changes_log_entry: all_file_changes_log.extend(file_changes_log_entry) for desc, cnt in file_changes_log_entry[0]['detail'].items(): subdir_detail[desc] += cnt # 进度显示 if (idx + 1) % 200 == 0 or idx == 0 or (idx + 1) == total_files: print(f" 进度: {idx+1}/{total_files} | 已清洗: {files_cleaned} 个文件 | 修正: {total_changes_subdir} 处") dataset_stats[subdir] = { 'total_files': total_files, 'files_cleaned': files_cleaned, 'total_changes': total_changes_subdir, 'detail': dict(subdir_detail), } ds_name = get_dataset_name(subdir) print(f" ✓ {ds_name}: {files_cleaned}/{total_files} 个文件被修改，共修正 {total_changes_subdir} 处") # ============================================================ # 汇总 # ============================================================ print("\n" + "=" * 70) print(" 清洗完成 - 汇总统计") print("=" * 70) total_files_all = sum(s['total_files'] for s in dataset_stats.values()) total_cleaned_all = sum(s['files_cleaned'] for s in dataset_stats.values()) total_changes_all = sum(s['total_changes'] for s in dataset_stats.values()) print(f" 总文件数: {total_files_all}") print(f" 被修改文件: {total_cleaned_all}") print(f" 总修正处: {total_changes_all}") print(f"\n 错误类型分布:") for desc, cnt in sorted(stats_counter.items(), key=lambda x: -x[1]): print(f" {desc}: {cnt}") # ============================================================ # 清洗后验证 - 零残留 # ============================================================ print("\n" + "=" * 70) print(" 清洗后残留验证") print("=" * 70) residual_counts, residual_files = validate_residuals() if residual_counts: print(f" ⚠ 发现 {sum(residual_counts.values())} 处残留:") for desc, cnt in sorted(residual_counts.items(), key=lambda x: -x[1]): print(f" {desc}: {cnt} 处 (共 {len(residual_files.get(desc, []))} 个文件)") residual_status = "有残留" else: print(" ✓ 零残留！所有规则均已正确应用。") residual_status = "零残留 ✓" # ============================================================ # 生成清洗报告 # ============================================================ print(f"\n 生成报告 → {REPORT_FILE}") os.makedirs(REPORT_DIR, exist_ok=True) lines = [] lines.append("# OCR/文本杂质清洗报告") lines.append("") lines.append(f"**数据目录：** `{SOURCE_DIR}`") lines.append(f"**子目录：** 疾病/ 方剂/ 中药材/ 针灸穴位/ 症状/ 术语/ 药膳食疗") lines.append(f"**清洗时间：** {__import__('datetime').datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") lines.append("") lines.append("---") lines.append("") lines.append("## 一、汇总统计") lines.append("") lines.append("| 数据集名 | 总文件数 | 清洗文件数 | 总修正处 | 错误类型分布 |") lines.append("|----------|----------|------------|----------|--------------|") for subdir in SUBDIRS: if subdir not in dataset_stats: continue s = dataset_stats[subdir] ds_name = get_dataset_name(subdir) detail_str = "; ".join([f"{d}: {c}" for d, c in sorted(s['detail'].items(), key=lambda x: -x[1])]) if not detail_str: detail_str = "-" lines.append(f"| {ds_name} | {s['total_files']} | {s['files_cleaned']} | {s['total_changes']} | {detail_str} |") lines.append(f"| **合计** | **{total_files_all}** | **{total_cleaned_all}** | **{total_changes_all}** | **{'; '.join([f'{d}: {c}' for d, c in sorted(stats_counter.items(), key=lambda x: -x[1])])}** |") lines.append("") lines.append("## 二、清洗规则详情") lines.append("") lines.append("| 优先级 | 规则 | 模式 | 替换为 | 修正次数 |") lines.append("|--------|------|------|--------|----------|") for i, (pattern, replacement, desc, is_regex) in enumerate(CLEAN_RULES, 1): pat_str = pattern.pattern if is_regex else pattern cnt = stats_counter.get(desc, 0) lines.append(f"| {i} | {desc} | `{pat_str}` | `{replacement}` | {cnt} |") lines.append("") lines.append("## 三、清洗后验证") lines.append("") if residual_counts: lines.append(f"### ⚠ 残留检测结果：{sum(residual_counts.values())} 处残留") lines.append("") lines.append("| 规则 | 残留数 | 涉及文件数 |") lines.append("|------|--------|------------|") for desc, cnt in sorted(residual_counts.items(), key=lambda x: -x[1]): lines.append(f"| {desc} | {cnt} | {len(residual_files.get(desc, []))} |") lines.append("") lines.append("**注意：** 残留可能存在于非字符串字段（如JSON key）或特殊格式中，需人工复核。") else: lines.append("**验证结果：零残留 ✓** — 所有规则均已正确应用，无未处理的匹配项。") lines.append("") lines.append("## 四、错误类型分布（全部）") lines.append("") lines.append("| 错误类型 | 修正次数 | 占比 |") lines.append("|----------|----------|------|") total = sum(stats_counter.values()) for desc, cnt in sorted(stats_counter.items(), key=lambda x: -x[1]): pct = f"{cnt/total*100:.1f}%" if total > 0 else "-" lines.append(f"| {desc} | {cnt} | {pct} |") lines.append(f"| **总计** | **{total}** | **100%** |") lines.append("") lines.append("## 五、被修改文件列表") lines.append("") lines.append(f"共 {len(all_file_changes_log)} 个文件被修改：") lines.append("") # 按数据集分组显示 by_dataset = defaultdict(list) for entry in all_file_changes_log: fpath = entry['filepath'] for subdir in SUBDIRS: if f'/01_来源数据/{subdir}/' in fpath: by_dataset[subdir].append(entry) break for subdir in SUBDIRS: entries = by_dataset.get(subdir, []) if not entries: continue ds_name = get_dataset_name(subdir) lines.append(f"### {ds_name}（{len(entries)} 个文件）") lines.append("") lines.append("| 文件名 | 修正总数 | 修正详情 |") lines.append("|--------|----------|----------|") for entry in sorted(entries, key=lambda x: -x['changes']): fname = os.path.basename(entry['filepath']) detail_str = "; ".join([f"{d}: {c}" for d, c in sorted(entry['detail'].items(), key=lambda x: -x[1])]) lines.append(f"| {fname} | {entry['changes']} | {detail_str} |") lines.append("") # 写入报告 report_content = "\n".join(lines) with open(REPORT_FILE, 'w', encoding='utf-8') as f: f.write(report_content) print(f"\n{'='*70}") print(f" 报告已生成: {REPORT_FILE}") print(f" 总文件: {total_files_all}") print(f" 被修改: {total_cleaned_all}") print(f" 总修正: {total_changes_all}") print(f" 残留验证: {residual_status}") print(f"{'='*70}") return total_files_all, total_cleaned_all, total_changes_all, residual_status if __name__ == "__main__": main()