608 lines
22 KiB
Python
608 lines
22 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import hashlib
|
|
import os
|
|
import re
|
|
import secrets
|
|
import subprocess
|
|
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
|
|
|
|
|
|
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")
|
|
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")
|
|
|
|
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)
|
|
PACS_ANNOTATION_TABLE_SQL = ident(PACS_ANNOTATION_TABLE)
|
|
UPP_ASSET_TABLE_SQL = ident(UPP_ASSET_TABLE)
|
|
UPP_STL_TABLE_SQL = ident(UPP_STL_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 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 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()
|
|
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 '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 'lower_abdomen' THEN 4
|
|
WHEN 'pelvis' THEN 5
|
|
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
|
|
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,
|
|
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 ? 'lower_abdomen' THEN '下腹部' END,
|
|
CASE WHEN ps.body_parts ? 'pelvis' THEN '盆腔' END
|
|
], NULL)) AS dicom_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.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
|
|
)
|
|
"""
|
|
|
|
|
|
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", "lower_abdomen", "pelvis"}
|
|
if dicom_part not in valid_parts:
|
|
raise HTTPException(status_code=400, detail="invalid DICOM part filter")
|
|
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,
|
|
"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,
|
|
},
|
|
}
|
|
|
|
|
|
@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:
|
|
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 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,
|
|
"tables": {
|
|
"dicom": PACS_TABLE,
|
|
"dicom_summary": PACS_SUMMARY_TABLE,
|
|
"dicom_annotations": PACS_ANNOTATION_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_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]]:
|
|
where = relation_where(q, status, algorithm_model, dicom_part)
|
|
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_body_parts, dicom_annotation_labels, 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)}
|
|
OFFSET {int(offset)}
|
|
""",
|
|
timeout=24,
|
|
)
|
|
|
|
|
|
@app.get("/api/relations/{ct_number}")
|
|
def relation_detail(ct_number: str, user: dict[str, str] = Depends(current_user)) -> 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]
|