364 lines
13 KiB
Python
364 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from fastapi import FastAPI, HTTPException, Query
|
|
from fastapi.responses import FileResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
|
|
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")
|
|
UPP_ASSET_TABLE = os.getenv("UPP_ASSET_TABLE", "upp_exam_assets")
|
|
UPP_STL_TABLE = os.getenv("UPP_STL_TABLE", "upp_stl_files")
|
|
|
|
IDENT_RE = re.compile(r"^[A-Za-z_][A-Za-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)
|
|
UPP_ASSET_TABLE_SQL = ident(UPP_ASSET_TABLE)
|
|
UPP_STL_TABLE_SQL = ident(UPP_STL_TABLE)
|
|
|
|
app = FastAPI(title="PACS UPP Database Visualizer")
|
|
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
|
|
|
|
|
|
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 normalize_ct(value: str) -> str:
|
|
return re.sub(r"\s+", "", str(value or "").upper())
|
|
|
|
|
|
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 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}
|
|
),
|
|
u_raw AS (
|
|
SELECT
|
|
u.*,
|
|
upper(u.ct_number) AS ct_key
|
|
FROM public.{UPP_ASSET_TABLE_SQL} u
|
|
),
|
|
u AS (
|
|
SELECT DISTINCT ON (ct_key) *
|
|
FROM u_raw
|
|
ORDER BY ct_key, stl_present DESC, list_present DESC, updated_at DESC NULLS LAST
|
|
),
|
|
s_raw AS (
|
|
SELECT
|
|
s.*,
|
|
upper(s.ct_number) AS ct_key
|
|
FROM public.{UPP_STL_TABLE_SQL} s
|
|
),
|
|
s AS (
|
|
SELECT DISTINCT ON (ct_key) *
|
|
FROM s_raw
|
|
ORDER BY ct_key, file_count DESC, updated_at DESC NULLS LAST
|
|
),
|
|
keys AS (
|
|
SELECT ct_key FROM p
|
|
UNION
|
|
SELECT ct_key FROM u
|
|
),
|
|
relation AS (
|
|
SELECT
|
|
k.ct_key,
|
|
p.ct_number AS pacs_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 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,
|
|
to_jsonb(array_remove(ARRAY[
|
|
CASE WHEN ps.body_parts ? 'upper_abdomen' THEN '肝胆模型' END,
|
|
CASE WHEN ps.body_parts ? 'chest' THEN '胸外模型' END,
|
|
CASE WHEN ps.body_parts ? 'upper_abdomen'
|
|
OR ps.body_parts ? 'lower_abdomen'
|
|
OR ps.body_parts ? 'pelvis' THEN '泌尿模型' END
|
|
], NULL)) AS dicom_models,
|
|
u.patient_name AS upp_patient_name,
|
|
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,
|
|
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,
|
|
u.selected_list_record,
|
|
u.list_records,
|
|
u.stl_candidates,
|
|
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.file_names,
|
|
s.files AS stl_files,
|
|
s.updated_at AS stl_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 u ON u.ct_key = k.ct_key
|
|
LEFT JOIN s ON s.ct_key = k.ct_key
|
|
)
|
|
"""
|
|
|
|
|
|
def relation_where(q: str, status: str, algorithm_model: str, dicom_model: 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}",
|
|
]
|
|
).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 == "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_model:
|
|
clauses.append(f"dicom_models ? {sql_literal(dicom_model)}")
|
|
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.get("/api/status")
|
|
def status() -> dict[str, Any]:
|
|
ok, message = db_available()
|
|
counts: dict[str, Any] = {}
|
|
if ok:
|
|
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,
|
|
COALESCE(sum(COALESCE(undetermined_series, 0)), 0)::int AS undetermined_series
|
|
FROM relation
|
|
""",
|
|
timeout=24,
|
|
)
|
|
counts = rows[0] if rows else {}
|
|
return {
|
|
"database": {"ok": ok, "message": message, "host": PGHOST, "database": PGDATABASE},
|
|
"tables": {
|
|
"pacs": PACS_TABLE,
|
|
"pacs_summary": PACS_SUMMARY_TABLE,
|
|
"upp_assets": UPP_ASSET_TABLE,
|
|
"upp_stl": UPP_STL_TABLE,
|
|
},
|
|
"counts": counts,
|
|
}
|
|
|
|
|
|
@app.get("/api/relations")
|
|
def relations(
|
|
q: str = "",
|
|
status: str = "all",
|
|
algorithm_model: str = "",
|
|
dicom_model: str = "",
|
|
limit: int = Query(default=300, ge=1, le=1000),
|
|
) -> list[dict[str, Any]]:
|
|
where = relation_where(q, status, algorithm_model, dicom_model)
|
|
return 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, 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_models, segment_categories, segment_families,
|
|
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)}
|
|
""",
|
|
timeout=24,
|
|
)
|
|
|
|
|
|
@app.get("/api/relations/{ct_number}")
|
|
def relation_detail(ct_number: str) -> dict[str, Any]:
|
|
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 rows[0]
|