239 lines
9.3 KiB
Python
Executable File
239 lines
9.3 KiB
Python
Executable File
#!/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 "-列表归档结果"
|
||
official_root = root / normalize_text(config.get("processed_input_root"))
|
||
search_root = official_root if official_root.exists() else root
|
||
archives: list[dict[str, Any]] = []
|
||
batch_summaries: list[dict[str, Any]] = []
|
||
raw_records: list[dict[str, Any]] = []
|
||
|
||
for result_dir in sorted(search_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()
|