#!/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 concurrent.futures import ThreadPoolExecutor, as_completed
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]] = {}
SERIES_SORT_CACHE: dict[str, tuple[float, list[Path]]] = {}
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
                CASE
                    WHEN part.value IN ('lower_abdomen', 'pelvis') THEN 'abdomen_pelvis'
                    ELSE part.value
                END
            ) 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 'abdomen_pelvis' 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=20, ge=1, le=80),
    offset: int = Query(default=0, ge=0),
    _: str = Depends(require_auth),
) -> 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", "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 ""
    total = int(pg_scalar(
        f"""
        SELECT count(*)::int
        FROM ({relation_select()}) relation
        {where_sql}
        """,
        timeout=20,
    ) or 0)
    rows = 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)}
        OFFSET {int(offset)}
        """,
        timeout=20,
    )
    return {
        "rows": rows,
        "total": total,
        "limit": int(limit),
        "offset": int(offset),
        "has_more": int(offset) + len(rows) < total,
    }


@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 read_headers_bulk(paths: list[Path], max_workers: int = 10) -> dict[Path, dict[str, str]]:
    unique_paths = list(dict.fromkeys(paths))
    if not unique_paths:
        return {}
    results: dict[Path, dict[str, str]] = {}
    workers = max(1, min(max_workers, len(unique_paths)))
    with ThreadPoolExecutor(max_workers=workers) as executor:
        future_map = {executor.submit(read_header, path): path for path in unique_paths}
        for future in as_completed(future_map):
            path = future_map[future]
            try:
                results[path] = future.result()
            except Exception:
                continue
    return results


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():
        direct = PROCESSED_ROOT / target_folder
        if direct.exists():
            return direct
        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 in {"abdomen_pelvis", "lower_abdomen", "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[Path]] = defaultdict(list)
    for path in root.rglob("*.dcm"):
        grouped[path.parent.name].append(path)
    annotations = get_annotations(ct_number)
    series = []
    file_map = {}
    summary_paths: list[Path] = []
    for paths in grouped.values():
        paths = sorted(paths)
        if not paths:
            continue
        summary_paths.append(paths[0])
    header_map = read_headers_bulk(summary_paths)
    for uid, paths in grouped.items():
        paths = sorted(paths)
        if not paths:
            continue
        first = header_map.get(paths[0])
        if not first:
            continue
        last = first
        actual_uid = first.get("SeriesInstanceUID") or uid
        file_map[actual_uid] = paths
        annotation = annotations.get(actual_uid) or annotations.get(uid) or {}
        series.append(
            {
                "series_uid": actual_uid,
                "description": first.get("SeriesDescription") or "未命名序列",
                "series_number": first.get("SeriesNumber") or "",
                "count": len(paths),
                "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 序列不存在")
    cache_key = f"{normalize_ct(ct_number)}|{series_uid}"
    cached = SERIES_SORT_CACHE.get(cache_key)
    if cached:
        SERIES_SORT_CACHE[cache_key] = (time.time(), cached[1])
        return cached[1]
    header_map = read_headers_bulk(list(files), max_workers=12)
    items: list[tuple[Path, dict[str, str]]] = [(path, meta) for path, meta in header_map.items()]
    if not items:
        raise HTTPException(status_code=404, detail="DICOM 序列文件不可读")
    sorted_files = [path for path, _ in sorted(items, key=sort_key)]
    data["files"][series_uid] = sorted_files
    SERIES_SORT_CACHE[cache_key] = (time.time(), sorted_files)
    if len(SERIES_SORT_CACHE) > 8:
        oldest = sorted(SERIES_SORT_CACHE.items(), key=lambda item: item[1][0])[0][0]
        SERIES_SORT_CACHE.pop(oldest, None)
    return sorted_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.LANCZOS)
    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",
    detail: str = "high",
    texture_size: int | None = Query(default=None, ge=128, le=1024),
    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 = {"low": 24, "medium": 48, "high": 72}.get(str(detail).lower(), 48)
    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))
    focus_index = min(max(center_index, start), end)
    indices = sorted({*indices, start, end, focus_index})
    sample_ds = datasets[focus_index]
    center, width = window_values(sample_ds, window)
    max_texture_size = int(texture_size or {"low": 384, "medium": 512, "high": 768}.get(str(detail).lower(), 512))
    frames = []
    frame_width = 0
    frame_height = 0
    for index in indices:
        png = render_array(stack[index], center, width, max_size=max_texture_size, 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": focus_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()
        SERIES_SORT_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}