1380 lines
53 KiB
Python
1380 lines
53 KiB
Python
#!/usr/bin/env python3
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import datetime as dt
|
||
import io
|
||
import json
|
||
import os
|
||
import re
|
||
import secrets
|
||
import shutil
|
||
import subprocess
|
||
import threading
|
||
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")
|
||
PACS_IMAGE_DB_ROOT = Path(os.getenv("PACS_IMAGE_DB_ROOT", "/home/wkmgc/Desktop/PACS影像数据库_192.168.3.3"))
|
||
PROCESSED_ROOT = Path(os.getenv("PACS_PROCESSED_ROOT", str(PACS_IMAGE_DB_ROOT / "PACS数据" / "DICOM数据" / "已处理_DICOM数据")))
|
||
REGISTRATION_CACHE_ROOT = Path(os.getenv("REGISTRATION_CACHE_ROOT", str(PACS_IMAGE_DB_ROOT / "PACS数据" / "DICOM与UPP配准数据")))
|
||
RUNTIME_DIR = Path(os.getenv("REGISTRATION_RUNTIME_DIR", str(APP_DIR / "runtime")))
|
||
SETTINGS_FILE = RUNTIME_DIR / "settings.json"
|
||
CACHE_STATUS_FILE = RUNTIME_DIR / "cache_refresh_status.json"
|
||
|
||
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]]] = {}
|
||
CACHE_REFRESH_LOCK = threading.Lock()
|
||
CACHE_REFRESH_THREAD: threading.Thread | None = None
|
||
SCHEDULER_THREAD: threading.Thread | None = None
|
||
DEFAULT_REFRESH_SETTINGS = {
|
||
"auto_refresh_enabled": True,
|
||
"refresh_time": "03:00",
|
||
"cache_root": str(REGISTRATION_CACHE_ROOT),
|
||
}
|
||
DEFAULT_CACHE_STATUS: dict[str, Any] = {
|
||
"running": False,
|
||
"stage": "idle",
|
||
"message": "尚未刷新",
|
||
"progress": 0,
|
||
"total_cases": 0,
|
||
"processed_cases": 0,
|
||
"copied_files": 0,
|
||
"skipped_files": 0,
|
||
"missing_files": 0,
|
||
"errors": [],
|
||
"last_started_at": "",
|
||
"last_finished_at": "",
|
||
"last_scheduled_date": "",
|
||
}
|
||
CACHE_REFRESH_STATUS: dict[str, Any] = {**DEFAULT_CACHE_STATUS}
|
||
|
||
|
||
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 = ""
|
||
|
||
|
||
class RefreshSettingsPayload(BaseModel):
|
||
auto_refresh_enabled: bool = True
|
||
refresh_time: str = "03:00"
|
||
|
||
|
||
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 now_text() -> str:
|
||
return dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
|
||
def ensure_runtime_dir() -> None:
|
||
RUNTIME_DIR.mkdir(parents=True, exist_ok=True)
|
||
|
||
|
||
def valid_refresh_time(value: str) -> str:
|
||
value = str(value or "").strip()
|
||
if re.fullmatch(r"(?:[01]\d|2[0-3]):[0-5]\d", value):
|
||
return value
|
||
return "03:00"
|
||
|
||
|
||
def read_refresh_settings() -> dict[str, Any]:
|
||
settings = {**DEFAULT_REFRESH_SETTINGS}
|
||
try:
|
||
if SETTINGS_FILE.exists():
|
||
loaded = json.loads(SETTINGS_FILE.read_text(encoding="utf-8"))
|
||
if isinstance(loaded, dict):
|
||
settings.update(loaded)
|
||
except Exception:
|
||
pass
|
||
settings["auto_refresh_enabled"] = bool(settings.get("auto_refresh_enabled", True))
|
||
settings["refresh_time"] = valid_refresh_time(str(settings.get("refresh_time") or "03:00"))
|
||
settings["cache_root"] = str(REGISTRATION_CACHE_ROOT)
|
||
return settings
|
||
|
||
|
||
def write_refresh_settings(settings: dict[str, Any]) -> dict[str, Any]:
|
||
payload = {
|
||
"auto_refresh_enabled": bool(settings.get("auto_refresh_enabled", True)),
|
||
"refresh_time": valid_refresh_time(str(settings.get("refresh_time") or "03:00")),
|
||
"cache_root": str(REGISTRATION_CACHE_ROOT),
|
||
}
|
||
ensure_runtime_dir()
|
||
SETTINGS_FILE.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
return payload
|
||
|
||
|
||
def read_cache_status_file() -> dict[str, Any]:
|
||
status = {**DEFAULT_CACHE_STATUS}
|
||
try:
|
||
if CACHE_STATUS_FILE.exists():
|
||
loaded = json.loads(CACHE_STATUS_FILE.read_text(encoding="utf-8"))
|
||
if isinstance(loaded, dict):
|
||
status.update(loaded)
|
||
except Exception:
|
||
pass
|
||
return status
|
||
|
||
|
||
def write_cache_status(status: dict[str, Any]) -> None:
|
||
ensure_runtime_dir()
|
||
CACHE_STATUS_FILE.write_text(json.dumps(status, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
|
||
|
||
def update_cache_status(**updates: Any) -> dict[str, Any]:
|
||
global CACHE_REFRESH_STATUS
|
||
with CACHE_REFRESH_LOCK:
|
||
CACHE_REFRESH_STATUS.update(updates)
|
||
status = {**CACHE_REFRESH_STATUS}
|
||
try:
|
||
write_cache_status(status)
|
||
except Exception:
|
||
pass
|
||
return status
|
||
|
||
|
||
def get_cache_status_snapshot() -> dict[str, Any]:
|
||
with CACHE_REFRESH_LOCK:
|
||
status = {**CACHE_REFRESH_STATUS}
|
||
settings = read_refresh_settings()
|
||
try:
|
||
refresh_hour, refresh_minute = [int(part) for part in settings["refresh_time"].split(":", 1)]
|
||
candidate = dt.datetime.now().replace(hour=refresh_hour, minute=refresh_minute, second=0, microsecond=0)
|
||
if candidate <= dt.datetime.now():
|
||
candidate += dt.timedelta(days=1)
|
||
status["next_scheduled_at"] = candidate.strftime("%Y-%m-%d %H:%M:%S") if settings["auto_refresh_enabled"] else ""
|
||
except Exception:
|
||
status["next_scheduled_at"] = ""
|
||
status["cache_root"] = str(REGISTRATION_CACHE_ROOT)
|
||
return status
|
||
|
||
|
||
def safe_path_token(value: Any) -> str:
|
||
text_value = re.sub(r"\s+", "_", str(value or "unknown").strip())
|
||
text_value = re.sub(r'[\\\\/:*?"<>|]+', "_", text_value)
|
||
return text_value[:120] or "unknown"
|
||
|
||
|
||
def mapped_path(value: Any) -> Path:
|
||
raw = str(value or "").strip()
|
||
if not raw:
|
||
return Path("")
|
||
path = Path(raw)
|
||
if path.exists():
|
||
return path
|
||
mappings = [
|
||
(
|
||
"/home/wkmgc/Desktop/PACS影像数据库_192.168.3.3",
|
||
PACS_IMAGE_DB_ROOT,
|
||
),
|
||
(
|
||
"/home/wkmgc/Desktop/PACS数据处理/PACS_DICOM处理/已处理_DICOM数据",
|
||
PACS_IMAGE_DB_ROOT / "PACS数据" / "DICOM数据" / "已处理_DICOM数据",
|
||
),
|
||
(
|
||
"/home/wkmgc/Desktop/Data_Disk_1/PACS数据/DICOM数据/已处理_DICOM数据",
|
||
PACS_IMAGE_DB_ROOT / "PACS数据" / "DICOM数据" / "已处理_DICOM数据",
|
||
),
|
||
(
|
||
"/home/wkmgc/Desktop/PACS数据处理/UPP_STL处理/已处理STL数据",
|
||
PACS_IMAGE_DB_ROOT / "PACS数据" / "重建STL数据" / "已处理STL数据",
|
||
),
|
||
(
|
||
"/home/wkmgc/Desktop/Data_Disk_1/PACS数据/重建STL数据/已处理STL数据",
|
||
PACS_IMAGE_DB_ROOT / "PACS数据" / "重建STL数据" / "已处理STL数据",
|
||
),
|
||
]
|
||
for source, target in mappings:
|
||
if raw == source or raw.startswith(source + "/"):
|
||
relative = raw.removeprefix(source).lstrip("/")
|
||
return target / relative
|
||
return path
|
||
|
||
|
||
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:
|
||
global CACHE_REFRESH_STATUS
|
||
CACHE_REFRESH_STATUS = read_cache_status_file()
|
||
if CACHE_REFRESH_STATUS.get("running"):
|
||
CACHE_REFRESH_STATUS.update({"running": False, "stage": "interrupted", "message": "上次刷新未正常结束，可手动重新刷新"})
|
||
write_refresh_settings(read_refresh_settings())
|
||
write_cache_status(CACHE_REFRESH_STATUS)
|
||
try:
|
||
ensure_registration_table()
|
||
except Exception:
|
||
pass
|
||
start_cache_scheduler()
|
||
|
||
|
||
@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"),
|
||
}
|
||
|
||
|
||
@app.get("/api/settings")
|
||
def settings(_: str = Depends(require_auth)) -> dict[str, Any]:
|
||
return {"refresh": read_refresh_settings(), "cache": get_cache_status_snapshot()}
|
||
|
||
|
||
@app.post("/api/settings")
|
||
def update_settings(payload: RefreshSettingsPayload, _: str = Depends(require_auth)) -> dict[str, Any]:
|
||
settings = write_refresh_settings(payload.dict())
|
||
return {"ok": True, "refresh": settings, "cache": get_cache_status_snapshot()}
|
||
|
||
|
||
@app.get("/api/cache/status")
|
||
def cache_status(_: str = Depends(require_auth)) -> dict[str, Any]:
|
||
return get_cache_status_snapshot()
|
||
|
||
|
||
@app.post("/api/cache/refresh")
|
||
def manual_cache_refresh(user: str = Depends(require_auth)) -> dict[str, Any]:
|
||
started = start_background_cache_refresh(reason="manual", user=user)
|
||
return {"started": started, "cache": get_cache_status_snapshot()}
|
||
|
||
|
||
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 = mapped_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 = mapped_path(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
|
||
resolved_path = mapped_path(path)
|
||
output.append(
|
||
{
|
||
"id": index,
|
||
"asset_id": asset_index,
|
||
"algorithm_model": row.get("algorithm_model") or "未指定模型",
|
||
"file_name": file_info.get("file_name") or resolved_path.name,
|
||
"segment_name": file_info.get("segment_name") or resolved_path.stem,
|
||
"family": file_info.get("family") or file_info.get("segment_name") or resolved_path.stem,
|
||
"category": file_info.get("category") or "未分类",
|
||
"size_bytes": int(file_info.get("size_bytes") or 0),
|
||
"path": str(resolved_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)
|
||
|
||
|
||
def copy_if_changed(source: Path, target: Path) -> bool:
|
||
if not source.exists() or not source.is_file():
|
||
raise FileNotFoundError(str(source))
|
||
target.parent.mkdir(parents=True, exist_ok=True)
|
||
if target.exists() and target.stat().st_size == source.stat().st_size:
|
||
return False
|
||
tmp = target.with_suffix(target.suffix + ".tmp")
|
||
shutil.copy2(source, tmp)
|
||
tmp.replace(target)
|
||
return True
|
||
|
||
|
||
def cache_one_case(row: dict[str, Any], cache_root: Path) -> dict[str, int]:
|
||
ct_number = normalize_ct(row.get("ct_number") or row.get("ct_key"))
|
||
algorithm_model = row.get("algorithm_model") or "未指定模型"
|
||
case_dir = cache_root / safe_path_token(ct_number) / safe_path_token(algorithm_model)
|
||
stl_dir = case_dir / "stl"
|
||
case_dir.mkdir(parents=True, exist_ok=True)
|
||
stl_dir.mkdir(parents=True, exist_ok=True)
|
||
(case_dir / "case.json").write_text(json.dumps(row, ensure_ascii=False, indent=2, default=str), encoding="utf-8")
|
||
|
||
counters = {"copied_files": 0, "skipped_files": 0, "missing_files": 0}
|
||
# DICOM pixels and per-slice headers stay on-demand; refresh caches only fast index data and STL materials.
|
||
series_payload: dict[str, Any] = {
|
||
"ct_number": ct_number,
|
||
"series_count": row.get("series_count"),
|
||
"dicom_file_count": row.get("dicom_file_count"),
|
||
"processed_path": str(mapped_path(row.get("processed_path") or "")),
|
||
"cached_at": now_text(),
|
||
}
|
||
(case_dir / "series.json").write_text(json.dumps(series_payload, ensure_ascii=False, indent=2, default=str), encoding="utf-8")
|
||
|
||
manifest = []
|
||
for file_info in stl_rows_for_ct(ct_number, algorithm_model):
|
||
source = Path(str(file_info.get("path") or ""))
|
||
target_name = f"{int(file_info.get('id') or 0):03d}_{safe_path_token(source.name or file_info.get('file_name'))}"
|
||
target = stl_dir / target_name
|
||
cached = {**file_info, "source_path": str(source), "cached_path": str(target), "cached_at": now_text()}
|
||
try:
|
||
if copy_if_changed(source, target):
|
||
counters["copied_files"] += 1
|
||
cached["cache_status"] = "copied"
|
||
else:
|
||
counters["skipped_files"] += 1
|
||
cached["cache_status"] = "unchanged"
|
||
except FileNotFoundError:
|
||
counters["missing_files"] += 1
|
||
cached["cache_status"] = "missing"
|
||
except Exception as exc:
|
||
counters["missing_files"] += 1
|
||
cached["cache_status"] = f"error: {exc}"
|
||
manifest.append(cached)
|
||
(case_dir / "stl_manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2, default=str), encoding="utf-8")
|
||
return counters
|
||
|
||
|
||
def relation_rows_for_cache() -> list[dict[str, Any]]:
|
||
return pg_json_rows(
|
||
f"""
|
||
SELECT *
|
||
FROM ({relation_select()}) relation
|
||
ORDER BY
|
||
registration_status = 'registered',
|
||
COALESCE(study_date, '') DESC,
|
||
COALESCE(study_time, '') DESC,
|
||
ct_key,
|
||
algorithm_model
|
||
""",
|
||
timeout=60,
|
||
)
|
||
|
||
|
||
def run_cache_refresh(reason: str = "manual", user: str = "system") -> None:
|
||
cache_root = REGISTRATION_CACHE_ROOT
|
||
update_cache_status(
|
||
running=True,
|
||
stage="starting",
|
||
message=f"{reason} 刷新启动",
|
||
progress=1,
|
||
total_cases=0,
|
||
processed_cases=0,
|
||
copied_files=0,
|
||
skipped_files=0,
|
||
missing_files=0,
|
||
errors=[],
|
||
last_started_at=now_text(),
|
||
)
|
||
try:
|
||
cache_root.mkdir(parents=True, exist_ok=True)
|
||
STUDY_CACHE.clear()
|
||
STACK_CACHE.clear()
|
||
rows = relation_rows_for_cache()
|
||
total = len(rows)
|
||
update_cache_status(stage="caching", message="正在缓存 DICOM 序列元数据与 STL 材料", total_cases=total, progress=4)
|
||
copied = skipped = missing = 0
|
||
errors: list[str] = []
|
||
for index, row in enumerate(rows, start=1):
|
||
label = f"{row.get('ct_number') or row.get('ct_key')} · {row.get('algorithm_model') or '未指定模型'}"
|
||
try:
|
||
counters = cache_one_case(row, cache_root)
|
||
copied += counters["copied_files"]
|
||
skipped += counters["skipped_files"]
|
||
missing += counters["missing_files"]
|
||
except Exception as exc:
|
||
errors.append(f"{label}: {exc}")
|
||
update_cache_status(
|
||
stage="caching",
|
||
message=f"正在缓存 {label}",
|
||
processed_cases=index,
|
||
copied_files=copied,
|
||
skipped_files=skipped,
|
||
missing_files=missing,
|
||
errors=errors[-20:],
|
||
progress=4 + int(index / max(total, 1) * 94),
|
||
)
|
||
update_cache_status(
|
||
running=False,
|
||
stage="finished",
|
||
message=f"刷新完成：{total} 个 CT/模型，复制 {copied} 个 STL，复用 {skipped} 个 STL",
|
||
progress=100,
|
||
total_cases=total,
|
||
processed_cases=total,
|
||
copied_files=copied,
|
||
skipped_files=skipped,
|
||
missing_files=missing,
|
||
errors=errors[-20:],
|
||
last_finished_at=now_text(),
|
||
)
|
||
except Exception as exc:
|
||
update_cache_status(
|
||
running=False,
|
||
stage="error",
|
||
message=f"刷新失败：{exc}",
|
||
progress=0,
|
||
errors=[str(exc)],
|
||
last_finished_at=now_text(),
|
||
)
|
||
|
||
|
||
def start_background_cache_refresh(reason: str = "manual", user: str = "system") -> bool:
|
||
global CACHE_REFRESH_THREAD
|
||
with CACHE_REFRESH_LOCK:
|
||
if CACHE_REFRESH_THREAD and CACHE_REFRESH_THREAD.is_alive():
|
||
return False
|
||
if CACHE_REFRESH_STATUS.get("running"):
|
||
return False
|
||
CACHE_REFRESH_THREAD = threading.Thread(target=run_cache_refresh, kwargs={"reason": reason, "user": user}, daemon=True)
|
||
CACHE_REFRESH_THREAD.start()
|
||
return True
|
||
|
||
|
||
def cache_scheduler_loop() -> None:
|
||
while True:
|
||
try:
|
||
settings = read_refresh_settings()
|
||
if settings.get("auto_refresh_enabled"):
|
||
refresh_time = settings.get("refresh_time") or "03:00"
|
||
today = dt.date.today().isoformat()
|
||
current = dt.datetime.now().strftime("%H:%M")
|
||
status = get_cache_status_snapshot()
|
||
if current == refresh_time and status.get("last_scheduled_date") != today:
|
||
update_cache_status(last_scheduled_date=today)
|
||
start_background_cache_refresh(reason="scheduled", user="system")
|
||
except Exception:
|
||
pass
|
||
time.sleep(30)
|
||
|
||
|
||
def start_cache_scheduler() -> None:
|
||
global SCHEDULER_THREAD
|
||
if SCHEDULER_THREAD and SCHEDULER_THREAD.is_alive():
|
||
return
|
||
SCHEDULER_THREAD = threading.Thread(target=cache_scheduler_loop, daemon=True)
|
||
SCHEDULER_THREAD.start()
|
||
|
||
|
||
@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}
|