commit 70215ce61107b76aa7af99faf0d7e9c598f69417 Author: Codex Date: Mon May 25 12:33:24 2026 +0800 Add PACS UPP OCR workflow diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..d83120c --- /dev/null +++ b/.gitignore @@ -0,0 +1,18 @@ +.env +*.env +__pycache__/ +*.pyc + +# 原始图片、预处理图片、OCR 缓存和结果数据不提交 +UPP列表处理/待处理-*图片集群/ +UPP列表处理/已处理-*图片集群/ +UPP列表处理/数据处理结果区/ + +# 本地专用流程和人工修正实数据不提交 +UPP列表处理/工作流_本地使用版.md +UPP列表处理/数据处理工作区/03_人工复核修正.json + +# 本地数据库配置或导出不提交 +UPP列表处理/数据处理工作区/06_PostgreSQL建表结构.sql +*.dump +*.sql.gz diff --git a/UPP列表处理/README.md b/UPP列表处理/README.md new file mode 100644 index 0000000..e69de29 diff --git a/UPP列表处理/工作流_Gitea版.md b/UPP列表处理/工作流_Gitea版.md new file mode 100644 index 0000000..e69de29 diff --git a/UPP列表处理/数据处理工作区/01_任务配置.json b/UPP列表处理/数据处理工作区/01_任务配置.json new file mode 100644 index 0000000..bdecbbf --- /dev/null +++ b/UPP列表处理/数据处理工作区/01_任务配置.json @@ -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" + } +} diff --git a/UPP列表处理/数据处理工作区/01_任务配置.template.json b/UPP列表处理/数据处理工作区/01_任务配置.template.json new file mode 100644 index 0000000..363d7a1 --- /dev/null +++ b/UPP列表处理/数据处理工作区/01_任务配置.template.json @@ -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" + } +} diff --git a/UPP列表处理/数据处理工作区/02_图片表格OCR归档.py b/UPP列表处理/数据处理工作区/02_图片表格OCR归档.py new file mode 100755 index 0000000..7d42779 --- /dev/null +++ b/UPP列表处理/数据处理工作区/02_图片表格OCR归档.py @@ -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() diff --git a/UPP列表处理/数据处理工作区/03_人工复核修正.template.json b/UPP列表处理/数据处理工作区/03_人工复核修正.template.json new file mode 100644 index 0000000..5c49a14 --- /dev/null +++ b/UPP列表处理/数据处理工作区/03_人工复核修正.template.json @@ -0,0 +1,13 @@ +[ + { + "图片路径": "待处理-[任务名]图片集群/[批次文件夹名]/[图片名].png", + "图片内行号": 1, + "记录信息": { + "记录编号": "人工确认值", + "字段1": "人工确认值", + "字段2": "人工确认值" + }, + "复核选项": {}, + "复核备注": "可选：说明为什么修正" + } +] diff --git a/UPP列表处理/数据处理工作区/04_合并批次结果.py b/UPP列表处理/数据处理工作区/04_合并批次结果.py new file mode 100755 index 0000000..724c8f6 --- /dev/null +++ b/UPP列表处理/数据处理工作区/04_合并批次结果.py @@ -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() diff --git a/UPP列表处理/数据处理工作区/05_同步PostgreSQL单表.py b/UPP列表处理/数据处理工作区/05_同步PostgreSQL单表.py new file mode 100755 index 0000000..6a0b350 --- /dev/null +++ b/UPP列表处理/数据处理工作区/05_同步PostgreSQL单表.py @@ -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() diff --git a/UPP列表处理/数据处理工作区/06_PostgreSQL建表结构.template.sql b/UPP列表处理/数据处理工作区/06_PostgreSQL建表结构.template.sql new file mode 100644 index 0000000..c997b83 --- /dev/null +++ b/UPP列表处理/数据处理工作区/06_PostgreSQL建表结构.template.sql @@ -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 '是否命中人工修正'; diff --git a/UPP列表处理/数据处理工作区/07_处理程序说明.md b/UPP列表处理/数据处理工作区/07_处理程序说明.md new file mode 100644 index 0000000..836a81c --- /dev/null +++ b/UPP列表处理/数据处理工作区/07_处理程序说明.md @@ -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` 保存业务字段；正式稳定后可把高频查询字段改成独立列。 diff --git a/UPP列表处理/数据处理工作区/08_本地运行流程.template.sh b/UPP列表处理/数据处理工作区/08_本地运行流程.template.sh new file mode 100755 index 0000000..841c315 --- /dev/null +++ b/UPP列表处理/数据处理工作区/08_本地运行流程.template.sh @@ -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 diff --git a/UPP列表处理/通用图片表格识别工作流模板.md b/UPP列表处理/通用图片表格识别工作流模板.md new file mode 100644 index 0000000..bd7776d --- /dev/null +++ b/UPP列表处理/通用图片表格识别工作流模板.md @@ -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 连接、目标表结构、业务字段、人工修正规则，都应在具体任务中由人工确认后再补充。