Add PACS UPP database visualizer
This commit is contained in:
Codex
341
数据库Web可视化/app.py
Normal file
341
数据库Web可视化/app.py
Normal file
@@ -0,0 +1,341 @@
|
||||
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"^D", "", 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.*,
|
||||
regexp_replace(upper(p.ct_number), '^D', '') AS norm_ct
|
||||
FROM public.{PACS_TABLE_SQL} p
|
||||
),
|
||||
ps AS (
|
||||
SELECT *
|
||||
FROM public.{PACS_SUMMARY_TABLE_SQL}
|
||||
),
|
||||
u_raw AS (
|
||||
SELECT
|
||||
u.*,
|
||||
regexp_replace(upper(u.ct_number), '^D', '') AS norm_ct
|
||||
FROM public.{UPP_ASSET_TABLE_SQL} u
|
||||
),
|
||||
u AS (
|
||||
SELECT DISTINCT ON (norm_ct) *
|
||||
FROM u_raw
|
||||
ORDER BY norm_ct, stl_present DESC, list_present DESC, updated_at DESC NULLS LAST
|
||||
),
|
||||
s_raw AS (
|
||||
SELECT
|
||||
s.*,
|
||||
regexp_replace(upper(s.ct_number), '^D', '') AS norm_ct
|
||||
FROM public.{UPP_STL_TABLE_SQL} s
|
||||
),
|
||||
s AS (
|
||||
SELECT DISTINCT ON (norm_ct) *
|
||||
FROM s_raw
|
||||
ORDER BY norm_ct, file_count DESC, updated_at DESC NULLS LAST
|
||||
),
|
||||
keys AS (
|
||||
SELECT norm_ct FROM p
|
||||
UNION
|
||||
SELECT norm_ct FROM u
|
||||
),
|
||||
relation AS (
|
||||
SELECT
|
||||
k.norm_ct,
|
||||
p.ct_number AS pacs_ct_number,
|
||||
u.ct_number AS upp_ct_number,
|
||||
s.ct_number AS stl_ct_number,
|
||||
p.ct_number IS NOT NULL AS pacs_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 u.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 u.ct_number IS NOT NULL THEN 'upp_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'
|
||||
WHEN p.ct_number IS NOT NULL AND u.ct_number IS NOT NULL THEN 'normalized'
|
||||
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,
|
||||
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.norm_ct = k.norm_ct
|
||||
LEFT JOIN ps ON ps.ct_number = p.ct_number
|
||||
LEFT JOIN u ON u.norm_ct = k.norm_ct
|
||||
LEFT JOIN s ON s.norm_ct = k.norm_ct
|
||||
)
|
||||
"""
|
||||
|
||||
|
||||
def relation_where(q: str, status: str) -> str:
|
||||
clauses: list[str] = []
|
||||
query = q.strip()
|
||||
if query:
|
||||
like = "%" + query.replace("%", "").replace("_", "") + "%"
|
||||
literal = sql_literal(like)
|
||||
clauses.append(
|
||||
" OR ".join(
|
||||
[
|
||||
f"norm_ct ILIKE {literal}",
|
||||
f"pacs_ct_number ILIKE {literal}",
|
||||
f"upp_ct_number ILIKE {literal}",
|
||||
f"pacs_patient_name ILIKE {literal}",
|
||||
f"upp_patient_name ILIKE {literal}",
|
||||
f"patient_id 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 == "upp_only":
|
||||
clauses.append("relation_status = 'upp_only'")
|
||||
elif status == "normalized":
|
||||
clauses.append("match_type = 'normalized'")
|
||||
elif status == "undetermined":
|
||||
clauses.append("COALESCE(undetermined_series, 0) > 0")
|
||||
elif status != "all":
|
||||
raise HTTPException(status_code=400, detail="invalid status filter")
|
||||
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 upp_present)::int AS upp_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 = 'upp_only')::int AS upp_only_count,
|
||||
count(*) FILTER (WHERE match_type = 'normalized')::int AS normalized_match_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", limit: int = Query(default=300, ge=1, le=1000)) -> list[dict[str, Any]]:
|
||||
where = relation_where(q, status)
|
||||
return pg_json_rows(
|
||||
f"""
|
||||
{relation_cte()}
|
||||
SELECT
|
||||
norm_ct, pacs_ct_number, upp_ct_number, stl_ct_number,
|
||||
pacs_present, upp_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, 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,
|
||||
norm_ct
|
||||
LIMIT {int(limit)}
|
||||
""",
|
||||
timeout=24,
|
||||
)
|
||||
|
||||
|
||||
@app.get("/api/relations/{ct_number}")
|
||||
def relation_detail(ct_number: str) -> dict[str, Any]:
|
||||
norm_ct = normalize_ct(ct_number)
|
||||
if not norm_ct:
|
||||
raise HTTPException(status_code=400, detail="invalid CT number")
|
||||
rows = pg_json_rows(
|
||||
f"""
|
||||
{relation_cte()}
|
||||
SELECT *
|
||||
FROM relation
|
||||
WHERE norm_ct = {sql_literal(norm_ct)}
|
||||
LIMIT 1
|
||||
""",
|
||||
timeout=24,
|
||||
)
|
||||
if not rows:
|
||||
raise HTTPException(status_code=404, detail="CT number not found")
|
||||
return rows[0]
|
||||
Reference in New Issue
Block a user