From 026f6ce83c9daced7fb80192099f3915685ece44 Mon Sep 17 00:00:00 2001 From: Codex Date: Thu, 28 May 2026 09:52:17 +0800 Subject: [PATCH] Add DICOM UPP patient name matching --- 数据库Web可视化/app.py | 103 +++++++++++++++++++++++++++++- 数据库Web可视化/requirements.txt | 1 + 数据库Web可视化/static/app.js | 16 ++++- 数据库Web可视化/static/styles.css | 21 ++++++ 4 files changed, 137 insertions(+), 4 deletions(-) diff --git a/数据库Web可视化/app.py b/数据库Web可视化/app.py index 23062ab..c7651bf 100644 --- a/数据库Web可视化/app.py +++ b/数据库Web可视化/app.py @@ -6,6 +6,8 @@ import os import re import secrets import subprocess +import unicodedata +from itertools import product from pathlib import Path from typing import Any @@ -14,6 +16,12 @@ from pydantic import BaseModel from fastapi.responses import FileResponse from fastapi.staticfiles import StaticFiles +try: + from pypinyin import Style, pinyin +except ImportError: # pragma: no cover - Docker image installs this dependency. + Style = None + pinyin = None + BASE_DIR = Path(__file__).resolve().parent STATIC_DIR = BASE_DIR / "static" @@ -33,6 +41,8 @@ WEB_ADMIN_PASSWORD = os.getenv("VISUALIZER_ADMIN_PASSWORD", "123456") PACS_VIEWER_URL = os.getenv("PACS_VIEWER_URL", "http://127.0.0.1:8107") IDENT_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") +CJK_RE = re.compile(r"[\u3400-\u9fff]") +LATIN_RE = re.compile(r"[a-z0-9]+") def ident(name: str) -> str: @@ -109,6 +119,94 @@ def normalize_ct(value: str) -> str: return re.sub(r"\s+", "", str(value or "").upper()) +def normalize_latin_name(value: Any) -> str: + text = unicodedata.normalize("NFKD", str(value or "").replace("ü", "u").replace("Ü", "U")) + text = "".join(char for char in text if not unicodedata.combining(char)).lower() + return "".join(LATIN_RE.findall(text)) + + +def normalize_cjk_name(value: Any) -> str: + return "".join(CJK_RE.findall(str(value or ""))) + + +def pinyin_name_candidates(value: Any) -> set[str]: + if pinyin is None or Style is None: + return set() + chinese = normalize_cjk_name(value) + if not chinese: + return set() + syllables = pinyin(chinese, style=Style.NORMAL, heteronym=True, errors="ignore") + choices: list[list[str]] = [] + for item in syllables[:6]: + normalized = sorted({normalize_latin_name(part) for part in item if normalize_latin_name(part)}) + if normalized: + choices.append(normalized[:4]) + if not choices: + return set() + return {"".join(parts) for parts in product(*choices)} + + +def edit_distance_with_cap(left: str, right: str, cap: int = 1) -> int: + if abs(len(left) - len(right)) > cap: + return cap + 1 + previous = list(range(len(right) + 1)) + for i, left_char in enumerate(left, 1): + current = [i] + row_min = current[0] + for j, right_char in enumerate(right, 1): + cost = 0 if left_char == right_char else 1 + value = min(previous[j] + 1, current[j - 1] + 1, previous[j - 1] + cost) + current.append(value) + row_min = min(row_min, value) + if row_min > cap: + return cap + 1 + previous = current + return previous[-1] + + +def near_pinyin_match(left: str, candidates: set[str]) -> bool: + if len(left) < 6: + return False + return any(len(candidate) >= 6 and edit_distance_with_cap(left, candidate, cap=1) <= 1 for candidate in candidates) + + +def enrich_patient_match(row: dict[str, Any]) -> dict[str, Any]: + pacs_name = row.get("pacs_patient_name") or "" + upp_name = row.get("upp_patient_name") or "" + if not pacs_name or not upp_name: + row["patient_name_match"] = "" + row["patient_name_match_note"] = "" + return row + + pacs_latin = normalize_latin_name(pacs_name) + upp_latin = normalize_latin_name(upp_name) + pacs_cjk = normalize_cjk_name(pacs_name) + upp_cjk = normalize_cjk_name(upp_name) + + if pacs_cjk and upp_cjk and pacs_cjk == upp_cjk: + row["patient_name_match"] = "same" + row["patient_name_match_note"] = "中文姓名一致" + elif pacs_latin and upp_latin and pacs_latin == upp_latin: + row["patient_name_match"] = "same" + row["patient_name_match_note"] = "拼音/拉丁姓名一致" + elif upp_latin and upp_latin in pinyin_name_candidates(pacs_name): + row["patient_name_match"] = "same" + row["patient_name_match_note"] = "DICOM中文姓名与UPP拼音匹配" + elif upp_latin and near_pinyin_match(upp_latin, pinyin_name_candidates(pacs_name)): + row["patient_name_match"] = "same" + row["patient_name_match_note"] = "DICOM中文姓名与UPP拼音近似匹配" + elif pacs_latin and pacs_latin in pinyin_name_candidates(upp_name): + row["patient_name_match"] = "same" + row["patient_name_match_note"] = "UPP中文姓名与DICOM拼音匹配" + elif pacs_latin and near_pinyin_match(pacs_latin, pinyin_name_candidates(upp_name)): + row["patient_name_match"] = "same" + row["patient_name_match_note"] = "UPP中文姓名与DICOM拼音近似匹配" + else: + row["patient_name_match"] = "different" + row["patient_name_match_note"] = f"DICOM患者：{pacs_name}；UPP患者：{upp_name}" + return row + + def db_available() -> tuple[bool, str]: try: pg_scalar("SELECT 1", timeout=4) @@ -559,7 +657,7 @@ def relations( user: dict[str, str] = Depends(current_user), ) -> list[dict[str, Any]]: where = relation_where(q, status, algorithm_model, dicom_part) - return pg_json_rows( + rows = pg_json_rows( f""" {relation_cte()} SELECT @@ -585,6 +683,7 @@ def relations( """, timeout=24, ) + return [enrich_patient_match(row) for row in rows] @app.get("/api/relations/{ct_number}") @@ -604,4 +703,4 @@ def relation_detail(ct_number: str, user: dict[str, str] = Depends(current_user) ) if not rows: raise HTTPException(status_code=404, detail="CT number not found") - return rows[0] + return enrich_patient_match(rows[0]) diff --git a/数据库Web可视化/requirements.txt b/数据库Web可视化/requirements.txt index babedd0..d221c0e 100644 --- a/数据库Web可视化/requirements.txt +++ b/数据库Web可视化/requirements.txt @@ -1,2 +1,3 @@ fastapi==0.116.1 uvicorn[standard]==0.35.0 +pypinyin==0.54.0 diff --git a/数据库Web可视化/static/app.js b/数据库Web可视化/static/app.js index b1a0298..3a64e43 100644 --- a/数据库Web可视化/static/app.js +++ b/数据库Web可视化/static/app.js @@ -145,6 +145,10 @@ function annotationLabels(row) { return labels.length ? labels : dicomParts(row); } +function patientNameMismatch(row) { + return row.pacs_present && row.stl_asset_present && row.patient_name_match === "different"; +} + function setDbStatus(data) { app.viewerUrl = data.viewer_url || app.viewerUrl; const pill = $("dbStatus"); @@ -210,7 +214,14 @@ function renderList() { ${escapeHtml(row.pacs_patient_name || row.upp_patient_name || "无姓名")} · ${escapeHtml(row.patient_id || row.patient_id_masked || "无ID")} DICOM ${Number(row.pacs_series_count || 0)} 序列${stlCount ? ` · STL ${stlCount} 个` : ""} - ${escapeHtml(statusLabel(row.relation_status))} +
+ ${escapeHtml(statusLabel(row.relation_status))} + ${ + patientNameMismatch(row) + ? `姓名需核对` + : "" + } +
${row.algorithm_model ? `${escapeHtml(row.algorithm_model)}` : ""} @@ -316,10 +327,11 @@ function renderDetail(row) { renderDl("stlDetails", [ ["CT号", row.stl_ct_number], + ["患者", row.upp_patient_name], ["STL状态", row.stl_present ? "存在" : "缺失"], ["重建模型", row.algorithm_model], ["UPP状态", row.upp_status], - ["患者", row.upp_patient_name], + ["姓名核对", row.patient_name_match === "different" ? row.patient_name_match_note : row.patient_name_match === "same" ? row.patient_name_match_note || "匹配" : ""], ["年龄/性别", [row.patient_age, row.upp_patient_sex].filter(Boolean).join(" / ")], ["检查时间", fmtDate(row.exam_date)], ["任务时间", fmtDate(row.task_created_at)], diff --git a/数据库Web可视化/static/styles.css b/数据库Web可视化/static/styles.css index 8c73769..e51fb33 100644 --- a/数据库Web可视化/static/styles.css +++ b/数据库Web可视化/static/styles.css @@ -430,6 +430,27 @@ button { gap: 7px; } +.card-status { + display: grid; + justify-items: end; + gap: 5px; +} + +.name-check-warning { + max-width: 84px; + padding: 2px 7px; + overflow: hidden; + border: 1px solid rgba(240, 181, 78, 0.58); + border-radius: 999px; + color: #ffe0a3; + background: rgba(240, 181, 78, 0.1); + font-size: 11px; + font-weight: 700; + line-height: 1.45; + text-overflow: ellipsis; + white-space: nowrap; +} + .mini-badge { flex: 0 0 auto; padding: 2px 7px;