257 lines
8.6 KiB
Python
257 lines
8.6 KiB
Python
#!/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()
|