176 lines
6.8 KiB
Python
Executable File
176 lines
6.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
对目标首页表做字段级体检。
|
|
|
|
输出：
|
|
- 数据处理结果区/05_质量体检/患者首页_数据库体检报告.txt
|
|
- 数据处理结果区/05_质量体检/患者首页_字段空值统计.csv
|
|
- 数据处理结果区/05_质量体检/患者首页_疑似异常记录.csv
|
|
"""
|
|
|
|
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_DIR = PROJECT_ROOT / "数据处理结果区" / "05_质量体检"
|
|
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) -> str:
|
|
if shutil.which("psql") is None:
|
|
raise RuntimeError("未找到 psql。请先安装 PostgreSQL 客户端。")
|
|
with tempfile.TemporaryDirectory(prefix="patient_front_audit_") as tmpdir:
|
|
sql_path = Path(tmpdir) / "audit.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 执行失败")
|
|
return completed.stdout
|
|
|
|
|
|
def build_empty_stat_sql(table: str) -> str:
|
|
monitored_columns = [
|
|
"inpatient_no",
|
|
"medical_record_no",
|
|
"patient_name",
|
|
"gender",
|
|
"birth_date",
|
|
"id_card_no",
|
|
"contact_address",
|
|
"admission_time",
|
|
"discharge_time",
|
|
"major_department",
|
|
"primary_diagnosis",
|
|
"primary_diagnosis_code",
|
|
"discharge_diagnoses",
|
|
"operations",
|
|
"quality_status",
|
|
"review_status",
|
|
]
|
|
selects = []
|
|
for column in monitored_columns:
|
|
identifier = quote_pg_identifier(column)
|
|
if column in {"discharge_diagnoses", "operations"}:
|
|
empty_expr = f"{identifier} IS NULL OR jsonb_array_length({identifier}) = 0"
|
|
else:
|
|
empty_expr = f"{identifier} IS NULL OR {identifier}::text = ''"
|
|
selects.append(f"SELECT '{column}' AS column_name, count(*) FILTER (WHERE {empty_expr}) AS empty_count, count(*) AS total_count FROM {table}")
|
|
return " UNION ALL ".join(selects)
|
|
|
|
|
|
def audit(args: argparse.Namespace) -> None:
|
|
output_dir = args.output_dir.resolve()
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
table = quote_pg_identifier(args.pg_table)
|
|
empty_csv = output_dir / "患者首页_字段空值统计.csv"
|
|
suspicious_csv = output_dir / "患者首页_疑似异常记录.csv"
|
|
report_path = output_dir / "患者首页_数据库体检报告.txt"
|
|
empty_stat_sql = build_empty_stat_sql(table)
|
|
suspicious_sql = (
|
|
"SELECT source_file, inpatient_no, medical_record_no, patient_name, review_status, "
|
|
"review_notes::text AS review_notes, contact_address, primary_diagnosis, primary_diagnosis_code "
|
|
f"FROM {table} "
|
|
"WHERE review_status <> 'auto_pass' "
|
|
"OR inpatient_no !~ '^ZY\\d{12}$' "
|
|
"OR medical_record_no !~ '^\\d{10}$' "
|
|
"OR contact_address ~ '急诊|门诊|其他医疗机构转入|医嘱离院' "
|
|
"OR primary_diagnosis_code IS NULL "
|
|
"OR primary_diagnosis_code = '' "
|
|
"ORDER BY source_file"
|
|
)
|
|
|
|
sql = f"""
|
|
\\set ON_ERROR_STOP on
|
|
\\pset pager off
|
|
\\o {str(report_path).replace("'", "''")}
|
|
SELECT '总记录数' AS item, count(*)::text AS value FROM {table}
|
|
UNION ALL
|
|
SELECT '需复核记录数', count(*)::text FROM {table} WHERE review_status <> 'auto_pass'
|
|
UNION ALL
|
|
SELECT '缺字段注释数', count(*)::text
|
|
FROM pg_class c
|
|
JOIN pg_namespace n ON n.oid = c.relnamespace
|
|
JOIN pg_attribute a ON a.attrelid = c.oid
|
|
WHERE n.nspname = 'public'
|
|
AND c.relname = {args.pg_table!r}
|
|
AND a.attnum > 0
|
|
AND NOT a.attisdropped
|
|
AND col_description(c.oid, a.attnum) IS NULL;
|
|
\\o
|
|
\\copy ({empty_stat_sql}) TO '{str(empty_csv).replace("'", "''")}' WITH (FORMAT csv, HEADER true, ENCODING 'UTF8')
|
|
\\copy ({suspicious_sql}) TO '{str(suspicious_csv).replace("'", "''")}' WITH (FORMAT csv, HEADER true, ENCODING 'UTF8')
|
|
""".strip()
|
|
stdout = run_psql(sql, args)
|
|
if stdout.strip():
|
|
print(stdout.strip())
|
|
print(f"体检报告：{report_path}")
|
|
print(f"字段空值统计：{empty_csv}")
|
|
print(f"疑似异常记录：{suspicious_csv}")
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description="核验目标首页表字段注释、空值和疑似错位记录。")
|
|
parser.add_argument("--pg-host", default=os.environ.get("PGHOST", ""), help="PostgreSQL 主机；也可用 PGHOST")
|
|
parser.add_argument("--pg-port", default=os.environ.get("PGPORT", "5432"), help="PostgreSQL 端口；也可用 PGPORT")
|
|
parser.add_argument("--pg-database", default=os.environ.get("PGDATABASE", ""), help="PostgreSQL 数据库；也可用 PGDATABASE")
|
|
parser.add_argument("--pg-user", default=os.environ.get("PGUSER", ""), help="PostgreSQL 用户名；也可用 PGUSER")
|
|
parser.add_argument("--pg-password", default=os.environ.get("PGPASSWORD", ""), help="PostgreSQL 密码；也可用 PGPASSWORD")
|
|
parser.add_argument("--pg-table", default=os.environ.get("PG_PATIENT_TABLE", DEFAULT_TABLE), help="PostgreSQL 表名")
|
|
parser.add_argument("-o", "--output-dir", type=Path, default=DEFAULT_OUTPUT_DIR, help="体检结果输出目录")
|
|
return parser
|
|
|
|
|
|
def main() -> int:
|
|
audit(build_parser().parse_args())
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
raise SystemExit(main())
|
|
except Exception as exc: # noqa: BLE001
|
|
print(f"错误：{exc}", file=sys.stderr)
|
|
raise SystemExit(1)
|