#!/usr/bin/env python3 # -*- coding: utf-8 -*- """Sync consolidated HIS patient records into PostgreSQL single table. Database credentials are read from environment variables by default: HIS_DB_HOST, HIS_DB_PORT, HIS_DB_NAME, HIS_DB_USER, HIS_DB_PASSWORD. """ from __future__ import annotations import argparse import csv import json import os import re import subprocess import tempfile from pathlib import Path from typing import Any DEFAULT_RESULT_JSON = Path("数据处理结果区/合并_患者列表_结构化.json") DEFAULT_SCHEMA_SQL = Path("数据处理工作区/06_PostgreSQL建表结构.sql") TABLE_NAME = '"Patient_Lists"' DATE_TIME_RE = re.compile(r"^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$") PATIENT_FIELDS = [ "batch_name", "major_department", "sub_department", "source_folder", "image_path", "image_name", "image_row_no", "patient_name", "gender", "age", "inpatient_no", "diagnosis", "admission_time", "last_write_time", "hospital_days", "discharge_time", "postoperative_days", "review_status", "review_notes", "manual_corrected", ] def scalar(value: Any) -> str: if value is None: return "" if isinstance(value, bool): return "true" if value else "false" return str(value) def clean_integer(value: Any) -> tuple[str, str]: text = scalar(value).strip() if not text: return "", "" if text.isdigit(): return text, "" return "", f"住院天数非数字，原识别值: {text}" def review_note_items(review: dict[str, Any]) -> list[str]: tips = review.get("提示", []) if isinstance(tips, list): note_items = [str(tip) for tip in tips if str(tip) and str(tip) != "缺少出院时间"] else: note_items = [str(tips)] if str(tips) else [] if review.get("人工备注"): note_items.append(f"人工备注: {review['人工备注']}") return note_items def admission_after_discharge(admission_time: Any, discharge_time: Any) -> bool: admission = scalar(admission_time).strip() discharge = scalar(discharge_time).strip() if not admission or not discharge: return False if not DATE_TIME_RE.match(admission) or not DATE_TIME_RE.match(discharge): return False return admission > discharge def psql_quote(path: Path) -> str: return "'" + str(path.resolve()).replace("'", "''") + "'" def flatten_patient_record(record: dict[str, Any]) -> dict[str, Any]: patient = record["患者信息"] image = record["图片信息"] review = record["复核"] note_items = review_note_items(review) hospital_days, hospital_days_note = clean_integer(patient["住院天数"]) if hospital_days_note: note_items.append(hospital_days_note) review_status = review["状态"] patient_name = scalar(patient["姓名"]).strip() if not patient_name: patient_name = "缺少姓名" review_status = "需人工复核" if "缺少姓名" not in note_items: note_items.append("缺少姓名") if admission_after_discharge(patient["入院时间"], patient["出院时间"]): review_status = "需人工复核" if "入院时间晚于出院时间" not in note_items: note_items.append("入院时间晚于出院时间") review_notes = "；".join(note_items) return { "batch_name": record["处理批次"], "major_department": record["大科室"], "sub_department": record["子科室"], "source_folder": record["来源文件夹"], "image_path": image["图片路径"], "image_name": image["图片名"], "image_row_no": image["图片内行号"], "patient_name": patient_name, "gender": patient["性别"], "age": patient["年龄"], "inpatient_no": patient["住院号"], "diagnosis": patient["诊断"], "admission_time": patient["入院时间"], "last_write_time": patient["最后书写时间"], "hospital_days": hospital_days, "discharge_time": patient["出院时间"], "postoperative_days": patient["手术后天数"], "review_status": review_status, "review_notes": review_notes, "manual_corrected": bool(review.get("人工修正")), } def prepare_unique_rows(rows: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], list[str], list[str]]: kept_by_no: dict[str, dict[str, Any]] = {} kept_order: list[str] = [] replacements: list[str] = [] invalid: list[str] = [] for row in rows: inpatient_no = str(row.get("inpatient_no", "")).strip() location = f"{row['batch_name']} / {row['image_path']} / 第{row['image_row_no']}行" if not inpatient_no: invalid.append(f"{location}: 空") continue if inpatient_no not in kept_by_no: kept_order.append(inpatient_no) else: previous = kept_by_no[inpatient_no] replacements.append( f"{inpatient_no}: {previous['batch_name']} / {previous['image_path']} / 第{previous['image_row_no']}行 -> {location}" ) kept_by_no[inpatient_no] = row return [kept_by_no[inpatient_no] for inpatient_no in kept_order], replacements, invalid def load_rows(result_json: Path) -> tuple[list[dict[str, Any]], list[str], list[str]]: merged = json.loads(result_json.read_text(encoding="utf-8")) rows = [flatten_patient_record(record) for record in merged.get("患者记录", [])] return prepare_unique_rows(rows) def write_temp_csv(rows: list[dict[str, Any]]) -> Path: temp_file = tempfile.NamedTemporaryFile( mode="w", encoding="utf-8", newline="", suffix="_his_patient_lists.csv", delete=False, ) with temp_file: writer = csv.DictWriter(temp_file, fieldnames=PATIENT_FIELDS) writer.writeheader() for row in rows: writer.writerow({field: scalar(row.get(field, "")) for field in PATIENT_FIELDS}) return Path(temp_file.name) def sync_to_postgres(args: argparse.Namespace, csv_path: Path, row_count: int) -> None: sql = f"""\\set ON_ERROR_STOP on \\i {psql_quote(Path(args.schema))} TRUNCATE TABLE {TABLE_NAME} RESTART IDENTITY; \\copy {TABLE_NAME}({','.join(PATIENT_FIELDS)}) FROM {psql_quote(csv_path)} WITH (FORMAT csv, HEADER true, NULL '') """ env = os.environ.copy() password = args.password or env.get("HIS_DB_PASSWORD") if password: env["PGPASSWORD"] = 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": "Patient_Lists", "records": row_count}, ensure_ascii=False)) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--input", default=str(DEFAULT_RESULT_JSON), help="合并后的患者记录 JSON") parser.add_argument("--schema", default=str(DEFAULT_SCHEMA_SQL), help="单表建表 SQL") parser.add_argument("--host", default=os.getenv("HIS_DB_HOST", "DB_HOST")) parser.add_argument("--port", default=os.getenv("HIS_DB_PORT", "5432")) parser.add_argument("--dbname", default=os.getenv("HIS_DB_NAME", "DB_NAME")) parser.add_argument("--user", default=os.getenv("HIS_DB_USER", "DB_USER")) parser.add_argument("--password", default="", help="数据库密码；建议改用 HIS_DB_PASSWORD 环境变量") return parser.parse_args() def main() -> None: args = parse_args() rows, replacements, invalid = load_rows(Path(args.input)) csv_path = write_temp_csv(rows) try: sync_to_postgres(args, csv_path, len(rows)) if invalid: print( json.dumps( { "skipped_empty_inpatient_no": len(invalid), "empty_examples": invalid[:20], }, ensure_ascii=False, ) ) if replacements: print( json.dumps( { "deduplicated_by_inpatient_no": len(replacements), "replacement_examples": replacements[:20], }, ensure_ascii=False, ) ) finally: csv_path.unlink(missing_ok=True) if __name__ == "__main__": main()