#!/usr/bin/env python3 # -*- coding: utf-8 -*- """合并通用图片表格 OCR 批次结果。""" from __future__ import annotations import argparse import csv import json import re import unicodedata from pathlib import Path from typing import Any DEFAULT_CONFIG = Path("数据处理工作区/01_任务配置.json") TEMPLATE_CONFIG = Path("数据处理工作区/01_任务配置.template.json") def normalize_text(value: Any) -> str: if value is None: return "" text = unicodedata.normalize("NFKC", str(value)).replace("\u3000", " ") return re.sub(r"\s+", " ", text).strip() def read_json(path: Path) -> Any: return json.loads(path.read_text(encoding="utf-8")) def write_json(path: Path, data: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") def write_jsonl(path: Path, records: list[dict[str, Any]]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="utf-8") as file: for record in records: file.write(json.dumps(record, ensure_ascii=False) + "\n") def load_config(path: Path) -> dict[str, Any]: if path.exists(): return read_json(path) if TEMPLATE_CONFIG.exists(): return read_json(TEMPLATE_CONFIG) return {} def configured_fields(config: dict[str, Any], archives: list[dict[str, Any]]) -> list[str]: fields = [normalize_text(item.get("name")) for item in config.get("fields", []) if normalize_text(item.get("name"))] if fields: return fields seen: list[str] = [] for archive in archives: for record in archive.get("图片表格记录", []): for key in record.get("记录信息", {}): if key not in seen: seen.append(key) return seen def summarize_record(record: dict[str, Any], unique_key: str) -> dict[str, Any]: image = record.get("图片信息", {}) return { "处理批次": record.get("处理批次", ""), "来源文件夹": record.get("来源文件夹", ""), "图片路径": image.get("图片路径", ""), "图片名": image.get("图片名", ""), "图片内行号": image.get("图片内行号", ""), "主唯一键字段": unique_key, "主唯一键值": record.get("记录信息", {}).get(unique_key, ""), "复核状态": record.get("复核", {}).get("状态", ""), } def deduplicate_records( records: list[dict[str, Any]], unique_key: str, ) -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]]]: if not unique_key: return records, [], [] kept_by_key: dict[str, dict[str, Any]] = {} order: list[str] = [] missing_records: list[dict[str, Any]] = [] missing_report: list[dict[str, Any]] = [] duplicates: list[dict[str, Any]] = [] for record in records: key = normalize_text(record.get("记录信息", {}).get(unique_key, "")) if not key: missing_records.append(record) missing_report.append({"记录": summarize_record(record, unique_key), "规则": "主唯一键为空"}) continue if key not in kept_by_key: order.append(key) else: duplicates.append( { "主唯一键字段": unique_key, "主唯一键值": key, "保留记录": summarize_record(record, unique_key), "剔除记录": summarize_record(kept_by_key[key], unique_key), "规则": "主唯一键重复，后出现记录覆盖先出现记录", } ) kept_by_key[key] = record return missing_records + [kept_by_key[key] for key in order], duplicates, missing_report def write_csv(path: Path, records: list[dict[str, Any]], fields: list[str]) -> None: fieldnames = [ "处理批次", "业务分类1", "业务分类2", "来源文件夹", "图片路径", "图片名", "图片内行号", *fields, "复核状态", "复核提示", "人工修正", ] path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="utf-8-sig", newline="") as file: writer = csv.DictWriter(file, fieldnames=fieldnames) writer.writeheader() for record in records: image = record.get("图片信息", {}) review = record.get("复核", {}) writer.writerow( { "处理批次": record.get("处理批次", ""), "业务分类1": record.get("业务分类1", ""), "业务分类2": record.get("业务分类2", ""), "来源文件夹": record.get("来源文件夹", ""), "图片路径": image.get("图片路径", ""), "图片名": image.get("图片名", ""), "图片内行号": image.get("图片内行号", ""), **{field: record.get("记录信息", {}).get(field, "") for field in fields}, "复核状态": review.get("状态", ""), "复核提示": "；".join(str(item) for item in review.get("提示", [])), "人工修正": bool(review.get("人工修正")), } ) def write_dict_csv(path: Path, rows: list[dict[str, Any]]) -> None: path.parent.mkdir(parents=True, exist_ok=True) fieldnames: list[str] = [] for row in rows: for key in row: if key not in fieldnames and not isinstance(row.get(key), (list, dict)): fieldnames.append(key) with path.open("w", encoding="utf-8-sig", newline="") as file: if not fieldnames: file.write("") return writer = csv.DictWriter(file, fieldnames=fieldnames) writer.writeheader() for row in rows: writer.writerow({field: row.get(field, "") for field in fieldnames}) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--config", default=str(DEFAULT_CONFIG), help="任务配置 JSON") parser.add_argument("--root", default="数据处理结果区", help="结果根目录") parser.add_argument("--result-suffix", default="", help="批次结果目录后缀") return parser.parse_args() def main() -> None: args = parse_args() config = load_config(Path(args.config)) root = Path(args.root) suffix = args.result_suffix or config.get("result_suffix") or "-列表归档结果" archives: list[dict[str, Any]] = [] batch_summaries: list[dict[str, Any]] = [] raw_records: list[dict[str, Any]] = [] for result_dir in sorted(root.rglob(f"*{suffix}")): archive_path = result_dir / "图片表格_结构化.json" if not archive_path.exists(): continue archive = read_json(archive_path) archives.append(archive) summary = dict(archive.get("汇总", {})) summary["结果目录"] = str(result_dir) batch_summaries.append(summary) raw_records.extend(archive.get("图片表格记录", [])) fields = configured_fields(config, archives) unique_key = normalize_text(config.get("unique_key") or (archives[0].get("任务配置", {}).get("unique_key") if archives else "")) records, duplicate_records, missing_key_records = deduplicate_records(raw_records, unique_key) need_review = [record for record in records if record.get("复核", {}).get("状态") == "需人工复核"] manual_records = [record for record in records if record.get("复核", {}).get("人工修正")] global_summary = { "批次数": len(batch_summaries), "合并前记录数": len(raw_records), "记录数": len(records), "需人工复核记录数": len(need_review), "人工修正记录数": len(manual_records), "重复主键剔除记录数": len(duplicate_records), "缺少主键记录数": len(missing_key_records), "主唯一键": unique_key, "批次汇总": batch_summaries, } merged_archive = { "任务配置": { "project_name": config.get("project_name", ""), "record_name": config.get("record_name", ""), "unique_key": unique_key, "fields": config.get("fields", []), "result_suffix": suffix, }, "汇总": global_summary, "重复主键记录": duplicate_records, "缺少主键记录": missing_key_records, "图片表格记录": records, } info_dir = root / "信息记录" write_json(root / "合并_图片表格_结构化.json", merged_archive) write_jsonl(root / "合并_图片表格_记录.jsonl", records) write_csv(root / "合并_图片表格_记录.csv", records, fields) write_json(info_dir / "全局汇总.json", global_summary) write_json(info_dir / "重复主键报告.json", duplicate_records) write_json(info_dir / "缺少主键报告.json", missing_key_records) write_dict_csv(info_dir / "批次汇总.csv", batch_summaries) print(json.dumps(global_summary, ensure_ascii=False, indent=2), flush=True) if __name__ == "__main__": main()