#!/usr/bin/env python3
from __future__ import annotations

import base64
import io
import json
import os
import re
import secrets
import subprocess
import time
from collections import Counter, defaultdict
from pathlib import Path
from typing import Any

import numpy as np
import pydicom
from fastapi import Depends, FastAPI, Header, HTTPException, Query, Response
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
from PIL import Image


APP_DIR = Path(__file__).resolve().parent
STATIC_DIR = APP_DIR / "static"


def load_env_file() -> None:
    env_file = APP_DIR / ".env"
    if not env_file.exists():
        return
    for line in env_file.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'"))


load_env_file()

PGHOST = os.getenv("PGHOST", "192.168.3.3")
PGPORT = os.getenv("PGPORT", "5432")
PGUSER = os.getenv("PGUSER", "his_user")
PGDATABASE = os.getenv("PGDATABASE", "pacs_db")
PACS_TABLE = os.getenv("PACS_TABLE", "pacs_dicom_files")
PACS_SUMMARY_TABLE = os.getenv("PACS_SUMMARY_TABLE", "pacs_dicom_study_summaries")
PACS_ANNOTATION_TABLE = os.getenv("PACS_ANNOTATION_TABLE", "pacs_dicom_series_annotations")
UPP_ASSET_TABLE = os.getenv("UPP_ASSET_TABLE", "upp_exam_assets")
UPP_STL_TABLE = os.getenv("UPP_STL_TABLE", "upp_stl_files")
REGISTRATION_TABLE = os.getenv("REGISTRATION_TABLE", "dicom_upp_registrations")
WEB_USER = os.getenv("REGISTRATION_WEB_USER", "admin")
WEB_PASSWORD = os.getenv("REGISTRATION_WEB_PASSWORD", "123456")
PACS_VIEWER_URL = os.getenv("PACS_VIEWER_URL", "http://127.0.0.1:8107")
RELATION_VIEWER_URL = os.getenv("RELATION_VIEWER_URL", "http://127.0.0.1:8108")
PROCESSED_ROOT = Path(os.getenv("PACS_PROCESSED_ROOT", "/home/wkmgc/Desktop/Data_Disk_1/PACS数据/DICOM数据/已处理_DICOM数据"))

IDENT_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
WINDOWS = {
    "default": None,
    "bone": (500.0, 2000.0),
    "soft": (40.0, 400.0),
    "contrast": (80.0, 180.0),
}
DEFAULT_POSE = {
    "rotateX": 0.0,
    "rotateY": 0.0,
    "rotateZ": 0.0,
    "translateX": 0.0,
    "translateY": 0.0,
    "translateZ": 0.0,
    "scale": 1.0,
    "flipX": False,
    "flipY": False,
    "flipZ": False,
}
DICOM_TAGS = [
    "SeriesInstanceUID",
    "StudyInstanceUID",
    "SeriesNumber",
    "SeriesDescription",
    "InstanceNumber",
    "SliceLocation",
    "ImagePositionPatient",
    "AcquisitionTime",
    "ContentTime",
    "SeriesTime",
    "Modality",
    "BodyPartExamined",
    "Rows",
    "Columns",
    "PixelSpacing",
    "SliceThickness",
    "SpacingBetweenSlices",
    "WindowCenter",
    "WindowWidth",
]


def ident(name: str) -> str:
    if not IDENT_RE.fullmatch(name):
        raise RuntimeError(f"invalid SQL identifier: {name}")
    return name


PACS_TABLE_SQL = ident(PACS_TABLE)
PACS_SUMMARY_TABLE_SQL = ident(PACS_SUMMARY_TABLE)
PACS_ANNOTATION_TABLE_SQL = ident(PACS_ANNOTATION_TABLE)
UPP_ASSET_TABLE_SQL = ident(UPP_ASSET_TABLE)
UPP_STL_TABLE_SQL = ident(UPP_STL_TABLE)
REGISTRATION_TABLE_SQL = ident(REGISTRATION_TABLE)

app = FastAPI(title="DICOM UPP Registration Workspace")
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
TOKENS: dict[str, str] = {}
STUDY_CACHE: dict[str, dict[str, Any]] = {}
STACK_CACHE: dict[str, tuple[float, dict[str, Any]]] = {}


class LoginPayload(BaseModel):
    username: str
    password: str


class RegistrationPayload(BaseModel):
    ct_number: str
    algorithm_model: str = "未指定模型"
    registration_status: str = "unregistered"
    series_instance_uid: str = ""
    series_description: str = ""
    selected_stl_files: list[dict[str, Any]] = []
    transform: dict[str, Any] = {}
    module_styles: dict[str, Any] = {}
    dicom_reference: dict[str, Any] = {}
    model_reference: dict[str, Any] = {}
    notes: str = ""


def pg_env() -> dict[str, str]:
    env = os.environ.copy()
    env.update({"PGHOST": PGHOST, "PGPORT": PGPORT, "PGUSER": PGUSER, "PGDATABASE": PGDATABASE})
    if os.getenv("PGPASSWORD"):
        env["PGPASSWORD"] = os.environ["PGPASSWORD"]
    return env


def run_psql(sql: str, timeout: int = 20) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        ["psql", "-X", "-q", "-t", "-A", "-v", "ON_ERROR_STOP=1", "-c", sql],
        text=True,
        capture_output=True,
        timeout=timeout,
        env=pg_env(),
    )


def pg_scalar(sql: str, timeout: int = 20) -> str:
    result = run_psql(sql, timeout=timeout)
    if result.returncode != 0:
        raise RuntimeError(result.stderr.strip() or result.stdout.strip())
    return result.stdout.strip()


def pg_json_rows(select_sql: str, timeout: int = 24) -> list[dict[str, Any]]:
    payload = pg_scalar(f"SELECT COALESCE(json_agg(row_to_json(q)), '[]'::json)::text FROM ({select_sql}) q", timeout=timeout)
    return json.loads(payload or "[]")


def sql_literal(value: Any) -> str:
    if value is None:
        return "NULL"
    return "'" + str(value).replace("'", "''") + "'"


def json_sql(value: Any) -> str:
    return f"{sql_literal(json.dumps(value, ensure_ascii=False, separators=(',', ':')))}::jsonb"


def normalize_ct(value: Any) -> str:
    return re.sub(r"\s+", "", str(value or "")).upper()


def parse_json_list(value: Any) -> list[Any]:
    if isinstance(value, list):
        return value
    if isinstance(value, str) and value:
        try:
            parsed = json.loads(value)
            return parsed if isinstance(parsed, list) else []
        except Exception:
            return []
    return []


def ensure_registration_table() -> None:
    pg_scalar(
        f"""
        CREATE TABLE IF NOT EXISTS public.{REGISTRATION_TABLE_SQL} (
            id bigserial PRIMARY KEY,
            ct_number text NOT NULL,
            algorithm_model text NOT NULL DEFAULT '未指定模型',
            registration_status text NOT NULL DEFAULT 'unregistered',
            series_instance_uid text NOT NULL DEFAULT '',
            series_description text NOT NULL DEFAULT '',
            selected_stl_files jsonb NOT NULL DEFAULT '[]'::jsonb,
            transform jsonb NOT NULL DEFAULT '{{}}'::jsonb,
            module_styles jsonb NOT NULL DEFAULT '{{}}'::jsonb,
            dicom_reference jsonb NOT NULL DEFAULT '{{}}'::jsonb,
            model_reference jsonb NOT NULL DEFAULT '{{}}'::jsonb,
            notes text NOT NULL DEFAULT '',
            updated_by text NOT NULL DEFAULT 'admin',
            created_at timestamptz NOT NULL DEFAULT now(),
            updated_at timestamptz NOT NULL DEFAULT now(),
            UNIQUE (ct_number, algorithm_model)
        );
        ALTER TABLE public.{REGISTRATION_TABLE_SQL}
            ADD COLUMN IF NOT EXISTS algorithm_model text NOT NULL DEFAULT '未指定模型',
            ADD COLUMN IF NOT EXISTS registration_status text NOT NULL DEFAULT 'unregistered',
            ADD COLUMN IF NOT EXISTS series_instance_uid text NOT NULL DEFAULT '',
            ADD COLUMN IF NOT EXISTS series_description text NOT NULL DEFAULT '',
            ADD COLUMN IF NOT EXISTS selected_stl_files jsonb NOT NULL DEFAULT '[]'::jsonb,
            ADD COLUMN IF NOT EXISTS transform jsonb NOT NULL DEFAULT '{{}}'::jsonb,
            ADD COLUMN IF NOT EXISTS module_styles jsonb NOT NULL DEFAULT '{{}}'::jsonb,
            ADD COLUMN IF NOT EXISTS dicom_reference jsonb NOT NULL DEFAULT '{{}}'::jsonb,
            ADD COLUMN IF NOT EXISTS model_reference jsonb NOT NULL DEFAULT '{{}}'::jsonb,
            ADD COLUMN IF NOT EXISTS notes text NOT NULL DEFAULT '',
            ADD COLUMN IF NOT EXISTS locked boolean NOT NULL DEFAULT false,
            ADD COLUMN IF NOT EXISTS updated_by text NOT NULL DEFAULT 'admin',
            ADD COLUMN IF NOT EXISTS created_at timestamptz NOT NULL DEFAULT now(),
            ADD COLUMN IF NOT EXISTS updated_at timestamptz NOT NULL DEFAULT now();
        UPDATE public.{REGISTRATION_TABLE_SQL}
        SET
            ct_number = upper(regexp_replace(COALESCE(ct_number, ''), '\\s+', '', 'g')),
            algorithm_model = COALESCE(NULLIF(btrim(algorithm_model), ''), '未指定模型'),
            registration_status = CASE
                WHEN registration_status IN ('registered', 'unregistered') THEN registration_status
                WHEN COALESCE(locked, false) IS TRUE THEN 'registered'
                ELSE 'unregistered'
            END
        WHERE ct_number <> upper(regexp_replace(COALESCE(ct_number, ''), '\\s+', '', 'g'))
           OR algorithm_model = ''
           OR registration_status NOT IN ('registered', 'unregistered');
        CREATE TABLE IF NOT EXISTS public.{REGISTRATION_TABLE_SQL}_duplicate_archive (
            archived_at timestamptz NOT NULL DEFAULT now(),
            reason text NOT NULL,
            row_data jsonb NOT NULL
        );
        WITH ranked AS (
            SELECT id,
                   row_number() OVER (
                       PARTITION BY ct_number, algorithm_model
                       ORDER BY registration_status = 'registered' DESC, updated_at DESC NULLS LAST, id DESC
                   ) AS rn
            FROM public.{REGISTRATION_TABLE_SQL}
        ),
        duplicates AS (
            SELECT t.*
            FROM public.{REGISTRATION_TABLE_SQL} t
            JOIN ranked r ON r.id = t.id
            WHERE r.rn > 1
        )
        INSERT INTO public.{REGISTRATION_TABLE_SQL}_duplicate_archive(reason, row_data)
        SELECT 'ct_algorithm_unique_migration', row_to_json(duplicates)::jsonb
        FROM duplicates;
        WITH ranked AS (
            SELECT id,
                   row_number() OVER (
                       PARTITION BY ct_number, algorithm_model
                       ORDER BY registration_status = 'registered' DESC, updated_at DESC NULLS LAST, id DESC
                   ) AS rn
            FROM public.{REGISTRATION_TABLE_SQL}
        )
        DELETE FROM public.{REGISTRATION_TABLE_SQL} t
        USING ranked r
        WHERE t.id = r.id AND r.rn > 1;
        DO $$
        DECLARE
            item record;
        BEGIN
            FOR item IN
                SELECT conname
                FROM pg_constraint
                WHERE conrelid = 'public.{REGISTRATION_TABLE_SQL}'::regclass
                  AND contype = 'u'
                  AND pg_get_constraintdef(oid) NOT LIKE '%(ct_number, algorithm_model)%'
            LOOP
                EXECUTE format('ALTER TABLE public.{REGISTRATION_TABLE_SQL} DROP CONSTRAINT %I', item.conname);
            END LOOP;
            IF NOT EXISTS (
                SELECT 1
                FROM pg_constraint
                WHERE conrelid = 'public.{REGISTRATION_TABLE_SQL}'::regclass
                  AND contype = 'u'
                  AND conname = '{REGISTRATION_TABLE_SQL}_ct_algorithm_key'
            ) THEN
                ALTER TABLE public.{REGISTRATION_TABLE_SQL}
                    ADD CONSTRAINT {REGISTRATION_TABLE_SQL}_ct_algorithm_key UNIQUE (ct_number, algorithm_model);
            END IF;
        END $$;
        CREATE INDEX IF NOT EXISTS idx_{REGISTRATION_TABLE_SQL}_ct ON public.{REGISTRATION_TABLE_SQL}(ct_number);
        CREATE INDEX IF NOT EXISTS idx_{REGISTRATION_TABLE_SQL}_status ON public.{REGISTRATION_TABLE_SQL}(registration_status);
        CREATE INDEX IF NOT EXISTS idx_{REGISTRATION_TABLE_SQL}_algorithm ON public.{REGISTRATION_TABLE_SQL}(algorithm_model);
        """,
        timeout=14,
    )


def db_available() -> tuple[bool, str]:
    if not os.getenv("PGPASSWORD"):
        return False, "PGPASSWORD 未设置"
    try:
        pg_scalar("SELECT 1", timeout=4)
        return True, "connected"
    except Exception as exc:
        return False, str(exc)


def require_auth(authorization: str | None = Header(default=None), access_token: str = Query(default="")) -> str:
    token = access_token.strip()
    if not token and authorization and authorization.startswith("Bearer "):
        token = authorization.removeprefix("Bearer ").strip()
    user = TOKENS.get(token)
    if not user:
        raise HTTPException(status_code=401, detail="unauthorized")
    return user


@app.on_event("startup")
def startup() -> None:
    try:
        ensure_registration_table()
    except Exception:
        pass


@app.get("/")
def index() -> FileResponse:
    return FileResponse(STATIC_DIR / "index.html")


@app.get("/health")
def health() -> str:
    return "ok"


@app.post("/api/auth/login")
def login(payload: LoginPayload) -> dict[str, str]:
    username = payload.username.strip()
    if username != WEB_USER or payload.password != WEB_PASSWORD:
        raise HTTPException(status_code=401, detail="用户名或密码错误")
    token = secrets.token_urlsafe(32)
    TOKENS[token] = username
    return {"token": token, "username": username, "role": "管理员"}


@app.get("/api/auth/me")
def me(user: str = Depends(require_auth)) -> dict[str, str]:
    return {"username": user, "role": "管理员"}


@app.get("/api/status")
def status(_: str = Depends(require_auth)) -> dict[str, Any]:
    ok, message = db_available()
    counts = {"complete_cases": 0, "registered": 0, "unregistered": 0}
    if ok:
        ensure_registration_table()
        rows = pg_json_rows(
            f"""
            WITH complete AS (
                SELECT DISTINCT p.ct_number, COALESCE(NULLIF(u.algorithm_model, ''), '未指定模型') AS algorithm_model
                FROM public.{PACS_TABLE_SQL} p
                JOIN public.{UPP_ASSET_TABLE_SQL} u ON upper(u.ct_number) = upper(p.ct_number)
                LEFT JOIN public.{UPP_STL_TABLE_SQL} s ON upper(s.ct_number) = upper(p.ct_number)
                WHERE COALESCE(u.stl_present, false) OR COALESCE(u.stl_file_count, 0) > 0 OR s.ct_number IS NOT NULL
            )
            SELECT
                count(*)::int AS complete_cases,
                count(*) FILTER (WHERE r.registration_status = 'registered')::int AS registered,
                count(*) FILTER (WHERE COALESCE(r.registration_status, 'unregistered') <> 'registered')::int AS unregistered
            FROM complete c
            LEFT JOIN public.{REGISTRATION_TABLE_SQL} r
              ON r.ct_number = c.ct_number AND r.algorithm_model = c.algorithm_model
            """,
            timeout=10,
        )
        if rows:
            counts = rows[0]
    return {
        "database": {"ok": ok, "message": message, "host": PGHOST, "database": PGDATABASE},
        "counts": counts,
        "links": {"pacs_viewer_url": PACS_VIEWER_URL, "relation_viewer_url": RELATION_VIEWER_URL},
        "server_time": time.strftime("%Y-%m-%d %H:%M:%S"),
    }


def annotation_labels_sql() -> str:
    return f"""
        SELECT
            a.ct_number,
            COALESCE(jsonb_agg(DISTINCT part.value) FILTER (WHERE part.value IS NOT NULL), '[]'::jsonb) AS body_part_keys,
            COALESCE(jsonb_agg(DISTINCT
                CASE part.value
                    WHEN 'head_neck' THEN '头颈部'
                    WHEN 'chest' THEN '胸部'
                    WHEN 'upper_abdomen' THEN '上腹部'
                    WHEN 'lower_abdomen' THEN '下腹部'
                    WHEN 'pelvis' THEN '盆腔'
                    ELSE NULL
                END
            ) FILTER (WHERE part.value IS NOT NULL), '[]'::jsonb) AS body_part_labels
        FROM public.{PACS_ANNOTATION_TABLE_SQL} a
        LEFT JOIN LATERAL jsonb_array_elements_text(COALESCE(a.body_parts, '[]'::jsonb)) AS part(value) ON true
        WHERE COALESCE(a.skipped, false) IS NOT TRUE
        GROUP BY a.ct_number
    """


def relation_select(where_sql: str = "") -> str:
    return f"""
        WITH ann AS ({annotation_labels_sql()}),
        complete AS (
            SELECT
                upper(p.ct_number) AS ct_key,
                p.ct_number,
                COALESCE(NULLIF(u.algorithm_model, ''), '未指定模型') AS algorithm_model,
                p.batch_name,
                COALESCE(NULLIF(p.source_patient_name, ''), NULLIF(p.patient_name_dicom, '')) AS patient_name,
                p.patient_id,
                p.study_date,
                p.study_time,
                p.study_description,
                p.series_count,
                p.dicom_file_count,
                p.processed_path,
                u.patient_name AS upp_patient_name,
                u.upp_status,
                COALESCE(u.processed_stl_dir, '') AS processed_stl_dir,
                COALESCE(u.stl_file_count, s.file_count, 0)::int AS stl_file_count,
                COALESCE(u.stl_total_bytes, s.total_bytes, 0)::bigint AS stl_total_bytes,
                COALESCE(ann.body_part_keys, ps.body_parts, '[]'::jsonb) AS body_part_keys,
                COALESCE(ann.body_part_labels, '[]'::jsonb) AS body_part_labels
            FROM public.{PACS_TABLE_SQL} p
            JOIN public.{UPP_ASSET_TABLE_SQL} u ON upper(u.ct_number) = upper(p.ct_number)
            LEFT JOIN public.{UPP_STL_TABLE_SQL} s ON upper(s.ct_number) = upper(p.ct_number)
            LEFT JOIN public.{PACS_SUMMARY_TABLE_SQL} ps ON ps.ct_number = p.ct_number
            LEFT JOIN ann ON ann.ct_number = p.ct_number
            WHERE COALESCE(u.stl_present, false) OR COALESCE(u.stl_file_count, 0) > 0 OR s.ct_number IS NOT NULL
        )
        SELECT
            c.*,
            COALESCE(r.registration_status, 'unregistered') AS registration_status,
            r.series_instance_uid,
            r.series_description,
            r.transform,
            r.selected_stl_files,
            r.notes,
            r.updated_at AS registration_updated_at
        FROM complete c
        LEFT JOIN public.{REGISTRATION_TABLE_SQL} r
          ON r.ct_number = c.ct_number AND r.algorithm_model = c.algorithm_model
        {where_sql}
    """


@app.get("/api/cases")
def cases(
    q: str = "",
    status_filter: str = Query(default="", alias="status"),
    body_part: str = "",
    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)}")
    where_sql = "WHERE " + " AND ".join(clauses) if clauses else ""
    return pg_json_rows(
        f"""
        SELECT *
        FROM ({relation_select()}) relation
        {where_sql}
        ORDER BY
            registration_status = 'registered',
            COALESCE(study_date, '') DESC,
            COALESCE(study_time, '') DESC,
            ct_key
        LIMIT {int(limit)}
        """,
        timeout=20,
    )


@app.get("/api/cases/{ct_number}")
def case_detail(ct_number: str, algorithm_model: str = "未指定模型", _: str = Depends(require_auth)) -> dict[str, Any]:
    rows = pg_json_rows(
        f"""
        SELECT *
        FROM ({relation_select()}) relation
        WHERE ct_key = {sql_literal(normalize_ct(ct_number))}
          AND algorithm_model = {sql_literal(algorithm_model or '未指定模型')}
        LIMIT 1
        """,
        timeout=16,
    )
    if not rows:
        raise HTTPException(status_code=404, detail="未找到完整关联 CT")
    return rows[0]


def numeric(value: Any, fallback: float = 0.0) -> float:
    try:
        return float(str(value).strip().split("\\")[0])
    except Exception:
        return fallback


def text(value: Any) -> str:
    return "" if value is None else str(value).strip()


def read_header(path: Path) -> dict[str, str]:
    ds = pydicom.dcmread(str(path), stop_before_pixels=True, force=True, specific_tags=DICOM_TAGS + ["SpecificCharacterSet"])
    return {tag: text(getattr(ds, tag, "")) for tag in DICOM_TAGS}


def sort_key(item: tuple[Path, dict[str, str]]) -> tuple[float, float, str]:
    path, meta = item
    z = numeric(meta.get("SliceLocation"), 0.0)
    position = meta.get("ImagePositionPatient", "")
    if position and not meta.get("SliceLocation"):
        try:
            z = float(str(position).strip("[]").split(",")[-1])
        except Exception:
            z = 0.0
    return (z, numeric(meta.get("InstanceNumber"), 0.0), str(path))


def get_study_record(ct_number: str) -> dict[str, Any]:
    rows = pg_json_rows(
        f"""
        SELECT *
        FROM public.{PACS_TABLE_SQL}
        WHERE upper(ct_number) = {sql_literal(normalize_ct(ct_number))}
        LIMIT 1
        """,
        timeout=12,
    )
    if not rows:
        raise HTTPException(status_code=404, detail="DICOM 检查不存在")
    return rows[0]


def resolve_study_root(study: dict[str, Any]) -> Path:
    root = Path(study.get("processed_path") or "")
    if root.exists():
        return root
    target_folder = str(study.get("target_folder_name") or "")
    if target_folder and PROCESSED_ROOT.exists():
        found = next(PROCESSED_ROOT.rglob(target_folder), None)
        if found:
            return found
    return root


def get_annotations(ct_number: str) -> dict[str, dict[str, Any]]:
    try:
        rows = pg_json_rows(
            f"""
            SELECT *
            FROM public.{PACS_ANNOTATION_TABLE_SQL}
            WHERE upper(ct_number) = {sql_literal(normalize_ct(ct_number))}
            """,
            timeout=12,
        )
    except Exception:
        return {}
    return {str(row.get("series_instance_uid") or ""): row for row in rows}


def annotation_labels(row: dict[str, Any]) -> list[str]:
    if not row:
        return []
    if row.get("skipped"):
        return ["略过/不采用"]
    labels = []
    for part in parse_json_list(row.get("body_parts")):
        if part == "head_neck":
            labels.append("头颈部")
        elif part == "chest":
            labels.append("胸部")
        elif part == "upper_abdomen":
            phase = {"plain": "平扫", "arterial": "动脉期", "portal_venous": "门脉期", "delayed": "延迟期", "unknown": "无法判别"}.get(row.get("upper_abdomen_phase") or "unknown", "无法判别")
            labels.append(f"上腹部-{phase}")
        elif part == "lower_abdomen":
            labels.append("下腹部")
        elif part == "pelvis":
            labels.append("盆腔")
    return labels


def scan_study(ct_number: str) -> dict[str, Any]:
    key = normalize_ct(ct_number)
    cached = STUDY_CACHE.get(key)
    if cached and time.time() - cached["cached_at"] < 600:
        return cached
    study = get_study_record(ct_number)
    root = resolve_study_root(study)
    if not root.exists():
        raise HTTPException(status_code=404, detail=f"DICOM 目录不存在：{root}")
    grouped: dict[str, list[tuple[Path, dict[str, str]]]] = defaultdict(list)
    for path in root.rglob("*.dcm"):
        try:
            meta = read_header(path)
        except Exception:
            continue
        uid = meta.get("SeriesInstanceUID") or path.parent.name
        grouped[uid].append((path, meta))
    annotations = get_annotations(ct_number)
    series = []
    file_map = {}
    for uid, items in grouped.items():
        items.sort(key=sort_key)
        first = items[0][1]
        last = items[-1][1]
        file_map[uid] = [path for path, _ in items]
        annotation = annotations.get(uid) or {}
        series.append(
            {
                "series_uid": uid,
                "description": first.get("SeriesDescription") or "未命名序列",
                "series_number": first.get("SeriesNumber") or "",
                "count": len(items),
                "modality": first.get("Modality") or "",
                "rows": first.get("Rows") or "",
                "columns": first.get("Columns") or "",
                "body_part_dicom": first.get("BodyPartExamined") or "",
                "series_time": first.get("SeriesTime") or first.get("AcquisitionTime") or first.get("ContentTime") or "",
                "first_time": first.get("AcquisitionTime") or first.get("ContentTime") or "",
                "last_time": last.get("AcquisitionTime") or last.get("ContentTime") or "",
                "pixel_spacing": first.get("PixelSpacing") or "",
                "slice_thickness": first.get("SliceThickness") or "",
                "spacing_between_slices": first.get("SpacingBetweenSlices") or "",
                "annotation_labels": annotation_labels(annotation),
            }
        )
    series.sort(key=lambda item: (str(item.get("first_time") or item.get("series_time") or ""), numeric(item.get("series_number"), 999999), item.get("description") or ""))
    payload = {"cached_at": time.time(), "study": study, "root": str(root), "series": series, "files": file_map}
    STUDY_CACHE[key] = payload
    return payload


@app.get("/api/cases/{ct_number}/series")
def series(ct_number: str, _: str = Depends(require_auth)) -> dict[str, Any]:
    data = scan_study(ct_number)
    return {"study": data["study"], "root": data["root"], "series": data["series"]}


def get_series_files(ct_number: str, series_uid: str) -> list[Path]:
    data = scan_study(ct_number)
    files = data["files"].get(series_uid)
    if not files:
        raise HTTPException(status_code=404, detail="DICOM 序列不存在")
    return files


def window_values(ds: pydicom.Dataset, preset: str) -> tuple[float, float]:
    if preset in WINDOWS and WINDOWS[preset]:
        return WINDOWS[preset]  # type: ignore[return-value]
    center = getattr(ds, "WindowCenter", 40)
    width = getattr(ds, "WindowWidth", 400)
    if isinstance(center, pydicom.multival.MultiValue):
        center = center[0]
    if isinstance(width, pydicom.multival.MultiValue):
        width = width[0]
    return numeric(center, 40.0), numeric(width, 400.0)


def pixel_spacing_from_ds(ds: pydicom.Dataset) -> tuple[float, float]:
    spacing = getattr(ds, "PixelSpacing", None)
    if spacing and len(spacing) >= 2:
        return max(numeric(spacing[0], 1.0), 0.001), max(numeric(spacing[1], 1.0), 0.001)
    return 1.0, 1.0


def dicom_to_hu(ds: pydicom.Dataset) -> np.ndarray:
    arr = ds.pixel_array.astype(np.float32)
    slope = float(getattr(ds, "RescaleSlope", 1) or 1)
    intercept = float(getattr(ds, "RescaleIntercept", 0) or 0)
    return arr * slope + intercept


def slice_spacing_from_datasets(datasets: list[pydicom.Dataset]) -> float:
    positions = []
    for ds in datasets:
        position = getattr(ds, "ImagePositionPatient", None)
        if position and len(position) >= 3:
            positions.append(np.array([numeric(position[0]), numeric(position[1]), numeric(position[2])], dtype=np.float32))
    distances = [float(np.linalg.norm(b - a)) for a, b in zip(positions, positions[1:])]
    distances = [item for item in distances if item > 0.001]
    if distances:
        return max(float(np.median(distances)), 0.001)
    if datasets:
        spacing = numeric(getattr(datasets[0], "SpacingBetweenSlices", 0), 0.0)
        if spacing > 0.001:
            return spacing
        thickness = numeric(getattr(datasets[0], "SliceThickness", 0), 0.0)
        if thickness > 0.001:
            return thickness
    return 1.0


def resize_for_spacing(pil: Image.Image, row_spacing: float, col_spacing: float) -> Image.Image:
    if abs(row_spacing - col_spacing) < 0.01:
        return pil
    unit = min(row_spacing, col_spacing)
    width = max(1, int(round(pil.width * col_spacing / unit)))
    height = max(1, int(round(pil.height * row_spacing / unit)))
    scale = min(1.0, 1800 / max(width, height))
    target = (max(1, int(round(width * scale))), max(1, int(round(height * scale))))
    return pil if target == pil.size else pil.resize(target, Image.Resampling.BILINEAR)


def render_array(arr: np.ndarray, center: float, width: float, max_size: int = 900, spacing: tuple[float, float] = (1.0, 1.0), invert: bool = False) -> bytes:
    low = center - width / 2.0
    high = center + width / 2.0
    img = ((np.clip(arr, low, high) - low) / max(high - low, 1.0) * 255.0).astype(np.uint8)
    if invert:
        img = 255 - img
    pil = Image.fromarray(img)
    pil = resize_for_spacing(pil, spacing[0], spacing[1])
    if max(pil.size) > max_size:
        pil.thumbnail((max_size, max_size), Image.Resampling.BILINEAR)
    output = io.BytesIO()
    pil.save(output, format="PNG", optimize=True)
    return output.getvalue()


def load_stack_data(ct_number: str, series_uid: str) -> dict[str, Any]:
    key = f"{normalize_ct(ct_number)}|{series_uid}"
    cached = STACK_CACHE.get(key)
    if cached:
        STACK_CACHE[key] = (time.time(), cached[1])
        return cached[1]
    files = get_series_files(ct_number, series_uid)
    arrays = []
    datasets = []
    for path in files:
        ds = pydicom.dcmread(str(path), force=True)
        datasets.append(ds)
        arrays.append(dicom_to_hu(ds))
    stack = np.stack(arrays, axis=0)
    row_spacing, col_spacing = pixel_spacing_from_ds(datasets[min(len(datasets) - 1, len(datasets) // 2)])
    payload = {"stack": stack, "datasets": datasets, "row_spacing": row_spacing, "col_spacing": col_spacing, "slice_spacing": slice_spacing_from_datasets(datasets)}
    STACK_CACHE[key] = (time.time(), payload)
    if len(STACK_CACHE) > 2:
        oldest = sorted(STACK_CACHE.items(), key=lambda item: item[1][0])[0][0]
        STACK_CACHE.pop(oldest, None)
    return payload


@app.get("/api/dicom/image")
def dicom_image(
    ct_number: str,
    series_uid: str,
    index: int = 0,
    window: str = "default",
    _: str = Depends(require_auth),
) -> Response:
    files = get_series_files(ct_number, series_uid)
    index = min(max(0, index), len(files) - 1)
    ds = pydicom.dcmread(str(files[index]), force=True)
    center, width = window_values(ds, window)
    payload = render_array(dicom_to_hu(ds), center, width, spacing=pixel_spacing_from_ds(ds), invert=getattr(ds, "PhotometricInterpretation", "") == "MONOCHROME1")
    return Response(payload, media_type="image/png")


@app.get("/api/dicom/fusion-volume")
def dicom_fusion_volume(
    ct_number: str,
    series_uid: str,
    center_index: int = 0,
    window: str = "soft",
    radius: int = Query(default=96, ge=4, le=512),
    range_start: int | None = None,
    range_end: int | None = None,
    _: str = Depends(require_auth),
) -> dict[str, Any]:
    stack_data = load_stack_data(ct_number, series_uid)
    stack = stack_data["stack"]
    datasets = stack_data["datasets"]
    total = int(stack.shape[0])
    center_index = min(max(0, center_index), total - 1)
    if range_start is not None and range_end is not None:
        start = min(max(0, int(range_start)), total - 1)
        end = min(max(0, int(range_end)), total - 1)
        if start > end:
            start, end = end, start
    else:
        start = max(0, center_index - radius)
        end = min(total - 1, center_index + radius)
    max_frames = 72
    if end - start + 1 > max_frames:
        step = int(np.ceil((end - start + 1) / max_frames))
        indices = list(range(start, end + 1, step))
    else:
        indices = list(range(start, end + 1))
    sample_ds = datasets[center_index]
    center, width = window_values(sample_ds, window)
    frames = []
    frame_width = 0
    frame_height = 0
    for index in indices:
        png = render_array(stack[index], center, width, max_size=256, spacing=(stack_data["row_spacing"], stack_data["col_spacing"]))
        frames.append("data:image/png;base64," + base64.b64encode(png).decode("ascii"))
        if not frame_width:
            with Image.open(io.BytesIO(png)) as pil:
                frame_width, frame_height = pil.size
    return {
        "frames": frames,
        "indices": indices,
        "start": start,
        "end": end,
        "center": center_index,
        "total": total,
        "width": frame_width,
        "height": frame_height,
        "spacing": {"row": stack_data["row_spacing"], "column": stack_data["col_spacing"], "slice": stack_data["slice_spacing"]},
        "physicalSize": {
            "width": int(stack.shape[2]) * stack_data["col_spacing"],
            "height": int(stack.shape[1]) * stack_data["row_spacing"],
            "depth": total * stack_data["slice_spacing"],
            "unit": "mm",
        },
    }


def stl_rows_for_ct(ct_number: str, algorithm_model: str = "") -> list[dict[str, Any]]:
    rows = pg_json_rows(
        f"""
        SELECT
            COALESCE(NULLIF(u.algorithm_model, ''), '未指定模型') AS algorithm_model,
            u.processed_stl_dir,
            CASE
                WHEN jsonb_typeof(u.stl_files) = 'array' AND jsonb_array_length(u.stl_files) > 0 THEN u.stl_files
                ELSE COALESCE(s.files, '[]'::jsonb)
            END AS files
        FROM public.{UPP_ASSET_TABLE_SQL} u
        LEFT JOIN public.{UPP_STL_TABLE_SQL} s ON upper(s.ct_number) = upper(u.ct_number)
        WHERE upper(u.ct_number) = {sql_literal(normalize_ct(ct_number))}
          AND ({sql_literal(algorithm_model)} = '' OR COALESCE(NULLIF(u.algorithm_model, ''), '未指定模型') = {sql_literal(algorithm_model or '未指定模型')})
        ORDER BY COALESCE(NULLIF(u.algorithm_model, ''), '未指定模型'), u.updated_at DESC NULLS LAST
        """,
        timeout=14,
    )
    output: list[dict[str, Any]] = []
    index = 0
    for asset_index, row in enumerate(rows):
        files = parse_json_list(row.get("files"))
        if not files and row.get("processed_stl_dir"):
            base = Path(str(row["processed_stl_dir"]))
            if base.exists():
                files = [
                    {
                        "file_name": path.name,
                        "segment_name": path.stem,
                        "family": path.stem,
                        "category": "未分类",
                        "processed_file_path": str(path),
                        "size_bytes": path.stat().st_size,
                    }
                    for path in sorted(base.glob("*.stl"))
                ]
        for file_info in files:
            path = file_info.get("processed_file_path") or file_info.get("source_file_path")
            if not path:
                continue
            output.append(
                {
                    "id": index,
                    "asset_id": asset_index,
                    "algorithm_model": row.get("algorithm_model") or "未指定模型",
                    "file_name": file_info.get("file_name") or Path(path).name,
                    "segment_name": file_info.get("segment_name") or Path(path).stem,
                    "family": file_info.get("family") or file_info.get("segment_name") or Path(path).stem,
                    "category": file_info.get("category") or "未分类",
                    "size_bytes": int(file_info.get("size_bytes") or 0),
                    "path": path,
                }
            )
            index += 1
    return output


@app.get("/api/cases/{ct_number}/stl")
def stl_assets(ct_number: str, algorithm_model: str = "", _: str = Depends(require_auth)) -> dict[str, Any]:
    files = stl_rows_for_ct(ct_number, algorithm_model)
    return {
        "ct_number": ct_number,
        "files": files,
        "families": sorted({str(item.get("family") or "") for item in files if item.get("family")}),
        "algorithm_models": sorted({str(item.get("algorithm_model") or "") for item in files if item.get("algorithm_model")}),
    }


@app.get("/api/stl/file")
def stl_file(ct_number: str, file_id: int, algorithm_model: str = "", _: str = Depends(require_auth)) -> FileResponse:
    files = stl_rows_for_ct(ct_number, algorithm_model)
    selected = next((item for item in files if int(item["id"]) == int(file_id)), None)
    if not selected:
        raise HTTPException(status_code=404, detail="STL 文件不存在")
    path = Path(str(selected["path"]))
    if not path.exists() or not path.is_file():
        raise HTTPException(status_code=404, detail=f"STL 路径不存在：{path}")
    return FileResponse(path, media_type="model/stl", filename=path.name)


@app.get("/api/registrations/{ct_number}")
def get_registration(ct_number: str, algorithm_model: str = "未指定模型", _: str = Depends(require_auth)) -> dict[str, Any]:
    ensure_registration_table()
    rows = pg_json_rows(
        f"""
        SELECT *
        FROM public.{REGISTRATION_TABLE_SQL}
        WHERE ct_number = {sql_literal(normalize_ct(ct_number))}
          AND algorithm_model = {sql_literal(algorithm_model or '未指定模型')}
        LIMIT 1
        """,
        timeout=12,
    )
    return rows[0] if rows else {
        "ct_number": normalize_ct(ct_number),
        "algorithm_model": algorithm_model or "未指定模型",
        "registration_status": "unregistered",
        "transform": DEFAULT_POSE,
        "selected_stl_files": [],
        "notes": "",
    }


@app.post("/api/registrations")
def save_registration(payload: RegistrationPayload, user: str = Depends(require_auth)) -> dict[str, Any]:
    ensure_registration_table()
    ct_number = normalize_ct(payload.ct_number)
    if not ct_number:
        raise HTTPException(status_code=400, detail="CT号不能为空")
    algorithm_model = payload.algorithm_model.strip() or "未指定模型"
    registration_status = payload.registration_status if payload.registration_status in {"registered", "unregistered"} else "unregistered"
    transform = {**DEFAULT_POSE, **(payload.transform or {})}
    pg_scalar(
        f"""
        INSERT INTO public.{REGISTRATION_TABLE_SQL} (
            ct_number, algorithm_model, registration_status, series_instance_uid, series_description,
            selected_stl_files, transform, module_styles, dicom_reference, model_reference,
            notes, updated_by, updated_at
        )
        VALUES (
            {sql_literal(ct_number)},
            {sql_literal(algorithm_model)},
            {sql_literal(registration_status)},
            {sql_literal(payload.series_instance_uid.strip())},
            {sql_literal(payload.series_description.strip())},
            {json_sql(payload.selected_stl_files)},
            {json_sql(transform)},
            {json_sql(payload.module_styles)},
            {json_sql(payload.dicom_reference)},
            {json_sql(payload.model_reference)},
            {sql_literal(payload.notes.strip())},
            {sql_literal(user)},
            now()
        )
        ON CONFLICT (ct_number, algorithm_model) DO UPDATE SET
            registration_status = EXCLUDED.registration_status,
            series_instance_uid = EXCLUDED.series_instance_uid,
            series_description = EXCLUDED.series_description,
            selected_stl_files = EXCLUDED.selected_stl_files,
            transform = EXCLUDED.transform,
            module_styles = EXCLUDED.module_styles,
            dicom_reference = EXCLUDED.dicom_reference,
            model_reference = EXCLUDED.model_reference,
            notes = EXCLUDED.notes,
            updated_by = EXCLUDED.updated_by,
            updated_at = now()
        """,
        timeout=14,
    )
    return {"ok": True, "ct_number": ct_number, "algorithm_model": algorithm_model, "registration_status": registration_status}