admin/HIS
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Improve patient list OCR cleanup

Browse Source
This commit is contained in:
admin
2026-06-11 20:31:11 +08:00
parent fde8e11cd1
commit 939c34be19
3 changed files with 155 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

112
患者列表处理/数据处理工作区/02_患者列表OCR归档.py
View File

@@ -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("复核备注"):