diff --git a/患者列表处理/数据处理工作区/02_患者列表OCR归档.py b/患者列表处理/数据处理工作区/02_患者列表OCR归档.py index c91fe50..0390ad5 100644 --- a/患者列表处理/数据处理工作区/02_患者列表OCR归档.py +++ b/患者列表处理/数据处理工作区/02_患者列表OCR归档.py @@ -463,12 +463,30 @@ def ocr_response_to_rows(response: dict[str, Any], engine: str, image_width: int 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) + text = re.sub( + r"(\d{4})-(\d{1,2})-(\d{1,2})\s+(\d{1,4}):(\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() + year, month, day, hour_text, minute_text, second_text = match.groups() + if second_text is None and len(hour_text) > 2: + if len(hour_text) == 3: + hour = int(hour_text[:2]) + minute = int(hour_text[2:]) + else: + hour = int(hour_text[:-2]) + minute = int(hour_text[-2:]) + second = int(minute_text) + else: + hour = int(hour_text) + minute = int(minute_text) + second = int(second_text or 0) + if not (0 <= hour <= 23 and 0 <= minute <= 59 and 0 <= second <= 59): + return match.group(0) return f"{int(year):04d}-{int(month):02d}-{int(day):02d} {int(hour):02d}:{int(minute):02d}:{int(second):02d}" @@ -485,19 +503,24 @@ def clean_patient_row(row: list[str]) -> dict[str, Any]: 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 normalize_inpatient_no(value: Any) -> str: + inpatient_no = re.sub(r"\s+", "", normalize_text(value)).upper() + if inpatient_no.startswith("ZV"): + inpatient_no = "ZY" + inpatient_no[2:] + if inpatient_no.startswith("ZYS"): + inpatient_no = "ZY5" + inpatient_no[3:] + if len(inpatient_no) > 2: + prefix, number = inpatient_no[:2], inpatient_no[2:] + inpatient_no = prefix + number.translate(str.maketrans({"O": "0", "I": "1", "L": "1", "S": "5"})) + return inpatient_no + + values["住院号"] = normalize_inpatient_no(values["住院号"]) + 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), + r"\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2}", + normalize_date(str(value)), ) ) @@ -510,8 +533,47 @@ def clean_patient_row(row: list[str]) -> dict[str, Any]: values["入院时间"] = normalize_date(values["入院时间"]) values["最后书写时间"] = normalize_date(values["最后书写时间"]) values["出院时间"] = normalize_date(values["出院时间"]) + + long_name_shift = re.fullmatch(r"(.+?)(?:\.\.\.)?\s*(男|女)", values["姓名"]) + if ( + long_name_shift + and values["性别"].endswith("岁") + and re.fullmatch(r"Z[YV]\w+", values["年龄"], re.IGNORECASE) + and values["住院号"] + ): + old_sex = values["性别"] + old_age = values["年龄"] + old_diagnosis = values["住院号"] + values["姓名"] = normalize_text(long_name_shift.group(1)) + values["性别"] = long_name_shift.group(2) + values["年龄"] = old_sex + values["住院号"] = normalize_inpatient_no(old_age) + if looks_datetime(values["诊断"]) and looks_datetime(values["入院时间"]) and looks_days(values["最后书写时间"]): + old_admission = values["诊断"] + old_last_write = values["入院时间"] + old_days = values["最后书写时间"] + old_discharge = values["住院天数"] + old_postop = values["出院时间"] + values["诊断"] = old_diagnosis + values["入院时间"] = normalize_date(old_admission) + values["最后书写时间"] = normalize_date(old_last_write) + values["住院天数"] = old_days + values["出院时间"] = normalize_date(old_discharge) if looks_datetime(old_discharge) else "" + if looks_postop(old_postop) and not values["手术后天数"]: + values["手术后天数"] = old_postop + elif not values["诊断"]: + values["诊断"] = old_diagnosis + + admission_with_diagnosis = re.search( + r"(.+?)\s*-*\s*(\d{4}[-/.]\d{1,2}[-/.]\d{1,2}\s+\d{1,4}:\d{1,2}(?::\d{1,2})?)\s*$", + values["入院时间"], + ) + if admission_with_diagnosis and not looks_datetime(values["入院时间"]): + values["诊断"] = normalize_text(f"{values['诊断']} {admission_with_diagnosis.group(1)}") + values["入院时间"] = normalize_date(admission_with_diagnosis.group(2)) + 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})$", + r"(\d{4}[-/.]\d{1,2}[-/.]\d{1,2}\s+\d{1,4}:\d{1,2}(?::\d{1,2})?)\s*[)）\]]*$", values["诊断"], ) if ( @@ -539,6 +601,25 @@ def clean_patient_row(row: list[str]) -> dict[str, Any]: values["诊断"] = values["诊断"][: shifted_by_long_diagnosis.start()].strip() values["入院时间"] = normalize_date(shifted_by_long_diagnosis.group(1)) + if ( + not looks_datetime(values["入院时间"]) + and looks_datetime(values["最后书写时间"]) + and looks_datetime(values["住院天数"]) + and looks_days(values["出院时间"]) + ): + diagnosis_extra = values["入院时间"] + old_admission = values["最后书写时间"] + old_last_write = values["住院天数"] + old_days = values["出院时间"] + old_discharge = values["手术后天数"] + if diagnosis_extra: + values["诊断"] = normalize_text(f"{values['诊断']} {diagnosis_extra}") + values["入院时间"] = normalize_date(old_admission) + values["最后书写时间"] = normalize_date(old_last_write) + values["住院天数"] = old_days + values["出院时间"] = normalize_date(old_discharge) if looks_datetime(old_discharge) else "" + values["手术后天数"] = "" + 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["住院天数"]), @@ -551,6 +632,10 @@ def clean_patient_row(row: list[str]) -> dict[str, Any]: if looks_postop(values["出院时间"]) and not values["手术后天数"]: values["手术后天数"] = values["出院时间"] values["出院时间"] = "" + if values["出院时间"] in {"手术日", "当日", "当天"}: + if not values["手术后天数"]: + values["手术后天数"] = values["出院时间"] + values["出院时间"] = "" if looks_postop(values["住院天数"]) and not values["手术后天数"]: values["手术后天数"] = values["住院天数"] values["住院天数"] = "" @@ -694,6 +779,7 @@ def apply_corrections(records: list[dict[str, Any]], corrections: dict[tuple[str continue correction = corrections[key] record["患者信息"].update(correction.get("患者信息", {})) + record["图片信息"].update(correction.get("图片信息", {})) record["复核"]["人工修正"] = True record["复核"]["复核选项"] = correction.get("复核选项", {}) if correction.get("复核备注"): diff --git a/患者列表处理/数据处理工作区/05_同步PostgreSQL单表.py b/患者列表处理/数据处理工作区/05_同步PostgreSQL单表.py index 277da04..dcf3cdb 100644 --- a/患者列表处理/数据处理工作区/05_同步PostgreSQL单表.py +++ b/患者列表处理/数据处理工作区/05_同步PostgreSQL单表.py @@ -12,6 +12,7 @@ import argparse import csv import json import os +import re import subprocess import tempfile from pathlib import Path @@ -22,6 +23,7 @@ DEFAULT_RESULT_JSON = Path("数据处理结果区/合并_患者列表_结构化. 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", @@ -75,6 +77,16 @@ def review_note_items(review: dict[str, Any]) -> list[str]: 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("'", "''") + "'" @@ -87,6 +99,17 @@ def flatten_patient_record(record: dict[str, Any]) -> dict[str, Any]: 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["处理批次"], @@ -96,7 +119,7 @@ def flatten_patient_record(record: dict[str, Any]) -> dict[str, Any]: "image_path": image["图片路径"], "image_name": image["图片名"], "image_row_no": image["图片内行号"], - "patient_name": patient["姓名"], + "patient_name": patient_name, "gender": patient["性别"], "age": patient["年龄"], "inpatient_no": patient["住院号"], @@ -106,7 +129,7 @@ def flatten_patient_record(record: dict[str, Any]) -> dict[str, Any]: "hospital_days": hospital_days, "discharge_time": patient["出院时间"], "postoperative_days": patient["手术后天数"], - "review_status": review["状态"], + "review_status": review_status, "review_notes": review_notes, "manual_corrected": bool(review.get("人工修正")), } diff --git a/患者列表处理/数据处理工作区/06_PostgreSQL建表结构.sql b/患者列表处理/数据处理工作区/06_PostgreSQL建表结构.sql index 967b8a9..c00afb8 100644 --- a/患者列表处理/数据处理工作区/06_PostgreSQL建表结构.sql +++ b/患者列表处理/数据处理工作区/06_PostgreSQL建表结构.sql @@ -49,6 +49,37 @@ CREATE INDEX IF NOT EXISTS idx_patient_lists_patient_name ON "Patient_Lists"(pat CREATE INDEX IF NOT EXISTS idx_patient_lists_review_status ON "Patient_Lists"(review_status); CREATE INDEX IF NOT EXISTS idx_patient_lists_audit_result ON "Patient_Lists"(audit_result); +DO $$ +BEGIN + IF NOT EXISTS ( + SELECT 1 FROM pg_constraint + WHERE conrelid = '"Patient_Lists"'::regclass + AND conname = 'ck_patient_lists_inpatient_no_present' + ) THEN + ALTER TABLE "Patient_Lists" + ADD CONSTRAINT ck_patient_lists_inpatient_no_present + CHECK (btrim(inpatient_no) <> '') NOT VALID; + END IF; + + IF NOT EXISTS ( + SELECT 1 FROM pg_constraint + WHERE conrelid = '"Patient_Lists"'::regclass + AND conname = 'ck_patient_lists_admission_before_discharge' + ) THEN + ALTER TABLE "Patient_Lists" + ADD CONSTRAINT ck_patient_lists_admission_before_discharge + CHECK ( + COALESCE(discharge_time, '') = '' + OR COALESCE(admission_time, '') = '' + OR admission_time !~ '^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$' + OR discharge_time !~ '^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$' + OR admission_time <= discharge_time + OR review_status = '需人工复核' + ) NOT VALID; + END IF; +END +$$; + COMMENT ON TABLE "Patient_Lists" IS 'HIS患者列表图片OCR归档后的正式患者记录单表'; COMMENT ON COLUMN "Patient_Lists".record_id IS '患者记录主键，数据库自动生成'; COMMENT ON COLUMN "Patient_Lists".batch_name IS '处理批次名称，通常为原始图片集群文件夹名';