1347 lines
55 KiB
Python
1347 lines
55 KiB
Python
#!/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,2}):(\d{1,2}):(\d{1,2})", date_repl, text)
|
||
return text
|
||
|
||
|
||
def date_repl(match: re.Match[str]) -> str:
|
||
year, month, day, hour, minute, second = match.groups()
|
||
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)
|
||
values["住院号"] = re.sub(r"\s+", "", values["住院号"]).upper()
|
||
if values["住院号"].startswith("ZV"):
|
||
values["住院号"] = "ZY" + values["住院号"][2:]
|
||
if values["住院号"].startswith("ZYS"):
|
||
values["住院号"] = "ZY5" + values["住院号"][3:]
|
||
if len(values["住院号"]) > 2:
|
||
prefix, number = values["住院号"][:2], values["住院号"][2:]
|
||
values["住院号"] = prefix + number.translate(str.maketrans({"O": "0", "I": "1", "L": "1", "S": "5"}))
|
||
def looks_datetime(value: Any) -> bool:
|
||
return bool(
|
||
re.fullmatch(
|
||
r"\d{4}[-/.]\d{1,2}[-/.]\d{1,2}\s+\d{1,2}:\d{1,2}:\d{1,2}",
|
||
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["出院时间"])
|
||
shifted_by_long_diagnosis = re.search(
|
||
r"(\d{4}[-/.]\d{1,2}[-/.]\d{1,2}\s+\d{1,2}:\d{1,2}:\d{1,2})$",
|
||
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))
|
||
|
||
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 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["复核"]["人工修正"] = 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()
|