#!/usr/bin/env python3 # -*- coding: utf-8 -*- """Archive HIS patient-list screenshots into structured JSON. Tencent OCR credentials are read from environment variables: TENCENTCLOUD_SECRET_ID and TENCENTCLOUD_SECRET_KEY. """ from __future__ import annotations import argparse import base64 import concurrent.futures import csv import datetime as dt import hashlib import hmac import json import os import re import signal import subprocess import threading import time import unicodedata import urllib.error import urllib.request from contextlib import contextmanager from dataclasses import dataclass from pathlib import Path from typing import Any from PIL import Image COLUMNS = [ "姓名", "性别", "年龄", "住院号", "诊断", "入院时间", "最后书写时间", "住院天数", "出院时间", "手术后天数", ] IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".bmp"} OCR_ENGINE_LABELS = { "table-v3": "腾讯云 RecognizeTableAccurateOCR 表格识别 V3", } GENERAL_COLUMN_ANCHORS = [0, 96, 160, 230, 450, 820, 1075, 1305, 1395, 1590] GENERAL_REFERENCE_WIDTH = 1842 GENERAL_ROW_Y_THRESHOLD = 18 class NonRetryableOcrError(RuntimeError): pass @dataclass(frozen=True) class Department: major: str sub: str raw_name: str def natural_key(path: Path) -> tuple[Any, ...]: parts = re.split(r"(\d+)", path.stem) key: list[Any] = [] for part in parts: if part.isdigit(): key.append(int(part)) else: key.append(part) return tuple(key) def normalize_text(value: Any) -> str: if value is None: return "" text = unicodedata.normalize("NFKC", str(value)) text = text.replace("\u3000", " ") return re.sub(r"\s+", " ", text).strip() def normalize_folder_name(folder_name: str) -> str: name = re.sub(r"^\d{4}[_-]\d{1,2}[_-]\d{1,2}~\d{4}[_-]\d{1,2}[_-]\d{1,2}_", "", folder_name) name = re.sub(r"^\d{4}[_-]\d{1,2}[_-]\d{1,2}_", "", name) name = re.sub(r"^\d{4}年\d{1,2}月\d{1,2}日[_-]?", "", name) name = re.sub(r"病房|病区", "", name) return normalize_text(name) def load_departments(path: Path) -> tuple[dict[str, str], dict[str, str]]: data = json.loads(path.read_text(encoding="utf-8")) sub_to_major: dict[str, str] = {} for group in data["大科室列表"]: major = group["大科室"] for sub in group["子科室"]: if sub in sub_to_major: raise ValueError(f"重复子科室: {sub}") sub_to_major[sub] = major aliases: dict[str, str] = {} for alias, sub in data.get("aliases", {}).items(): if sub not in sub_to_major: raise ValueError(f"别名 {alias} 指向不存在的子科室 {sub}") aliases[normalize_text(alias)] = sub for sub in sub_to_major: aliases[normalize_text(sub)] = sub return sub_to_major, aliases def classify_department(folder_name: str, sub_to_major: dict[str, str], aliases: dict[str, str]) -> Department: raw = normalize_folder_name(folder_name) if raw in aliases: sub = aliases[raw] return Department(sub_to_major[sub], sub, raw) compact = raw.replace("科", "").replace("外", "外") for alias, sub in sorted(aliases.items(), key=lambda item: len(item[0]), reverse=True): alias_compact = alias.replace("科", "") if alias and (alias in raw or alias_compact in compact): return Department(sub_to_major[sub], sub, raw) return Department("未分类", "未分类", raw) def batched(items: list[Path], size: int) -> list[list[Path]]: return [items[i : i + size] for i in range(0, len(items), size)] def merge_images(image_paths: list[Path], output_path: Path, padding_y: int = 0) -> 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 = [] 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 infer_rows_for_image(image_path: Path, override_rows: int = 0) -> int: if override_rows > 0: return override_rows with Image.open(image_path) as image: # HIS list screenshots in this batch use about 39 px per row: # 700 px -> 18 rows, 784 px -> 20 rows. return max(1, round(image.height / 39.0)) def locate_source_row(table_row_index: int, row_counts: list[int]) -> tuple[int, int]: offset = 0 for image_index, row_count in enumerate(row_counts): if table_row_index < offset + row_count: return image_index, table_row_index - offset offset += row_count return len(row_counts) - 1, max(0, table_row_index - sum(row_counts[:-1])) def get_credentials() -> tuple[str, str]: secret_id = os.getenv("TENCENTCLOUD_SECRET_ID") or os.getenv("TENCENT_SECRET_ID") secret_key = os.getenv("TENCENTCLOUD_SECRET_KEY") or os.getenv("TENCENT_SECRET_KEY") if not secret_id or not secret_key: raise RuntimeError("请先设置 TENCENTCLOUD_SECRET_ID 和 TENCENTCLOUD_SECRET_KEY 环境变量") 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")) @contextmanager def wall_clock_timeout(seconds: int): if seconds <= 0 or threading.current_thread() is not threading.main_thread(): yield return def handle_timeout(_signum: int, _frame: Any) -> None: raise TimeoutError(f"OCR请求超过 {seconds} 秒") previous_handler = signal.getsignal(signal.SIGALRM) previous_timer = signal.setitimer(signal.ITIMER_REAL, 0) signal.signal(signal.SIGALRM, handle_timeout) signal.setitimer(signal.ITIMER_REAL, seconds) try: yield finally: signal.setitimer(signal.ITIMER_REAL, 0) signal.signal(signal.SIGALRM, previous_handler) if previous_timer[0] > 0: signal.setitimer(signal.ITIMER_REAL, previous_timer[0], previous_timer[1]) def call_tencent_ocr( action: str, 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 json.loads(cache_path.read_text(encoding="utf-8")) if not secret_id or not secret_key: raise RuntimeError(f"OCR缓存不存在，且当前为仅重建模式: {cache_path}") image_base64 = base64.b64encode(image_path.read_bytes()).decode("ascii") payload = {"ImageBase64": image_base64, "UseNewModel": True} if action == "GeneralAccurateOCR": payload = {"ImageBase64": image_base64} last_error: str | None = None for attempt in range(max_retries + 1): try: with wall_clock_timeout(timeout): data = tc3_request(action, payload, secret_id, secret_key, region, timeout) response = data.get("Response", {}) if "Error" in response: error_text = json.dumps(response["Error"], ensure_ascii=False) if response["Error"].get("Code") == "FailedOperation.OcrFailed": raise NonRetryableOcrError(error_text) raise RuntimeError(error_text) response.pop("Data", None) cache_path.parent.mkdir(parents=True, exist_ok=True) cache_path.write_text(json.dumps(response, ensure_ascii=False, indent=2), encoding="utf-8") return response except NonRetryableOcrError: raise 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 call_table_v3( image_path: Path, cache_path: Path, secret_id: str, secret_key: str, region: str, timeout: int, force: bool, max_retries: int, ) -> dict[str, Any]: return call_tencent_ocr( "RecognizeTableAccurateOCR", image_path, cache_path, secret_id, secret_key, region, timeout, force, max_retries, ) def call_general_accurate( image_path: Path, cache_path: Path, secret_id: str, secret_key: str, region: str, timeout: int, force: bool, max_retries: int, ) -> dict[str, Any]: return call_tencent_ocr( "GeneralAccurateOCR", image_path, cache_path, secret_id, secret_key, region, timeout, force, max_retries, ) def cells_to_rows(table_response: dict[str, Any]) -> list[list[str]]: tables = table_response.get("TableDetections") or [] cells: list[dict[str, Any]] = [] for table in tables: cells.extend(table.get("Cells") or []) if not cells: return [] max_row = max(int(cell.get("RowTl", 0)) for cell in cells) max_col = max(int(cell.get("ColTl", 0)) for cell in cells) rows = [["" for _ in range(max(max_col + 1, len(COLUMNS)))] for _ in range(max_row + 1)] for cell in cells: row = int(cell.get("RowTl", 0)) col = int(cell.get("ColTl", 0)) text = normalize_text(cell.get("Text", "")) if col >= len(rows[row]): rows[row].extend([""] * (col - len(rows[row]) + 1)) if rows[row][col]: rows[row][col] = normalize_text(rows[row][col] + " " + text) else: rows[row][col] = text return [row[: len(COLUMNS)] + [""] * max(0, len(COLUMNS) - len(row)) for row in rows] def general_column_index(x: float, image_width: int) -> int: scale = image_width / GENERAL_REFERENCE_WIDTH if image_width else 1.0 anchors = [value * scale for value in GENERAL_COLUMN_ANCHORS] boundaries = [(anchors[index] + anchors[index + 1]) / 2 for index in range(len(anchors) - 1)] for index, boundary in enumerate(boundaries): if x < boundary: return index return len(anchors) - 1 def general_detections_to_rows(general_response: dict[str, Any], image_width: int) -> list[list[str]]: detections: list[dict[str, Any]] = [] for item in general_response.get("TextDetections") or []: text = normalize_text(item.get("DetectedText", "")) if not text: continue polygon = item.get("ItemPolygon") or {} x = float(polygon.get("X", 0) or 0) y = float(polygon.get("Y", 0) or 0) height = float(polygon.get("Height", 0) or 0) detections.append({"x": x, "y_center": y + height / 2, "text": text}) detections.sort(key=lambda item: (item["y_center"], item["x"])) grouped_rows: list[dict[str, Any]] = [] for detection in detections: if not grouped_rows or abs(detection["y_center"] - grouped_rows[-1]["y_center"]) > GENERAL_ROW_Y_THRESHOLD: grouped_rows.append({"y_center": detection["y_center"], "items": []}) else: count = len(grouped_rows[-1]["items"]) grouped_rows[-1]["y_center"] = (grouped_rows[-1]["y_center"] * count + detection["y_center"]) / (count + 1) grouped_rows[-1]["items"].append(detection) rows: list[list[str]] = [] for grouped_row in grouped_rows: columns: list[list[str]] = [[] for _ in COLUMNS] for detection in sorted(grouped_row["items"], key=lambda item: item["x"]): columns[general_column_index(detection["x"], image_width)].append(detection["text"]) rows.append([normalize_text(" ".join(values)) for values in columns]) return rows def ocr_response_to_rows(response: dict[str, Any], engine: str, image_width: int) -> list[list[str]]: if engine == "general-accurate": return general_detections_to_rows(response, image_width) return cells_to_rows(response) def normalize_date(text: str) -> str: text = normalize_text(text).replace("/", "-").replace(".", "-") text = re.sub( r"(\d{4})-(\d{1,2})-(\d{1,2})\s+(\d{1,4}):(\d{1,2})(?::(\d{1,2}))?", date_repl, text, ) return text def date_repl(match: re.Match[str]) -> str: year, month, day, hour_text, minute_text, second_text = match.groups() if second_text is None and len(hour_text) > 2: if len(hour_text) == 3: hour = int(hour_text[:2]) minute = int(hour_text[2:]) else: hour = int(hour_text[:-2]) minute = int(hour_text[-2:]) second = int(minute_text) else: hour = int(hour_text) minute = int(minute_text) second = int(second_text or 0) if not (0 <= hour <= 23 and 0 <= minute <= 59 and 0 <= second <= 59): return match.group(0) return f"{int(year):04d}-{int(month):02d}-{int(day):02d} {int(hour):02d}:{int(minute):02d}:{int(second):02d}" def clean_patient_row(row: list[str]) -> dict[str, Any]: values = {column: normalize_text(row[index]) for index, column in enumerate(COLUMNS)} if values["姓名"] and not values["性别"]: name_gender = re.fullmatch(r"(.+?)(男|女)", values["姓名"]) if name_gender: values["姓名"] = normalize_text(name_gender.group(1)) values["性别"] = name_gender.group(2) if values["性别"] not in {"男", "女"}: sex_age = re.fullmatch(r"(男|女)\s*(\d{1,3}岁)", values["性别"]) if sex_age: values["性别"] = sex_age.group(1) if not values["年龄"]: values["年龄"] = sex_age.group(2) def normalize_inpatient_no(value: Any) -> str: inpatient_no = re.sub(r"\s+", "", normalize_text(value)).upper() if inpatient_no.startswith("ZV"): inpatient_no = "ZY" + inpatient_no[2:] if inpatient_no.startswith("ZYS"): inpatient_no = "ZY5" + inpatient_no[3:] if len(inpatient_no) > 2: prefix, number = inpatient_no[:2], inpatient_no[2:] inpatient_no = prefix + number.translate(str.maketrans({"O": "0", "I": "1", "L": "1", "S": "5"})) return inpatient_no values["住院号"] = normalize_inpatient_no(values["住院号"]) def looks_datetime(value: Any) -> bool: return bool( re.fullmatch( r"\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2}", normalize_date(str(value)), ) ) def looks_days(value: Any) -> bool: return bool(re.fullmatch(r"\d+", str(value))) def looks_postop(value: Any) -> bool: return bool(re.fullmatch(r"后\d+天", str(value))) values["入院时间"] = normalize_date(values["入院时间"]) values["最后书写时间"] = normalize_date(values["最后书写时间"]) values["出院时间"] = normalize_date(values["出院时间"]) long_name_shift = re.fullmatch(r"(.+?)(?:\.\.\.)?\s*(男|女)", values["姓名"]) if ( long_name_shift and values["性别"].endswith("岁") and re.fullmatch(r"Z[YV]\w+", values["年龄"], re.IGNORECASE) and values["住院号"] ): old_sex = values["性别"] old_age = values["年龄"] old_diagnosis = values["住院号"] values["姓名"] = normalize_text(long_name_shift.group(1)) values["性别"] = long_name_shift.group(2) values["年龄"] = old_sex values["住院号"] = normalize_inpatient_no(old_age) if looks_datetime(values["诊断"]) and looks_datetime(values["入院时间"]) and looks_days(values["最后书写时间"]): old_admission = values["诊断"] old_last_write = values["入院时间"] old_days = values["最后书写时间"] old_discharge = values["住院天数"] old_postop = values["出院时间"] values["诊断"] = old_diagnosis values["入院时间"] = normalize_date(old_admission) values["最后书写时间"] = normalize_date(old_last_write) values["住院天数"] = old_days values["出院时间"] = normalize_date(old_discharge) if looks_datetime(old_discharge) else "" if looks_postop(old_postop) and not values["手术后天数"]: values["手术后天数"] = old_postop elif not values["诊断"]: values["诊断"] = old_diagnosis admission_with_diagnosis = re.search( r"(.+?)\s*-*\s*(\d{4}[-/.]\d{1,2}[-/.]\d{1,2}\s+\d{1,4}:\d{1,2}(?::\d{1,2})?)\s*$", values["入院时间"], ) if admission_with_diagnosis and not looks_datetime(values["入院时间"]): values["诊断"] = normalize_text(f"{values['诊断']} {admission_with_diagnosis.group(1)}") values["入院时间"] = normalize_date(admission_with_diagnosis.group(2)) shifted_by_long_diagnosis = re.search( r"(\d{4}[-/.]\d{1,2}[-/.]\d{1,2}\s+\d{1,4}:\d{1,2}(?::\d{1,2})?)\s*[)）\]]*$", values["诊断"], ) if ( shifted_by_long_diagnosis and looks_datetime(values["入院时间"]) and looks_days(values["最后书写时间"]) and (not values["住院天数"] or looks_datetime(values["住院天数"]) or looks_postop(values["住院天数"])) ): shifted_admission = normalize_date(shifted_by_long_diagnosis.group(1)) old_last_write = values["入院时间"] old_hospital_days = values["最后书写时间"] old_discharge_or_postop = values["住院天数"] old_postop = values["出院时间"] values["诊断"] = values["诊断"][: shifted_by_long_diagnosis.start()].strip() values["入院时间"] = shifted_admission values["最后书写时间"] = normalize_date(old_last_write) values["住院天数"] = old_hospital_days values["出院时间"] = normalize_date(old_discharge_or_postop) if looks_datetime(old_discharge_or_postop) else "" if not values["手术后天数"]: if looks_postop(old_discharge_or_postop): values["手术后天数"] = old_discharge_or_postop elif looks_postop(old_postop): values["手术后天数"] = old_postop if shifted_by_long_diagnosis and not values["入院时间"]: values["诊断"] = values["诊断"][: shifted_by_long_diagnosis.start()].strip() values["入院时间"] = normalize_date(shifted_by_long_diagnosis.group(1)) if ( not looks_datetime(values["入院时间"]) and looks_datetime(values["最后书写时间"]) and looks_datetime(values["住院天数"]) and looks_days(values["出院时间"]) ): diagnosis_extra = values["入院时间"] old_admission = values["最后书写时间"] old_last_write = values["住院天数"] old_days = values["出院时间"] old_discharge = values["手术后天数"] if diagnosis_extra: values["诊断"] = normalize_text(f"{values['诊断']} {diagnosis_extra}") values["入院时间"] = normalize_date(old_admission) values["最后书写时间"] = normalize_date(old_last_write) values["住院天数"] = old_days values["出院时间"] = normalize_date(old_discharge) if looks_datetime(old_discharge) else "" values["手术后天数"] = "" days_with_discharge = re.fullmatch( r"(\d+)\s+(\d{4}[-/.]\d{1,2}[-/.]\d{1,2}\s+\d{1,2}:\d{1,2}:\d{1,2})", str(values["住院天数"]), ) if days_with_discharge: values["住院天数"] = days_with_discharge.group(1) if not values["出院时间"]: values["出院时间"] = normalize_date(days_with_discharge.group(2)) if looks_postop(values["出院时间"]) and not values["手术后天数"]: values["手术后天数"] = values["出院时间"] values["出院时间"] = "" if values["出院时间"] in {"手术日", "当日", "当天"}: if not values["手术后天数"]: values["手术后天数"] = values["出院时间"] values["出院时间"] = "" if looks_postop(values["住院天数"]) and not values["手术后天数"]: values["手术后天数"] = values["住院天数"] values["住院天数"] = "" if looks_days(values["最后书写时间"]) and looks_datetime(values["住院天数"]): old_days = values["最后书写时间"] old_discharge = values["住院天数"] old_postop = values["出院时间"] values["最后书写时间"] = "" values["住院天数"] = old_days values["出院时间"] = normalize_date(old_discharge) if looks_postop(old_postop) and not values["手术后天数"]: values["手术后天数"] = old_postop elif looks_datetime(values["住院天数"]) and (not values["出院时间"] or looks_postop(values["出院时间"])): old_discharge = values["住院天数"] old_postop = values["出院时间"] values["住院天数"] = "" values["出院时间"] = normalize_date(old_discharge) if looks_postop(old_postop) and not values["手术后天数"]: values["手术后天数"] = old_postop last_write_with_postop = re.fullmatch(r"(.+?)\s+(后\d+天)", values["最后书写时间"]) if last_write_with_postop and not values["手术后天数"]: values["最后书写时间"] = last_write_with_postop.group(1) values["手术后天数"] = last_write_with_postop.group(2) if re.fullmatch(r"后\d+天", values["最后书写时间"]) and not values["手术后天数"]: values["手术后天数"] = values["最后书写时间"] values["最后书写时间"] = "" if values["住院天数"].isdigit(): values["住院天数"] = int(values["住院天数"]) return values def validate_patient_row(values: dict[str, Any]) -> list[str]: warnings: list[str] = [] if not values.get("姓名"): warnings.append("缺少姓名") if values.get("性别") not in {"男", "女"}: warnings.append("性别异常") if values.get("年龄") and not re.fullmatch(r"\d{1,3}岁", str(values["年龄"])): warnings.append("年龄格式异常") if not normalize_text(values.get("住院号", "")): warnings.append("缺少住院号") if not values.get("入院时间"): warnings.append("缺少入院时间") elif not re.fullmatch(r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}", str(values["入院时间"])): warnings.append("入院时间格式异常") if values.get("出院时间") and not re.fullmatch( r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}", str(values["出院时间"]) ): warnings.append("出院时间格式异常") if ( re.fullmatch(r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}", str(values.get("入院时间", ""))) and re.fullmatch(r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}", str(values.get("出院时间", ""))) and str(values["入院时间"]) > str(values["出院时间"]) ): warnings.append("出院时间早于入院时间") if values.get("最后书写时间") and not re.fullmatch( r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}", str(values["最后书写时间"]) ): warnings.append("最后书写时间格式异常") if values.get("住院天数") != "" and not isinstance(values.get("住院天数"), int): warnings.append("住院天数格式异常") return warnings def is_blank_or_footer(row: dict[str, Any]) -> bool: if row.get("住院号"): return False filled = [value for value in row.values() if value not in ("", None)] return len(filled) == 0 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 write_csv(path: Path, records: list[dict[str, Any]]) -> None: fieldnames = [ "大科室", "子科室", "来源文件夹", "图片名", "图片内行号", *COLUMNS, "复核状态", "复核提示", ] 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: patient = record["患者信息"] writer.writerow( { "大科室": record["大科室"], "子科室": record["子科室"], "来源文件夹": record["来源文件夹"], "图片名": record["图片信息"]["图片名"], "图片内行号": record["图片信息"]["图片内行号"], **patient, "复核状态": record["复核"]["状态"], "复核提示": "；".join(record["复核"]["提示"]), } ) def load_corrections(path: Path) -> dict[tuple[str, int], dict[str, Any]]: if not path.exists(): return {} data = json.loads(path.read_text(encoding="utf-8")) corrections: dict[tuple[str, int], dict[str, Any]] = {} for item in data: image_path = item["图片路径"] row_no = int(item["图片内行号"]) corrections[(image_path, row_no)] = item for prefix in ["已处理-患者目录图片集群/", "待处理-患者目录图片集群/"]: corrections[(prefix + image_path, row_no)] = item return corrections def apply_review_options(warnings: list[str], correction: dict[str, Any]) -> list[str]: return warnings def apply_corrections(records: list[dict[str, Any]], corrections: dict[tuple[str, int], dict[str, Any]]) -> None: for record in records: key = (record["图片信息"]["图片路径"], int(record["图片信息"]["图片内行号"])) if key not in corrections: record["复核"]["提示"] = validate_patient_row(record["患者信息"]) record["复核"]["状态"] = "需人工复核" if record["复核"]["提示"] else "自动复核通过" continue correction = corrections[key] record["患者信息"].update(correction.get("患者信息", {})) record["图片信息"].update(correction.get("图片信息", {})) record["复核"]["人工修正"] = True record["复核"]["复核选项"] = correction.get("复核选项", {}) if correction.get("复核备注"): record["复核"]["人工备注"] = correction.get("复核备注", "") record["复核"]["提示"] = apply_review_options(validate_patient_row(record["患者信息"]), correction) record["复核"]["状态"] = "需人工复核" if record["复核"]["提示"] else "人工复核通过" def record_quality_rank(record: dict[str, Any]) -> tuple[int, int, int, int]: patient = record["患者信息"] review = record["复核"] review_ok = 0 if review.get("状态") == "需人工复核" else 1 manual_corrected = 1 if review.get("人工修正") else 0 date_count = int(bool(patient.get("入院时间"))) + int(bool(patient.get("出院时间"))) filled_count = sum(1 for column in COLUMNS if patient.get(column) not in ("", None)) return (review_ok, manual_corrected, date_count, filled_count) def summarize_record_for_duplicate(record: dict[str, Any]) -> dict[str, Any]: patient = record["患者信息"] image = record["图片信息"] return { "大科室": record["大科室"], "子科室": record["子科室"], "来源文件夹": record["来源文件夹"], "图片路径": image["图片路径"], "图片名": image["图片名"], "图片内行号": image["图片内行号"], "姓名": patient.get("姓名", ""), "住院号": patient.get("住院号", ""), "入院时间": patient.get("入院时间", ""), "出院时间": patient.get("出院时间", ""), "复核状态": record["复核"].get("状态", ""), } def deduplicate_records_by_inpatient_no( records: list[dict[str, Any]], ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: kept_by_no: dict[str, dict[str, Any]] = {} kept_order: list[str] = [] dropped: list[dict[str, Any]] = [] for record in records: inpatient_no = normalize_text(record["患者信息"].get("住院号", "")) if not inpatient_no: dropped.append( { "住院号": "", "保留记录": {}, "剔除记录": summarize_record_for_duplicate(record), "规则": "住院号为空，未纳入归档结果和数据库", } ) continue if inpatient_no not in kept_by_no: kept_order.append(inpatient_no) else: dropped.append( { "住院号": inpatient_no, "保留记录": summarize_record_for_duplicate(record), "剔除记录": summarize_record_for_duplicate(kept_by_no[inpatient_no]), "规则": "住院号重复，后出现记录覆盖先出现记录", } ) kept_by_no[inpatient_no] = record return [kept_by_no[inpatient_no] for inpatient_no in kept_order], dropped def process_folder( folder: Path, output_root: Path, department: Department, args: argparse.Namespace, secret_id: str, secret_key: str, ) -> tuple[list[dict[str, Any]], dict[str, Any]]: images = sorted( [path for path in folder.iterdir() if path.is_file() and path.suffix.lower() in IMAGE_EXTENSIONS], key=natural_key, ) folder_key = folder.name composites_dir = output_root / "merged_images" / folder_key raw_dir = output_root / "raw_ocr" / folder_key records: list[dict[str, Any]] = [] folder_warnings: list[str] = [] group_infos: list[dict[str, Any]] = [] def cache_label(label: str) -> str: if args.ocr_engine != "table-v3": label = f"{args.ocr_engine.replace('-', '_')}_{label}" if args.image_padding_y > 0: return f"{label}_pady{args.image_padding_y}" return label def records_from_rows( rows: list[list[str]], group_paths: list[Path], group_index_value: int, composite_path_value: Path, cache_path_value: Path, request_id: str | None, ) -> list[dict[str, Any]]: built_records: list[dict[str, Any]] = [] row_counts = [infer_rows_for_image(path, args.rows_per_image) for path in group_paths] for table_row_index, row in enumerate(rows): patient = clean_patient_row(row) if is_blank_or_footer(patient): continue source_index, image_row = locate_source_row(table_row_index, row_counts) source_path = group_paths[source_index] warnings = validate_patient_row(patient) record = { "大科室": department.major, "子科室": department.sub, "来源文件夹": folder.name, "标准化文件夹科室名": department.raw_name, "患者信息": patient, "图片信息": { "图片路径": str(source_path), "图片名": source_path.name, "图片序号": natural_key(source_path), "图片内行号": image_row + 1, "拼接组序号": group_index_value, "拼接图片路径": str(composite_path_value), "OCR缓存路径": str(cache_path_value), "OCR请求ID": request_id, }, "复核": { "状态": "需人工复核" if warnings else "自动复核通过", "提示": warnings, }, } built_records.append(record) return built_records def expected_row_count(group_paths: list[Path]) -> int: return sum(infer_rows_for_image(path, args.rows_per_image) for path in group_paths) def attempt_ocr_group( group_paths: list[Path], group_index_value: int, label: str, display_label: str, prebuilt_merge_info: dict[str, Any] | None = None, ) -> tuple[dict[str, Any], list[dict[str, Any]]]: composite_path = composites_dir / f"{label}.png" merge_info = prebuilt_merge_info or merge_images(group_paths, composite_path, args.image_padding_y) cache_path = raw_dir / f"{label}.json" print(f" {display_label}: {len(group_paths)} images -> {composite_path}", flush=True) if args.ocr_engine == "general-accurate": response = call_general_accurate( composite_path, cache_path, secret_id, secret_key, args.region, args.timeout, args.force, args.max_retries, ) else: response = call_table_v3( composite_path, cache_path, secret_id, secret_key, args.region, args.timeout, args.force, args.max_retries, ) rows = ocr_response_to_rows(response, args.ocr_engine, int(merge_info.get("width", 0))) expected_rows = expected_row_count(group_paths) if len(group_paths) > 1 and len(rows) < expected_rows: raise RuntimeError(f"识别行数偏少: {len(rows)} / {expected_rows}") request_id = response.get("RequestId") print(f" rows: {len(rows)} request: {request_id}", flush=True) info = { "label": label, "merged_image": merge_info, "ocr_cache": str(cache_path), "ocr_request_id": request_id, "row_count": len(rows), "expected_row_count": expected_rows, "image_count": len(group_paths), } return info, records_from_rows(rows, group_paths, group_index_value, composite_path, cache_path, request_id) def run_chunked_fallback( group: list[Path], group_index_value: int, main_label: str, fallback_size: int, ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: candidate_infos: list[dict[str, Any]] = [] candidate_records: list[dict[str, Any]] = [] chunks = list(enumerate(batched(group, fallback_size))) def run_chunk(item: tuple[int, list[Path]]) -> tuple[dict[str, Any], list[dict[str, Any]]]: chunk_index, chunk = item chunk_label = f"{main_label}_fallback{fallback_size}_{chunk_index:02d}" chunk_info, chunk_records = attempt_ocr_group( chunk, group_index_value, chunk_label, chunk_label, ) chunk_info["part_index"] = chunk_index return chunk_info, chunk_records if args.workers > 1 and len(chunks) > 1: with concurrent.futures.ThreadPoolExecutor(max_workers=min(args.workers, len(chunks))) as executor: results = list(executor.map(run_chunk, chunks)) else: results = [run_chunk(item) for item in chunks] for chunk_info, chunk_records in results: candidate_infos.append(chunk_info) candidate_records.extend(chunk_records) return candidate_infos, candidate_records def single_cache_paths(main_label: str, group: list[Path]) -> list[Path]: return [raw_dir / f"{main_label}_part_{part_index:02d}.json" for part_index in range(len(group))] def chunk_cache_paths(main_label: str, group: list[Path], fallback_size: int) -> list[Path]: chunks = batched(group, fallback_size) return [raw_dir / f"{main_label}_fallback{fallback_size}_{chunk_index:02d}.json" for chunk_index in range(len(chunks))] def run_single_fallback( group: list[Path], group_index_value: int, main_label: str, main_merge_info: dict[str, Any], main_cache_path: Path, initial_error: str, fallback_errors: list[str], adaptive_from_previous: bool = False, ) -> None: print(f" 尝试单张OCR回退", flush=True) single_infos: list[dict[str, Any]] = [] for part_index, single_path in enumerate(group): single_label = f"{main_label}_part_{part_index:02d}" single_composite_path = composites_dir / f"{single_label}.png" single_cache_path = raw_dir / f"{single_label}.json" single_merge_info = merge_images([single_path], single_composite_path, args.image_padding_y) try: single_info, single_records = attempt_ocr_group( [single_path], group_index_value, single_label, single_label, single_merge_info, ) single_info["part_index"] = part_index records.extend(single_records) single_infos.append(single_info) except RuntimeError as single_exc: single_message = f"{single_label} OCR失败: {single_exc}" print(f" {single_message}", flush=True) folder_warnings.append(single_message) single_infos.append( { "label": single_label, "part_index": part_index, "merged_image": single_merge_info, "ocr_cache": str(single_cache_path), "ocr_request_id": None, "row_count": 0, "expected_row_count": expected_row_count([single_path]), "image_count": 1, "error": str(single_exc), } ) group_infos.append( { "group_index": group_index_value, "label": main_label, "merged_image": main_merge_info, "ocr_cache": str(main_cache_path), "ocr_request_id": None, "row_count": sum(item["row_count"] for item in single_infos), "expected_row_count": expected_row_count(group), "fallback": True, "fallback_strategy": "single_image", "fallback_parts": single_infos, "initial_error": initial_error, "fallback_errors": fallback_errors, "adaptive_from_previous": adaptive_from_previous, } ) groups = batched(images, args.batch_size) if args.limit_groups_per_folder: groups = groups[: args.limit_groups_per_folder] preferred_fallback_size: int | None = None preferred_fallback_failures = 0 for group_index, group in enumerate(groups): main_label = cache_label(f"group_{group_index:04d}") main_composite_path = composites_dir / f"{main_label}.png" main_cache_path = raw_dir / f"{main_label}.json" main_merge_info = merge_images(group, main_composite_path, args.image_padding_y) part_cache_paths = single_cache_paths(main_label, group) if not main_cache_path.exists() and all(path.exists() for path in part_cache_paths): message = f"{main_label} 使用已有单图缓存重建" print(f" {message}", flush=True) run_single_fallback( group, group_index, main_label, main_merge_info, main_cache_path, message, [], adaptive_from_previous=True, ) time.sleep(args.sleep) continue if preferred_fallback_size and len(group) > 1: use_preferred_fallback = True if args.rebuild_from_cache: if preferred_fallback_size == 1: preferred_paths = single_cache_paths(main_label, group) else: preferred_paths = chunk_cache_paths(main_label, group, int(preferred_fallback_size)) use_preferred_fallback = all(path.exists() for path in preferred_paths) if not use_preferred_fallback: print(f" {main_label} 当前回退缓存不完整，先尝试主缓存", flush=True) if not use_preferred_fallback: preferred_fallback_size = None elif preferred_fallback_size == 1: message = f"{main_label} 依据前序组结果，直接使用单张OCR" print(f" {message}", flush=True) run_single_fallback( group, group_index, main_label, main_merge_info, main_cache_path, message, [], adaptive_from_previous=True, ) time.sleep(args.sleep) continue elif 1 < preferred_fallback_size < len(group): message = f"{main_label} 依据前序组结果，直接使用 {preferred_fallback_size} 张拼接" print(f" {message}", flush=True) try: candidate_infos, candidate_records = run_chunked_fallback( group, group_index, main_label, int(preferred_fallback_size), ) records.extend(candidate_records) group_infos.append( { "group_index": group_index, "label": main_label, "merged_image": main_merge_info, "ocr_cache": str(main_cache_path), "ocr_request_id": None, "row_count": sum(item["row_count"] for item in candidate_infos), "expected_row_count": expected_row_count(group), "fallback": True, "fallback_strategy": f"{preferred_fallback_size}_images", "fallback_parts": candidate_infos, "initial_error": message, "fallback_errors": [], "adaptive_from_previous": True, } ) time.sleep(args.sleep) preferred_fallback_failures = 0 continue except RuntimeError as adaptive_exc: failed_size = int(preferred_fallback_size) preferred_fallback_failures += 1 adaptive_message = f"{main_label} 前序 {failed_size} 张策略未通过，本组改用单张OCR: {adaptive_exc}" print(f" {adaptive_message}", flush=True) folder_warnings.append(adaptive_message) if preferred_fallback_failures >= args.fallback_demote_threshold: preferred_fallback_size = 1 demote_message = ( f"{main_label} 连续 {preferred_fallback_failures} 个拼接组未通过，" "后续直接使用单张OCR" ) print(f" {demote_message}", flush=True) folder_warnings.append(demote_message) else: preferred_fallback_size = failed_size run_single_fallback( group, group_index, main_label, main_merge_info, main_cache_path, adaptive_message, [], adaptive_from_previous=True, ) time.sleep(args.sleep) continue try: main_info, main_records = attempt_ocr_group( group, group_index, main_label, main_label, main_merge_info, ) main_info.update( { "group_index": group_index, "fallback": False, "fallback_strategy": None, } ) group_infos.append(main_info) records.extend(main_records) except RuntimeError as exc: if len(group) == 1: message = f"{main_label} OCR失败: {exc}" print(f" {message}", flush=True) folder_warnings.append(message) group_infos.append( { "group_index": group_index, "label": main_label, "merged_image": main_merge_info, "ocr_cache": str(main_cache_path), "ocr_request_id": None, "row_count": 0, "expected_row_count": expected_row_count(group), "fallback": False, "fallback_strategy": None, "error": str(exc), } ) time.sleep(args.sleep) continue message = f"{main_label} {len(group)}张拼接OCR未通过，开始降档: {exc}" print(f" {message}", flush=True) folder_warnings.append(message) fallback_errors: list[str] = [] fallback_success = False for fallback_size in (4, 3, 2): if not 1 < fallback_size < len(group): continue print(f" 尝试 {fallback_size} 张拼接回退", flush=True) try: candidate_infos, candidate_records = run_chunked_fallback(group, group_index, main_label, fallback_size) except RuntimeError as fallback_exc: fallback_message = f"{main_label} {fallback_size}张拼接仍未通过: {fallback_exc}" print(f" {fallback_message}", flush=True) folder_warnings.append(fallback_message) fallback_errors.append(fallback_message) continue records.extend(candidate_records) preferred_fallback_size = fallback_size preferred_fallback_failures = 0 group_infos.append( { "group_index": group_index, "label": main_label, "merged_image": main_merge_info, "ocr_cache": str(main_cache_path), "ocr_request_id": None, "row_count": sum(item["row_count"] for item in candidate_infos), "expected_row_count": expected_row_count(group), "fallback": True, "fallback_strategy": f"{fallback_size}_images", "fallback_parts": candidate_infos, "initial_error": str(exc), "fallback_errors": fallback_errors, } ) fallback_success = True break if fallback_success: time.sleep(args.sleep) continue preferred_fallback_size = 1 run_single_fallback(group, group_index, main_label, main_merge_info, main_cache_path, str(exc), fallback_errors) time.sleep(args.sleep) summary = { "来源文件夹": folder.name, "大科室": department.major, "子科室": department.sub, "图片数": len(images), "记录数": len(records), "拼接组数": len(group_infos), "拼接组": group_infos, "提示": folder_warnings, } return records, summary def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--input", default="列表", help="患者列表图片根目录") parser.add_argument("--output", default="数据处理工作区/列表归档结果", help="输出目录") parser.add_argument("--departments", default="数据处理工作区/01_科室分类规则.json", help="科室分类 JSON") parser.add_argument("--batch-size", type=int, default=3, help="每次拼接图片数量，建议 3 或 4") parser.add_argument("--image-padding-y", type=int, default=24, help="每张图拼接前添加的上下白色边界像素数") parser.add_argument("--rows-per-image", type=int, default=0, help="每张 HIS 列表截图的行数；0 表示按图片高度自动推断") parser.add_argument("--corrections", default="数据处理工作区/03_人工复核修正.json", help="人工复核修正 JSON") parser.add_argument( "--ocr-engine", choices=sorted(OCR_ENGINE_LABELS), default="table-v3", help="OCR 引擎：table-v3 表格识别 V3", ) parser.add_argument("--region", default="ap-shanghai", help="腾讯云 OCR 地域") parser.add_argument("--timeout", type=int, default=60, help="单次 OCR 超时秒数") parser.add_argument("--sleep", type=float, default=0.2, help="OCR 调用间隔秒数") parser.add_argument("--max-retries", type=int, default=0, help="OCR 调用失败重试次数") parser.add_argument( "--fallback-demote-threshold", type=int, default=3, help="同一科室连续多少个拼接组失败后，才把后续组整体降为单张OCR", ) parser.add_argument("--workers", type=int, default=1, help="同一拼接组内并发OCR分片数；建议 1-2") parser.add_argument("--folder-workers", type=int, default=1, help="并发处理科室目录数；OCR接口稳定时可设为 2-4") parser.add_argument("--force", 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 个拼接组") parser.add_argument("--rebuild-from-cache", action="store_true", help="只用已有 OCR 缓存重建结果，不发起 OCR 请求") return parser.parse_args() def main() -> None: args = parse_args() if args.batch_size < 1: raise ValueError("--batch-size 必须大于 0") if args.workers < 1: raise ValueError("--workers 必须大于 0") if args.folder_workers < 1: raise ValueError("--folder-workers 必须大于 0") if args.image_padding_y < 0: raise ValueError("--image-padding-y 不能小于 0") input_root = Path(args.input) output_root = Path(args.output) sub_to_major, aliases = load_departments(Path(args.departments)) if args.rebuild_from_cache: secret_id, secret_key = "", "" else: secret_id, secret_key = get_credentials() corrections = load_corrections(Path(args.corrections)) folders = sorted([path for path in input_root.iterdir() if path.is_dir()], key=lambda path: path.name) if args.limit_folders: folders = folders[: args.limit_folders] all_records: list[dict[str, Any]] = [] folder_summaries: list[dict[str, Any]] = [] classifications: list[dict[str, Any]] = [] folder_jobs: list[tuple[Path, Department]] = [] for folder in folders: department = classify_department(folder.name, sub_to_major, aliases) classifications.append( { "来源文件夹": folder.name, "标准化文件夹科室名": department.raw_name, "大科室": department.major, "子科室": department.sub, } ) folder_jobs.append((folder, department)) def run_folder_job(item: tuple[Path, Department]) -> tuple[list[dict[str, Any]], dict[str, Any]]: folder, department = item print(f"[{folder.name}] -> {department.major} / {department.sub}", flush=True) return process_folder(folder, output_root, department, args, secret_id, secret_key) if args.folder_workers > 1 and len(folder_jobs) > 1: with concurrent.futures.ThreadPoolExecutor(max_workers=min(args.folder_workers, len(folder_jobs))) as executor: folder_results = list(executor.map(run_folder_job, folder_jobs)) else: folder_results = [run_folder_job(item) for item in folder_jobs] for records, summary in folder_results: all_records.extend(records) folder_summaries.append(summary) apply_corrections(all_records, corrections) record_count_before_dedup = len(all_records) all_records, duplicate_records = deduplicate_records_by_inpatient_no(all_records) issue_records = [record for record in all_records if record["复核"]["状态"] == "需人工复核"] corrected_records = [record for record in all_records if record["复核"].get("人工修正")] archive = { "生成时间": dt.datetime.now().isoformat(timespec="seconds"), "输入目录": str(input_root), "OCR引擎": OCR_ENGINE_LABELS[args.ocr_engine], "拼接设置": { "batch_size": args.batch_size, "rows_per_image": args.rows_per_image, "image_padding_y": args.image_padding_y, }, "科室归类": classifications, "汇总": { "科室目录数": len(folders), "图片数": sum(item["图片数"] for item in folder_summaries), "去重前患者记录数": record_count_before_dedup, "患者记录数": len(all_records), "需人工复核记录数": len(issue_records), "人工修正记录数": len(corrected_records), "重复住院号剔除记录数": len(duplicate_records), }, "科室汇总": folder_summaries, "重复住院号剔除记录": duplicate_records, "患者记录": all_records, } review_report = { "生成时间": archive["生成时间"], "汇总": archive["汇总"], "需人工复核记录": issue_records, "人工修正记录": corrected_records, "重复住院号剔除记录": duplicate_records, "科室级提示": [ {"来源文件夹": item["来源文件夹"], "提示": item["提示"]} for item in folder_summaries if item["提示"] ], } write_json(output_root / "列表_科室归类.json", classifications) write_json(output_root / "患者列表_结构化.json", archive) write_jsonl(output_root / "患者列表_记录.jsonl", all_records) write_csv(output_root / "患者列表_记录.csv", all_records) write_json(output_root / "复核报告.json", review_report) write_json(output_root / "重复住院号报告.json", duplicate_records) print(json.dumps(archive["汇总"], ensure_ascii=False, indent=2), flush=True) if __name__ == "__main__": main()