881 lines
34 KiB
Python
Executable File
881 lines
34 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""通用图片表格 OCR 归档脚本。
|
||
|
||
腾讯云密钥通过环境变量读取：
|
||
TENCENTCLOUD_SECRET_ID 和 TENCENTCLOUD_SECRET_KEY。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import base64
|
||
import csv
|
||
import datetime as dt
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
import math
|
||
import os
|
||
import re
|
||
import subprocess
|
||
import time
|
||
import unicodedata
|
||
import urllib.error
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from PIL import Image
|
||
|
||
|
||
IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".bmp", ".tif", ".tiff"}
|
||
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 natural_key(path: Path) -> tuple[Any, ...]:
|
||
parts = re.split(r"(\d+)", path.stem)
|
||
key: list[Any] = []
|
||
for part in parts:
|
||
key.append(int(part) if part.isdigit() else part)
|
||
return tuple(key)
|
||
|
||
|
||
def safe_filename(value: str) -> str:
|
||
value = normalize_text(value) or "root"
|
||
return re.sub(r'[\\/:*?"<>|]+', "_", value)
|
||
|
||
|
||
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():
|
||
print(f"配置不存在，暂用模板: {TEMPLATE_CONFIG}", flush=True)
|
||
return read_json(TEMPLATE_CONFIG)
|
||
raise FileNotFoundError(f"找不到配置文件: {path}")
|
||
|
||
|
||
def field_names(config: dict[str, Any]) -> list[str]:
|
||
fields = config.get("fields") or []
|
||
names = [normalize_text(field.get("name")) for field in fields if normalize_text(field.get("name"))]
|
||
if not names:
|
||
raise ValueError("配置 fields 不能为空")
|
||
return names
|
||
|
||
|
||
def list_images(folder: Path) -> list[Path]:
|
||
return sorted(
|
||
[path for path in folder.iterdir() if path.is_file() and path.suffix.lower() in IMAGE_EXTENSIONS],
|
||
key=natural_key,
|
||
)
|
||
|
||
|
||
def find_source_folders(input_root: Path) -> list[Path]:
|
||
if list_images(input_root):
|
||
return [input_root]
|
||
|
||
folders = [path for path in input_root.iterdir() if path.is_dir() and list_images(path)]
|
||
if folders:
|
||
return sorted(folders, key=lambda item: natural_key(item))
|
||
|
||
nested = {path.parent for path in input_root.rglob("*") if path.is_file() and path.suffix.lower() in IMAGE_EXTENSIONS}
|
||
return sorted(nested, key=lambda item: str(item))
|
||
|
||
|
||
def batched(items: list[Path], size: int) -> list[list[Path]]:
|
||
size = max(1, int(size))
|
||
return [items[index : index + size] for index in range(0, len(items), size)]
|
||
|
||
|
||
def merge_images(image_paths: list[Path], output_path: Path, padding_y: int) -> dict[str, Any]:
|
||
padding_y = max(0, int(padding_y))
|
||
opened = [Image.open(path).convert("RGB") for path in image_paths]
|
||
try:
|
||
width = max(image.width for image in opened)
|
||
height = sum(image.height + padding_y * 2 for image in opened)
|
||
merged = Image.new("RGB", (width, height), "white")
|
||
y = 0
|
||
source_images: list[dict[str, Any]] = []
|
||
for path, image in zip(image_paths, opened):
|
||
image_y = y + padding_y
|
||
merged.paste(image, (0, image_y))
|
||
source_images.append(
|
||
{
|
||
"path": str(path),
|
||
"name": path.name,
|
||
"width": image.width,
|
||
"height": image.height,
|
||
"y_offset": image_y,
|
||
"block_y_offset": y,
|
||
"block_height": image.height + padding_y * 2,
|
||
"padding_top": padding_y,
|
||
"padding_bottom": padding_y,
|
||
}
|
||
)
|
||
y += image.height + padding_y * 2
|
||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||
merged.save(output_path)
|
||
return {
|
||
"path": str(output_path),
|
||
"width": width,
|
||
"height": height,
|
||
"padding_y": padding_y,
|
||
"source_images": source_images,
|
||
}
|
||
finally:
|
||
for image in opened:
|
||
image.close()
|
||
|
||
|
||
def read_credentials() -> tuple[str, str]:
|
||
secret_id = os.getenv("TENCENTCLOUD_SECRET_ID") or os.getenv("TENCENT_SECRET_ID") or ""
|
||
secret_key = os.getenv("TENCENTCLOUD_SECRET_KEY") or os.getenv("TENCENT_SECRET_KEY") or ""
|
||
return secret_id, secret_key
|
||
|
||
|
||
def tc3_request(
|
||
action: str,
|
||
payload: dict[str, Any],
|
||
secret_id: str,
|
||
secret_key: str,
|
||
region: str,
|
||
timeout: int,
|
||
) -> dict[str, Any]:
|
||
service = "ocr"
|
||
host = "ocr.tencentcloudapi.com"
|
||
endpoint = f"https://{host}"
|
||
version = "2018-11-19"
|
||
body = json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
|
||
algorithm = "TC3-HMAC-SHA256"
|
||
timestamp = int(dt.datetime.now(dt.timezone.utc).timestamp())
|
||
date = dt.datetime.fromtimestamp(timestamp, dt.timezone.utc).strftime("%Y-%m-%d")
|
||
content_type = "application/json; charset=utf-8"
|
||
|
||
canonical_headers = f"content-type:{content_type}\nhost:{host}\n"
|
||
signed_headers = "content-type;host"
|
||
hashed_payload = hashlib.sha256(body.encode("utf-8")).hexdigest()
|
||
canonical_request = "\n".join(["POST", "/", "", canonical_headers, signed_headers, hashed_payload])
|
||
credential_scope = f"{date}/{service}/tc3_request"
|
||
string_to_sign = "\n".join(
|
||
[
|
||
algorithm,
|
||
str(timestamp),
|
||
credential_scope,
|
||
hashlib.sha256(canonical_request.encode("utf-8")).hexdigest(),
|
||
]
|
||
)
|
||
|
||
def sign(key: bytes, message: str) -> bytes:
|
||
return hmac.new(key, message.encode("utf-8"), hashlib.sha256).digest()
|
||
|
||
secret_date = sign(("TC3" + secret_key).encode("utf-8"), date)
|
||
secret_service = sign(secret_date, service)
|
||
secret_signing = sign(secret_service, "tc3_request")
|
||
signature = hmac.new(secret_signing, string_to_sign.encode("utf-8"), hashlib.sha256).hexdigest()
|
||
authorization = (
|
||
f"{algorithm} Credential={secret_id}/{credential_scope}, "
|
||
f"SignedHeaders={signed_headers}, Signature={signature}"
|
||
)
|
||
|
||
headers = {
|
||
"Authorization": authorization,
|
||
"Content-Type": content_type,
|
||
"Host": host,
|
||
"X-TC-Action": action,
|
||
"X-TC-Timestamp": str(timestamp),
|
||
"X-TC-Version": version,
|
||
"X-TC-Region": region,
|
||
}
|
||
command = [
|
||
"curl",
|
||
"-sS",
|
||
"--connect-timeout",
|
||
str(min(10, max(1, timeout))),
|
||
"--max-time",
|
||
str(max(1, timeout)),
|
||
"-X",
|
||
"POST",
|
||
endpoint,
|
||
]
|
||
for key, value in headers.items():
|
||
command.extend(["-H", f"{key}: {value}"])
|
||
command.extend(["--data-binary", "@-"])
|
||
completed = subprocess.run(
|
||
command,
|
||
input=body.encode("utf-8"),
|
||
capture_output=True,
|
||
timeout=max(1, timeout) + 5,
|
||
check=False,
|
||
)
|
||
if completed.returncode != 0:
|
||
error_text = completed.stderr.decode("utf-8", errors="replace").strip()
|
||
raise urllib.error.URLError(error_text or f"curl return code {completed.returncode}")
|
||
return json.loads(completed.stdout.decode("utf-8"))
|
||
|
||
|
||
def call_tencent_table_ocr(
|
||
image_path: Path,
|
||
cache_path: Path,
|
||
secret_id: str,
|
||
secret_key: str,
|
||
region: str,
|
||
timeout: int,
|
||
force: bool,
|
||
max_retries: int,
|
||
) -> dict[str, Any]:
|
||
if cache_path.exists() and not force:
|
||
return read_json(cache_path)
|
||
if not secret_id or not secret_key:
|
||
raise RuntimeError(
|
||
f"OCR缓存不存在，且未设置 TENCENTCLOUD_SECRET_ID / TENCENTCLOUD_SECRET_KEY: {cache_path}"
|
||
)
|
||
|
||
image_base64 = base64.b64encode(image_path.read_bytes()).decode("ascii")
|
||
payload = {"ImageBase64": image_base64, "UseNewModel": True}
|
||
last_error = ""
|
||
for attempt in range(max_retries + 1):
|
||
try:
|
||
data = tc3_request(
|
||
"RecognizeTableAccurateOCR",
|
||
payload,
|
||
secret_id,
|
||
secret_key,
|
||
region,
|
||
timeout,
|
||
)
|
||
response = data.get("Response", {})
|
||
if "Error" in response:
|
||
raise RuntimeError(json.dumps(response["Error"], ensure_ascii=False))
|
||
cache_path.parent.mkdir(parents=True, exist_ok=True)
|
||
write_json(cache_path, response)
|
||
return response
|
||
except (urllib.error.URLError, TimeoutError, OSError, RuntimeError) as exc:
|
||
last_error = str(exc)
|
||
if attempt >= max_retries:
|
||
break
|
||
time.sleep(2**attempt)
|
||
raise RuntimeError(f"OCR 调用失败: {image_path} {last_error}")
|
||
|
||
|
||
def cells_to_rows(response: dict[str, Any], expected_columns: int) -> list[list[str]]:
|
||
cells: list[dict[str, Any]] = []
|
||
for table in response.get("TableDetections") or []:
|
||
cells.extend(table.get("Cells") or [])
|
||
if not cells:
|
||
return []
|
||
|
||
max_row = max(int(cell.get("RowTl", 0) or 0) for cell in cells)
|
||
max_col = max(int(cell.get("ColTl", 0) or 0) for cell in cells)
|
||
column_count = max(expected_columns, max_col + 1)
|
||
rows = [["" for _ in range(column_count)] for _ in range(max_row + 1)]
|
||
for cell in cells:
|
||
row_index = int(cell.get("RowTl", 0) or 0)
|
||
col_index = int(cell.get("ColTl", 0) or 0)
|
||
text = normalize_text(cell.get("Text", ""))
|
||
if rows[row_index][col_index]:
|
||
rows[row_index][col_index] = normalize_text(rows[row_index][col_index] + " " + text)
|
||
else:
|
||
rows[row_index][col_index] = text
|
||
return [row[:expected_columns] + [""] * max(0, expected_columns - len(row)) for row in rows]
|
||
|
||
|
||
def looks_like_header(row: list[str], fields: list[str]) -> bool:
|
||
normalized_row = [normalize_text(item) for item in row]
|
||
hits = sum(1 for field in fields if field in normalized_row)
|
||
return hits >= max(2, math.ceil(len(fields) * 0.5))
|
||
|
||
|
||
def normalize_date_like(text: str, target_type: str) -> str:
|
||
text = normalize_text(text).replace("/", "-").replace(".", "-")
|
||
match = re.fullmatch(
|
||
r"(\d{4})-(\d{1,2})-(\d{1,2})(?:\s+(\d{1,2}):(\d{1,2})(?::(\d{1,2}))?)?",
|
||
text,
|
||
)
|
||
if not match:
|
||
return text
|
||
year, month, day, hour, minute, second = match.groups()
|
||
date_part = f"{int(year):04d}-{int(month):02d}-{int(day):02d}"
|
||
if target_type == "date":
|
||
return date_part
|
||
if hour is None:
|
||
return date_part
|
||
return f"{date_part} {int(hour):02d}:{int(minute or 0):02d}:{int(second or 0):02d}"
|
||
|
||
|
||
def clean_field_value(value: Any, field: dict[str, Any]) -> Any:
|
||
text = normalize_text(value)
|
||
for rule in field.get("clean") or []:
|
||
if rule == "remove_spaces":
|
||
text = re.sub(r"\s+", "", text)
|
||
elif rule == "upper":
|
||
text = text.upper()
|
||
elif rule == "lower":
|
||
text = text.lower()
|
||
elif rule == "strip":
|
||
text = text.strip()
|
||
field_type = normalize_text(field.get("type") or "text").lower()
|
||
if field_type in {"date", "datetime"}:
|
||
text = normalize_date_like(text, field_type)
|
||
if field_type == "integer" and re.fullmatch(r"\d+", text):
|
||
return int(text)
|
||
if field_type == "number":
|
||
try:
|
||
return float(text) if text else ""
|
||
except ValueError:
|
||
return text
|
||
return text
|
||
|
||
|
||
def row_to_record_info(row: list[str], fields: list[dict[str, Any]]) -> dict[str, Any]:
|
||
values: dict[str, Any] = {}
|
||
for index, field in enumerate(fields):
|
||
name = normalize_text(field.get("name"))
|
||
values[name] = clean_field_value(row[index] if index < len(row) else "", field)
|
||
return values
|
||
|
||
|
||
def validate_record_info(
|
||
record_info: dict[str, Any],
|
||
fields: list[dict[str, Any]],
|
||
unique_key: str,
|
||
) -> list[str]:
|
||
warnings: list[str] = []
|
||
for field in fields:
|
||
name = normalize_text(field.get("name"))
|
||
value = record_info.get(name, "")
|
||
text = normalize_text(value)
|
||
field_type = normalize_text(field.get("type") or "text").lower()
|
||
if field.get("required") and not text:
|
||
warnings.append(f"缺少{name}")
|
||
if text and field_type == "integer" and not isinstance(value, int):
|
||
warnings.append(f"{name}非整数")
|
||
if text and field_type == "number":
|
||
try:
|
||
float(text)
|
||
except ValueError:
|
||
warnings.append(f"{name}非数字")
|
||
if text and field_type == "date" and not re.fullmatch(r"\d{4}-\d{2}-\d{2}", text):
|
||
warnings.append(f"{name}日期格式异常")
|
||
if text and field_type == "datetime" and not re.fullmatch(r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}", text):
|
||
warnings.append(f"{name}时间格式异常")
|
||
pattern = normalize_text(field.get("pattern"))
|
||
if text and pattern and not re.fullmatch(pattern, text):
|
||
warnings.append(f"{name}格式不符合规则")
|
||
if unique_key and not normalize_text(record_info.get(unique_key, "")):
|
||
warnings.append(f"缺少主唯一键: {unique_key}")
|
||
return warnings
|
||
|
||
|
||
def is_blank_record(record_info: dict[str, Any]) -> bool:
|
||
return all(value in ("", None) for value in record_info.values())
|
||
|
||
|
||
def classify_folder(folder: Path, config: dict[str, Any]) -> dict[str, str]:
|
||
classification = config.get("classification") or {}
|
||
default = classification.get("default") or {}
|
||
result = {
|
||
"业务分类1": normalize_text(default.get("业务分类1") or "未分类"),
|
||
"业务分类2": normalize_text(default.get("业务分类2") or "未分类"),
|
||
}
|
||
if not classification.get("enabled", True):
|
||
return result
|
||
folder_name = normalize_text(folder.name)
|
||
for rule in classification.get("folder_rules") or []:
|
||
contains = normalize_text(rule.get("contains"))
|
||
equals = normalize_text(rule.get("equals"))
|
||
if (contains and contains in folder_name) or (equals and equals == folder_name):
|
||
result["业务分类1"] = normalize_text(rule.get("业务分类1") or result["业务分类1"])
|
||
result["业务分类2"] = normalize_text(rule.get("业务分类2") or result["业务分类2"])
|
||
return result
|
||
return result
|
||
|
||
|
||
def infer_rows_for_image(image_path: Path, rows_per_image: int, row_height_px: float) -> int:
|
||
if rows_per_image > 0:
|
||
return rows_per_image
|
||
if row_height_px > 0:
|
||
with Image.open(image_path) as image:
|
||
return max(1, round(image.height / row_height_px))
|
||
return 0
|
||
|
||
|
||
def locate_source_row(
|
||
row_index: int,
|
||
row_counts: list[int],
|
||
total_rows: int,
|
||
image_count: int,
|
||
) -> tuple[int, int]:
|
||
if sum(row_counts) > 0:
|
||
offset = 0
|
||
for image_index, row_count in enumerate(row_counts):
|
||
if row_index < offset + row_count:
|
||
return image_index, row_index - offset
|
||
offset += row_count
|
||
return len(row_counts) - 1, max(0, row_index - sum(row_counts[:-1]))
|
||
rows_per_image = max(1, math.ceil(max(1, total_rows) / max(1, image_count)))
|
||
image_index = min(image_count - 1, row_index // rows_per_image)
|
||
return image_index, row_index - image_index * rows_per_image
|
||
|
||
|
||
def correction_keys(image_path: str, row_no: int) -> list[tuple[str, int]]:
|
||
path = normalize_text(image_path)
|
||
return [
|
||
(path, row_no),
|
||
(Path(path).name, row_no),
|
||
]
|
||
|
||
|
||
def load_corrections(path: Path) -> dict[tuple[str, int], dict[str, Any]]:
|
||
if not path.exists():
|
||
return {}
|
||
data = read_json(path)
|
||
if isinstance(data, dict):
|
||
items = data.get("records") or data.get("修正记录") or []
|
||
else:
|
||
items = data
|
||
corrections: dict[tuple[str, int], dict[str, Any]] = {}
|
||
for item in items:
|
||
image_path = normalize_text(item.get("图片路径"))
|
||
row_no = int(item.get("图片内行号") or 0)
|
||
if not image_path or row_no <= 0:
|
||
continue
|
||
for key in correction_keys(image_path, row_no):
|
||
corrections[key] = item
|
||
return corrections
|
||
|
||
|
||
def apply_corrections(
|
||
records: list[dict[str, Any]],
|
||
corrections: dict[tuple[str, int], dict[str, Any]],
|
||
fields: list[dict[str, Any]],
|
||
unique_key: str,
|
||
) -> None:
|
||
for record in records:
|
||
image_path = record["图片信息"]["图片路径"]
|
||
row_no = int(record["图片信息"]["图片内行号"])
|
||
correction = None
|
||
for key in correction_keys(image_path, row_no):
|
||
if key in corrections:
|
||
correction = corrections[key]
|
||
break
|
||
if correction:
|
||
record_info = correction.get("记录信息") or correction.get("患者信息") or {}
|
||
record["记录信息"].update(record_info)
|
||
record["复核"]["人工修正"] = True
|
||
if correction.get("复核备注"):
|
||
record["复核"]["人工备注"] = correction.get("复核备注")
|
||
warnings = validate_record_info(record["记录信息"], fields, unique_key)
|
||
record["复核"]["提示"] = warnings
|
||
if warnings:
|
||
record["复核"]["状态"] = "需人工复核"
|
||
elif record["复核"].get("人工修正"):
|
||
record["复核"]["状态"] = "人工复核通过"
|
||
else:
|
||
record["复核"]["状态"] = "自动复核通过"
|
||
|
||
|
||
def summarize_record(record: dict[str, Any], unique_key: str) -> dict[str, Any]:
|
||
image = record["图片信息"]
|
||
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: list[dict[str, Any]] = []
|
||
missing_records: list[dict[str, Any]] = []
|
||
duplicates: list[dict[str, Any]] = []
|
||
for record in records:
|
||
key = normalize_text(record["记录信息"].get(unique_key, ""))
|
||
if not key:
|
||
missing.append({"记录": summarize_record(record, unique_key), "规则": "主唯一键为空"})
|
||
missing_records.append(record)
|
||
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
|
||
|
||
|
||
def records_to_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["图片信息"]
|
||
review = record["复核"]
|
||
writer.writerow(
|
||
{
|
||
"处理批次": record.get("处理批次", ""),
|
||
"业务分类1": record.get("业务分类1", ""),
|
||
"业务分类2": record.get("业务分类2", ""),
|
||
"来源文件夹": record.get("来源文件夹", ""),
|
||
"图片路径": image.get("图片路径", ""),
|
||
"图片名": image.get("图片名", ""),
|
||
"图片内行号": image.get("图片内行号", ""),
|
||
**{field: record["记录信息"].get(field, "") for field in fields},
|
||
"复核状态": review.get("状态", ""),
|
||
"复核提示": "；".join(str(item) for item in review.get("提示", [])),
|
||
"人工修正": bool(review.get("人工修正")),
|
||
}
|
||
)
|
||
|
||
|
||
def build_records_from_rows(
|
||
rows: list[list[str]],
|
||
group_paths: list[Path],
|
||
fields: list[dict[str, Any]],
|
||
config: dict[str, Any],
|
||
batch_name: str,
|
||
source_folder: Path,
|
||
categories: dict[str, str],
|
||
group_index: int,
|
||
merge_info: dict[str, Any],
|
||
cache_path: Path,
|
||
request_id: str,
|
||
) -> list[dict[str, Any]]:
|
||
ocr_config = config.get("ocr") or {}
|
||
names = field_names(config)
|
||
indexed_rows = list(enumerate(rows))
|
||
skip_header_rows = int(ocr_config.get("skip_header_rows") or 0)
|
||
if skip_header_rows > 0:
|
||
indexed_rows = indexed_rows[skip_header_rows:]
|
||
elif ocr_config.get("auto_skip_header", True):
|
||
indexed_rows = [(index, row) for index, row in indexed_rows if not looks_like_header(row, names)]
|
||
|
||
rows_per_image = int(ocr_config.get("rows_per_image") or 0)
|
||
row_height_px = float(ocr_config.get("row_height_px") or 0)
|
||
row_counts = [infer_rows_for_image(path, rows_per_image, row_height_px) for path in group_paths]
|
||
total_rows = len(indexed_rows)
|
||
records: list[dict[str, Any]] = []
|
||
unique_key = normalize_text(config.get("unique_key"))
|
||
for output_row_index, (_raw_row_index, row) in enumerate(indexed_rows):
|
||
record_info = row_to_record_info(row, fields)
|
||
if is_blank_record(record_info):
|
||
continue
|
||
source_index, image_row = locate_source_row(output_row_index, row_counts, total_rows, len(group_paths))
|
||
source_path = group_paths[source_index]
|
||
warnings = validate_record_info(record_info, fields, unique_key)
|
||
records.append(
|
||
{
|
||
"处理批次": batch_name,
|
||
"业务分类1": categories.get("业务分类1", ""),
|
||
"业务分类2": categories.get("业务分类2", ""),
|
||
"来源文件夹": source_folder.name,
|
||
"记录信息": record_info,
|
||
"图片信息": {
|
||
"图片路径": str(source_path),
|
||
"图片名": source_path.name,
|
||
"图片序号": list(natural_key(source_path)),
|
||
"图片内行号": image_row + 1,
|
||
"拼接组序号": group_index,
|
||
"拼接图片路径": merge_info.get("path", ""),
|
||
"OCR缓存路径": str(cache_path),
|
||
"OCR请求ID": request_id,
|
||
},
|
||
"复核": {
|
||
"状态": "需人工复核" if warnings else "自动复核通过",
|
||
"提示": warnings,
|
||
"人工修正": False,
|
||
},
|
||
}
|
||
)
|
||
return records
|
||
|
||
|
||
def process_folder(
|
||
folder: Path,
|
||
output_root: Path,
|
||
config: dict[str, Any],
|
||
args: argparse.Namespace,
|
||
secret_id: str,
|
||
secret_key: str,
|
||
batch_name: str,
|
||
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
|
||
ocr_config = config.get("ocr") or {}
|
||
fields = config.get("fields") or []
|
||
images = list_images(folder)
|
||
batch_size = args.batch_size or int(ocr_config.get("batch_size") or 6)
|
||
padding_y = args.image_padding_y if args.image_padding_y is not None else int(ocr_config.get("image_padding_y") or 0)
|
||
timeout = args.timeout or int(ocr_config.get("timeout") or 90)
|
||
max_retries = args.max_retries if args.max_retries is not None else int(ocr_config.get("max_retries") or 1)
|
||
min_row_ratio = float(ocr_config.get("min_row_ratio") or 0)
|
||
region = args.region or os.getenv("TENCENTCLOUD_REGION") or ocr_config.get("region") or "ap-shanghai"
|
||
sleep_seconds = args.sleep if args.sleep is not None else float(ocr_config.get("sleep") or 0)
|
||
|
||
folder_key = safe_filename(folder.name)
|
||
merged_dir = output_root / "merged_images" / folder_key
|
||
raw_dir = output_root / "raw_ocr" / folder_key
|
||
categories = classify_folder(folder, config)
|
||
records: list[dict[str, Any]] = []
|
||
group_infos: list[dict[str, Any]] = []
|
||
errors: list[dict[str, Any]] = []
|
||
|
||
def attempt_group(group: list[Path], group_index: int, label: str) -> tuple[dict[str, Any], list[dict[str, Any]]]:
|
||
merged_path = merged_dir / f"{label}.png"
|
||
cache_path = raw_dir / f"{label}.json"
|
||
merge_info = merge_images(group, merged_path, padding_y)
|
||
print(f" OCR: {folder.name} / {label} / {len(group)} 张", flush=True)
|
||
response = call_tencent_table_ocr(
|
||
merged_path,
|
||
cache_path,
|
||
secret_id,
|
||
secret_key,
|
||
region,
|
||
timeout,
|
||
args.force,
|
||
max_retries,
|
||
)
|
||
rows = cells_to_rows(response, len(fields))
|
||
rows_per_image = int(ocr_config.get("rows_per_image") or 0)
|
||
row_height_px = float(ocr_config.get("row_height_px") or 0)
|
||
expected_rows = sum(infer_rows_for_image(path, rows_per_image, row_height_px) for path in group)
|
||
if expected_rows > 0 and len(group) > 1 and len(rows) < expected_rows * min_row_ratio:
|
||
raise RuntimeError(f"识别行数偏少: {len(rows)} / {expected_rows}")
|
||
request_id = normalize_text(response.get("RequestId"))
|
||
built_records = build_records_from_rows(
|
||
rows,
|
||
group,
|
||
fields,
|
||
config,
|
||
batch_name,
|
||
folder,
|
||
categories,
|
||
group_index,
|
||
merge_info,
|
||
cache_path,
|
||
request_id,
|
||
)
|
||
info = {
|
||
"拼接组序号": group_index,
|
||
"标签": label,
|
||
"图片数": len(group),
|
||
"识别行数": len(rows),
|
||
"生成记录数": len(built_records),
|
||
"预估行数": expected_rows,
|
||
"拼接图片路径": str(merged_path),
|
||
"OCR缓存路径": str(cache_path),
|
||
"OCR请求ID": request_id,
|
||
}
|
||
time.sleep(max(0, sleep_seconds))
|
||
return info, built_records
|
||
|
||
groups = batched(images, batch_size)
|
||
if args.limit_groups_per_folder:
|
||
groups = groups[: args.limit_groups_per_folder]
|
||
for group_index, group in enumerate(groups, start=1):
|
||
label = f"{folder_key}_group_{group_index:04d}_n{len(group)}_pady{padding_y}"
|
||
try:
|
||
info, group_records = attempt_group(group, group_index, label)
|
||
group_infos.append(info)
|
||
records.extend(group_records)
|
||
except Exception as exc:
|
||
message = str(exc)
|
||
errors.append({"拼接组序号": group_index, "标签": label, "错误": message})
|
||
print(f" 拼接组失败，尝试单张回退: {message}", flush=True)
|
||
if len(group) == 1:
|
||
continue
|
||
for part_index, image_path in enumerate(group):
|
||
single_label = f"{label}_part_{part_index:02d}"
|
||
try:
|
||
info, single_records = attempt_group([image_path], group_index, single_label)
|
||
info["回退"] = "single"
|
||
group_infos.append(info)
|
||
records.extend(single_records)
|
||
except Exception as single_exc:
|
||
errors.append(
|
||
{
|
||
"拼接组序号": group_index,
|
||
"标签": single_label,
|
||
"图片": str(image_path),
|
||
"错误": str(single_exc),
|
||
}
|
||
)
|
||
|
||
summary = {
|
||
"来源文件夹": folder.name,
|
||
"业务分类1": categories.get("业务分类1", ""),
|
||
"业务分类2": categories.get("业务分类2", ""),
|
||
"图片数": len(images),
|
||
"拼接组数": len(group_infos),
|
||
"记录数": len(records),
|
||
"错误数": len(errors),
|
||
"拼接组": group_infos,
|
||
"错误": errors,
|
||
}
|
||
return records, summary
|
||
|
||
|
||
def parse_args() -> argparse.Namespace:
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument("--config", default=str(DEFAULT_CONFIG), help="任务配置 JSON")
|
||
parser.add_argument("--input", required=True, help="待处理图片批次目录")
|
||
parser.add_argument("--output", required=True, help="批次输出目录")
|
||
parser.add_argument("--corrections", default="数据处理工作区/03_人工复核修正.json", help="人工修正 JSON")
|
||
parser.add_argument("--batch-name", default="", help="处理批次名；默认使用 input 目录名")
|
||
parser.add_argument("--ocr-engine", default="", help="当前模板仅支持 table-v3")
|
||
parser.add_argument("--region", default="", help="腾讯云 OCR 地域")
|
||
parser.add_argument("--batch-size", type=int, default=0, help="每组拼接图片数")
|
||
parser.add_argument("--image-padding-y", type=int, default=None, help="每张图上下白边像素")
|
||
parser.add_argument("--timeout", type=int, default=0, help="单次 OCR 超时秒数")
|
||
parser.add_argument("--sleep", type=float, default=None, help="OCR 调用间隔秒数")
|
||
parser.add_argument("--max-retries", type=int, default=None, help="OCR 失败重试次数")
|
||
parser.add_argument("--force", action="store_true", help="忽略 OCR 缓存重新识别")
|
||
parser.add_argument("--rebuild-from-cache", action="store_true", help="只用已有 OCR 缓存重建结果")
|
||
parser.add_argument("--limit-folders", type=int, default=0, help="调试：只处理前 N 个来源文件夹")
|
||
parser.add_argument("--limit-groups-per-folder", type=int, default=0, help="调试：每个文件夹只处理前 N 个拼接组")
|
||
return parser.parse_args()
|
||
|
||
|
||
def main() -> None:
|
||
args = parse_args()
|
||
config = load_config(Path(args.config))
|
||
engine = args.ocr_engine or (config.get("ocr") or {}).get("engine", "table-v3")
|
||
if engine != "table-v3":
|
||
raise ValueError("当前通用模板仅实现 table-v3，即腾讯云 RecognizeTableAccurateOCR")
|
||
|
||
input_root = Path(args.input)
|
||
output_root = Path(args.output)
|
||
batch_name = args.batch_name or input_root.name
|
||
if not input_root.exists():
|
||
raise FileNotFoundError(f"输入目录不存在: {input_root}")
|
||
output_root.mkdir(parents=True, exist_ok=True)
|
||
|
||
secret_id, secret_key = ("", "") if args.rebuild_from_cache else read_credentials()
|
||
folders = find_source_folders(input_root)
|
||
if args.limit_folders:
|
||
folders = folders[: args.limit_folders]
|
||
if not folders:
|
||
raise RuntimeError(f"未发现图片文件: {input_root}")
|
||
|
||
all_records: list[dict[str, Any]] = []
|
||
folder_summaries: list[dict[str, Any]] = []
|
||
for folder in folders:
|
||
print(f"处理来源文件夹: {folder}", flush=True)
|
||
folder_records, summary = process_folder(folder, output_root, config, args, secret_id, secret_key, batch_name)
|
||
all_records.extend(folder_records)
|
||
folder_summaries.append(summary)
|
||
|
||
fields = config.get("fields") or []
|
||
names = field_names(config)
|
||
unique_key = normalize_text(config.get("unique_key"))
|
||
corrections = load_corrections(Path(args.corrections))
|
||
apply_corrections(all_records, corrections, fields, unique_key)
|
||
kept_records, duplicate_records, missing_key_records = deduplicate_records(all_records, unique_key)
|
||
|
||
need_review = [record for record in kept_records if record["复核"]["状态"] == "需人工复核"]
|
||
manual_records = [record for record in kept_records if record["复核"].get("人工修正")]
|
||
error_count = sum(int(summary.get("错误数", 0)) for summary in folder_summaries)
|
||
summary = {
|
||
"项目名称": config.get("project_name", ""),
|
||
"记录对象": config.get("record_name", ""),
|
||
"处理批次": batch_name,
|
||
"来源文件夹数": len(folder_summaries),
|
||
"图片数": sum(int(item.get("图片数", 0)) for item in folder_summaries),
|
||
"去重前记录数": len(all_records),
|
||
"记录数": len(kept_records),
|
||
"需人工复核记录数": len(need_review),
|
||
"人工修正记录数": len(manual_records),
|
||
"重复主键剔除记录数": len(duplicate_records),
|
||
"缺少主键记录数": len(missing_key_records),
|
||
"错误数": error_count,
|
||
"主唯一键": unique_key,
|
||
}
|
||
archive = {
|
||
"任务配置": {
|
||
"project_name": config.get("project_name", ""),
|
||
"record_name": config.get("record_name", ""),
|
||
"unique_key": unique_key,
|
||
"fields": fields,
|
||
"result_suffix": config.get("result_suffix", "-列表归档结果"),
|
||
},
|
||
"汇总": summary,
|
||
"来源文件夹汇总": folder_summaries,
|
||
"重复主键记录": duplicate_records,
|
||
"缺少主键记录": missing_key_records,
|
||
"图片表格记录": kept_records,
|
||
}
|
||
review_report = {
|
||
"汇总": summary,
|
||
"需人工复核记录": need_review,
|
||
"人工修正记录": manual_records,
|
||
"重复主键记录": duplicate_records,
|
||
"缺少主键记录": missing_key_records,
|
||
"来源文件夹汇总": folder_summaries,
|
||
}
|
||
|
||
write_json(output_root / "图片表格_结构化.json", archive)
|
||
write_jsonl(output_root / "图片表格_记录.jsonl", kept_records)
|
||
records_to_csv(output_root / "图片表格_记录.csv", kept_records, names)
|
||
write_json(output_root / "复核报告.json", review_report)
|
||
write_json(output_root / "重复主键报告.json", duplicate_records)
|
||
write_json(output_root / "缺少主键报告.json", missing_key_records)
|
||
write_json(output_root / "信息记录" / "汇总.json", summary)
|
||
write_json(output_root / "信息记录" / "来源文件夹汇总.json", folder_summaries)
|
||
print(json.dumps(summary, ensure_ascii=False, indent=2), flush=True)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|