225 lines
9.6 KiB
Python
Executable File
225 lines
9.6 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
从 PostgreSQL 导出患者首页复核表，并把人工复核结果回写到目标首页表。
|
||
|
||
依赖系统命令 psql；数据库密码通过 PGPASSWORD 环境变量或 --pg-password 传入。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import os
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
from pathlib import Path
|
||
|
||
|
||
PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
||
DEFAULT_OUTPUT = PROJECT_ROOT / "数据处理结果区" / "04_复核与人工校正" / "患者首页_人工复核表.csv"
|
||
DEFAULT_TABLE = os.environ.get("PGTABLE") or os.environ.get("PG_PATIENT_TABLE", "")
|
||
|
||
|
||
def quote_pg_identifier(identifier: str) -> str:
|
||
if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", identifier):
|
||
raise ValueError(f"非法 PostgreSQL 表名：{identifier}")
|
||
return '"' + identifier.replace('"', '""') + '"'
|
||
|
||
|
||
def connection_env(args: argparse.Namespace) -> dict[str, str]:
|
||
env = os.environ.copy()
|
||
env.update(
|
||
{
|
||
"PGHOST": args.pg_host or os.environ.get("PGHOST", ""),
|
||
"PGPORT": str(args.pg_port or os.environ.get("PGPORT", "")),
|
||
"PGDATABASE": args.pg_database or os.environ.get("PGDATABASE", ""),
|
||
"PGUSER": args.pg_user or os.environ.get("PGUSER", ""),
|
||
"PGPASSWORD": args.pg_password or os.environ.get("PGPASSWORD", ""),
|
||
}
|
||
)
|
||
missing = [name for name in ["PGHOST", "PGPORT", "PGDATABASE", "PGUSER", "PGPASSWORD"] if not env.get(name)]
|
||
if missing:
|
||
raise RuntimeError("缺少 PostgreSQL 连接配置：" + "、".join(missing))
|
||
return env
|
||
|
||
|
||
def run_psql(sql: str, args: argparse.Namespace) -> None:
|
||
if shutil.which("psql") is None:
|
||
raise RuntimeError("未找到 psql。请先安装 PostgreSQL 客户端。")
|
||
with tempfile.TemporaryDirectory(prefix="patient_front_review_") as tmpdir:
|
||
sql_path = Path(tmpdir) / "review.sql"
|
||
sql_path.write_text(sql, encoding="utf-8")
|
||
completed = subprocess.run(
|
||
["psql", "--no-password", "--file", str(sql_path)],
|
||
check=False,
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.PIPE,
|
||
text=True,
|
||
encoding="utf-8",
|
||
errors="replace",
|
||
env=connection_env(args),
|
||
)
|
||
if completed.returncode != 0:
|
||
raise RuntimeError(completed.stderr.strip() or completed.stdout.strip() or "psql 执行失败")
|
||
if completed.stdout.strip():
|
||
print(completed.stdout.strip())
|
||
|
||
|
||
def export_review(args: argparse.Namespace) -> None:
|
||
output = args.output.resolve()
|
||
output.parent.mkdir(parents=True, exist_ok=True)
|
||
escaped_output = str(output).replace("'", "''")
|
||
table = quote_pg_identifier(args.pg_table)
|
||
where_sql = "" if args.all else "WHERE review_status <> 'auto_pass' OR manual_corrected IS TRUE"
|
||
select_sql = (
|
||
"SELECT inpatient_no, source_file, medical_record_no, front_page_medical_record_no, patient_name, "
|
||
"gender, birth_date, discharge_dept, major_department, primary_diagnosis, "
|
||
"primary_diagnosis_code, total_cost, self_pay_amount, review_status, "
|
||
"review_notes::text AS review_notes, auto_corrections::text AS auto_corrections, "
|
||
"manual_corrected, contact_address, contact_phone, discharge_diagnoses::text AS discharge_diagnoses, "
|
||
"operations::text AS operations, pathology_diagnosis, pathology_diagnosis_code, pathology_no, "
|
||
"discharge_disposition_code, readmission_plan_code, quality_status, quality_notes::text AS quality_notes, "
|
||
"'' AS manual_review_notes, '' AS corrected_medical_record_no, "
|
||
"'' AS corrected_patient_name, '' AS corrected_major_department, "
|
||
"'' AS corrected_primary_diagnosis, '' AS corrected_primary_diagnosis_code, "
|
||
"'' AS corrected_total_cost, '' AS corrected_self_pay_amount, '' AS mark_review_status "
|
||
f"FROM {table} {where_sql} ORDER BY source_file"
|
||
)
|
||
sql = f"\\set ON_ERROR_STOP on\n\\copy ({select_sql}) TO '{escaped_output}' WITH (FORMAT csv, HEADER true, ENCODING 'UTF8')\n"
|
||
run_psql(sql, args)
|
||
print(f"已导出复核表：{output}")
|
||
|
||
|
||
def import_review(args: argparse.Namespace) -> None:
|
||
input_path = args.input.resolve()
|
||
if not input_path.exists():
|
||
raise RuntimeError(f"复核表不存在：{input_path}")
|
||
escaped_input = str(input_path).replace("'", "''")
|
||
table = quote_pg_identifier(args.pg_table)
|
||
sql = f"""
|
||
\\set ON_ERROR_STOP on
|
||
CREATE TEMP TABLE patient_front_review_import (
|
||
inpatient_no TEXT,
|
||
source_file TEXT,
|
||
medical_record_no TEXT,
|
||
front_page_medical_record_no TEXT,
|
||
patient_name TEXT,
|
||
gender TEXT,
|
||
birth_date TEXT,
|
||
discharge_dept TEXT,
|
||
major_department TEXT,
|
||
primary_diagnosis TEXT,
|
||
primary_diagnosis_code TEXT,
|
||
total_cost TEXT,
|
||
self_pay_amount TEXT,
|
||
review_status TEXT,
|
||
review_notes TEXT,
|
||
auto_corrections TEXT,
|
||
manual_corrected TEXT,
|
||
contact_address TEXT,
|
||
contact_phone TEXT,
|
||
discharge_diagnoses TEXT,
|
||
operations TEXT,
|
||
pathology_diagnosis TEXT,
|
||
pathology_diagnosis_code TEXT,
|
||
pathology_no TEXT,
|
||
discharge_disposition_code TEXT,
|
||
readmission_plan_code TEXT,
|
||
quality_status TEXT,
|
||
quality_notes TEXT,
|
||
manual_review_notes TEXT,
|
||
corrected_medical_record_no TEXT,
|
||
corrected_patient_name TEXT,
|
||
corrected_major_department TEXT,
|
||
corrected_primary_diagnosis TEXT,
|
||
corrected_primary_diagnosis_code TEXT,
|
||
corrected_total_cost TEXT,
|
||
corrected_self_pay_amount TEXT,
|
||
mark_review_status TEXT
|
||
);
|
||
|
||
\\copy patient_front_review_import FROM '{escaped_input}' WITH (FORMAT csv, HEADER true, ENCODING 'UTF8')
|
||
|
||
UPDATE {table} AS t
|
||
SET
|
||
medical_record_no = COALESCE(NULLIF(i.corrected_medical_record_no, ''), t.medical_record_no),
|
||
patient_name = COALESCE(NULLIF(i.corrected_patient_name, ''), t.patient_name),
|
||
major_department = COALESCE(NULLIF(i.corrected_major_department, ''), t.major_department),
|
||
primary_diagnosis = COALESCE(NULLIF(i.corrected_primary_diagnosis, ''), t.primary_diagnosis),
|
||
primary_diagnosis_code = COALESCE(NULLIF(i.corrected_primary_diagnosis_code, ''), t.primary_diagnosis_code),
|
||
total_cost = COALESCE(NULLIF(i.corrected_total_cost, '')::numeric, t.total_cost),
|
||
self_pay_amount = COALESCE(NULLIF(i.corrected_self_pay_amount, '')::numeric, t.self_pay_amount),
|
||
review_status = COALESCE(NULLIF(i.mark_review_status, ''), 'reviewed'),
|
||
review_notes = CASE
|
||
WHEN NULLIF(i.manual_review_notes, '') IS NULL THEN t.review_notes
|
||
ELSE t.review_notes || jsonb_build_array(i.manual_review_notes)
|
||
END,
|
||
manual_corrected = CASE
|
||
WHEN NULLIF(i.corrected_medical_record_no, '') IS NOT NULL
|
||
OR NULLIF(i.corrected_patient_name, '') IS NOT NULL
|
||
OR NULLIF(i.corrected_major_department, '') IS NOT NULL
|
||
OR NULLIF(i.corrected_primary_diagnosis, '') IS NOT NULL
|
||
OR NULLIF(i.corrected_primary_diagnosis_code, '') IS NOT NULL
|
||
OR NULLIF(i.corrected_total_cost, '') IS NOT NULL
|
||
OR NULLIF(i.corrected_self_pay_amount, '') IS NOT NULL
|
||
OR lower(COALESCE(i.manual_corrected, '')) IN ('true', 't', '1', 'yes', 'y')
|
||
THEN true
|
||
ELSE t.manual_corrected
|
||
END
|
||
FROM patient_front_review_import AS i
|
||
WHERE (
|
||
NULLIF(i.inpatient_no, '') IS NOT NULL
|
||
AND t.inpatient_no = i.inpatient_no
|
||
) OR (
|
||
NULLIF(i.inpatient_no, '') IS NULL
|
||
AND t.source_file = i.source_file
|
||
);
|
||
""".strip()
|
||
run_psql(sql, args)
|
||
print(f"已回写人工复核表：{input_path}")
|
||
|
||
|
||
def build_parser() -> argparse.ArgumentParser:
|
||
parser = argparse.ArgumentParser(description="导出/回写患者首页 PostgreSQL 人工复核表。")
|
||
subparsers = parser.add_subparsers(dest="command", required=True)
|
||
|
||
def add_pg_options(subparser: argparse.ArgumentParser) -> None:
|
||
subparser.add_argument("--pg-host", default=os.environ.get("PGHOST", ""), help="PostgreSQL 主机；也可用 PGHOST")
|
||
subparser.add_argument("--pg-port", default=os.environ.get("PGPORT", "5432"), help="PostgreSQL 端口；也可用 PGPORT")
|
||
subparser.add_argument("--pg-database", default=os.environ.get("PGDATABASE", ""), help="PostgreSQL 数据库；也可用 PGDATABASE")
|
||
subparser.add_argument("--pg-user", default=os.environ.get("PGUSER", ""), help="PostgreSQL 用户名；也可用 PGUSER")
|
||
subparser.add_argument("--pg-password", default=os.environ.get("PGPASSWORD", ""), help="PostgreSQL 密码；也可用 PGPASSWORD")
|
||
subparser.add_argument("--pg-table", default=os.environ.get("PG_PATIENT_TABLE", DEFAULT_TABLE), help="PostgreSQL 表名")
|
||
|
||
export_parser = subparsers.add_parser("export", help="从 PostgreSQL 导出待复核 CSV")
|
||
add_pg_options(export_parser)
|
||
export_parser.add_argument("-o", "--output", type=Path, default=DEFAULT_OUTPUT, help="复核 CSV 输出路径")
|
||
export_parser.add_argument("--all", action="store_true", help="导出全部记录，而不仅是需复核/已人工修正记录")
|
||
|
||
import_parser = subparsers.add_parser("import", help="把人工复核 CSV 回写 PostgreSQL")
|
||
add_pg_options(import_parser)
|
||
import_parser.add_argument("-i", "--input", type=Path, default=DEFAULT_OUTPUT, help="人工复核 CSV 路径")
|
||
return parser
|
||
|
||
|
||
def main() -> int:
|
||
args = build_parser().parse_args()
|
||
if args.command == "export":
|
||
export_review(args)
|
||
elif args.command == "import":
|
||
import_review(args)
|
||
else:
|
||
raise RuntimeError(f"未知命令：{args.command}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
raise SystemExit(main())
|
||
except Exception as exc: # noqa: BLE001 - 命令行工具需要清楚输出错误
|
||
print(f"错误：{exc}", file=sys.stderr)
|
||
raise SystemExit(1)
|