1009 lines
39 KiB
Python
1009 lines
39 KiB
Python
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import io
|
|
import json
|
|
import os
|
|
import re
|
|
import secrets
|
|
import subprocess
|
|
import time
|
|
from collections import Counter, defaultdict
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import numpy as np
|
|
import pydicom
|
|
from fastapi import Depends, FastAPI, Header, HTTPException, Query, Response
|
|
from fastapi.responses import FileResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from pydantic import BaseModel
|
|
from PIL import Image
|
|
|
|
|
|
APP_DIR = Path(__file__).resolve().parent
|
|
STATIC_DIR = APP_DIR / "static"
|
|
|
|
|
|
def load_env_file() -> None:
|
|
env_file = APP_DIR / ".env"
|
|
if not env_file.exists():
|
|
return
|
|
for line in env_file.read_text(encoding="utf-8").splitlines():
|
|
line = line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
key, value = line.split("=", 1)
|
|
os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'"))
|
|
|
|
|
|
load_env_file()
|
|
|
|
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")
|
|
WEB_USER = os.getenv("REGISTRATION_WEB_USER", "admin")
|
|
WEB_PASSWORD = os.getenv("REGISTRATION_WEB_PASSWORD", "123456")
|
|
PACS_VIEWER_URL = os.getenv("PACS_VIEWER_URL", "http://127.0.0.1:8107")
|
|
RELATION_VIEWER_URL = os.getenv("RELATION_VIEWER_URL", "http://127.0.0.1:8108")
|
|
PROCESSED_ROOT = Path(os.getenv("PACS_PROCESSED_ROOT", "/home/wkmgc/Desktop/Data_Disk_1/PACS数据/DICOM数据/已处理_DICOM数据"))
|
|
|
|
IDENT_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
|
|
WINDOWS = {
|
|
"default": None,
|
|
"bone": (500.0, 2000.0),
|
|
"soft": (40.0, 400.0),
|
|
"contrast": (80.0, 180.0),
|
|
}
|
|
DEFAULT_POSE = {
|
|
"rotateX": 0.0,
|
|
"rotateY": 0.0,
|
|
"rotateZ": 0.0,
|
|
"translateX": 0.0,
|
|
"translateY": 0.0,
|
|
"translateZ": 0.0,
|
|
"scale": 1.0,
|
|
"flipX": False,
|
|
"flipY": False,
|
|
"flipZ": False,
|
|
}
|
|
DICOM_TAGS = [
|
|
"SeriesInstanceUID",
|
|
"StudyInstanceUID",
|
|
"SeriesNumber",
|
|
"SeriesDescription",
|
|
"InstanceNumber",
|
|
"SliceLocation",
|
|
"ImagePositionPatient",
|
|
"AcquisitionTime",
|
|
"ContentTime",
|
|
"SeriesTime",
|
|
"Modality",
|
|
"BodyPartExamined",
|
|
"Rows",
|
|
"Columns",
|
|
"PixelSpacing",
|
|
"SliceThickness",
|
|
"SpacingBetweenSlices",
|
|
"WindowCenter",
|
|
"WindowWidth",
|
|
]
|
|
|
|
|
|
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)
|
|
|
|
app = FastAPI(title="DICOM UPP Registration Workspace")
|
|
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
|
|
TOKENS: dict[str, str] = {}
|
|
STUDY_CACHE: dict[str, dict[str, Any]] = {}
|
|
STACK_CACHE: dict[str, tuple[float, dict[str, Any]]] = {}
|
|
|
|
|
|
class LoginPayload(BaseModel):
|
|
username: str
|
|
password: str
|
|
|
|
|
|
class RegistrationPayload(BaseModel):
|
|
ct_number: str
|
|
algorithm_model: str = "未指定模型"
|
|
registration_status: str = "unregistered"
|
|
series_instance_uid: str = ""
|
|
series_description: str = ""
|
|
selected_stl_files: list[dict[str, Any]] = []
|
|
transform: dict[str, Any] = {}
|
|
module_styles: dict[str, Any] = {}
|
|
dicom_reference: dict[str, Any] = {}
|
|
model_reference: dict[str, Any] = {}
|
|
notes: str = ""
|
|
|
|
|
|
def pg_env() -> dict[str, str]:
|
|
env = os.environ.copy()
|
|
env.update({"PGHOST": PGHOST, "PGPORT": PGPORT, "PGUSER": PGUSER, "PGDATABASE": PGDATABASE})
|
|
if os.getenv("PGPASSWORD"):
|
|
env["PGPASSWORD"] = os.environ["PGPASSWORD"]
|
|
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:
|
|
if value is None:
|
|
return "NULL"
|
|
return "'" + str(value).replace("'", "''") + "'"
|
|
|
|
|
|
def json_sql(value: Any) -> str:
|
|
return f"{sql_literal(json.dumps(value, ensure_ascii=False, separators=(',', ':')))}::jsonb"
|
|
|
|
|
|
def normalize_ct(value: Any) -> str:
|
|
return re.sub(r"\s+", "", str(value or "")).upper()
|
|
|
|
|
|
def parse_json_list(value: Any) -> list[Any]:
|
|
if isinstance(value, list):
|
|
return value
|
|
if isinstance(value, str) and value:
|
|
try:
|
|
parsed = json.loads(value)
|
|
return parsed if isinstance(parsed, list) else []
|
|
except Exception:
|
|
return []
|
|
return []
|
|
|
|
|
|
def ensure_registration_table() -> None:
|
|
pg_scalar(
|
|
f"""
|
|
CREATE TABLE IF NOT EXISTS public.{REGISTRATION_TABLE_SQL} (
|
|
id bigserial PRIMARY KEY,
|
|
ct_number text NOT NULL,
|
|
algorithm_model text NOT NULL DEFAULT '未指定模型',
|
|
registration_status text NOT NULL DEFAULT 'unregistered',
|
|
series_instance_uid text NOT NULL DEFAULT '',
|
|
series_description text NOT NULL DEFAULT '',
|
|
selected_stl_files jsonb NOT NULL DEFAULT '[]'::jsonb,
|
|
transform jsonb NOT NULL DEFAULT '{{}}'::jsonb,
|
|
module_styles jsonb NOT NULL DEFAULT '{{}}'::jsonb,
|
|
dicom_reference jsonb NOT NULL DEFAULT '{{}}'::jsonb,
|
|
model_reference jsonb NOT NULL DEFAULT '{{}}'::jsonb,
|
|
notes text NOT NULL DEFAULT '',
|
|
updated_by text NOT NULL DEFAULT 'admin',
|
|
created_at timestamptz NOT NULL DEFAULT now(),
|
|
updated_at timestamptz NOT NULL DEFAULT now(),
|
|
UNIQUE (ct_number, algorithm_model)
|
|
);
|
|
ALTER TABLE public.{REGISTRATION_TABLE_SQL}
|
|
ADD COLUMN IF NOT EXISTS algorithm_model text NOT NULL DEFAULT '未指定模型',
|
|
ADD COLUMN IF NOT EXISTS registration_status text NOT NULL DEFAULT 'unregistered',
|
|
ADD COLUMN IF NOT EXISTS series_instance_uid text NOT NULL DEFAULT '',
|
|
ADD COLUMN IF NOT EXISTS series_description text NOT NULL DEFAULT '',
|
|
ADD COLUMN IF NOT EXISTS selected_stl_files jsonb NOT NULL DEFAULT '[]'::jsonb,
|
|
ADD COLUMN IF NOT EXISTS transform jsonb NOT NULL DEFAULT '{{}}'::jsonb,
|
|
ADD COLUMN IF NOT EXISTS module_styles jsonb NOT NULL DEFAULT '{{}}'::jsonb,
|
|
ADD COLUMN IF NOT EXISTS dicom_reference jsonb NOT NULL DEFAULT '{{}}'::jsonb,
|
|
ADD COLUMN IF NOT EXISTS model_reference jsonb NOT NULL DEFAULT '{{}}'::jsonb,
|
|
ADD COLUMN IF NOT EXISTS notes text NOT NULL DEFAULT '',
|
|
ADD COLUMN IF NOT EXISTS locked boolean NOT NULL DEFAULT false,
|
|
ADD COLUMN IF NOT EXISTS updated_by text NOT NULL DEFAULT 'admin',
|
|
ADD COLUMN IF NOT EXISTS created_at timestamptz NOT NULL DEFAULT now(),
|
|
ADD COLUMN IF NOT EXISTS updated_at timestamptz NOT NULL DEFAULT now();
|
|
UPDATE public.{REGISTRATION_TABLE_SQL}
|
|
SET
|
|
ct_number = upper(regexp_replace(COALESCE(ct_number, ''), '\\s+', '', 'g')),
|
|
algorithm_model = COALESCE(NULLIF(btrim(algorithm_model), ''), '未指定模型'),
|
|
registration_status = CASE
|
|
WHEN registration_status IN ('registered', 'unregistered') THEN registration_status
|
|
WHEN COALESCE(locked, false) IS TRUE THEN 'registered'
|
|
ELSE 'unregistered'
|
|
END
|
|
WHERE ct_number <> upper(regexp_replace(COALESCE(ct_number, ''), '\\s+', '', 'g'))
|
|
OR algorithm_model = ''
|
|
OR registration_status NOT IN ('registered', 'unregistered');
|
|
CREATE TABLE IF NOT EXISTS public.{REGISTRATION_TABLE_SQL}_duplicate_archive (
|
|
archived_at timestamptz NOT NULL DEFAULT now(),
|
|
reason text NOT NULL,
|
|
row_data jsonb NOT NULL
|
|
);
|
|
WITH ranked AS (
|
|
SELECT id,
|
|
row_number() OVER (
|
|
PARTITION BY ct_number, algorithm_model
|
|
ORDER BY registration_status = 'registered' DESC, updated_at DESC NULLS LAST, id DESC
|
|
) AS rn
|
|
FROM public.{REGISTRATION_TABLE_SQL}
|
|
),
|
|
duplicates AS (
|
|
SELECT t.*
|
|
FROM public.{REGISTRATION_TABLE_SQL} t
|
|
JOIN ranked r ON r.id = t.id
|
|
WHERE r.rn > 1
|
|
)
|
|
INSERT INTO public.{REGISTRATION_TABLE_SQL}_duplicate_archive(reason, row_data)
|
|
SELECT 'ct_algorithm_unique_migration', row_to_json(duplicates)::jsonb
|
|
FROM duplicates;
|
|
WITH ranked AS (
|
|
SELECT id,
|
|
row_number() OVER (
|
|
PARTITION BY ct_number, algorithm_model
|
|
ORDER BY registration_status = 'registered' DESC, updated_at DESC NULLS LAST, id DESC
|
|
) AS rn
|
|
FROM public.{REGISTRATION_TABLE_SQL}
|
|
)
|
|
DELETE FROM public.{REGISTRATION_TABLE_SQL} t
|
|
USING ranked r
|
|
WHERE t.id = r.id AND r.rn > 1;
|
|
DO $$
|
|
DECLARE
|
|
item record;
|
|
BEGIN
|
|
FOR item IN
|
|
SELECT conname
|
|
FROM pg_constraint
|
|
WHERE conrelid = 'public.{REGISTRATION_TABLE_SQL}'::regclass
|
|
AND contype = 'u'
|
|
AND pg_get_constraintdef(oid) NOT LIKE '%(ct_number, algorithm_model)%'
|
|
LOOP
|
|
EXECUTE format('ALTER TABLE public.{REGISTRATION_TABLE_SQL} DROP CONSTRAINT %I', item.conname);
|
|
END LOOP;
|
|
IF NOT EXISTS (
|
|
SELECT 1
|
|
FROM pg_constraint
|
|
WHERE conrelid = 'public.{REGISTRATION_TABLE_SQL}'::regclass
|
|
AND contype = 'u'
|
|
AND conname = '{REGISTRATION_TABLE_SQL}_ct_algorithm_key'
|
|
) THEN
|
|
ALTER TABLE public.{REGISTRATION_TABLE_SQL}
|
|
ADD CONSTRAINT {REGISTRATION_TABLE_SQL}_ct_algorithm_key UNIQUE (ct_number, algorithm_model);
|
|
END IF;
|
|
END $$;
|
|
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}_status ON public.{REGISTRATION_TABLE_SQL}(registration_status);
|
|
CREATE INDEX IF NOT EXISTS idx_{REGISTRATION_TABLE_SQL}_algorithm ON public.{REGISTRATION_TABLE_SQL}(algorithm_model);
|
|
""",
|
|
timeout=14,
|
|
)
|
|
|
|
|
|
def db_available() -> tuple[bool, str]:
|
|
if not os.getenv("PGPASSWORD"):
|
|
return False, "PGPASSWORD 未设置"
|
|
try:
|
|
pg_scalar("SELECT 1", timeout=4)
|
|
return True, "connected"
|
|
except Exception as exc:
|
|
return False, str(exc)
|
|
|
|
|
|
def require_auth(authorization: str | None = Header(default=None), access_token: str = Query(default="")) -> str:
|
|
token = access_token.strip()
|
|
if not token and authorization and authorization.startswith("Bearer "):
|
|
token = authorization.removeprefix("Bearer ").strip()
|
|
user = TOKENS.get(token)
|
|
if not user:
|
|
raise HTTPException(status_code=401, detail="unauthorized")
|
|
return user
|
|
|
|
|
|
@app.on_event("startup")
|
|
def startup() -> None:
|
|
try:
|
|
ensure_registration_table()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
@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()
|
|
if username != WEB_USER or payload.password != WEB_PASSWORD:
|
|
raise HTTPException(status_code=401, detail="用户名或密码错误")
|
|
token = secrets.token_urlsafe(32)
|
|
TOKENS[token] = username
|
|
return {"token": token, "username": username, "role": "管理员"}
|
|
|
|
|
|
@app.get("/api/auth/me")
|
|
def me(user: str = Depends(require_auth)) -> dict[str, str]:
|
|
return {"username": user, "role": "管理员"}
|
|
|
|
|
|
@app.get("/api/status")
|
|
def status(_: str = Depends(require_auth)) -> dict[str, Any]:
|
|
ok, message = db_available()
|
|
counts = {"complete_cases": 0, "registered": 0, "unregistered": 0}
|
|
if ok:
|
|
ensure_registration_table()
|
|
rows = pg_json_rows(
|
|
f"""
|
|
WITH complete AS (
|
|
SELECT DISTINCT p.ct_number, COALESCE(NULLIF(u.algorithm_model, ''), '未指定模型') AS algorithm_model
|
|
FROM public.{PACS_TABLE_SQL} p
|
|
JOIN public.{UPP_ASSET_TABLE_SQL} u ON upper(u.ct_number) = upper(p.ct_number)
|
|
LEFT JOIN public.{UPP_STL_TABLE_SQL} s ON upper(s.ct_number) = upper(p.ct_number)
|
|
WHERE COALESCE(u.stl_present, false) OR COALESCE(u.stl_file_count, 0) > 0 OR s.ct_number IS NOT NULL
|
|
)
|
|
SELECT
|
|
count(*)::int AS complete_cases,
|
|
count(*) FILTER (WHERE r.registration_status = 'registered')::int AS registered,
|
|
count(*) FILTER (WHERE COALESCE(r.registration_status, 'unregistered') <> 'registered')::int AS unregistered
|
|
FROM complete c
|
|
LEFT JOIN public.{REGISTRATION_TABLE_SQL} r
|
|
ON r.ct_number = c.ct_number AND r.algorithm_model = c.algorithm_model
|
|
""",
|
|
timeout=10,
|
|
)
|
|
if rows:
|
|
counts = rows[0]
|
|
return {
|
|
"database": {"ok": ok, "message": message, "host": PGHOST, "database": PGDATABASE},
|
|
"counts": counts,
|
|
"links": {"pacs_viewer_url": PACS_VIEWER_URL, "relation_viewer_url": RELATION_VIEWER_URL},
|
|
"server_time": time.strftime("%Y-%m-%d %H:%M:%S"),
|
|
}
|
|
|
|
|
|
def annotation_labels_sql() -> str:
|
|
return f"""
|
|
SELECT
|
|
a.ct_number,
|
|
COALESCE(jsonb_agg(DISTINCT part.value) FILTER (WHERE part.value IS NOT NULL), '[]'::jsonb) AS body_part_keys,
|
|
COALESCE(jsonb_agg(DISTINCT
|
|
CASE part.value
|
|
WHEN 'head_neck' THEN '头颈部'
|
|
WHEN 'chest' THEN '胸部'
|
|
WHEN 'upper_abdomen' THEN '上腹部'
|
|
WHEN 'lower_abdomen' THEN '下腹部'
|
|
WHEN 'pelvis' THEN '盆腔'
|
|
ELSE NULL
|
|
END
|
|
) FILTER (WHERE part.value IS NOT NULL), '[]'::jsonb) AS body_part_labels
|
|
FROM public.{PACS_ANNOTATION_TABLE_SQL} a
|
|
LEFT JOIN LATERAL jsonb_array_elements_text(COALESCE(a.body_parts, '[]'::jsonb)) AS part(value) ON true
|
|
WHERE COALESCE(a.skipped, false) IS NOT TRUE
|
|
GROUP BY a.ct_number
|
|
"""
|
|
|
|
|
|
def relation_select(where_sql: str = "") -> str:
|
|
return f"""
|
|
WITH ann AS ({annotation_labels_sql()}),
|
|
complete AS (
|
|
SELECT
|
|
upper(p.ct_number) AS ct_key,
|
|
p.ct_number,
|
|
COALESCE(NULLIF(u.algorithm_model, ''), '未指定模型') AS algorithm_model,
|
|
p.batch_name,
|
|
COALESCE(NULLIF(p.source_patient_name, ''), NULLIF(p.patient_name_dicom, '')) AS patient_name,
|
|
p.patient_id,
|
|
p.study_date,
|
|
p.study_time,
|
|
p.study_description,
|
|
p.series_count,
|
|
p.dicom_file_count,
|
|
p.processed_path,
|
|
u.patient_name AS upp_patient_name,
|
|
u.upp_status,
|
|
COALESCE(u.processed_stl_dir, '') AS processed_stl_dir,
|
|
COALESCE(u.stl_file_count, s.file_count, 0)::int AS stl_file_count,
|
|
COALESCE(u.stl_total_bytes, s.total_bytes, 0)::bigint AS stl_total_bytes,
|
|
COALESCE(ann.body_part_keys, ps.body_parts, '[]'::jsonb) AS body_part_keys,
|
|
COALESCE(ann.body_part_labels, '[]'::jsonb) AS body_part_labels
|
|
FROM public.{PACS_TABLE_SQL} p
|
|
JOIN public.{UPP_ASSET_TABLE_SQL} u ON upper(u.ct_number) = upper(p.ct_number)
|
|
LEFT JOIN public.{UPP_STL_TABLE_SQL} s ON upper(s.ct_number) = upper(p.ct_number)
|
|
LEFT JOIN public.{PACS_SUMMARY_TABLE_SQL} ps ON ps.ct_number = p.ct_number
|
|
LEFT JOIN ann ON ann.ct_number = p.ct_number
|
|
WHERE COALESCE(u.stl_present, false) OR COALESCE(u.stl_file_count, 0) > 0 OR s.ct_number IS NOT NULL
|
|
)
|
|
SELECT
|
|
c.*,
|
|
COALESCE(r.registration_status, 'unregistered') AS registration_status,
|
|
r.series_instance_uid,
|
|
r.series_description,
|
|
r.transform,
|
|
r.selected_stl_files,
|
|
r.notes,
|
|
r.updated_at AS registration_updated_at
|
|
FROM complete c
|
|
LEFT JOIN public.{REGISTRATION_TABLE_SQL} r
|
|
ON r.ct_number = c.ct_number AND r.algorithm_model = c.algorithm_model
|
|
{where_sql}
|
|
"""
|
|
|
|
|
|
@app.get("/api/cases")
|
|
def cases(
|
|
q: str = "",
|
|
status_filter: str = Query(default="", alias="status"),
|
|
body_part: str = "",
|
|
algorithm_model: str = "",
|
|
limit: int = Query(default=120, ge=1, le=400),
|
|
_: str = Depends(require_auth),
|
|
) -> list[dict[str, Any]]:
|
|
ensure_registration_table()
|
|
clauses = []
|
|
if q.strip():
|
|
like = "%" + q.strip().replace("%", "").replace("_", "") + "%"
|
|
clauses.append(
|
|
"("
|
|
+ " OR ".join(
|
|
[
|
|
f"ct_number ILIKE {sql_literal(like)}",
|
|
f"patient_name ILIKE {sql_literal(like)}",
|
|
f"patient_id ILIKE {sql_literal(like)}",
|
|
f"algorithm_model ILIKE {sql_literal(like)}",
|
|
f"upp_patient_name ILIKE {sql_literal(like)}",
|
|
]
|
|
)
|
|
+ ")"
|
|
)
|
|
if status_filter in {"registered", "unregistered"}:
|
|
clauses.append(f"registration_status = {sql_literal(status_filter)}")
|
|
if body_part in {"head_neck", "chest", "upper_abdomen", "lower_abdomen", "pelvis"}:
|
|
clauses.append(f"body_part_keys ? {sql_literal(body_part)}")
|
|
if algorithm_model.strip():
|
|
clauses.append(f"algorithm_model ILIKE {sql_literal('%' + algorithm_model.strip().replace('%', '').replace('_', '') + '%')}")
|
|
where_sql = "WHERE " + " AND ".join(clauses) if clauses else ""
|
|
return pg_json_rows(
|
|
f"""
|
|
SELECT *
|
|
FROM ({relation_select()}) relation
|
|
{where_sql}
|
|
ORDER BY
|
|
registration_status = 'registered',
|
|
COALESCE(study_date, '') DESC,
|
|
COALESCE(study_time, '') DESC,
|
|
ct_key
|
|
LIMIT {int(limit)}
|
|
""",
|
|
timeout=20,
|
|
)
|
|
|
|
|
|
@app.get("/api/cases/{ct_number}")
|
|
def case_detail(ct_number: str, algorithm_model: str = "未指定模型", _: str = Depends(require_auth)) -> dict[str, Any]:
|
|
rows = pg_json_rows(
|
|
f"""
|
|
SELECT *
|
|
FROM ({relation_select()}) relation
|
|
WHERE ct_key = {sql_literal(normalize_ct(ct_number))}
|
|
AND algorithm_model = {sql_literal(algorithm_model or '未指定模型')}
|
|
LIMIT 1
|
|
""",
|
|
timeout=16,
|
|
)
|
|
if not rows:
|
|
raise HTTPException(status_code=404, detail="未找到完整关联 CT")
|
|
return rows[0]
|
|
|
|
|
|
def numeric(value: Any, fallback: float = 0.0) -> float:
|
|
try:
|
|
return float(str(value).strip().split("\\")[0])
|
|
except Exception:
|
|
return fallback
|
|
|
|
|
|
def text(value: Any) -> str:
|
|
return "" if value is None else str(value).strip()
|
|
|
|
|
|
def read_header(path: Path) -> dict[str, str]:
|
|
ds = pydicom.dcmread(str(path), stop_before_pixels=True, force=True, specific_tags=DICOM_TAGS + ["SpecificCharacterSet"])
|
|
return {tag: text(getattr(ds, tag, "")) for tag in DICOM_TAGS}
|
|
|
|
|
|
def sort_key(item: tuple[Path, dict[str, str]]) -> tuple[float, float, str]:
|
|
path, meta = item
|
|
z = numeric(meta.get("SliceLocation"), 0.0)
|
|
position = meta.get("ImagePositionPatient", "")
|
|
if position and not meta.get("SliceLocation"):
|
|
try:
|
|
z = float(str(position).strip("[]").split(",")[-1])
|
|
except Exception:
|
|
z = 0.0
|
|
return (z, numeric(meta.get("InstanceNumber"), 0.0), str(path))
|
|
|
|
|
|
def get_study_record(ct_number: str) -> dict[str, Any]:
|
|
rows = pg_json_rows(
|
|
f"""
|
|
SELECT *
|
|
FROM public.{PACS_TABLE_SQL}
|
|
WHERE upper(ct_number) = {sql_literal(normalize_ct(ct_number))}
|
|
LIMIT 1
|
|
""",
|
|
timeout=12,
|
|
)
|
|
if not rows:
|
|
raise HTTPException(status_code=404, detail="DICOM 检查不存在")
|
|
return rows[0]
|
|
|
|
|
|
def resolve_study_root(study: dict[str, Any]) -> Path:
|
|
root = Path(study.get("processed_path") or "")
|
|
if root.exists():
|
|
return root
|
|
target_folder = str(study.get("target_folder_name") or "")
|
|
if target_folder and PROCESSED_ROOT.exists():
|
|
found = next(PROCESSED_ROOT.rglob(target_folder), None)
|
|
if found:
|
|
return found
|
|
return root
|
|
|
|
|
|
def get_annotations(ct_number: str) -> dict[str, dict[str, Any]]:
|
|
try:
|
|
rows = pg_json_rows(
|
|
f"""
|
|
SELECT *
|
|
FROM public.{PACS_ANNOTATION_TABLE_SQL}
|
|
WHERE upper(ct_number) = {sql_literal(normalize_ct(ct_number))}
|
|
""",
|
|
timeout=12,
|
|
)
|
|
except Exception:
|
|
return {}
|
|
return {str(row.get("series_instance_uid") or ""): row for row in rows}
|
|
|
|
|
|
def annotation_labels(row: dict[str, Any]) -> list[str]:
|
|
if not row:
|
|
return []
|
|
if row.get("skipped"):
|
|
return ["略过/不采用"]
|
|
labels = []
|
|
for part in parse_json_list(row.get("body_parts")):
|
|
if part == "head_neck":
|
|
labels.append("头颈部")
|
|
elif part == "chest":
|
|
labels.append("胸部")
|
|
elif part == "upper_abdomen":
|
|
phase = {"plain": "平扫", "arterial": "动脉期", "portal_venous": "门脉期", "delayed": "延迟期", "unknown": "无法判别"}.get(row.get("upper_abdomen_phase") or "unknown", "无法判别")
|
|
labels.append(f"上腹部-{phase}")
|
|
elif part == "lower_abdomen":
|
|
labels.append("下腹部")
|
|
elif part == "pelvis":
|
|
labels.append("盆腔")
|
|
return labels
|
|
|
|
|
|
def scan_study(ct_number: str) -> dict[str, Any]:
|
|
key = normalize_ct(ct_number)
|
|
cached = STUDY_CACHE.get(key)
|
|
if cached and time.time() - cached["cached_at"] < 600:
|
|
return cached
|
|
study = get_study_record(ct_number)
|
|
root = resolve_study_root(study)
|
|
if not root.exists():
|
|
raise HTTPException(status_code=404, detail=f"DICOM 目录不存在：{root}")
|
|
grouped: dict[str, list[tuple[Path, dict[str, str]]]] = defaultdict(list)
|
|
for path in root.rglob("*.dcm"):
|
|
try:
|
|
meta = read_header(path)
|
|
except Exception:
|
|
continue
|
|
uid = meta.get("SeriesInstanceUID") or path.parent.name
|
|
grouped[uid].append((path, meta))
|
|
annotations = get_annotations(ct_number)
|
|
series = []
|
|
file_map = {}
|
|
for uid, items in grouped.items():
|
|
items.sort(key=sort_key)
|
|
first = items[0][1]
|
|
last = items[-1][1]
|
|
file_map[uid] = [path for path, _ in items]
|
|
annotation = annotations.get(uid) or {}
|
|
series.append(
|
|
{
|
|
"series_uid": uid,
|
|
"description": first.get("SeriesDescription") or "未命名序列",
|
|
"series_number": first.get("SeriesNumber") or "",
|
|
"count": len(items),
|
|
"modality": first.get("Modality") or "",
|
|
"rows": first.get("Rows") or "",
|
|
"columns": first.get("Columns") or "",
|
|
"body_part_dicom": first.get("BodyPartExamined") or "",
|
|
"series_time": first.get("SeriesTime") or first.get("AcquisitionTime") or first.get("ContentTime") or "",
|
|
"first_time": first.get("AcquisitionTime") or first.get("ContentTime") or "",
|
|
"last_time": last.get("AcquisitionTime") or last.get("ContentTime") or "",
|
|
"pixel_spacing": first.get("PixelSpacing") or "",
|
|
"slice_thickness": first.get("SliceThickness") or "",
|
|
"spacing_between_slices": first.get("SpacingBetweenSlices") or "",
|
|
"annotation_labels": annotation_labels(annotation),
|
|
}
|
|
)
|
|
series.sort(key=lambda item: (str(item.get("first_time") or item.get("series_time") or ""), numeric(item.get("series_number"), 999999), item.get("description") or ""))
|
|
payload = {"cached_at": time.time(), "study": study, "root": str(root), "series": series, "files": file_map}
|
|
STUDY_CACHE[key] = payload
|
|
return payload
|
|
|
|
|
|
@app.get("/api/cases/{ct_number}/series")
|
|
def series(ct_number: str, _: str = Depends(require_auth)) -> dict[str, Any]:
|
|
data = scan_study(ct_number)
|
|
return {"study": data["study"], "root": data["root"], "series": data["series"]}
|
|
|
|
|
|
def get_series_files(ct_number: str, series_uid: str) -> list[Path]:
|
|
data = scan_study(ct_number)
|
|
files = data["files"].get(series_uid)
|
|
if not files:
|
|
raise HTTPException(status_code=404, detail="DICOM 序列不存在")
|
|
return files
|
|
|
|
|
|
def window_values(ds: pydicom.Dataset, preset: str) -> tuple[float, float]:
|
|
if preset in WINDOWS and WINDOWS[preset]:
|
|
return WINDOWS[preset] # type: ignore[return-value]
|
|
center = getattr(ds, "WindowCenter", 40)
|
|
width = getattr(ds, "WindowWidth", 400)
|
|
if isinstance(center, pydicom.multival.MultiValue):
|
|
center = center[0]
|
|
if isinstance(width, pydicom.multival.MultiValue):
|
|
width = width[0]
|
|
return numeric(center, 40.0), numeric(width, 400.0)
|
|
|
|
|
|
def pixel_spacing_from_ds(ds: pydicom.Dataset) -> tuple[float, float]:
|
|
spacing = getattr(ds, "PixelSpacing", None)
|
|
if spacing and len(spacing) >= 2:
|
|
return max(numeric(spacing[0], 1.0), 0.001), max(numeric(spacing[1], 1.0), 0.001)
|
|
return 1.0, 1.0
|
|
|
|
|
|
def dicom_to_hu(ds: pydicom.Dataset) -> np.ndarray:
|
|
arr = ds.pixel_array.astype(np.float32)
|
|
slope = float(getattr(ds, "RescaleSlope", 1) or 1)
|
|
intercept = float(getattr(ds, "RescaleIntercept", 0) or 0)
|
|
return arr * slope + intercept
|
|
|
|
|
|
def slice_spacing_from_datasets(datasets: list[pydicom.Dataset]) -> float:
|
|
positions = []
|
|
for ds in datasets:
|
|
position = getattr(ds, "ImagePositionPatient", None)
|
|
if position and len(position) >= 3:
|
|
positions.append(np.array([numeric(position[0]), numeric(position[1]), numeric(position[2])], dtype=np.float32))
|
|
distances = [float(np.linalg.norm(b - a)) for a, b in zip(positions, positions[1:])]
|
|
distances = [item for item in distances if item > 0.001]
|
|
if distances:
|
|
return max(float(np.median(distances)), 0.001)
|
|
if datasets:
|
|
spacing = numeric(getattr(datasets[0], "SpacingBetweenSlices", 0), 0.0)
|
|
if spacing > 0.001:
|
|
return spacing
|
|
thickness = numeric(getattr(datasets[0], "SliceThickness", 0), 0.0)
|
|
if thickness > 0.001:
|
|
return thickness
|
|
return 1.0
|
|
|
|
|
|
def resize_for_spacing(pil: Image.Image, row_spacing: float, col_spacing: float) -> Image.Image:
|
|
if abs(row_spacing - col_spacing) < 0.01:
|
|
return pil
|
|
unit = min(row_spacing, col_spacing)
|
|
width = max(1, int(round(pil.width * col_spacing / unit)))
|
|
height = max(1, int(round(pil.height * row_spacing / unit)))
|
|
scale = min(1.0, 1800 / max(width, height))
|
|
target = (max(1, int(round(width * scale))), max(1, int(round(height * scale))))
|
|
return pil if target == pil.size else pil.resize(target, Image.Resampling.BILINEAR)
|
|
|
|
|
|
def render_array(arr: np.ndarray, center: float, width: float, max_size: int = 900, spacing: tuple[float, float] = (1.0, 1.0), invert: bool = False) -> bytes:
|
|
low = center - width / 2.0
|
|
high = center + width / 2.0
|
|
img = ((np.clip(arr, low, high) - low) / max(high - low, 1.0) * 255.0).astype(np.uint8)
|
|
if invert:
|
|
img = 255 - img
|
|
pil = Image.fromarray(img)
|
|
pil = resize_for_spacing(pil, spacing[0], spacing[1])
|
|
if max(pil.size) > max_size:
|
|
pil.thumbnail((max_size, max_size), Image.Resampling.BILINEAR)
|
|
output = io.BytesIO()
|
|
pil.save(output, format="PNG", optimize=True)
|
|
return output.getvalue()
|
|
|
|
|
|
def load_stack_data(ct_number: str, series_uid: str) -> dict[str, Any]:
|
|
key = f"{normalize_ct(ct_number)}|{series_uid}"
|
|
cached = STACK_CACHE.get(key)
|
|
if cached:
|
|
STACK_CACHE[key] = (time.time(), cached[1])
|
|
return cached[1]
|
|
files = get_series_files(ct_number, series_uid)
|
|
arrays = []
|
|
datasets = []
|
|
for path in files:
|
|
ds = pydicom.dcmread(str(path), force=True)
|
|
datasets.append(ds)
|
|
arrays.append(dicom_to_hu(ds))
|
|
stack = np.stack(arrays, axis=0)
|
|
row_spacing, col_spacing = pixel_spacing_from_ds(datasets[min(len(datasets) - 1, len(datasets) // 2)])
|
|
payload = {"stack": stack, "datasets": datasets, "row_spacing": row_spacing, "col_spacing": col_spacing, "slice_spacing": slice_spacing_from_datasets(datasets)}
|
|
STACK_CACHE[key] = (time.time(), payload)
|
|
if len(STACK_CACHE) > 2:
|
|
oldest = sorted(STACK_CACHE.items(), key=lambda item: item[1][0])[0][0]
|
|
STACK_CACHE.pop(oldest, None)
|
|
return payload
|
|
|
|
|
|
@app.get("/api/dicom/image")
|
|
def dicom_image(
|
|
ct_number: str,
|
|
series_uid: str,
|
|
index: int = 0,
|
|
window: str = "default",
|
|
_: str = Depends(require_auth),
|
|
) -> Response:
|
|
files = get_series_files(ct_number, series_uid)
|
|
index = min(max(0, index), len(files) - 1)
|
|
ds = pydicom.dcmread(str(files[index]), force=True)
|
|
center, width = window_values(ds, window)
|
|
payload = render_array(dicom_to_hu(ds), center, width, spacing=pixel_spacing_from_ds(ds), invert=getattr(ds, "PhotometricInterpretation", "") == "MONOCHROME1")
|
|
return Response(payload, media_type="image/png")
|
|
|
|
|
|
@app.get("/api/dicom/fusion-volume")
|
|
def dicom_fusion_volume(
|
|
ct_number: str,
|
|
series_uid: str,
|
|
center_index: int = 0,
|
|
window: str = "soft",
|
|
radius: int = Query(default=96, ge=4, le=512),
|
|
range_start: int | None = None,
|
|
range_end: int | None = None,
|
|
_: str = Depends(require_auth),
|
|
) -> dict[str, Any]:
|
|
stack_data = load_stack_data(ct_number, series_uid)
|
|
stack = stack_data["stack"]
|
|
datasets = stack_data["datasets"]
|
|
total = int(stack.shape[0])
|
|
center_index = min(max(0, center_index), total - 1)
|
|
if range_start is not None and range_end is not None:
|
|
start = min(max(0, int(range_start)), total - 1)
|
|
end = min(max(0, int(range_end)), total - 1)
|
|
if start > end:
|
|
start, end = end, start
|
|
else:
|
|
start = max(0, center_index - radius)
|
|
end = min(total - 1, center_index + radius)
|
|
max_frames = 72
|
|
if end - start + 1 > max_frames:
|
|
step = int(np.ceil((end - start + 1) / max_frames))
|
|
indices = list(range(start, end + 1, step))
|
|
else:
|
|
indices = list(range(start, end + 1))
|
|
sample_ds = datasets[center_index]
|
|
center, width = window_values(sample_ds, window)
|
|
frames = []
|
|
frame_width = 0
|
|
frame_height = 0
|
|
for index in indices:
|
|
png = render_array(stack[index], center, width, max_size=256, spacing=(stack_data["row_spacing"], stack_data["col_spacing"]))
|
|
frames.append("data:image/png;base64," + base64.b64encode(png).decode("ascii"))
|
|
if not frame_width:
|
|
with Image.open(io.BytesIO(png)) as pil:
|
|
frame_width, frame_height = pil.size
|
|
return {
|
|
"frames": frames,
|
|
"indices": indices,
|
|
"start": start,
|
|
"end": end,
|
|
"center": center_index,
|
|
"total": total,
|
|
"width": frame_width,
|
|
"height": frame_height,
|
|
"spacing": {"row": stack_data["row_spacing"], "column": stack_data["col_spacing"], "slice": stack_data["slice_spacing"]},
|
|
"physicalSize": {
|
|
"width": int(stack.shape[2]) * stack_data["col_spacing"],
|
|
"height": int(stack.shape[1]) * stack_data["row_spacing"],
|
|
"depth": total * stack_data["slice_spacing"],
|
|
"unit": "mm",
|
|
},
|
|
}
|
|
|
|
|
|
def stl_rows_for_ct(ct_number: str, algorithm_model: str = "") -> list[dict[str, Any]]:
|
|
rows = pg_json_rows(
|
|
f"""
|
|
SELECT
|
|
COALESCE(NULLIF(u.algorithm_model, ''), '未指定模型') AS algorithm_model,
|
|
u.processed_stl_dir,
|
|
CASE
|
|
WHEN jsonb_typeof(u.stl_files) = 'array' AND jsonb_array_length(u.stl_files) > 0 THEN u.stl_files
|
|
ELSE COALESCE(s.files, '[]'::jsonb)
|
|
END AS files
|
|
FROM public.{UPP_ASSET_TABLE_SQL} u
|
|
LEFT JOIN public.{UPP_STL_TABLE_SQL} s ON upper(s.ct_number) = upper(u.ct_number)
|
|
WHERE upper(u.ct_number) = {sql_literal(normalize_ct(ct_number))}
|
|
AND ({sql_literal(algorithm_model)} = '' OR COALESCE(NULLIF(u.algorithm_model, ''), '未指定模型') = {sql_literal(algorithm_model or '未指定模型')})
|
|
ORDER BY COALESCE(NULLIF(u.algorithm_model, ''), '未指定模型'), u.updated_at DESC NULLS LAST
|
|
""",
|
|
timeout=14,
|
|
)
|
|
output: list[dict[str, Any]] = []
|
|
index = 0
|
|
for asset_index, row in enumerate(rows):
|
|
files = parse_json_list(row.get("files"))
|
|
if not files and row.get("processed_stl_dir"):
|
|
base = Path(str(row["processed_stl_dir"]))
|
|
if base.exists():
|
|
files = [
|
|
{
|
|
"file_name": path.name,
|
|
"segment_name": path.stem,
|
|
"family": path.stem,
|
|
"category": "未分类",
|
|
"processed_file_path": str(path),
|
|
"size_bytes": path.stat().st_size,
|
|
}
|
|
for path in sorted(base.glob("*.stl"))
|
|
]
|
|
for file_info in files:
|
|
path = file_info.get("processed_file_path") or file_info.get("source_file_path")
|
|
if not path:
|
|
continue
|
|
output.append(
|
|
{
|
|
"id": index,
|
|
"asset_id": asset_index,
|
|
"algorithm_model": row.get("algorithm_model") or "未指定模型",
|
|
"file_name": file_info.get("file_name") or Path(path).name,
|
|
"segment_name": file_info.get("segment_name") or Path(path).stem,
|
|
"family": file_info.get("family") or file_info.get("segment_name") or Path(path).stem,
|
|
"category": file_info.get("category") or "未分类",
|
|
"size_bytes": int(file_info.get("size_bytes") or 0),
|
|
"path": path,
|
|
}
|
|
)
|
|
index += 1
|
|
return output
|
|
|
|
|
|
@app.get("/api/cases/{ct_number}/stl")
|
|
def stl_assets(ct_number: str, algorithm_model: str = "", _: str = Depends(require_auth)) -> dict[str, Any]:
|
|
files = stl_rows_for_ct(ct_number, algorithm_model)
|
|
return {
|
|
"ct_number": ct_number,
|
|
"files": files,
|
|
"families": sorted({str(item.get("family") or "") for item in files if item.get("family")}),
|
|
"algorithm_models": sorted({str(item.get("algorithm_model") or "") for item in files if item.get("algorithm_model")}),
|
|
}
|
|
|
|
|
|
@app.get("/api/stl/file")
|
|
def stl_file(ct_number: str, file_id: int, algorithm_model: str = "", _: str = Depends(require_auth)) -> FileResponse:
|
|
files = stl_rows_for_ct(ct_number, algorithm_model)
|
|
selected = next((item for item in files if int(item["id"]) == int(file_id)), None)
|
|
if not selected:
|
|
raise HTTPException(status_code=404, detail="STL 文件不存在")
|
|
path = Path(str(selected["path"]))
|
|
if not path.exists() or not path.is_file():
|
|
raise HTTPException(status_code=404, detail=f"STL 路径不存在：{path}")
|
|
return FileResponse(path, media_type="model/stl", filename=path.name)
|
|
|
|
|
|
@app.get("/api/registrations/{ct_number}")
|
|
def get_registration(ct_number: str, algorithm_model: str = "未指定模型", _: str = Depends(require_auth)) -> dict[str, Any]:
|
|
ensure_registration_table()
|
|
rows = pg_json_rows(
|
|
f"""
|
|
SELECT *
|
|
FROM public.{REGISTRATION_TABLE_SQL}
|
|
WHERE ct_number = {sql_literal(normalize_ct(ct_number))}
|
|
AND algorithm_model = {sql_literal(algorithm_model or '未指定模型')}
|
|
LIMIT 1
|
|
""",
|
|
timeout=12,
|
|
)
|
|
return rows[0] if rows else {
|
|
"ct_number": normalize_ct(ct_number),
|
|
"algorithm_model": algorithm_model or "未指定模型",
|
|
"registration_status": "unregistered",
|
|
"transform": DEFAULT_POSE,
|
|
"selected_stl_files": [],
|
|
"notes": "",
|
|
}
|
|
|
|
|
|
@app.post("/api/registrations")
|
|
def save_registration(payload: RegistrationPayload, user: str = Depends(require_auth)) -> dict[str, Any]:
|
|
ensure_registration_table()
|
|
ct_number = normalize_ct(payload.ct_number)
|
|
if not ct_number:
|
|
raise HTTPException(status_code=400, detail="CT号不能为空")
|
|
algorithm_model = payload.algorithm_model.strip() or "未指定模型"
|
|
registration_status = payload.registration_status if payload.registration_status in {"registered", "unregistered"} else "unregistered"
|
|
transform = {**DEFAULT_POSE, **(payload.transform or {})}
|
|
pg_scalar(
|
|
f"""
|
|
INSERT INTO public.{REGISTRATION_TABLE_SQL} (
|
|
ct_number, algorithm_model, registration_status, series_instance_uid, series_description,
|
|
selected_stl_files, transform, module_styles, dicom_reference, model_reference,
|
|
notes, updated_by, updated_at
|
|
)
|
|
VALUES (
|
|
{sql_literal(ct_number)},
|
|
{sql_literal(algorithm_model)},
|
|
{sql_literal(registration_status)},
|
|
{sql_literal(payload.series_instance_uid.strip())},
|
|
{sql_literal(payload.series_description.strip())},
|
|
{json_sql(payload.selected_stl_files)},
|
|
{json_sql(transform)},
|
|
{json_sql(payload.module_styles)},
|
|
{json_sql(payload.dicom_reference)},
|
|
{json_sql(payload.model_reference)},
|
|
{sql_literal(payload.notes.strip())},
|
|
{sql_literal(user)},
|
|
now()
|
|
)
|
|
ON CONFLICT (ct_number, algorithm_model) DO UPDATE SET
|
|
registration_status = EXCLUDED.registration_status,
|
|
series_instance_uid = EXCLUDED.series_instance_uid,
|
|
series_description = EXCLUDED.series_description,
|
|
selected_stl_files = EXCLUDED.selected_stl_files,
|
|
transform = EXCLUDED.transform,
|
|
module_styles = EXCLUDED.module_styles,
|
|
dicom_reference = EXCLUDED.dicom_reference,
|
|
model_reference = EXCLUDED.model_reference,
|
|
notes = EXCLUDED.notes,
|
|
updated_by = EXCLUDED.updated_by,
|
|
updated_at = now()
|
|
""",
|
|
timeout=14,
|
|
)
|
|
return {"ok": True, "ct_number": ct_number, "algorithm_model": algorithm_model, "registration_status": registration_status}
|