921 lines
39 KiB
Python
921 lines
39 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
import hashlib
|
||
import os
|
||
import re
|
||
import secrets
|
||
import subprocess
|
||
import unicodedata
|
||
from itertools import product
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from fastapi import Depends, FastAPI, Header, HTTPException, Query
|
||
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"
|
||
|
||
PGHOST = os.getenv("PGHOST", "192.168.3.3")
|
||
PGPORT = os.getenv("PGPORT", "5432")
|
||
PGUSER = os.getenv("PGUSER", "his_user")
|
||
PGDATABASE = os.getenv("PGDATABASE", "pacs_db")
|
||
PACS_TABLE = os.getenv("PACS_TABLE", "pacs_dicom_files")
|
||
PACS_SUMMARY_TABLE = os.getenv("PACS_SUMMARY_TABLE", "pacs_dicom_study_summaries")
|
||
PACS_ANNOTATION_TABLE = os.getenv("PACS_ANNOTATION_TABLE", "pacs_dicom_series_annotations")
|
||
UPP_ASSET_TABLE = os.getenv("UPP_ASSET_TABLE", "upp_exam_assets")
|
||
UPP_STL_TABLE = os.getenv("UPP_STL_TABLE", "upp_stl_files")
|
||
REGISTRATION_TABLE = os.getenv("REGISTRATION_TABLE", "dicom_upp_registrations")
|
||
USER_TABLE = os.getenv("VISUALIZER_USER_TABLE", "db_visualizer_users")
|
||
WEB_ADMIN_USER = os.getenv("VISUALIZER_ADMIN_USER", "admin")
|
||
WEB_ADMIN_PASSWORD = os.getenv("VISUALIZER_ADMIN_PASSWORD", "123456")
|
||
PACS_VIEWER_URL = os.getenv("PACS_VIEWER_URL", "http://127.0.0.1:8107")
|
||
REGISTRATION_URL = os.getenv("REGISTRATION_URL", "http://127.0.0.1:8109")
|
||
|
||
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:
|
||
if not IDENT_RE.fullmatch(name):
|
||
raise RuntimeError(f"invalid SQL identifier: {name}")
|
||
return name
|
||
|
||
|
||
PACS_TABLE_SQL = ident(PACS_TABLE)
|
||
PACS_SUMMARY_TABLE_SQL = ident(PACS_SUMMARY_TABLE)
|
||
PACS_ANNOTATION_TABLE_SQL = ident(PACS_ANNOTATION_TABLE)
|
||
UPP_ASSET_TABLE_SQL = ident(UPP_ASSET_TABLE)
|
||
UPP_STL_TABLE_SQL = ident(UPP_STL_TABLE)
|
||
REGISTRATION_TABLE_SQL = ident(REGISTRATION_TABLE)
|
||
USER_TABLE_SQL = ident(USER_TABLE)
|
||
|
||
app = FastAPI(title="DICOM UPP Database Visualizer")
|
||
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
|
||
TOKENS: dict[str, dict[str, str]] = {}
|
||
|
||
|
||
class LoginPayload(BaseModel):
|
||
username: str
|
||
password: str
|
||
|
||
|
||
class UserPayload(BaseModel):
|
||
username: str
|
||
password: str = ""
|
||
role: str = "阅片员"
|
||
status: str = "启用"
|
||
|
||
|
||
class PasswordPayload(BaseModel):
|
||
password: str
|
||
|
||
|
||
def pg_env() -> dict[str, str]:
|
||
env = os.environ.copy()
|
||
env.update({"PGHOST": PGHOST, "PGPORT": PGPORT, "PGUSER": PGUSER, "PGDATABASE": PGDATABASE})
|
||
return env
|
||
|
||
|
||
def run_psql(sql: str, timeout: int = 20) -> subprocess.CompletedProcess[str]:
|
||
return subprocess.run(
|
||
["psql", "-X", "-q", "-t", "-A", "-v", "ON_ERROR_STOP=1", "-c", sql],
|
||
text=True,
|
||
capture_output=True,
|
||
timeout=timeout,
|
||
env=pg_env(),
|
||
)
|
||
|
||
|
||
def pg_scalar(sql: str, timeout: int = 20) -> str:
|
||
result = run_psql(sql, timeout=timeout)
|
||
if result.returncode != 0:
|
||
raise RuntimeError(result.stderr.strip() or result.stdout.strip())
|
||
return result.stdout.strip()
|
||
|
||
|
||
def pg_json_rows(select_sql: str, timeout: int = 24) -> list[dict[str, Any]]:
|
||
payload = pg_scalar(f"SELECT COALESCE(json_agg(row_to_json(q)), '[]'::json)::text FROM ({select_sql}) q", timeout=timeout)
|
||
return json.loads(payload or "[]")
|
||
|
||
|
||
def sql_literal(value: Any) -> str:
|
||
return "'" + str(value).replace("'", "''") + "'"
|
||
|
||
|
||
def password_hash(password: str) -> str:
|
||
return hashlib.sha256(password.encode("utf-8")).hexdigest()
|
||
|
||
|
||
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 match_patient_names(pacs_name: Any, upp_name: Any) -> tuple[str, str]:
|
||
pacs_text = str(pacs_name or "")
|
||
upp_text = str(upp_name or "")
|
||
if not pacs_text or not upp_text:
|
||
return "", ""
|
||
|
||
pacs_latin = normalize_latin_name(pacs_text)
|
||
upp_latin = normalize_latin_name(upp_text)
|
||
pacs_cjk = normalize_cjk_name(pacs_text)
|
||
upp_cjk = normalize_cjk_name(upp_text)
|
||
|
||
if pacs_cjk and upp_cjk and pacs_cjk == upp_cjk:
|
||
return "same", "中文姓名一致"
|
||
if pacs_latin and upp_latin and pacs_latin == upp_latin:
|
||
return "same", "拼音/拉丁姓名一致"
|
||
if upp_latin and upp_latin in pinyin_name_candidates(pacs_text):
|
||
return "same", "DICOM中文姓名与UPP拼音匹配"
|
||
if upp_latin and near_pinyin_match(upp_latin, pinyin_name_candidates(pacs_text)):
|
||
return "same", "DICOM中文姓名与UPP拼音近似匹配"
|
||
if pacs_latin and pacs_latin in pinyin_name_candidates(upp_text):
|
||
return "same", "UPP中文姓名与DICOM拼音匹配"
|
||
if pacs_latin and near_pinyin_match(pacs_latin, pinyin_name_candidates(upp_text)):
|
||
return "same", "UPP中文姓名与DICOM拼音近似匹配"
|
||
return "different", f"DICOM患者：{pacs_text}；UPP患者：{upp_text}"
|
||
|
||
|
||
def enrich_patient_match(row: dict[str, Any]) -> dict[str, Any]:
|
||
pacs_name = row.get("pacs_patient_name") or ""
|
||
upp_names = [row.get("upp_patient_name"), *(row.get("upp_patient_names") or [])]
|
||
upp_names = list(dict.fromkeys([str(name) for name in upp_names if name]))
|
||
if not pacs_name or not upp_names:
|
||
row["patient_name_match"] = ""
|
||
row["patient_name_match_note"] = ""
|
||
return row
|
||
|
||
mismatch_notes = []
|
||
for upp_name in upp_names:
|
||
status, note = match_patient_names(pacs_name, upp_name)
|
||
if status == "same":
|
||
row["patient_name_match"] = "same"
|
||
row["patient_name_match_note"] = note
|
||
row["upp_patient_name"] = row.get("upp_patient_name") or upp_name
|
||
return row
|
||
if note:
|
||
mismatch_notes.append(note)
|
||
row["patient_name_match"] = "different"
|
||
row["patient_name_match_note"] = "；".join(mismatch_notes[:3])
|
||
return row
|
||
|
||
|
||
def db_available() -> tuple[bool, str]:
|
||
try:
|
||
pg_scalar("SELECT 1", timeout=4)
|
||
return True, "connected"
|
||
except Exception as exc: # noqa: BLE001
|
||
return False, str(exc)
|
||
|
||
|
||
def ensure_user_table() -> None:
|
||
run_psql(
|
||
f"""
|
||
CREATE TABLE IF NOT EXISTS public.{USER_TABLE_SQL} (
|
||
username text PRIMARY KEY,
|
||
password_hash text NOT NULL,
|
||
role text NOT NULL DEFAULT '阅片员',
|
||
status text NOT NULL DEFAULT '启用',
|
||
created_at timestamptz NOT NULL DEFAULT now(),
|
||
updated_at timestamptz NOT NULL DEFAULT now()
|
||
);
|
||
INSERT INTO public.{USER_TABLE_SQL} (username, password_hash, role, status)
|
||
VALUES ({sql_literal(WEB_ADMIN_USER)}, {sql_literal(password_hash(WEB_ADMIN_PASSWORD))}, '管理员', '启用')
|
||
ON CONFLICT (username) DO NOTHING;
|
||
""",
|
||
timeout=10,
|
||
)
|
||
|
||
|
||
def ensure_registration_table() -> None:
|
||
run_psql(
|
||
f"""
|
||
CREATE TABLE IF NOT EXISTS public.{REGISTRATION_TABLE_SQL} (
|
||
id bigserial PRIMARY KEY,
|
||
ct_number text NOT NULL,
|
||
series_instance_uid text NOT NULL DEFAULT '',
|
||
series_description text NOT NULL DEFAULT '',
|
||
algorithm_model text NOT NULL DEFAULT '',
|
||
stl_family text NOT NULL DEFAULT '',
|
||
view_mode text NOT NULL DEFAULT 'family',
|
||
selected_stl_files jsonb NOT NULL DEFAULT '[]'::jsonb,
|
||
transform jsonb NOT NULL DEFAULT '{{}}'::jsonb,
|
||
dicom_reference jsonb NOT NULL DEFAULT '{{}}'::jsonb,
|
||
model_reference jsonb NOT NULL DEFAULT '{{}}'::jsonb,
|
||
segmentation jsonb NOT NULL DEFAULT '{{}}'::jsonb,
|
||
notes text NOT NULL DEFAULT '',
|
||
locked boolean NOT NULL DEFAULT false,
|
||
locked_by text,
|
||
locked_at timestamptz,
|
||
updated_by text NOT NULL DEFAULT 'admin',
|
||
created_at timestamptz NOT NULL DEFAULT now(),
|
||
updated_at timestamptz NOT NULL DEFAULT now(),
|
||
UNIQUE (ct_number, series_instance_uid, algorithm_model, stl_family)
|
||
);
|
||
CREATE INDEX IF NOT EXISTS idx_{REGISTRATION_TABLE_SQL}_ct ON public.{REGISTRATION_TABLE_SQL}(ct_number);
|
||
CREATE INDEX IF NOT EXISTS idx_{REGISTRATION_TABLE_SQL}_locked ON public.{REGISTRATION_TABLE_SQL}(locked);
|
||
""",
|
||
timeout=10,
|
||
)
|
||
|
||
|
||
def fetch_user(username: str) -> dict[str, Any] | None:
|
||
ensure_user_table()
|
||
rows = pg_json_rows(
|
||
f"""
|
||
SELECT username, password_hash, role, status, created_at, updated_at
|
||
FROM public.{USER_TABLE_SQL}
|
||
WHERE username = {sql_literal(username)}
|
||
LIMIT 1
|
||
""",
|
||
timeout=10,
|
||
)
|
||
return rows[0] if rows else None
|
||
|
||
|
||
def current_user(authorization: str | None = Header(default=None)) -> dict[str, str]:
|
||
if not authorization or not authorization.startswith("Bearer "):
|
||
raise HTTPException(status_code=401, detail="unauthorized")
|
||
token = authorization.removeprefix("Bearer ").strip()
|
||
user = TOKENS.get(token)
|
||
if not user:
|
||
raise HTTPException(status_code=401, detail="unauthorized")
|
||
return user
|
||
|
||
|
||
def admin_user(user: dict[str, str] = Depends(current_user)) -> dict[str, str]:
|
||
if user.get("role") != "管理员":
|
||
raise HTTPException(status_code=403, detail="forbidden")
|
||
return user
|
||
|
||
|
||
@app.on_event("startup")
|
||
def startup() -> None:
|
||
try:
|
||
ensure_user_table()
|
||
ensure_registration_table()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def relation_cte() -> str:
|
||
return f"""
|
||
WITH
|
||
p AS (
|
||
SELECT
|
||
p.*,
|
||
upper(p.ct_number) AS ct_key
|
||
FROM public.{PACS_TABLE_SQL} p
|
||
),
|
||
ps AS (
|
||
SELECT *
|
||
FROM public.{PACS_SUMMARY_TABLE_SQL}
|
||
),
|
||
a_label_raw AS (
|
||
SELECT
|
||
a.ct_number,
|
||
CASE part.value
|
||
WHEN 'head_neck' THEN '头颈部'
|
||
WHEN 'chest' THEN '胸部-' || CASE COALESCE(NULLIF(a.chest_window, ''), 'unknown')
|
||
WHEN 'lung' THEN '肺窗'
|
||
WHEN 'mediastinal' THEN '纵隔窗'
|
||
ELSE '无法判别'
|
||
END
|
||
WHEN 'upper_abdomen' THEN '上腹部-' || CASE COALESCE(NULLIF(a.upper_abdomen_phase, ''), 'unknown')
|
||
WHEN 'plain' THEN '平扫'
|
||
WHEN 'arterial' THEN '动脉期'
|
||
WHEN 'portal_venous' THEN '门脉期'
|
||
WHEN 'delayed' THEN '延迟期'
|
||
ELSE '无法判别'
|
||
END
|
||
WHEN 'abdomen_pelvis' THEN '腹盆部'
|
||
WHEN 'lower_abdomen' THEN '腹盆部'
|
||
WHEN 'pelvis' THEN '腹盆部'
|
||
ELSE NULL
|
||
END AS label,
|
||
CASE part.value
|
||
WHEN 'head_neck' THEN 1
|
||
WHEN 'chest' THEN 2
|
||
WHEN 'upper_abdomen' THEN 3
|
||
WHEN 'abdomen_pelvis' THEN 4
|
||
WHEN 'lower_abdomen' THEN 4
|
||
WHEN 'pelvis' THEN 4
|
||
ELSE 99
|
||
END AS sort_key
|
||
FROM public.{PACS_ANNOTATION_TABLE_SQL} a
|
||
CROSS JOIN LATERAL jsonb_array_elements_text(a.body_parts) AS part(value)
|
||
WHERE COALESCE(a.skipped, false) IS NOT TRUE
|
||
),
|
||
a_label_distinct AS (
|
||
SELECT ct_number, label, min(sort_key) AS sort_key
|
||
FROM a_label_raw
|
||
WHERE label IS NOT NULL
|
||
GROUP BY ct_number, label
|
||
),
|
||
a_labels AS (
|
||
SELECT
|
||
ct_number,
|
||
to_jsonb(array_agg(label ORDER BY sort_key, label)) AS dicom_annotation_labels
|
||
FROM a_label_distinct
|
||
GROUP BY ct_number
|
||
),
|
||
u_raw AS (
|
||
SELECT
|
||
u.*,
|
||
upper(u.ct_number) AS ct_key,
|
||
COALESCE(NULLIF(u.algorithm_model, ''), '未指定模型') AS algorithm_model_key
|
||
FROM public.{UPP_ASSET_TABLE_SQL} u
|
||
),
|
||
u_model_ranked AS (
|
||
SELECT
|
||
u_raw.*,
|
||
row_number() OVER (
|
||
PARTITION BY ct_key, algorithm_model_key
|
||
ORDER BY
|
||
COALESCE(stl_present, false) DESC,
|
||
COALESCE(stl_file_count, 0) DESC,
|
||
updated_at DESC NULLS LAST
|
||
) AS model_rank
|
||
FROM u_raw
|
||
),
|
||
u_model AS (
|
||
SELECT *
|
||
FROM u_model_ranked
|
||
WHERE model_rank = 1
|
||
),
|
||
u_duplicate AS (
|
||
SELECT
|
||
ct_key,
|
||
COALESCE(sum(model_count - 1), 0)::int AS duplicate_model_asset_count
|
||
FROM (
|
||
SELECT ct_key, algorithm_model_key, count(*)::int AS model_count
|
||
FROM u_raw
|
||
GROUP BY ct_key, algorithm_model_key
|
||
HAVING count(*) > 1
|
||
) duplicate_models
|
||
GROUP BY ct_key
|
||
),
|
||
u AS (
|
||
SELECT
|
||
u_model.ct_key,
|
||
(array_agg(u_model.ct_number ORDER BY u_model.updated_at DESC NULLS LAST) FILTER (WHERE u_model.ct_number IS NOT NULL))[1] AS ct_number,
|
||
bool_or(COALESCE(u_model.list_present, false)) AS list_present,
|
||
bool_or(COALESCE(u_model.stl_present, false)) AS stl_present,
|
||
(array_agg(u_model.patient_name ORDER BY u_model.updated_at DESC NULLS LAST) FILTER (WHERE NULLIF(u_model.patient_name, '') IS NOT NULL))[1] AS patient_name,
|
||
to_jsonb(array_remove(array_agg(DISTINCT NULLIF(u_model.patient_name, '')), NULL)) AS patient_names,
|
||
(array_agg(u_model.patient_sex ORDER BY u_model.updated_at DESC NULLS LAST) FILTER (WHERE NULLIF(u_model.patient_sex, '') IS NOT NULL))[1] AS patient_sex,
|
||
(array_agg(u_model.patient_age ORDER BY u_model.updated_at DESC NULLS LAST) FILTER (WHERE NULLIF(u_model.patient_age, '') IS NOT NULL))[1] AS patient_age,
|
||
(array_agg(u_model.patient_id_masked ORDER BY u_model.updated_at DESC NULLS LAST) FILTER (WHERE NULLIF(u_model.patient_id_masked, '') IS NOT NULL))[1] AS patient_id_masked,
|
||
max(u_model.exam_date) AS exam_date,
|
||
max(u_model.task_created_at) AS task_created_at,
|
||
(array_agg(u_model.exam_description ORDER BY u_model.updated_at DESC NULLS LAST) FILTER (WHERE NULLIF(u_model.exam_description, '') IS NOT NULL))[1] AS exam_description,
|
||
(array_agg(u_model.exam_device ORDER BY u_model.updated_at DESC NULLS LAST) FILTER (WHERE NULLIF(u_model.exam_device, '') IS NOT NULL))[1] AS exam_device,
|
||
array_to_string(array_remove(array_agg(DISTINCT NULLIF(u_model.algorithm_model, '')), NULL), '、') AS algorithm_model,
|
||
to_jsonb(array_remove(array_agg(DISTINCT NULLIF(u_model.algorithm_model, '')), NULL)) AS algorithm_models,
|
||
array_to_string(array_remove(array_agg(DISTINCT NULLIF(u_model.upp_status, '')), NULL), '、') AS upp_status,
|
||
COALESCE(sum(COALESCE(u_model.list_record_count, 0)), 0)::int AS list_record_count,
|
||
(array_agg(u_model.processed_stl_dir ORDER BY u_model.updated_at DESC NULLS LAST) FILTER (WHERE NULLIF(u_model.processed_stl_dir, '') IS NOT NULL))[1] AS processed_stl_dir,
|
||
COALESCE(sum(COALESCE(u_model.stl_file_count, 0)), 0)::int AS stl_file_count,
|
||
COALESCE(sum(COALESCE(u_model.stl_total_bytes, 0)), 0)::bigint AS stl_total_bytes,
|
||
max(u_model.updated_at) AS updated_at,
|
||
count(*)::int AS upp_asset_count,
|
||
COALESCE(max(u_duplicate.duplicate_model_asset_count), 0)::int AS duplicate_model_asset_count,
|
||
jsonb_agg(
|
||
jsonb_build_object(
|
||
'ct_number', u_model.ct_number,
|
||
'algorithm_model', u_model.algorithm_model,
|
||
'upp_status', u_model.upp_status,
|
||
'file_count', u_model.stl_file_count,
|
||
'total_bytes', u_model.stl_total_bytes,
|
||
'processed_stl_dir', u_model.processed_stl_dir,
|
||
'updated_at', u_model.updated_at
|
||
)
|
||
ORDER BY u_model.updated_at DESC NULLS LAST
|
||
) AS upp_assets
|
||
FROM u_model
|
||
LEFT JOIN u_duplicate ON u_duplicate.ct_key = u_model.ct_key
|
||
GROUP BY u_model.ct_key
|
||
),
|
||
s_raw AS (
|
||
SELECT
|
||
s.*,
|
||
upper(s.ct_number) AS ct_key
|
||
FROM public.{UPP_STL_TABLE_SQL} s
|
||
),
|
||
s_segment_names AS (
|
||
SELECT
|
||
ct_key,
|
||
COALESCE(jsonb_agg(DISTINCT name.value) FILTER (WHERE name.value IS NOT NULL), '[]'::jsonb) AS segment_names
|
||
FROM s_raw
|
||
CROSS JOIN LATERAL jsonb_array_elements_text(COALESCE(s_raw.segment_names, '[]'::jsonb)) AS name(value)
|
||
GROUP BY ct_key
|
||
),
|
||
s_segment_families AS (
|
||
SELECT
|
||
ct_key,
|
||
COALESCE(jsonb_agg(DISTINCT family.value) FILTER (WHERE family.value IS NOT NULL), '[]'::jsonb) AS segment_families
|
||
FROM s_raw
|
||
CROSS JOIN LATERAL jsonb_array_elements_text(COALESCE(s_raw.segment_families, '[]'::jsonb)) AS family(value)
|
||
GROUP BY ct_key
|
||
),
|
||
s_segment_categories AS (
|
||
SELECT
|
||
ct_key,
|
||
COALESCE(jsonb_agg(DISTINCT category.value) FILTER (WHERE category.value IS NOT NULL), '[]'::jsonb) AS segment_categories
|
||
FROM s_raw
|
||
CROSS JOIN LATERAL jsonb_array_elements_text(COALESCE(s_raw.segment_categories, '[]'::jsonb)) AS category(value)
|
||
GROUP BY ct_key
|
||
),
|
||
s_segments AS (
|
||
SELECT
|
||
s_raw.ct_key,
|
||
COALESCE(s_segment_names.segment_names, '[]'::jsonb) AS segment_names,
|
||
COALESCE(s_segment_families.segment_families, '[]'::jsonb) AS segment_families,
|
||
COALESCE(s_segment_categories.segment_categories, '[]'::jsonb) AS segment_categories
|
||
FROM (SELECT DISTINCT ct_key FROM s_raw) s_raw
|
||
LEFT JOIN s_segment_names ON s_segment_names.ct_key = s_raw.ct_key
|
||
LEFT JOIN s_segment_families ON s_segment_families.ct_key = s_raw.ct_key
|
||
LEFT JOIN s_segment_categories ON s_segment_categories.ct_key = s_raw.ct_key
|
||
),
|
||
s AS (
|
||
SELECT
|
||
s_raw.ct_key,
|
||
(array_agg(s_raw.ct_number ORDER BY s_raw.updated_at DESC NULLS LAST) FILTER (WHERE s_raw.ct_number IS NOT NULL))[1] AS ct_number,
|
||
COALESCE(sum(COALESCE(s_raw.file_count, 0)), 0)::int AS file_count,
|
||
COALESCE(sum(COALESCE(s_raw.total_bytes, 0)), 0)::bigint AS total_bytes,
|
||
COALESCE((array_agg(s_segments.segment_names) FILTER (WHERE s_segments.segment_names IS NOT NULL))[1], '[]'::jsonb) AS segment_names,
|
||
COALESCE((array_agg(s_segments.segment_families) FILTER (WHERE s_segments.segment_families IS NOT NULL))[1], '[]'::jsonb) AS segment_families,
|
||
COALESCE((array_agg(s_segments.segment_categories) FILTER (WHERE s_segments.segment_categories IS NOT NULL))[1], '[]'::jsonb) AS segment_categories,
|
||
max(s_raw.updated_at) AS updated_at,
|
||
count(*)::int AS stl_asset_count,
|
||
jsonb_agg(
|
||
jsonb_build_object(
|
||
'ct_number', s_raw.ct_number,
|
||
'file_count', s_raw.file_count,
|
||
'total_bytes', s_raw.total_bytes,
|
||
'updated_at', s_raw.updated_at,
|
||
'segment_categories', s_raw.segment_categories,
|
||
'segment_families', s_raw.segment_families
|
||
)
|
||
ORDER BY s_raw.updated_at DESC NULLS LAST
|
||
) AS stl_assets
|
||
FROM s_raw
|
||
LEFT JOIN s_segments ON s_segments.ct_key = s_raw.ct_key
|
||
GROUP BY s_raw.ct_key
|
||
),
|
||
r AS (
|
||
SELECT
|
||
upper(ct_number) AS ct_key,
|
||
count(*) FILTER (WHERE COALESCE(registration_status, 'unregistered') = 'registered')::int AS registration_count,
|
||
count(DISTINCT series_instance_uid) FILTER (
|
||
WHERE COALESCE(registration_status, 'unregistered') = 'registered'
|
||
AND series_instance_uid <> ''
|
||
)::int AS registered_series,
|
||
count(*) FILTER (
|
||
WHERE COALESCE(registration_status, 'unregistered') = 'registered'
|
||
AND COALESCE(locked, false)
|
||
)::int AS locked_count,
|
||
max(updated_at) FILTER (WHERE COALESCE(registration_status, 'unregistered') = 'registered') AS registration_updated_at
|
||
FROM public.{REGISTRATION_TABLE_SQL}
|
||
GROUP BY upper(ct_number)
|
||
),
|
||
keys AS (
|
||
SELECT ct_key FROM p
|
||
UNION
|
||
SELECT ct_key FROM u
|
||
UNION
|
||
SELECT ct_key FROM s
|
||
),
|
||
relation AS (
|
||
SELECT
|
||
k.ct_key,
|
||
p.ct_number AS pacs_ct_number,
|
||
COALESCE(s.ct_number, u.ct_number) AS stl_ct_number,
|
||
u.ct_number AS upp_ct_number,
|
||
s.ct_number AS stl_file_ct_number,
|
||
p.ct_number IS NOT NULL AS pacs_present,
|
||
u.ct_number IS NOT NULL OR s.ct_number IS NOT NULL AS stl_asset_present,
|
||
u.ct_number IS NOT NULL AS upp_present,
|
||
COALESCE(u.stl_present, false) OR s.ct_number IS NOT NULL AS stl_present,
|
||
CASE
|
||
WHEN p.ct_number IS NOT NULL AND (COALESCE(u.stl_present, false) OR s.ct_number IS NOT NULL) THEN 'complete'
|
||
WHEN p.ct_number IS NOT NULL AND u.ct_number IS NOT NULL THEN 'no_stl'
|
||
WHEN p.ct_number IS NOT NULL THEN 'pacs_only'
|
||
WHEN (COALESCE(u.stl_present, false) OR s.ct_number IS NOT NULL) THEN 'stl_only'
|
||
WHEN u.ct_number IS NOT NULL THEN 'list_only'
|
||
ELSE 'unknown'
|
||
END AS relation_status,
|
||
CASE
|
||
WHEN p.ct_number IS NOT NULL AND u.ct_number IS NOT NULL AND p.ct_number = u.ct_number THEN 'exact'
|
||
ELSE ''
|
||
END AS match_type,
|
||
p.batch_name,
|
||
p.source_patient_name,
|
||
p.patient_name_dicom,
|
||
COALESCE(NULLIF(p.source_patient_name, ''), NULLIF(p.patient_name_dicom, '')) AS pacs_patient_name,
|
||
p.patient_id,
|
||
p.patient_sex AS pacs_patient_sex,
|
||
p.study_date,
|
||
p.study_time,
|
||
p.study_description,
|
||
p.modality,
|
||
p.series_count AS pacs_series_count,
|
||
p.dicom_file_count,
|
||
p.processed_path,
|
||
p.updated_at AS pacs_updated_at,
|
||
ps.annotated_series,
|
||
ps.undetermined_series,
|
||
ps.completed,
|
||
ps.body_parts,
|
||
COALESCE(a_labels.dicom_annotation_labels, '[]'::jsonb) AS dicom_annotation_labels,
|
||
to_jsonb(array_remove(ARRAY[
|
||
CASE WHEN ps.body_parts ? 'head_neck' THEN '头颈部' END,
|
||
CASE WHEN ps.body_parts ? 'chest' THEN '胸部' END,
|
||
CASE WHEN ps.body_parts ? 'upper_abdomen' THEN '上腹部' END,
|
||
CASE WHEN ps.body_parts ? 'abdomen_pelvis' OR ps.body_parts ? 'lower_abdomen' OR ps.body_parts ? 'pelvis' THEN '腹盆部' END
|
||
], NULL)) AS dicom_body_parts,
|
||
u.patient_name AS upp_patient_name,
|
||
COALESCE(u.patient_names, '[]'::jsonb) AS upp_patient_names,
|
||
u.patient_sex AS upp_patient_sex,
|
||
u.patient_age,
|
||
u.patient_id_masked,
|
||
u.exam_date,
|
||
u.task_created_at,
|
||
u.exam_description,
|
||
u.exam_device,
|
||
u.algorithm_model,
|
||
COALESCE(u.algorithm_models, '[]'::jsonb) AS algorithm_models,
|
||
u.upp_status,
|
||
u.list_present,
|
||
u.list_record_count,
|
||
u.processed_stl_dir,
|
||
u.stl_file_count,
|
||
u.stl_total_bytes,
|
||
u.updated_at AS upp_updated_at,
|
||
COALESCE(u.upp_asset_count, 0) AS upp_asset_count,
|
||
COALESCE(u.duplicate_model_asset_count, 0) AS duplicate_model_asset_count,
|
||
COALESCE(u.upp_assets, '[]'::jsonb) AS upp_assets,
|
||
s.file_count AS stl_file_count_agg,
|
||
s.total_bytes AS stl_total_bytes_agg,
|
||
s.segment_names,
|
||
s.segment_families,
|
||
s.segment_categories,
|
||
s.updated_at AS stl_updated_at,
|
||
COALESCE(s.stl_asset_count, 0) AS stl_asset_count,
|
||
COALESCE(s.stl_assets, '[]'::jsonb) AS stl_assets,
|
||
COALESCE(r.registration_count, 0)::int AS registration_count,
|
||
COALESCE(r.registered_series, 0)::int AS registered_series,
|
||
COALESCE(r.locked_count, 0)::int AS locked_count,
|
||
r.registration_updated_at,
|
||
CASE
|
||
WHEN COALESCE(NULLIF(p.source_patient_name, ''), NULLIF(p.patient_name_dicom, '')) IS NULL
|
||
OR NULLIF(u.patient_name, '') IS NULL THEN ''
|
||
WHEN regexp_replace(COALESCE(NULLIF(p.source_patient_name, ''), p.patient_name_dicom), '\\s+', '', 'g')
|
||
= regexp_replace(u.patient_name, '\\s+', '', 'g') THEN 'same'
|
||
ELSE 'different'
|
||
END AS patient_name_match
|
||
FROM keys k
|
||
LEFT JOIN p ON p.ct_key = k.ct_key
|
||
LEFT JOIN ps ON ps.ct_number = p.ct_number
|
||
LEFT JOIN a_labels ON a_labels.ct_number = p.ct_number
|
||
LEFT JOIN u ON u.ct_key = k.ct_key
|
||
LEFT JOIN s ON s.ct_key = k.ct_key
|
||
LEFT JOIN r ON r.ct_key = k.ct_key
|
||
)
|
||
"""
|
||
|
||
|
||
def relation_where(q: str, status: str, algorithm_model: str, dicom_part: str) -> str:
|
||
clauses: list[str] = []
|
||
query = q.strip()
|
||
if query:
|
||
like = "%" + query.replace("%", "").replace("_", "") + "%"
|
||
literal = sql_literal(like)
|
||
clauses.append(
|
||
" OR ".join(
|
||
[
|
||
f"ct_key ILIKE {literal}",
|
||
f"pacs_ct_number ILIKE {literal}",
|
||
f"stl_ct_number ILIKE {literal}",
|
||
f"pacs_patient_name ILIKE {literal}",
|
||
f"upp_patient_name ILIKE {literal}",
|
||
f"patient_id ILIKE {literal}",
|
||
f"algorithm_model ILIKE {literal}",
|
||
f"upp_status ILIKE {literal}",
|
||
f"exam_description ILIKE {literal}",
|
||
f"study_description ILIKE {literal}",
|
||
f"COALESCE(dicom_body_parts::text, '') ILIKE {literal}",
|
||
]
|
||
).join(("(", ")"))
|
||
)
|
||
if status == "complete":
|
||
clauses.append("relation_status = 'complete'")
|
||
elif status == "no_stl":
|
||
clauses.append("relation_status = 'no_stl'")
|
||
elif status == "pacs_only":
|
||
clauses.append("relation_status = 'pacs_only'")
|
||
elif status == "stl_only":
|
||
clauses.append("relation_status = 'stl_only'")
|
||
elif status == "list_only":
|
||
clauses.append("relation_status = 'list_only'")
|
||
elif status == "pending_dicom":
|
||
clauses.append("pacs_present AND COALESCE(completed, false) IS NOT TRUE")
|
||
elif status == "undetermined":
|
||
clauses.append("COALESCE(undetermined_series, 0) > 0")
|
||
elif status != "all":
|
||
raise HTTPException(status_code=400, detail="invalid status filter")
|
||
if algorithm_model:
|
||
clauses.append(f"algorithm_model = {sql_literal(algorithm_model)}")
|
||
if dicom_part:
|
||
valid_parts = {"head_neck", "chest", "upper_abdomen", "abdomen_pelvis"}
|
||
if dicom_part not in valid_parts:
|
||
raise HTTPException(status_code=400, detail="invalid DICOM part filter")
|
||
if dicom_part == "abdomen_pelvis":
|
||
clauses.append("(body_parts ? 'abdomen_pelvis' OR body_parts ? 'lower_abdomen' OR body_parts ? 'pelvis')")
|
||
else:
|
||
clauses.append(f"body_parts ? {sql_literal(dicom_part)}")
|
||
return "WHERE " + " AND ".join(clauses) if clauses else ""
|
||
|
||
|
||
@app.get("/")
|
||
def index() -> FileResponse:
|
||
return FileResponse(STATIC_DIR / "index.html")
|
||
|
||
|
||
@app.get("/health")
|
||
def health() -> str:
|
||
return "ok"
|
||
|
||
|
||
@app.post("/api/auth/login")
|
||
def login(payload: LoginPayload) -> dict[str, str]:
|
||
username = payload.username.strip()
|
||
try:
|
||
user = fetch_user(username)
|
||
except Exception:
|
||
user = None
|
||
if not user and username == WEB_ADMIN_USER:
|
||
user = {"username": WEB_ADMIN_USER, "password_hash": password_hash(WEB_ADMIN_PASSWORD), "role": "管理员", "status": "启用"}
|
||
if not user or user.get("status") != "启用" or user.get("password_hash") != password_hash(payload.password):
|
||
raise HTTPException(status_code=401, detail="用户名或密码错误")
|
||
token = secrets.token_urlsafe(32)
|
||
TOKENS[token] = {"username": user["username"], "role": user.get("role", "阅片员")}
|
||
return {"token": token, "username": user["username"], "role": user.get("role", "阅片员")}
|
||
|
||
|
||
@app.get("/api/auth/me")
|
||
def me(user: dict[str, str] = Depends(current_user)) -> dict[str, str]:
|
||
return user
|
||
|
||
|
||
@app.get("/api/settings")
|
||
def settings(user: dict[str, str] = Depends(admin_user)) -> dict[str, Any]:
|
||
rows = pg_json_rows(
|
||
f"""
|
||
SELECT username, role, status, created_at, updated_at
|
||
FROM public.{USER_TABLE_SQL}
|
||
ORDER BY username
|
||
""",
|
||
timeout=10,
|
||
)
|
||
return {
|
||
"user": user,
|
||
"viewer_url": PACS_VIEWER_URL,
|
||
"registration_url": REGISTRATION_URL,
|
||
"users": rows,
|
||
"database": {"host": PGHOST, "port": PGPORT, "database": PGDATABASE},
|
||
"tables": {
|
||
"dicom": PACS_TABLE,
|
||
"dicom_summary": PACS_SUMMARY_TABLE,
|
||
"dicom_annotations": PACS_ANNOTATION_TABLE,
|
||
"upp_assets": UPP_ASSET_TABLE,
|
||
"upp_stl": UPP_STL_TABLE,
|
||
"registration": REGISTRATION_TABLE,
|
||
},
|
||
}
|
||
|
||
|
||
@app.post("/api/settings/users")
|
||
def save_user(payload: UserPayload, user: dict[str, str] = Depends(admin_user)) -> dict[str, str]:
|
||
username = payload.username.strip()
|
||
if not username:
|
||
raise HTTPException(status_code=400, detail="账号不能为空")
|
||
role = payload.role if payload.role in {"管理员", "阅片员", "访客"} else "阅片员"
|
||
status = payload.status if payload.status in {"启用", "停用"} else "启用"
|
||
existing = fetch_user(username)
|
||
if existing and not payload.password:
|
||
run_psql(
|
||
f"""
|
||
UPDATE public.{USER_TABLE_SQL}
|
||
SET role = {sql_literal(role)}, status = {sql_literal(status)}, updated_at = now()
|
||
WHERE username = {sql_literal(username)}
|
||
""",
|
||
timeout=10,
|
||
)
|
||
else:
|
||
if not payload.password:
|
||
raise HTTPException(status_code=400, detail="新账号需要初始密码")
|
||
run_psql(
|
||
f"""
|
||
INSERT INTO public.{USER_TABLE_SQL} (username, password_hash, role, status, updated_at)
|
||
VALUES ({sql_literal(username)}, {sql_literal(password_hash(payload.password))}, {sql_literal(role)}, {sql_literal(status)}, now())
|
||
ON CONFLICT (username) DO UPDATE
|
||
SET password_hash = EXCLUDED.password_hash,
|
||
role = EXCLUDED.role,
|
||
status = EXCLUDED.status,
|
||
updated_at = now()
|
||
""",
|
||
timeout=10,
|
||
)
|
||
return {"status": "ok", "updated_by": user["username"]}
|
||
|
||
|
||
@app.put("/api/settings/users/{username}/password")
|
||
def change_password(username: str, payload: PasswordPayload, user: dict[str, str] = Depends(admin_user)) -> dict[str, str]:
|
||
if not payload.password:
|
||
raise HTTPException(status_code=400, detail="密码不能为空")
|
||
run_psql(
|
||
f"""
|
||
UPDATE public.{USER_TABLE_SQL}
|
||
SET password_hash = {sql_literal(password_hash(payload.password))}, updated_at = now()
|
||
WHERE username = {sql_literal(username)}
|
||
""",
|
||
timeout=10,
|
||
)
|
||
return {"status": "ok", "updated_by": user["username"]}
|
||
|
||
|
||
@app.get("/api/status")
|
||
def status(user: dict[str, str] = Depends(current_user)) -> dict[str, Any]:
|
||
ok, message = db_available()
|
||
counts: dict[str, Any] = {}
|
||
if ok:
|
||
ensure_registration_table()
|
||
rows = pg_json_rows(
|
||
f"""
|
||
{relation_cte()}
|
||
SELECT
|
||
count(*)::int AS total_ct,
|
||
count(*) FILTER (WHERE pacs_present)::int AS pacs_count,
|
||
count(*) FILTER (WHERE stl_asset_present)::int AS stl_asset_count,
|
||
count(*) FILTER (WHERE list_present)::int AS list_count,
|
||
count(*) FILTER (WHERE stl_present)::int AS stl_count,
|
||
count(*) FILTER (WHERE relation_status = 'complete')::int AS complete_count,
|
||
count(*) FILTER (WHERE relation_status = 'no_stl')::int AS no_stl_count,
|
||
count(*) FILTER (WHERE relation_status = 'pacs_only')::int AS pacs_only_count,
|
||
count(*) FILTER (WHERE relation_status = 'stl_only')::int AS stl_only_count,
|
||
count(*) FILTER (WHERE relation_status = 'list_only')::int AS list_only_count,
|
||
count(*) FILTER (WHERE registration_count > 0)::int AS registered_ct_count,
|
||
count(*) FILTER (WHERE locked_count > 0)::int AS locked_ct_count,
|
||
COALESCE(sum(registration_count), 0)::int AS registration_count,
|
||
COALESCE(sum(locked_count), 0)::int AS locked_registration_count,
|
||
count(*) FILTER (WHERE pacs_present AND COALESCE(completed, false) IS NOT TRUE)::int AS pending_dicom_count,
|
||
COALESCE(sum(CASE WHEN pacs_present THEN COALESCE(undetermined_series, 0) ELSE 0 END), 0)::int AS pending_dicom_series
|
||
FROM relation
|
||
""",
|
||
timeout=24,
|
||
)
|
||
counts = rows[0] if rows else {}
|
||
return {
|
||
"database": {"ok": ok, "message": message, "host": PGHOST, "database": PGDATABASE},
|
||
"viewer_url": PACS_VIEWER_URL,
|
||
"registration_url": REGISTRATION_URL,
|
||
"tables": {
|
||
"dicom": PACS_TABLE,
|
||
"dicom_summary": PACS_SUMMARY_TABLE,
|
||
"dicom_annotations": PACS_ANNOTATION_TABLE,
|
||
"upp_assets": UPP_ASSET_TABLE,
|
||
"upp_stl": UPP_STL_TABLE,
|
||
"registration": REGISTRATION_TABLE,
|
||
},
|
||
"counts": counts,
|
||
}
|
||
|
||
|
||
@app.get("/api/relations")
|
||
def relations(
|
||
q: str = "",
|
||
status: str = "all",
|
||
algorithm_model: str = "",
|
||
dicom_part: str = "",
|
||
limit: int = Query(default=300, ge=1, le=1000),
|
||
offset: int = Query(default=0, ge=0),
|
||
user: dict[str, str] = Depends(current_user),
|
||
) -> list[dict[str, Any]]:
|
||
ensure_registration_table()
|
||
where = relation_where(q, status, algorithm_model, dicom_part)
|
||
rows = pg_json_rows(
|
||
f"""
|
||
{relation_cte()}
|
||
SELECT
|
||
ct_key, pacs_ct_number, stl_ct_number,
|
||
pacs_present, stl_asset_present, list_present, stl_present, relation_status, match_type,
|
||
pacs_patient_name, upp_patient_name, upp_patient_names, patient_name_match, patient_id, patient_id_masked,
|
||
study_date, study_time, exam_date, task_created_at,
|
||
study_description, exam_description, algorithm_model, upp_status,
|
||
pacs_series_count, dicom_file_count, annotated_series, undetermined_series, completed,
|
||
list_present, list_record_count, stl_file_count, stl_file_count_agg,
|
||
body_parts, dicom_body_parts, dicom_annotation_labels, algorithm_models,
|
||
upp_asset_count, duplicate_model_asset_count, upp_assets, stl_asset_count, stl_assets,
|
||
segment_categories, segment_families,
|
||
registration_count, registered_series, locked_count, registration_updated_at,
|
||
pacs_updated_at, upp_updated_at, stl_updated_at
|
||
FROM relation
|
||
{where}
|
||
ORDER BY
|
||
pacs_present DESC,
|
||
relation_status,
|
||
COALESCE(study_date, '') DESC,
|
||
COALESCE(study_time, '') DESC,
|
||
ct_key
|
||
LIMIT {int(limit)}
|
||
OFFSET {int(offset)}
|
||
""",
|
||
timeout=24,
|
||
)
|
||
return [enrich_patient_match(row) for row in rows]
|
||
|
||
|
||
@app.get("/api/relations/{ct_number}")
|
||
def relation_detail(ct_number: str, user: dict[str, str] = Depends(current_user)) -> dict[str, Any]:
|
||
ensure_registration_table()
|
||
ct_key = normalize_ct(ct_number)
|
||
if not ct_key:
|
||
raise HTTPException(status_code=400, detail="invalid CT number")
|
||
rows = pg_json_rows(
|
||
f"""
|
||
{relation_cte()}
|
||
SELECT *
|
||
FROM relation
|
||
WHERE ct_key = {sql_literal(ct_key)}
|
||
LIMIT 1
|
||
""",
|
||
timeout=24,
|
||
)
|
||
if not rows:
|
||
raise HTTPException(status_code=404, detail="CT number not found")
|
||
return enrich_patient_match(rows[0])
|