admin/PACS
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Add PACS UPP OCR workflow

Browse Source
This commit is contained in:
Codex
2026-05-25 12:33:24 +08:00
commit 70215ce611
13 changed files with 2347 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
.gitignore vendored Normal file
View File

@@ -0,0 +1,18 @@
.env
*.env
__pycache__/
*.pyc
# 原始图片、预处理图片、OCR 缓存和结果数据不提交
UPP列表处理/待处理-*图片集群/
UPP列表处理/已处理-*图片集群/
UPP列表处理/数据处理结果区/
# 本地专用流程和人工修正实数据不提交
UPP列表处理/工作流_本地使用版.md
UPP列表处理/数据处理工作区/03_人工复核修正.json
# 本地数据库配置或导出不提交
UPP列表处理/数据处理工作区/06_PostgreSQL建表结构.sql
*.dump
*.sql.gz

0
UPP列表处理/README.md Normal file
View File

0
UPP列表处理/工作流_Gitea版.md Normal file
View File

135
UPP列表处理/数据处理工作区/01_任务配置.json Normal file
View File

@@ -0,0 +1,135 @@
{
"project_name": "PACS UPP列表处理",
"record_name": "UPP列表记录",
"input_root": "待处理-UPP列表图片集群",
"processed_input_root": "已处理-UPP列表图片集群",
"result_root": "数据处理结果区",
"result_suffix": "-列表归档结果",
"unique_key": "",
"missing_unique_key_action": "keep_for_review",
"duplicate_unique_key_action": "keep_later",
"fields": [
{
"name": "姓名",
"required": true,
"type": "text",
"clean": ["strip"],
"pattern": "",
"description": "患者姓名，按截图显示保留拼音或符号。"
},
{
"name": "性别",
"required": true,
"type": "text",
"clean": ["strip", "upper"],
"pattern": "^[MF男女]$",
"description": "性别。"
},
{
"name": "年龄",
"required": false,
"type": "text",
"clean": ["remove_spaces", "upper"],
"pattern": "",
"description": "年龄，保留原始单位，例如 049Y。"
},
{
"name": "患者号",
"required": false,
"type": "text",
"clean": ["remove_spaces"],
"pattern": "",
"description": "患者号，截图中可能带省略号。"
},
{
"name": "检查号",
"required": true,
"type": "text",
"clean": ["remove_spaces", "upper"],
"pattern": "",
"description": "检查号。"
},
{
"name": "检查日期",
"required": false,
"type": "datetime",
"clean": ["strip"],
"pattern": "",
"description": "检查日期时间。"
},
{
"name": "任务创建时间",
"required": false,
"type": "datetime",
"clean": ["strip"],
"pattern": "",
"description": "UPP 任务创建时间。"
},
{
"name": "检查描述",
"required": false,
"type": "text",
"clean": ["strip"],
"pattern": "",
"description": "检查描述。"
},
{
"name": "检查设备",
"required": false,
"type": "text",
"clean": ["strip", "upper"],
"pattern": "",
"description": "检查设备。"
},
{
"name": "算法模型",
"required": false,
"type": "text",
"clean": ["strip"],
"pattern": "",
"description": "算法模型。"
},
{
"name": "状态",
"required": false,
"type": "text",
"clean": ["strip"],
"pattern": "",
"description": "处理状态。"
}
],
"classification": {
"enabled": true,
"category_1_name": "业务分类1",
"category_2_name": "业务分类2",
"default": {
"业务分类1": "PACS",
"业务分类2": "UPP列表"
},
"folder_rules": [
{
"contains": "肝胆外科",
"业务分类1": "PACS",
"业务分类2": "肝胆外科"
}
]
},
"ocr": {
"engine": "table-v3",
"region": "ap-shanghai",
"batch_size": 1,
"image_padding_y": 0,
"rows_per_image": 20,
"row_height_px": 0,
"skip_header_rows": 0,
"auto_skip_header": true,
"min_row_ratio": 0.85,
"timeout": 90,
"sleep": 0.2,
"max_retries": 1
},
"postgres": {
"table_name": "PACS_UPP_List_Records",
"env_prefix": "WORKFLOW_DB"
}
}

78
UPP列表处理/数据处理工作区/01_任务配置.template.json Normal file
View File

@@ -0,0 +1,78 @@
{
"project_name": "通用图片表格识别任务",
"record_name": "图片表格记录",
"input_root": "待处理-[任务名]图片集群",
"processed_input_root": "已处理-[任务名]图片集群",
"result_root": "数据处理结果区",
"result_suffix": "-列表归档结果",
"unique_key": "记录编号",
"missing_unique_key_action": "keep_for_review",
"duplicate_unique_key_action": "keep_later",
"fields": [
{
"name": "记录编号",
"required": true,
"type": "text",
"clean": [
"remove_spaces",
"upper"
],
"pattern": "",
"description": "主唯一键。请按具体任务改成住院号、检查号、accession_no 或组合键字段。"
},
{
"name": "字段1",
"required": false,
"type": "text",
"clean": [
"strip"
],
"pattern": "",
"description": "待替换字段"
},
{
"name": "字段2",
"required": false,
"type": "text",
"clean": [
"strip"
],
"pattern": "",
"description": "待替换字段"
}
],
"classification": {
"enabled": true,
"category_1_name": "业务分类1",
"category_2_name": "业务分类2",
"default": {
"业务分类1": "未分类",
"业务分类2": "未分类"
},
"folder_rules": [
{
"contains": "示例",
"业务分类1": "示例大类",
"业务分类2": "示例小类"
}
]
},
"ocr": {
"engine": "table-v3",
"region": "ap-shanghai",
"batch_size": 6,
"image_padding_y": 24,
"rows_per_image": 0,
"row_height_px": 0,
"skip_header_rows": 0,
"auto_skip_header": true,
"min_row_ratio": 0.85,
"timeout": 90,
"sleep": 0.2,
"max_retries": 1
},
"postgres": {
"table_name": "Image_Table_Records",
"env_prefix": "WORKFLOW_DB"
}
}

880
UPP列表处理/数据处理工作区/02_图片表格OCR归档.py Executable file
View File

@@ -0,0 +1,880 @@
#!/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()

13
UPP列表处理/数据处理工作区/03_人工复核修正.template.json Normal file
View File

@@ -0,0 +1,13 @@
[
{
"图片路径": "待处理-[任务名]图片集群/[批次文件夹名]/[图片名].png",
"图片内行号": 1,
"记录信息": {
"记录编号": "人工确认值",
"字段1": "人工确认值",
"字段2": "人工确认值"
},
"复核选项": {},
"复核备注": "可选：说明为什么修正"
}
]

236
UPP列表处理/数据处理工作区/04_合并批次结果.py Executable file
View File

@@ -0,0 +1,236 @@
#!/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()

192
UPP列表处理/数据处理工作区/05_同步PostgreSQL单表.py Executable file
View File

@@ -0,0 +1,192 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""把通用图片表格 OCR 合并结果同步到 PostgreSQL 单表。"""
from __future__ import annotations
import argparse
import csv
import json
import os
import re
import subprocess
import tempfile
import unicodedata
from pathlib import Path
from typing import Any
DEFAULT_INPUT = Path("数据处理结果区/合并_图片表格_结构化.json")
DEFAULT_SCHEMA = Path("数据处理工作区/06_PostgreSQL建表结构.sql")
TEMPLATE_SCHEMA = Path("数据处理工作区/06_PostgreSQL建表结构.template.sql")
CSV_FIELDS = [
"batch_name",
"source_folder",
"category_1",
"category_2",
"image_path",
"image_name",
"image_row_no",
"unique_key",
"record_data",
"review_status",
"review_notes",
"manual_corrected",
]
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 scalar(value: Any) -> str:
if value is None:
return ""
if isinstance(value, bool):
return "true" if value else "false"
return str(value)
def sql_file(path: Path) -> Path:
if path.exists():
return path
if TEMPLATE_SCHEMA.exists():
return TEMPLATE_SCHEMA
raise FileNotFoundError(f"找不到 SQL 文件: {path}")
def quote_file(path: Path) -> str:
return "'" + str(path.resolve()).replace("'", "''") + "'"
def quote_table_name(name: str) -> str:
name = normalize_text(name)
if name.startswith('"') and name.endswith('"'):
return name
if re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", name):
return f'"{name}"'
raise ValueError(f"表名需要是简单标识符或已加双引号: {name}")
def flatten_record(record: dict[str, Any], unique_key_field: str) -> dict[str, Any]:
image = record.get("图片信息", {})
review = record.get("复核", {})
record_info = record.get("记录信息", {})
unique_key = normalize_text(record_info.get(unique_key_field, ""))
return {
"batch_name": record.get("处理批次", ""),
"source_folder": record.get("来源文件夹", ""),
"category_1": record.get("业务分类1", ""),
"category_2": record.get("业务分类2", ""),
"image_path": image.get("图片路径", ""),
"image_name": image.get("图片名", ""),
"image_row_no": image.get("图片内行号", ""),
"unique_key": unique_key,
"record_data": json.dumps(record_info, ensure_ascii=False, separators=(",", ":")),
"review_status": review.get("状态", ""),
"review_notes": "；".join(str(item) for item in review.get("提示", [])),
"manual_corrected": bool(review.get("人工修正")),
}
def load_rows(input_path: Path, allow_empty_key: bool) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
archive = read_json(input_path)
unique_key = normalize_text(archive.get("任务配置", {}).get("unique_key") or archive.get("汇总", {}).get("主唯一键"))
if not unique_key:
raise ValueError("合并结果中没有主唯一键配置，无法同步到默认 PostgreSQL 模板")
rows: list[dict[str, Any]] = []
skipped: list[dict[str, Any]] = []
for record in archive.get("图片表格记录", []):
row = flatten_record(record, unique_key)
if not row["unique_key"] and not allow_empty_key:
skipped.append(row)
continue
rows.append(row)
return rows, skipped
def write_temp_csv(rows: list[dict[str, Any]]) -> Path:
temp_file = tempfile.NamedTemporaryFile(
mode="w",
encoding="utf-8",
newline="",
suffix="_image_table_records.csv",
delete=False,
)
with temp_file:
writer = csv.DictWriter(temp_file, fieldnames=CSV_FIELDS)
writer.writeheader()
for row in rows:
writer.writerow({field: scalar(row.get(field, "")) for field in CSV_FIELDS})
return Path(temp_file.name)
def sync_to_postgres(args: argparse.Namespace, csv_path: Path, row_count: int) -> None:
table_name = quote_table_name(args.table)
truncate_sql = "" if args.append else f"TRUNCATE TABLE {table_name} RESTART IDENTITY;\n"
sql = f"""\\set ON_ERROR_STOP on
\\i {quote_file(sql_file(Path(args.schema)))}
{truncate_sql}\\copy {table_name}({','.join(CSV_FIELDS)}) FROM {quote_file(csv_path)} WITH (FORMAT csv, HEADER true, NULL '')
"""
env = os.environ.copy()
if args.password:
env["PGPASSWORD"] = args.password
elif env.get("WORKFLOW_DB_PASSWORD"):
env["PGPASSWORD"] = env["WORKFLOW_DB_PASSWORD"]
command = [
"psql",
"-h",
args.host,
"-p",
str(args.port),
"-U",
args.user,
"-d",
args.dbname,
"-v",
"ON_ERROR_STOP=1",
]
completed = subprocess.run(command, input=sql, text=True, env=env, capture_output=True)
print(completed.stdout, end="")
if completed.returncode != 0:
print(completed.stderr, end="")
completed.check_returncode()
print(json.dumps({"synced_table": args.table, "records": row_count}, ensure_ascii=False))
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--input", default=str(DEFAULT_INPUT), help="合并后的结构化 JSON")
parser.add_argument("--schema", default=str(DEFAULT_SCHEMA), help="建表 SQL")
parser.add_argument("--table", default=os.getenv("WORKFLOW_DB_TABLE", "Image_Table_Records"), help="目标表名")
parser.add_argument("--host", default=os.getenv("WORKFLOW_DB_HOST", "127.0.0.1"))
parser.add_argument("--port", default=os.getenv("WORKFLOW_DB_PORT", "5432"))
parser.add_argument("--dbname", default=os.getenv("WORKFLOW_DB_NAME", "postgres"))
parser.add_argument("--user", default=os.getenv("WORKFLOW_DB_USER", "postgres"))
parser.add_argument("--password", default="", help="数据库密码；建议用 WORKFLOW_DB_PASSWORD")
parser.add_argument("--append", action="store_true", help="追加导入，不清空表")
parser.add_argument("--allow-empty-key", action="store_true", help="允许空 unique_key；默认跳过")
return parser.parse_args()
def main() -> None:
args = parse_args()
rows, skipped = load_rows(Path(args.input), args.allow_empty_key)
csv_path = write_temp_csv(rows)
try:
sync_to_postgres(args, csv_path, len(rows))
if skipped:
print(json.dumps({"skipped_empty_unique_key": len(skipped), "examples": skipped[:10]}, ensure_ascii=False))
finally:
csv_path.unlink(missing_ok=True)
if __name__ == "__main__":
main()

43
UPP列表处理/数据处理工作区/06_PostgreSQL建表结构.template.sql Normal file
View File

@@ -0,0 +1,43 @@
CREATE TABLE IF NOT EXISTS "Image_Table_Records" (
record_id bigserial PRIMARY KEY,
batch_name text NOT NULL,
source_folder text NOT NULL,
category_1 text,
category_2 text,
image_path text NOT NULL,
image_name text NOT NULL,
image_row_no integer NOT NULL,
unique_key text NOT NULL,
record_data jsonb NOT NULL,
review_status text NOT NULL,
review_notes text,
manual_corrected boolean NOT NULL DEFAULT false,
imported_at timestamptz NOT NULL DEFAULT now(),
audit_result text,
audit_ai_feedback text,
audit_manual_feedback text,
audit_checked_by text,
audit_checked_at timestamptz,
CONSTRAINT uq_image_table_records_unique_key UNIQUE (unique_key),
CONSTRAINT ck_image_table_records_unique_key_present CHECK (btrim(unique_key) <> '')
);
CREATE INDEX IF NOT EXISTS idx_image_table_records_batch_name ON "Image_Table_Records"(batch_name);
CREATE INDEX IF NOT EXISTS idx_image_table_records_source_folder ON "Image_Table_Records"(source_folder);
CREATE INDEX IF NOT EXISTS idx_image_table_records_category ON "Image_Table_Records"(category_1, category_2);
CREATE INDEX IF NOT EXISTS idx_image_table_records_review_status ON "Image_Table_Records"(review_status);
CREATE INDEX IF NOT EXISTS idx_image_table_records_record_data_gin ON "Image_Table_Records" USING gin (record_data);
COMMENT ON TABLE "Image_Table_Records" IS '通用图片表格OCR归档记录表';
COMMENT ON COLUMN "Image_Table_Records".batch_name IS '处理批次名称';
COMMENT ON COLUMN "Image_Table_Records".source_folder IS '原始来源文件夹名';
COMMENT ON COLUMN "Image_Table_Records".category_1 IS '业务分类1，可按任务改名或改造成正式列';
COMMENT ON COLUMN "Image_Table_Records".category_2 IS '业务分类2，可按任务改名或改造成正式列';
COMMENT ON COLUMN "Image_Table_Records".image_path IS '原始图片路径';
COMMENT ON COLUMN "Image_Table_Records".image_name IS '原始图片文件名';
COMMENT ON COLUMN "Image_Table_Records".image_row_no IS '记录在原始图片内的行号';
COMMENT ON COLUMN "Image_Table_Records".unique_key IS '主唯一键，来自任务配置 unique_key 字段';
COMMENT ON COLUMN "Image_Table_Records".record_data IS '识别后的业务字段JSON';
COMMENT ON COLUMN "Image_Table_Records".review_status IS '自动复核或人工复核状态';
COMMENT ON COLUMN "Image_Table_Records".review_notes IS '复核提示';
COMMENT ON COLUMN "Image_Table_Records".manual_corrected IS '是否命中人工修正';

104
UPP列表处理/数据处理工作区/07_处理程序说明.md Normal file
View File

@@ -0,0 +1,104 @@
# 通用图片表格识别处理程序说明
本目录是一套可复用的图片表格 OCR 处理脚手架。默认只提供通用字段和通用数据库结构；用于 UPP、PACS、HIS 或其他任务前，先复制并修改 `01_任务配置.template.json`。
## 文件顺序
1. `01_任务配置.template.json`：任务名、字段顺序、主唯一键、分类规则、OCR 参数、数据库表名。
2. `02_图片表格OCR归档.py`：处理单个图片批次，输出结构化 JSON、CSV、复核报告和 OCR 缓存。
3. `03_人工复核修正.template.json`：人工修正文件模板。正式修正时复制为 `03_人工复核修正.json`，不要提交真实数据。
4. `04_合并批次结果.py`：扫描 `数据处理结果区`，合并所有 `*-列表归档结果`。
5. `05_同步PostgreSQL单表.py`：把合并结果同步到 PostgreSQL 通用 JSONB 单表。
6. `06_PostgreSQL建表结构.template.sql`：通用表结构模板，正式任务可改成业务正式列。
7. `07_处理程序说明.md`：当前说明。
8. `08_本地运行流程.template.sh`：按顺序执行的本地命令模板。
## 首次使用
```bash
cp 数据处理工作区/01_任务配置.template.json 数据处理工作区/01_任务配置.json
cp 数据处理工作区/03_人工复核修正.template.json 数据处理工作区/03_人工复核修正.json
cp 数据处理工作区/06_PostgreSQL建表结构.template.sql 数据处理工作区/06_PostgreSQL建表结构.sql
```
然后人工修改：
- `project_name`、`record_name`。
- `unique_key`。
- `fields` 字段顺序、字段类型、必填规则、正则规则。
- `classification.folder_rules`。
- `ocr.rows_per_image` 或 `ocr.row_height_px`，如果要启用行数漏识别校验。
- `postgres.table_name` 和 SQL 表名。
## 处理单个批次
```bash
export TENCENTCLOUD_SECRET_ID='填入腾讯云 SecretId'
export TENCENTCLOUD_SECRET_KEY='填入腾讯云 SecretKey'
python3 数据处理工作区/02_图片表格OCR归档.py \
--config 数据处理工作区/01_任务配置.json \
--input "待处理-[任务名]图片集群/[批次文件夹名]" \
--output "数据处理结果区/已处理-[任务名]图片集群/[批次文件夹名]-列表归档结果"
```
只用已有 OCR 缓存重建：
```bash
python3 数据处理工作区/02_图片表格OCR归档.py \
--config 数据处理工作区/01_任务配置.json \
--input "待处理-[任务名]图片集群/[批次文件夹名]" \
--output "数据处理结果区/已处理-[任务名]图片集群/[批次文件夹名]-列表归档结果" \
--rebuild-from-cache
```
## 合并和入库
```bash
python3 数据处理工作区/04_合并批次结果.py --config 数据处理工作区/01_任务配置.json
```
如需同步 PostgreSQL：
```bash
export WORKFLOW_DB_HOST='数据库主机'
export WORKFLOW_DB_PORT='5432'
export WORKFLOW_DB_NAME='数据库名'
export WORKFLOW_DB_USER='数据库用户'
export WORKFLOW_DB_PASSWORD='数据库密码'
python3 数据处理工作区/05_同步PostgreSQL单表.py \
--input 数据处理结果区/合并_图片表格_结构化.json \
--schema 数据处理工作区/06_PostgreSQL建表结构.sql
```
## 输出文件
单批次输出：
- `图片表格_结构化.json`
- `图片表格_记录.jsonl`
- `图片表格_记录.csv`
- `复核报告.json`
- `重复主键报告.json`
- `缺少主键报告.json`
- `merged_images/`
- `raw_ocr/`
- `信息记录/汇总.json`
合并输出：
- `数据处理结果区/合并_图片表格_结构化.json`
- `数据处理结果区/合并_图片表格_记录.jsonl`
- `数据处理结果区/合并_图片表格_记录.csv`
- `数据处理结果区/信息记录/全局汇总.json`
- `数据处理结果区/信息记录/批次汇总.csv`
- `数据处理结果区/信息记录/重复主键报告.json`
- `数据处理结果区/信息记录/缺少主键报告.json`
## 注意
- 腾讯云密钥、数据库密码只放环境变量。
- `03_人工复核修正.json` 可能含真实业务数据，不要提交。
- `raw_ocr/` 和 `merged_images/` 是过程文件，建议保留在本地，便于复核和缓存重建。
- 当前 PostgreSQL 模板采用 `record_data jsonb` 保存业务字段；正式稳定后可把高频查询字段改成独立列。

31
UPP列表处理/数据处理工作区/08_本地运行流程.template.sh Executable file
View File

@@ -0,0 +1,31 @@
#!/usr/bin/env bash
set -euo pipefail
# 复制本文件后再填写真实批次名、密钥和数据库配置。
CONFIG="数据处理工作区/01_任务配置.json"
BATCH_NAME="[批次文件夹名]"
INPUT_DIR="待处理-[任务名]图片集群/${BATCH_NAME}"
OUTPUT_DIR="数据处理结果区/已处理-[任务名]图片集群/${BATCH_NAME}-列表归档结果"
export TENCENTCLOUD_SECRET_ID="${TENCENTCLOUD_SECRET_ID:-填入腾讯云 SecretId}"
export TENCENTCLOUD_SECRET_KEY="${TENCENTCLOUD_SECRET_KEY:-填入腾讯云 SecretKey}"
python3 数据处理工作区/02_图片表格OCR归档.py \
--config "${CONFIG}" \
--input "${INPUT_DIR}" \
--output "${OUTPUT_DIR}"
python3 数据处理工作区/04_合并批次结果.py \
--config "${CONFIG}"
# 如需入库，先复制 06_PostgreSQL建表结构.template.sql 为 06_PostgreSQL建表结构.sql，并设置以下变量。
# export WORKFLOW_DB_HOST='数据库主机'
# export WORKFLOW_DB_PORT='5432'
# export WORKFLOW_DB_NAME='数据库名'
# export WORKFLOW_DB_USER='数据库用户'
# export WORKFLOW_DB_PASSWORD='数据库密码'
#
# python3 数据处理工作区/05_同步PostgreSQL单表.py \
# --input 数据处理结果区/合并_图片表格_结构化.json \
# --schema 数据处理工作区/06_PostgreSQL建表结构.sql

617
UPP列表处理/通用图片表格识别工作流模板.md Normal file
View File

@@ -0,0 +1,617 @@
# 通用图片表格识别工作流模板
本文档用于把一个空白目录整理成“图片列表 / 图片表格 OCR 识别 -> 结构化归档 -> 人工复核 -> 合并 -> 可选入库”的通用工作流模板。
模板抽象自 `~/Desktop/HIS数据处理/患者列表处理` 的成熟流程，但这里不绑定 HIS、患者列表、UPP 或任何具体业务字段。后续遇到类似“多张图片中包含规则表格，需要批量识别成结构化记录”的任务时，可先复制本模板，再人工补充腾讯云 API、Gitea、PostgreSQL、字段清洗规则和业务校验规则。
## 1. 适用场景
适合：
- 输入是一批或多批图片，图片中包含列表、目录、清单、统计表、检查单、排班表等规则表格。
- 需要保留每条结构化记录的来源图片、图片内行号、批次名等追溯信息。
- OCR 结果需要自动校验，也允许人工复核和人工修正。
- 最终结果既要保存在本地 JSON / CSV，也可能同步到 PostgreSQL 或其他数据库。
- 工作流需要可提交到 Gitea，但不能提交密钥、图片原件、OCR 缓存和识别结果。
不适合直接套用：
- 图片完全无表格结构，需要复杂版面理解。
- 每张图片字段位置差异极大，无法用统一列名和统一清洗规则描述。
- 任务要求实时在线处理，而不是批量离线归档。
## 2. 模板使用方式
新任务开始时，先人工填写本节占位信息。
```text
项目名称：[填写，例如 PACS UPP 列表处理]
记录对象：[填写，例如 UPP 列表记录 / 检查记录 / 患者记录]
输入图片根目录：[填写，例如 待处理-xxx图片集群/批次文件夹名]
输出结果根目录：[填写，例如 数据处理结果区/已处理-xxx图片集群/批次文件夹名-列表归档结果]
主唯一键：[填写，例如 住院号 / 检查号 / accession_no / 影像号 / 组合键]
业务分类维度：[填写，例如 大科室、子科室；或 设备、项目类型；可为空]
目标数据库表：[填写，例如 "Generic_Image_Table_Records"]
Gitea 仓库：[填写，例如 https://gitea.example.com/team/repo.git]
```
字段定义表：
| 序号 | 字段名 | 是否必填 | 清洗规则 | 校验规则 | 备注 |
| --- | --- | --- | --- | --- | --- |
| 1 | `[字段1]` | 是 / 否 | `[去空格、日期规范化等]` | `[不能为空、格式等]` | `[说明]` |
| 2 | `[字段2]` | 是 / 否 | `[待补充]` | `[待补充]` | `[说明]` |
| 3 | `[字段3]` | 是 / 否 | `[待补充]` | `[待补充]` | `[说明]` |
建议先明确字段顺序。图片表格 OCR 的后续清洗、CSV 导出、数据库列顺序、人工复核页面都会依赖这个顺序。
## 3. 推荐目录结构
```text
.
├── README.md
├── 工作流_本地使用版.md
├── 工作流_Gitea版.md
├── 通用图片表格识别工作流模板.md
├── 待处理-[任务名]图片集群/
│ └── [批次文件夹名]/
├── 已处理-[任务名]图片集群/
├── 数据处理工作区/
│ ├── 01_任务配置.template.json
│ ├── 01_任务配置.json
│ ├── 02_图片表格OCR归档.py
│ ├── 03_人工复核修正.template.json
│ ├── 03_人工复核修正.json
│ ├── 04_合并批次结果.py
│ ├── 05_同步PostgreSQL单表.py
│ ├── 06_PostgreSQL建表结构.template.sql
│ ├── 06_PostgreSQL建表结构.sql
│ ├── 07_处理程序说明.md
│ └── 08_本地运行流程.template.sh
├── 数据处理结果区/
│ ├── 已处理-[任务名]图片集群/
│ └── 信息记录/
└── 人工复核网页端/
```
目录职责：
- `待处理-[任务名]图片集群/`：新下载、尚未处理的原始图片批次。
- `已处理-[任务名]图片集群/`：确认完成归档后的原始图片批次。
- `数据处理工作区/`：任务配置、处理脚本、人工修正模板、数据库结构、说明和本地运行流程模板。
- `数据处理结果区/已处理-[任务名]图片集群/`：每个批次的 OCR 缓存、拼接图、结构化 JSON、CSV、复核报告。
- `数据处理结果区/信息记录/`：全局汇总、批次汇总、重复键报告、缺失键报告。
- `人工复核网页端/`：可选，用于可视化查看来源图片、修正字段、抽查和同步数据库。
提交到 Gitea 时，只提交程序、模板、空 README、SQL 结构和说明文档。不要提交图片、OCR 缓存、识别结果、人工修正实数据、`.env`、数据库导出文件。
## 4. 配置占位
### 4.1 腾讯云 OCR
推荐使用环境变量，不要写入脚本或文档。
```bash
export TENCENTCLOUD_SECRET_ID='填入腾讯云 SecretId'
export TENCENTCLOUD_SECRET_KEY='填入腾讯云 SecretKey'
export TENCENTCLOUD_REGION='ap-shanghai'
```
建议脚本支持：
```text
OCR 引擎：[table-v3 / general-accurate / 其他]
默认接口：RecognizeTableAccurateOCR
默认地域：ap-shanghai
单次超时：60-120 秒
失败重试：0-2 次
调用间隔：0.2-1 秒
```
### 4.2 PostgreSQL
数据库是可选模块。没有确认字段和表名之前，先只产出本地 JSON / CSV。
```bash
export WORKFLOW_DB_HOST='数据库主机'
export WORKFLOW_DB_PORT='5432'
export WORKFLOW_DB_NAME='数据库名'
export WORKFLOW_DB_USER='数据库用户'
export WORKFLOW_DB_PASSWORD='数据库密码'
```
待人工补充：
```text
正式表名：[填写]
主键策略：[bigserial / uuid / 来源字段组合]
唯一约束：[填写主唯一键或组合唯一键]
允许为空字段：[填写]
必须人工复核才允许入库的异常：[填写]
索引字段：[填写常用查询字段]
```
### 4.3 Gitea
```text
仓库地址：[填写]
默认分支：[main / master / 其他]
提交范围：程序、模板、说明、SQL、示例配置
排除范围：密钥、图片、OCR 缓存、处理结果、人工修正实数据、数据库密码
```
建议 `.gitignore` 至少包含：
```gitignore
.env
*.env
__pycache__/
*.pyc
待处理-*图片集群/
已处理-*图片集群/
数据处理结果区/
数据处理工作区/03_人工复核修正.json
人工复核网页端/instance/
```
## 5. 批处理脚本能力清单
`02_图片表格OCR归档.py` 建议具备以下能力：
1. 扫描输入目录，识别 `.png`、`.jpg`、`.jpeg`、`.bmp`。
2. 按自然序排序图片，例如 `第2页.png` 应排在 `第10页.png` 前面。
3. 可按第一层文件夹做业务分类，例如科室、设备、项目、来源系统。
4. 将多张图片纵向拼接，降低 OCR 调用次数。
5. 拼接时支持上下白边，例如 `--image-padding-y 24`，减少贴边表格行漏识别。
6. 调用 OCR 接口，响应写入 `raw_ocr/` 缓存。
7. 从 OCR 响应解析表格行和列，生成统一字段顺序的记录。
8. 将每条记录绑定来源图片、图片名、图片内行号、拼接组、OCR 请求 ID。
9. 对字段做清洗，例如空格归一、日期规范化、编号纠错、列错位修正。
10. 对记录做自动校验，输出 `自动复核通过` 或 `需人工复核`。
11. 支持读取 `03_人工复核修正.json` 后用缓存重建结果。
12. 支持主唯一键去重，输出重复键报告。
13. 输出结构化 JSON、JSONL、CSV、复核报告、汇总信息。
推荐命令参数：
```bash
python3 数据处理工作区/02_图片表格OCR归档.py \
--config "数据处理工作区/01_任务配置.json" \
--input "待处理-[任务名]图片集群/[批次文件夹名]" \
--output "数据处理结果区/已处理-[任务名]图片集群/[批次文件夹名]-列表归档结果" \
--corrections "数据处理工作区/03_人工复核修正.json" \
--ocr-engine table-v3 \
--batch-size 6 \
--image-padding-y 24 \
--workers 1 \
--folder-workers 2 \
--timeout 90 \
--max-retries 1
```
只用已有 OCR 缓存重建：
```bash
python3 数据处理工作区/02_图片表格OCR归档.py \
--config "数据处理工作区/01_任务配置.json" \
--input "待处理-[任务名]图片集群/[批次文件夹名]" \
--output "数据处理结果区/已处理-[任务名]图片集群/[批次文件夹名]-列表归档结果" \
--ocr-engine table-v3 \
--batch-size 6 \
--image-padding-y 24 \
--rebuild-from-cache
```
## 6. OCR 策略模板
建议默认策略：
- 图片预处理：保持原图，不裁切；必要时统一 RGB；拼接图背景使用白色。
- 拼接策略：先按 `batch-size=6` 纵向拼接；如果漏行、超时或接口失败，自动降到 `4/3/2/单张`。
- 白边策略：上下各加 `24px` 白边作为初始值；如果图片行距很小或边缘文字贴边，可增加到 `32px` 或 `40px`。
- 缓存策略：每个拼接图对应一个 OCR JSON 缓存；默认命中缓存不重新请求；加 `--force` 才重新识别。
- 行数校验：按图片高度、已知行数或模板行高估算预期行数，OCR 返回行数偏少时触发降档。
- 失败处理：单个拼接组失败时记录错误，不让整个批次无声失败；正式任务结束后必须检查错误列表。
输出目录建议：
```text
[批次]-列表归档结果/
├── merged_images/
│ └── [来源文件夹]/
├── raw_ocr/
│ └── [来源文件夹]/
├── 列表_分类归档.json
├── 图片表格_结构化.json
├── 图片表格_记录.jsonl
├── 图片表格_记录.csv
├── 复核报告.json
├── 重复主键报告.json
└── 信息记录/
├── 汇总.json
└── 分类汇总.csv
```
## 7. 结构化记录模板
建议每条记录采用分层结构，便于同时支持业务展示、复核和数据库同步。
```json
{
"处理批次": "[批次名]",
"业务分类1": "[例如大科室，可为空]",
"业务分类2": "[例如子科室，可为空]",
"来源文件夹": "[原始文件夹名]",
"记录信息": {
"[字段1]": "[值]",
"[字段2]": "[值]",
"[字段3]": "[值]"
},
"图片信息": {
"图片路径": "[原始图片路径]",
"图片名": "[原始图片名]",
"图片序号": "[自然排序序号或页码]",
"图片内行号": 1,
"拼接组序号": 1,
"拼接图片路径": "[merged_images 路径]",
"OCR缓存路径": "[raw_ocr 路径]",
"OCR请求ID": "[接口返回 RequestId]"
},
"复核": {
"状态": "自动复核通过",
"提示": [],
"人工修正": false,
"人工备注": ""
}
}
```
复核状态建议统一：
- `自动复核通过`：字段清洗和规则校验后没有发现明显问题。
- `需人工复核`：缺必填项、格式异常、行数可疑、主键缺失、时间逻辑异常等。
- `人工复核通过`：命中人工修正配置，修正后通过校验。
- `AI修改-待确认`：可作为网页端草稿状态，不应直接入库为人工确认结果。
## 8. 字段清洗和校验模板
每个任务都需要人工补充清洗规则，不建议把某个业务的规则硬套到新任务。
基础清洗：
- 全角转半角。
- 去除多余空格、换行、不可见字符。
- 统一日期格式，例如 `YYYY-MM-DD HH:MM:SS`。
- 统一编号格式，例如去空格、统一大小写、修正常见 OCR 误识别。
- 处理列错位，例如长文本吞入下一列、时间字段左移、数字字段被识别到备注列。
基础校验：
- 主唯一键不能为空。
- 必填字段不能为空。
- 数字字段必须可转成整数或小数。
- 日期字段必须符合目标格式。
- 起止时间不能倒置；如果倒置，应标记为 `需人工复核`。
- 同批次或全局主唯一键重复时，必须输出重复报告。
任务专用校验占位：
```text
[规则1]：
[规则2]：
[规则3]：
```
去重策略占位：
```text
主唯一键：[填写]
重复时保留：[后出现记录 / 字段更完整记录 / 人工复核通过记录 / 其他]
缺少主唯一键：[剔除 / 保留但需人工复核 / 禁止入库]
重复报告文件名：[填写]
```
## 9. 人工复核模板
人工修正文件建议只在本地存在，仓库只保留 `.template.json`。
`数据处理工作区/03_人工复核修正.template.json`：
```json
[
{
"图片路径": "待处理-[任务名]图片集群/[批次]/[图片名].png",
"图片内行号": 1,
"记录信息": {
"[字段1]": "[人工确认值]",
"[字段2]": "[人工确认值]"
},
"复核选项": {},
"复核备注": "[可选]"
}
]
```
人工复核顺序：
1. 打开批次结果中的 `复核报告.json`。
2. 根据 `图片路径`、`图片名`、`图片内行号` 回看原图或网页端裁剪图。
3. 将确认值写入 `03_人工复核修正.json`。
4. 使用 `--rebuild-from-cache` 重建，不重复调用 OCR。
5. 检查 `复核报告.json` 中需复核数量是否下降。
6. 重新合并全局结果。
如果使用网页端，建议它至少支持：
- 批次列表和汇总。
- 按复核状态过滤记录。
- 原图裁剪定位到图片内行。
- 编辑字段并保存到人工修正 JSON。
- 数据库可用时同步单条记录；数据库不可用时保留待同步状态。
- 抽查功能可选，AI 输出只作为辅助，不直接覆盖人工确认结果。
## 10. 合并批次模板
`04_合并批次结果.py` 建议扫描 `数据处理结果区/` 下所有 `*-列表归档结果`，合并每个批次的结构化结果。
推荐输出：
```text
数据处理结果区/
├── 合并_图片表格_结构化.json
├── 合并_图片表格_记录.jsonl
├── 合并_图片表格_记录.csv
└── 信息记录/
├── 全局汇总.json
├── 批次汇总.csv
├── 重复主键报告.json
└── 缺少主键报告.json
```
合并时建议重新执行全局去重，因为不同批次之间也可能出现重复主键。
运行：
```bash
python3 数据处理工作区/04_合并批次结果.py --config "数据处理工作区/01_任务配置.json"
```
合并后检查：
```text
批次数
总图片数
合并前记录数
合并后记录数
需人工复核记录数
人工修正记录数
重复主键剔除记录数
缺少主键剔除记录数
```
## 11. PostgreSQL 入库模板
`05_同步PostgreSQL单表.py` 建议只读取合并后的 JSON，生成临时 CSV，再用 `psql \copy` 导入正式表。
入库原则：
- 数据库只放正式查看和追溯需要的字段。
- OCR 请求号、OCR 缓存路径、拼接图路径、拼接组细节等中间态字段默认留在本地结果目录，不进入正式表，除非业务明确需要。
- 主唯一键应有唯一约束。
- 必填主键应有非空约束。
- 重要时间逻辑、状态逻辑可放入数据库 CHECK 约束。
建表示例骨架：
```sql
CREATE TABLE IF NOT EXISTS "[目标表名]" (
record_id bigserial PRIMARY KEY,
batch_name text NOT NULL,
source_folder text NOT NULL,
image_path text NOT NULL,
image_name text NOT NULL,
image_row_no integer NOT NULL,
unique_key text NOT NULL,
field_1 text,
field_2 text,
field_3 text,
review_status text NOT NULL,
review_notes text,
manual_corrected boolean NOT NULL DEFAULT false,
imported_at timestamptz NOT NULL DEFAULT now(),
audit_result text,
audit_ai_feedback text,
audit_manual_feedback text,
audit_checked_by text,
audit_checked_at timestamptz,
CONSTRAINT uq_target_unique_key UNIQUE (unique_key),
CONSTRAINT ck_target_unique_key_present CHECK (btrim(unique_key) <> '')
);
CREATE INDEX IF NOT EXISTS idx_target_batch_name ON "[目标表名]"(batch_name);
CREATE INDEX IF NOT EXISTS idx_target_source_folder ON "[目标表名]"(source_folder);
CREATE INDEX IF NOT EXISTS idx_target_review_status ON "[目标表名]"(review_status);
```
同步命令：
```bash
export WORKFLOW_DB_HOST='数据库主机'
export WORKFLOW_DB_PORT='5432'
export WORKFLOW_DB_NAME='数据库名'
export WORKFLOW_DB_USER='数据库用户'
export WORKFLOW_DB_PASSWORD='数据库密码'
python3 数据处理工作区/05_同步PostgreSQL单表.py \
--input "数据处理结果区/合并_图片表格_结构化.json" \
--schema "数据处理工作区/06_PostgreSQL建表结构.sql" \
--table '"[目标表名]"'
```
入库后核对：
```sql
SELECT count(*) FROM "[目标表名]";
SELECT review_status, manual_corrected, count(*)
FROM "[目标表名]"
GROUP BY review_status, manual_corrected;
SELECT unique_key
FROM "[目标表名]"
GROUP BY unique_key
HAVING count(*) > 1;
```
## 12. 本地完整流程
处理一个新批次：
```bash
# 1. 设置 OCR 密钥
export TENCENTCLOUD_SECRET_ID='填入腾讯云 SecretId'
export TENCENTCLOUD_SECRET_KEY='填入腾讯云 SecretKey'
# 2. 运行 OCR 归档
python3 数据处理工作区/02_图片表格OCR归档.py \
--config "数据处理工作区/01_任务配置.json" \
--input "待处理-[任务名]图片集群/[批次文件夹名]" \
--output "数据处理结果区/已处理-[任务名]图片集群/[批次文件夹名]-列表归档结果" \
--ocr-engine table-v3 \
--batch-size 6 \
--image-padding-y 24 \
--workers 1 \
--folder-workers 2 \
--timeout 90 \
--max-retries 1
# 3. 检查复核报告和重复报告
# 4. 必要时填写 03_人工复核修正.json
# 5. 用缓存重建
python3 数据处理工作区/02_图片表格OCR归档.py \
--config "数据处理工作区/01_任务配置.json" \
--input "待处理-[任务名]图片集群/[批次文件夹名]" \
--output "数据处理结果区/已处理-[任务名]图片集群/[批次文件夹名]-列表归档结果" \
--ocr-engine table-v3 \
--batch-size 6 \
--image-padding-y 24 \
--rebuild-from-cache
# 6. 合并所有批次
python3 数据处理工作区/04_合并批次结果.py --config "数据处理工作区/01_任务配置.json"
# 7. 可选：同步数据库
python3 数据处理工作区/05_同步PostgreSQL单表.py
```
确认无误后移动原始图片批次：
```bash
mv "待处理-[任务名]图片集群/[批次文件夹名]" "已处理-[任务名]图片集群/"
```
## 13. Gitea 提交流程
提交前检查：
```bash
git status --short --ignored
git diff --cached --name-only
rg -n "Secret|PASSWORD|密码|密钥|TENCENTCLOUD_SECRET|DB_PASSWORD" README.md 工作流*.md 数据处理工作区 人工复核网页端
```
推荐只提交：
```bash
git add \
.gitignore \
README.md \
工作流_本地使用版.md \
工作流_Gitea版.md \
通用图片表格识别工作流模板.md \
数据处理工作区/01_任务配置.template.json \
数据处理工作区/02_图片表格OCR归档.py \
数据处理工作区/03_人工复核修正.template.json \
数据处理工作区/04_合并批次结果.py \
数据处理工作区/05_同步PostgreSQL单表.py \
数据处理工作区/06_PostgreSQL建表结构.template.sql \
数据处理工作区/07_处理程序说明.md \
数据处理工作区/08_本地运行流程.template.sh \
人工复核网页端
git commit -m "Add generic image table OCR workflow"
git push origin main
```
不要提交：
```text
.env
数据处理工作区/03_人工复核修正.json
待处理-*图片集群/
已处理-*图片集群/
数据处理结果区/
raw_ocr/
merged_images/
*.csv
*.jsonl
包含真实业务数据的 *.json
```
## 14. 迁移到新任务时的改造清单
必须改：
- `[任务名]`、输入目录、输出目录。
- 字段顺序和字段名。
- OCR 行列解析逻辑。
- 字段清洗规则。
- 必填字段和复核规则。
- 主唯一键和去重策略。
- CSV 表头。
- PostgreSQL 表名、列名、约束和索引。
通常要改：
- 分类规则，例如科室、设备、项目类型、来源系统。
- 行数估算规则，例如每张图片有多少数据行。
- 拼接白边和默认 batch-size。
- 人工复核网页端展示字段。
- 抽查提示词和 AI 校验标准。
通常不用改：
- 环境变量读取方式。
- OCR 缓存机制。
- `--rebuild-from-cache` 重建方式。
- 批次合并流程。
- Gitea 排除真实数据和密钥的原则。
## 15. 完成标准
一个批次可以认为处理完成，需要同时满足：
- OCR 脚本运行结束，没有未处理的接口错误。
- `图片表格_结构化.json` 和 `图片表格_记录.csv` 已生成。
- `复核报告.json` 已检查，需人工复核项已处理或明确保留。
- 主唯一键缺失和重复报告已检查。
- 合并脚本已运行，全局汇总符合预期。
- 如果启用 PostgreSQL，入库记录数与合并结果一致。
- 原始图片批次已从 `待处理` 移到 `已处理`。
- Gitea 仓库没有密钥、图片、OCR 缓存、真实处理结果。
## 16. 给未来智能体的执行提示
遇到一个新图片表格任务时，先不要直接写死业务规则。建议按以下顺序推进：
1. 读取本模板和当前目录结构。
2. 盘点待处理图片批次、图片数量、图片命名规律。
3. 抽样查看 3-5 张图片，确认表格字段、行高、是否有表头、是否跨页。
4. 让用户确认字段顺序、主唯一键、是否需要数据库。
5. 基于模板生成或改造处理脚本。
6. 先小批量试跑，检查 OCR 响应、行数和列错位。
7. 固化清洗规则和复核规则。
8. 全批次运行，输出报告。
9. 人工复核后用缓存重建。
10. 合并、可选入库、归档。
本模板只定义工作流骨架。腾讯云 API 密钥、Gitea 地址、PostgreSQL 连接、目标表结构、业务字段、人工修正规则，都应在具体任务中由人工确认后再补充。