Add DICOM UPP patient name matching

数据库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])

数据库Web可视化/requirements.txt

@@ -1,2 +1,3 @@
 fastapi==0.116.1
 uvicorn[standard]==0.35.0
+pypinyin==0.54.0

数据库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() {
           <span>${escapeHtml(row.pacs_patient_name || row.upp_patient_name || "无姓名")} · ${escapeHtml(row.patient_id || row.patient_id_masked || "无ID")}</span>
           <small>DICOM ${Number(row.pacs_series_count || 0)} 序列${stlCount ? ` · STL ${stlCount} 个` : ""}</small>
         </div>
-        <i class="mini-badge ${escapeHtml(row.relation_status)}">${escapeHtml(statusLabel(row.relation_status))}</i>
+        <div class="card-status">
+          <i class="mini-badge ${escapeHtml(row.relation_status)}">${escapeHtml(statusLabel(row.relation_status))}</i>
+          ${
+            patientNameMismatch(row)
+              ? `<small class="name-check-warning" title="${escapeHtml(row.patient_name_match_note || "DICOM 与 UPP 患者姓名不一致")}">姓名需核对</small>`
+              : ""
+          }
+        </div>
       </div>
       <div class="card-tag-rail">
         ${row.algorithm_model ? `<em class="model-tag" title="${escapeHtml(row.algorithm_model)}">${escapeHtml(row.algorithm_model)}</em>` : ""}
@@ -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)],

数据库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;