from __future__ import annotations

import json
import hashlib
import os
import re
import secrets
import subprocess
import unicodedata
from itertools import product
from pathlib import Path
from typing import Any

from fastapi import Depends, FastAPI, Header, HTTPException, Query
from pydantic import BaseModel
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles

try:
    from pypinyin import Style, pinyin
except ImportError:  # pragma: no cover - Docker image installs this dependency.
    Style = None
    pinyin = None


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

PGHOST = os.getenv("PGHOST", "192.168.3.3")
PGPORT = os.getenv("PGPORT", "5432")
PGUSER = os.getenv("PGUSER", "his_user")
PGDATABASE = os.getenv("PGDATABASE", "pacs_db")
PACS_TABLE = os.getenv("PACS_TABLE", "pacs_dicom_files")
PACS_SUMMARY_TABLE = os.getenv("PACS_SUMMARY_TABLE", "pacs_dicom_study_summaries")
PACS_ANNOTATION_TABLE = os.getenv("PACS_ANNOTATION_TABLE", "pacs_dicom_series_annotations")
UPP_ASSET_TABLE = os.getenv("UPP_ASSET_TABLE", "upp_exam_assets")
UPP_STL_TABLE = os.getenv("UPP_STL_TABLE", "upp_stl_files")
REGISTRATION_TABLE = os.getenv("REGISTRATION_TABLE", "dicom_upp_registrations")
USER_TABLE = os.getenv("VISUALIZER_USER_TABLE", "db_visualizer_users")
WEB_ADMIN_USER = os.getenv("VISUALIZER_ADMIN_USER", "admin")
WEB_ADMIN_PASSWORD = os.getenv("VISUALIZER_ADMIN_PASSWORD", "123456")
PACS_VIEWER_URL = os.getenv("PACS_VIEWER_URL", "http://127.0.0.1:8107")
REGISTRATION_URL = os.getenv("REGISTRATION_URL", "http://127.0.0.1:8109")

IDENT_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
CJK_RE = re.compile(r"[\u3400-\u9fff]")
LATIN_RE = re.compile(r"[a-z0-9]+")


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


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

app = FastAPI(title="DICOM UPP Database Visualizer")
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
TOKENS: dict[str, dict[str, str]] = {}


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


class UserPayload(BaseModel):
    username: str
    password: str = ""
    role: str = "阅片员"
    status: str = "启用"


class PasswordPayload(BaseModel):
    password: str


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


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


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


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


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


def password_hash(password: str) -> str:
    return hashlib.sha256(password.encode("utf-8")).hexdigest()


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


def normalize_latin_name(value: Any) -> str:
    text = unicodedata.normalize("NFKD", str(value or "").replace("ü", "u").replace("Ü", "U"))
    text = "".join(char for char in text if not unicodedata.combining(char)).lower()
    return "".join(LATIN_RE.findall(text))


def normalize_cjk_name(value: Any) -> str:
    return "".join(CJK_RE.findall(str(value or "")))


def pinyin_name_candidates(value: Any) -> set[str]:
    if pinyin is None or Style is None:
        return set()
    chinese = normalize_cjk_name(value)
    if not chinese:
        return set()
    syllables = pinyin(chinese, style=Style.NORMAL, heteronym=True, errors="ignore")
    choices: list[list[str]] = []
    for item in syllables[:6]:
        normalized = sorted({normalize_latin_name(part) for part in item if normalize_latin_name(part)})
        if normalized:
            choices.append(normalized[:4])
    if not choices:
        return set()
    return {"".join(parts) for parts in product(*choices)}


def edit_distance_with_cap(left: str, right: str, cap: int = 1) -> int:
    if abs(len(left) - len(right)) > cap:
        return cap + 1
    previous = list(range(len(right) + 1))
    for i, left_char in enumerate(left, 1):
        current = [i]
        row_min = current[0]
        for j, right_char in enumerate(right, 1):
            cost = 0 if left_char == right_char else 1
            value = min(previous[j] + 1, current[j - 1] + 1, previous[j - 1] + cost)
            current.append(value)
            row_min = min(row_min, value)
        if row_min > cap:
            return cap + 1
        previous = current
    return previous[-1]


def near_pinyin_match(left: str, candidates: set[str]) -> bool:
    if len(left) < 6:
        return False
    return any(len(candidate) >= 6 and edit_distance_with_cap(left, candidate, cap=1) <= 1 for candidate in candidates)


def match_patient_names(pacs_name: Any, upp_name: Any) -> tuple[str, str]:
    pacs_text = str(pacs_name or "")
    upp_text = str(upp_name or "")
    if not pacs_text or not upp_text:
        return "", ""

    pacs_latin = normalize_latin_name(pacs_text)
    upp_latin = normalize_latin_name(upp_text)
    pacs_cjk = normalize_cjk_name(pacs_text)
    upp_cjk = normalize_cjk_name(upp_text)

    if pacs_cjk and upp_cjk and pacs_cjk == upp_cjk:
        return "same", "中文姓名一致"
    if pacs_latin and upp_latin and pacs_latin == upp_latin:
        return "same", "拼音/拉丁姓名一致"
    if upp_latin and upp_latin in pinyin_name_candidates(pacs_text):
        return "same", "DICOM中文姓名与UPP拼音匹配"
    if upp_latin and near_pinyin_match(upp_latin, pinyin_name_candidates(pacs_text)):
        return "same", "DICOM中文姓名与UPP拼音近似匹配"
    if pacs_latin and pacs_latin in pinyin_name_candidates(upp_text):
        return "same", "UPP中文姓名与DICOM拼音匹配"
    if pacs_latin and near_pinyin_match(pacs_latin, pinyin_name_candidates(upp_text)):
        return "same", "UPP中文姓名与DICOM拼音近似匹配"
    return "different", f"DICOM患者：{pacs_text}；UPP患者：{upp_text}"


def enrich_patient_match(row: dict[str, Any]) -> dict[str, Any]:
    pacs_name = row.get("pacs_patient_name") or ""
    upp_names = [row.get("upp_patient_name"), *(row.get("upp_patient_names") or [])]
    upp_names = list(dict.fromkeys([str(name) for name in upp_names if name]))
    if not pacs_name or not upp_names:
        row["patient_name_match"] = ""
        row["patient_name_match_note"] = ""
        return row

    mismatch_notes = []
    for upp_name in upp_names:
        status, note = match_patient_names(pacs_name, upp_name)
        if status == "same":
            row["patient_name_match"] = "same"
            row["patient_name_match_note"] = note
            row["upp_patient_name"] = row.get("upp_patient_name") or upp_name
            return row
        if note:
            mismatch_notes.append(note)
    row["patient_name_match"] = "different"
    row["patient_name_match_note"] = "；".join(mismatch_notes[:3])
    return row


def db_available() -> tuple[bool, str]:
    try:
        pg_scalar("SELECT 1", timeout=4)
        return True, "connected"
    except Exception as exc:  # noqa: BLE001
        return False, str(exc)


def ensure_user_table() -> None:
    run_psql(
        f"""
        CREATE TABLE IF NOT EXISTS public.{USER_TABLE_SQL} (
            username text PRIMARY KEY,
            password_hash text NOT NULL,
            role text NOT NULL DEFAULT '阅片员',
            status text NOT NULL DEFAULT '启用',
            created_at timestamptz NOT NULL DEFAULT now(),
            updated_at timestamptz NOT NULL DEFAULT now()
        );
        INSERT INTO public.{USER_TABLE_SQL} (username, password_hash, role, status)
        VALUES ({sql_literal(WEB_ADMIN_USER)}, {sql_literal(password_hash(WEB_ADMIN_PASSWORD))}, '管理员', '启用')
        ON CONFLICT (username) DO NOTHING;
        """,
        timeout=10,
    )


def ensure_registration_table() -> None:
    run_psql(
        f"""
        CREATE TABLE IF NOT EXISTS public.{REGISTRATION_TABLE_SQL} (
            id bigserial PRIMARY KEY,
            ct_number text NOT NULL,
            series_instance_uid text NOT NULL DEFAULT '',
            series_description text NOT NULL DEFAULT '',
            algorithm_model text NOT NULL DEFAULT '',
            stl_family text NOT NULL DEFAULT '',
            view_mode text NOT NULL DEFAULT 'family',
            selected_stl_files jsonb NOT NULL DEFAULT '[]'::jsonb,
            transform jsonb NOT NULL DEFAULT '{{}}'::jsonb,
            dicom_reference jsonb NOT NULL DEFAULT '{{}}'::jsonb,
            model_reference jsonb NOT NULL DEFAULT '{{}}'::jsonb,
            segmentation jsonb NOT NULL DEFAULT '{{}}'::jsonb,
            notes text NOT NULL DEFAULT '',
            locked boolean NOT NULL DEFAULT false,
            locked_by text,
            locked_at timestamptz,
            updated_by text NOT NULL DEFAULT 'admin',
            created_at timestamptz NOT NULL DEFAULT now(),
            updated_at timestamptz NOT NULL DEFAULT now(),
            UNIQUE (ct_number, series_instance_uid, algorithm_model, stl_family)
        );
        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}_locked ON public.{REGISTRATION_TABLE_SQL}(locked);
        """,
        timeout=10,
    )


def fetch_user(username: str) -> dict[str, Any] | None:
    ensure_user_table()
    rows = pg_json_rows(
        f"""
        SELECT username, password_hash, role, status, created_at, updated_at
        FROM public.{USER_TABLE_SQL}
        WHERE username = {sql_literal(username)}
        LIMIT 1
        """,
        timeout=10,
    )
    return rows[0] if rows else None


def current_user(authorization: str | None = Header(default=None)) -> dict[str, str]:
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="unauthorized")
    token = authorization.removeprefix("Bearer ").strip()
    user = TOKENS.get(token)
    if not user:
        raise HTTPException(status_code=401, detail="unauthorized")
    return user


def admin_user(user: dict[str, str] = Depends(current_user)) -> dict[str, str]:
    if user.get("role") != "管理员":
        raise HTTPException(status_code=403, detail="forbidden")
    return user


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


def relation_cte() -> str:
    return f"""
        WITH
        p AS (
            SELECT
                p.*,
                upper(p.ct_number) AS ct_key
            FROM public.{PACS_TABLE_SQL} p
        ),
        ps AS (
            SELECT *
            FROM public.{PACS_SUMMARY_TABLE_SQL}
        ),
        a_label_raw AS (
            SELECT
                a.ct_number,
                CASE part.value
                    WHEN 'head_neck' THEN '头颈部'
                    WHEN 'chest' THEN '胸部-' || CASE COALESCE(NULLIF(a.chest_window, ''), 'unknown')
                        WHEN 'lung' THEN '肺窗'
                        WHEN 'mediastinal' THEN '纵隔窗'
                        ELSE '无法判别'
                    END
                    WHEN 'upper_abdomen' THEN '上腹部-' || CASE COALESCE(NULLIF(a.upper_abdomen_phase, ''), 'unknown')
                        WHEN 'plain' THEN '平扫'
                        WHEN 'arterial' THEN '动脉期'
                        WHEN 'portal_venous' THEN '门脉期'
                        WHEN 'delayed' THEN '延迟期'
                        ELSE '无法判别'
                    END
                    WHEN 'abdomen_pelvis' THEN '腹盆部'
                    WHEN 'lower_abdomen' THEN '腹盆部'
                    WHEN 'pelvis' THEN '腹盆部'
                    ELSE NULL
                END AS label,
                CASE part.value
                    WHEN 'head_neck' THEN 1
                    WHEN 'chest' THEN 2
                    WHEN 'upper_abdomen' THEN 3
                    WHEN 'abdomen_pelvis' THEN 4
                    WHEN 'lower_abdomen' THEN 4
                    WHEN 'pelvis' THEN 4
                    ELSE 99
                END AS sort_key
            FROM public.{PACS_ANNOTATION_TABLE_SQL} a
            CROSS JOIN LATERAL jsonb_array_elements_text(a.body_parts) AS part(value)
            WHERE COALESCE(a.skipped, false) IS NOT TRUE
        ),
        a_label_distinct AS (
            SELECT ct_number, label, min(sort_key) AS sort_key
            FROM a_label_raw
            WHERE label IS NOT NULL
            GROUP BY ct_number, label
        ),
        a_labels AS (
            SELECT
                ct_number,
                to_jsonb(array_agg(label ORDER BY sort_key, label)) AS dicom_annotation_labels
            FROM a_label_distinct
            GROUP BY ct_number
        ),
        u_raw AS (
            SELECT
                u.*,
                upper(u.ct_number) AS ct_key,
                COALESCE(NULLIF(u.algorithm_model, ''), '未指定模型') AS algorithm_model_key
            FROM public.{UPP_ASSET_TABLE_SQL} u
        ),
        u_model_ranked AS (
            SELECT
                u_raw.*,
                row_number() OVER (
                    PARTITION BY ct_key, algorithm_model_key
                    ORDER BY
                        COALESCE(stl_present, false) DESC,
                        COALESCE(stl_file_count, 0) DESC,
                        updated_at DESC NULLS LAST
                ) AS model_rank
            FROM u_raw
        ),
        u_model AS (
            SELECT *
            FROM u_model_ranked
            WHERE model_rank = 1
        ),
        u_duplicate AS (
            SELECT
                ct_key,
                COALESCE(sum(model_count - 1), 0)::int AS duplicate_model_asset_count
            FROM (
                SELECT ct_key, algorithm_model_key, count(*)::int AS model_count
                FROM u_raw
                GROUP BY ct_key, algorithm_model_key
                HAVING count(*) > 1
            ) duplicate_models
            GROUP BY ct_key
        ),
        u AS (
            SELECT
                u_model.ct_key,
                (array_agg(u_model.ct_number ORDER BY u_model.updated_at DESC NULLS LAST) FILTER (WHERE u_model.ct_number IS NOT NULL))[1] AS ct_number,
                bool_or(COALESCE(u_model.list_present, false)) AS list_present,
                bool_or(COALESCE(u_model.stl_present, false)) AS stl_present,
                (array_agg(u_model.patient_name ORDER BY u_model.updated_at DESC NULLS LAST) FILTER (WHERE NULLIF(u_model.patient_name, '') IS NOT NULL))[1] AS patient_name,
                to_jsonb(array_remove(array_agg(DISTINCT NULLIF(u_model.patient_name, '')), NULL)) AS patient_names,
                (array_agg(u_model.patient_sex ORDER BY u_model.updated_at DESC NULLS LAST) FILTER (WHERE NULLIF(u_model.patient_sex, '') IS NOT NULL))[1] AS patient_sex,
                (array_agg(u_model.patient_age ORDER BY u_model.updated_at DESC NULLS LAST) FILTER (WHERE NULLIF(u_model.patient_age, '') IS NOT NULL))[1] AS patient_age,
                (array_agg(u_model.patient_id_masked ORDER BY u_model.updated_at DESC NULLS LAST) FILTER (WHERE NULLIF(u_model.patient_id_masked, '') IS NOT NULL))[1] AS patient_id_masked,
                max(u_model.exam_date) AS exam_date,
                max(u_model.task_created_at) AS task_created_at,
                (array_agg(u_model.exam_description ORDER BY u_model.updated_at DESC NULLS LAST) FILTER (WHERE NULLIF(u_model.exam_description, '') IS NOT NULL))[1] AS exam_description,
                (array_agg(u_model.exam_device ORDER BY u_model.updated_at DESC NULLS LAST) FILTER (WHERE NULLIF(u_model.exam_device, '') IS NOT NULL))[1] AS exam_device,
                array_to_string(array_remove(array_agg(DISTINCT NULLIF(u_model.algorithm_model, '')), NULL), '、') AS algorithm_model,
                to_jsonb(array_remove(array_agg(DISTINCT NULLIF(u_model.algorithm_model, '')), NULL)) AS algorithm_models,
                array_to_string(array_remove(array_agg(DISTINCT NULLIF(u_model.upp_status, '')), NULL), '、') AS upp_status,
                COALESCE(sum(COALESCE(u_model.list_record_count, 0)), 0)::int AS list_record_count,
                (array_agg(u_model.processed_stl_dir ORDER BY u_model.updated_at DESC NULLS LAST) FILTER (WHERE NULLIF(u_model.processed_stl_dir, '') IS NOT NULL))[1] AS processed_stl_dir,
                COALESCE(sum(COALESCE(u_model.stl_file_count, 0)), 0)::int AS stl_file_count,
                COALESCE(sum(COALESCE(u_model.stl_total_bytes, 0)), 0)::bigint AS stl_total_bytes,
                max(u_model.updated_at) AS updated_at,
                count(*)::int AS upp_asset_count,
                COALESCE(max(u_duplicate.duplicate_model_asset_count), 0)::int AS duplicate_model_asset_count,
                jsonb_agg(
                    jsonb_build_object(
                        'ct_number', u_model.ct_number,
                        'algorithm_model', u_model.algorithm_model,
                        'upp_status', u_model.upp_status,
                        'file_count', u_model.stl_file_count,
                        'total_bytes', u_model.stl_total_bytes,
                        'processed_stl_dir', u_model.processed_stl_dir,
                        'updated_at', u_model.updated_at
                    )
                    ORDER BY u_model.updated_at DESC NULLS LAST
                ) AS upp_assets
            FROM u_model
            LEFT JOIN u_duplicate ON u_duplicate.ct_key = u_model.ct_key
            GROUP BY u_model.ct_key
        ),
        s_raw AS (
            SELECT
                s.*,
                upper(s.ct_number) AS ct_key
            FROM public.{UPP_STL_TABLE_SQL} s
        ),
        s_segment_names AS (
            SELECT
                ct_key,
                COALESCE(jsonb_agg(DISTINCT name.value) FILTER (WHERE name.value IS NOT NULL), '[]'::jsonb) AS segment_names
            FROM s_raw
            CROSS JOIN LATERAL jsonb_array_elements_text(COALESCE(s_raw.segment_names, '[]'::jsonb)) AS name(value)
            GROUP BY ct_key
        ),
        s_segment_families AS (
            SELECT
                ct_key,
                COALESCE(jsonb_agg(DISTINCT family.value) FILTER (WHERE family.value IS NOT NULL), '[]'::jsonb) AS segment_families
            FROM s_raw
            CROSS JOIN LATERAL jsonb_array_elements_text(COALESCE(s_raw.segment_families, '[]'::jsonb)) AS family(value)
            GROUP BY ct_key
        ),
        s_segment_categories AS (
            SELECT
                ct_key,
                COALESCE(jsonb_agg(DISTINCT category.value) FILTER (WHERE category.value IS NOT NULL), '[]'::jsonb) AS segment_categories
            FROM s_raw
            CROSS JOIN LATERAL jsonb_array_elements_text(COALESCE(s_raw.segment_categories, '[]'::jsonb)) AS category(value)
            GROUP BY ct_key
        ),
        s_segments AS (
            SELECT
                s_raw.ct_key,
                COALESCE(s_segment_names.segment_names, '[]'::jsonb) AS segment_names,
                COALESCE(s_segment_families.segment_families, '[]'::jsonb) AS segment_families,
                COALESCE(s_segment_categories.segment_categories, '[]'::jsonb) AS segment_categories
            FROM (SELECT DISTINCT ct_key FROM s_raw) s_raw
            LEFT JOIN s_segment_names ON s_segment_names.ct_key = s_raw.ct_key
            LEFT JOIN s_segment_families ON s_segment_families.ct_key = s_raw.ct_key
            LEFT JOIN s_segment_categories ON s_segment_categories.ct_key = s_raw.ct_key
        ),
        s AS (
            SELECT
                s_raw.ct_key,
                (array_agg(s_raw.ct_number ORDER BY s_raw.updated_at DESC NULLS LAST) FILTER (WHERE s_raw.ct_number IS NOT NULL))[1] AS ct_number,
                COALESCE(sum(COALESCE(s_raw.file_count, 0)), 0)::int AS file_count,
                COALESCE(sum(COALESCE(s_raw.total_bytes, 0)), 0)::bigint AS total_bytes,
                COALESCE((array_agg(s_segments.segment_names) FILTER (WHERE s_segments.segment_names IS NOT NULL))[1], '[]'::jsonb) AS segment_names,
                COALESCE((array_agg(s_segments.segment_families) FILTER (WHERE s_segments.segment_families IS NOT NULL))[1], '[]'::jsonb) AS segment_families,
                COALESCE((array_agg(s_segments.segment_categories) FILTER (WHERE s_segments.segment_categories IS NOT NULL))[1], '[]'::jsonb) AS segment_categories,
                max(s_raw.updated_at) AS updated_at,
                count(*)::int AS stl_asset_count,
                jsonb_agg(
                    jsonb_build_object(
                        'ct_number', s_raw.ct_number,
                        'file_count', s_raw.file_count,
                        'total_bytes', s_raw.total_bytes,
                        'updated_at', s_raw.updated_at,
                        'segment_categories', s_raw.segment_categories,
                        'segment_families', s_raw.segment_families
                    )
                    ORDER BY s_raw.updated_at DESC NULLS LAST
                ) AS stl_assets
            FROM s_raw
            LEFT JOIN s_segments ON s_segments.ct_key = s_raw.ct_key
            GROUP BY s_raw.ct_key
        ),
        r AS (
            SELECT
                upper(ct_number) AS ct_key,
                count(*) FILTER (WHERE COALESCE(registration_status, 'unregistered') = 'registered')::int AS registration_count,
                count(DISTINCT series_instance_uid) FILTER (
                    WHERE COALESCE(registration_status, 'unregistered') = 'registered'
                      AND series_instance_uid <> ''
                )::int AS registered_series,
                count(*) FILTER (
                    WHERE COALESCE(registration_status, 'unregistered') = 'registered'
                      AND COALESCE(locked, false)
                )::int AS locked_count,
                max(updated_at) FILTER (WHERE COALESCE(registration_status, 'unregistered') = 'registered') AS registration_updated_at
            FROM public.{REGISTRATION_TABLE_SQL}
            GROUP BY upper(ct_number)
        ),
        keys AS (
            SELECT ct_key FROM p
            UNION
            SELECT ct_key FROM u
            UNION
            SELECT ct_key FROM s
        ),
        relation AS (
            SELECT
                k.ct_key,
                p.ct_number AS pacs_ct_number,
                COALESCE(s.ct_number, u.ct_number) AS stl_ct_number,
                u.ct_number AS upp_ct_number,
                s.ct_number AS stl_file_ct_number,
                p.ct_number IS NOT NULL AS pacs_present,
                u.ct_number IS NOT NULL OR s.ct_number IS NOT NULL AS stl_asset_present,
                u.ct_number IS NOT NULL AS upp_present,
                COALESCE(u.stl_present, false) OR s.ct_number IS NOT NULL AS stl_present,
                CASE
                    WHEN p.ct_number IS NOT NULL AND (COALESCE(u.stl_present, false) OR s.ct_number IS NOT NULL) THEN 'complete'
                    WHEN p.ct_number IS NOT NULL AND u.ct_number IS NOT NULL THEN 'no_stl'
                    WHEN p.ct_number IS NOT NULL THEN 'pacs_only'
                    WHEN (COALESCE(u.stl_present, false) OR s.ct_number IS NOT NULL) THEN 'stl_only'
                    WHEN u.ct_number IS NOT NULL THEN 'list_only'
                    ELSE 'unknown'
                END AS relation_status,
                CASE
                    WHEN p.ct_number IS NOT NULL AND u.ct_number IS NOT NULL AND p.ct_number = u.ct_number THEN 'exact'
                    ELSE ''
                END AS match_type,
                p.batch_name,
                p.source_patient_name,
                p.patient_name_dicom,
                COALESCE(NULLIF(p.source_patient_name, ''), NULLIF(p.patient_name_dicom, '')) AS pacs_patient_name,
                p.patient_id,
                p.patient_sex AS pacs_patient_sex,
                p.study_date,
                p.study_time,
                p.study_description,
                p.modality,
                p.series_count AS pacs_series_count,
                p.dicom_file_count,
                p.processed_path,
                p.updated_at AS pacs_updated_at,
                ps.annotated_series,
                ps.undetermined_series,
                ps.completed,
                ps.body_parts,
                COALESCE(a_labels.dicom_annotation_labels, '[]'::jsonb) AS dicom_annotation_labels,
                to_jsonb(array_remove(ARRAY[
                    CASE WHEN ps.body_parts ? 'head_neck' THEN '头颈部' END,
                    CASE WHEN ps.body_parts ? 'chest' THEN '胸部' END,
                    CASE WHEN ps.body_parts ? 'upper_abdomen' THEN '上腹部' END,
                    CASE WHEN ps.body_parts ? 'abdomen_pelvis' OR ps.body_parts ? 'lower_abdomen' OR ps.body_parts ? 'pelvis' THEN '腹盆部' END
                ], NULL)) AS dicom_body_parts,
                u.patient_name AS upp_patient_name,
                COALESCE(u.patient_names, '[]'::jsonb) AS upp_patient_names,
                u.patient_sex AS upp_patient_sex,
                u.patient_age,
                u.patient_id_masked,
                u.exam_date,
                u.task_created_at,
                u.exam_description,
                u.exam_device,
                u.algorithm_model,
                COALESCE(u.algorithm_models, '[]'::jsonb) AS algorithm_models,
                u.upp_status,
                u.list_present,
                u.list_record_count,
                u.processed_stl_dir,
                u.stl_file_count,
                u.stl_total_bytes,
                u.updated_at AS upp_updated_at,
                COALESCE(u.upp_asset_count, 0) AS upp_asset_count,
                COALESCE(u.duplicate_model_asset_count, 0) AS duplicate_model_asset_count,
                COALESCE(u.upp_assets, '[]'::jsonb) AS upp_assets,
                s.file_count AS stl_file_count_agg,
                s.total_bytes AS stl_total_bytes_agg,
                s.segment_names,
                s.segment_families,
                s.segment_categories,
                s.updated_at AS stl_updated_at,
                COALESCE(s.stl_asset_count, 0) AS stl_asset_count,
                COALESCE(s.stl_assets, '[]'::jsonb) AS stl_assets,
                COALESCE(r.registration_count, 0)::int AS registration_count,
                COALESCE(r.registered_series, 0)::int AS registered_series,
                COALESCE(r.locked_count, 0)::int AS locked_count,
                r.registration_updated_at,
                CASE
                    WHEN COALESCE(NULLIF(p.source_patient_name, ''), NULLIF(p.patient_name_dicom, '')) IS NULL
                      OR NULLIF(u.patient_name, '') IS NULL THEN ''
                    WHEN regexp_replace(COALESCE(NULLIF(p.source_patient_name, ''), p.patient_name_dicom), '\\s+', '', 'g')
                       = regexp_replace(u.patient_name, '\\s+', '', 'g') THEN 'same'
                    ELSE 'different'
                END AS patient_name_match
            FROM keys k
            LEFT JOIN p ON p.ct_key = k.ct_key
            LEFT JOIN ps ON ps.ct_number = p.ct_number
            LEFT JOIN a_labels ON a_labels.ct_number = p.ct_number
            LEFT JOIN u ON u.ct_key = k.ct_key
            LEFT JOIN s ON s.ct_key = k.ct_key
            LEFT JOIN r ON r.ct_key = k.ct_key
        )
    """


def relation_where(q: str, status: str, algorithm_model: str, dicom_part: str) -> str:
    clauses: list[str] = []
    query = q.strip()
    if query:
        like = "%" + query.replace("%", "").replace("_", "") + "%"
        literal = sql_literal(like)
        clauses.append(
            " OR ".join(
                [
                    f"ct_key ILIKE {literal}",
                    f"pacs_ct_number ILIKE {literal}",
                    f"stl_ct_number ILIKE {literal}",
                    f"pacs_patient_name ILIKE {literal}",
                    f"upp_patient_name ILIKE {literal}",
                    f"patient_id ILIKE {literal}",
                    f"algorithm_model ILIKE {literal}",
                    f"upp_status ILIKE {literal}",
                    f"exam_description ILIKE {literal}",
                    f"study_description ILIKE {literal}",
                    f"COALESCE(dicom_body_parts::text, '') ILIKE {literal}",
                ]
            ).join(("(", ")"))
        )
    if status == "complete":
        clauses.append("relation_status = 'complete'")
    elif status == "no_stl":
        clauses.append("relation_status = 'no_stl'")
    elif status == "pacs_only":
        clauses.append("relation_status = 'pacs_only'")
    elif status == "stl_only":
        clauses.append("relation_status = 'stl_only'")
    elif status == "list_only":
        clauses.append("relation_status = 'list_only'")
    elif status == "pending_dicom":
        clauses.append("pacs_present AND COALESCE(completed, false) IS NOT TRUE")
    elif status == "undetermined":
        clauses.append("COALESCE(undetermined_series, 0) > 0")
    elif status != "all":
        raise HTTPException(status_code=400, detail="invalid status filter")
    if algorithm_model:
        clauses.append(f"algorithm_model = {sql_literal(algorithm_model)}")
    if dicom_part:
        valid_parts = {"head_neck", "chest", "upper_abdomen", "abdomen_pelvis"}
        if dicom_part not in valid_parts:
            raise HTTPException(status_code=400, detail="invalid DICOM part filter")
        if dicom_part == "abdomen_pelvis":
            clauses.append("(body_parts ? 'abdomen_pelvis' OR body_parts ? 'lower_abdomen' OR body_parts ? 'pelvis')")
        else:
            clauses.append(f"body_parts ? {sql_literal(dicom_part)}")
    return "WHERE " + " AND ".join(clauses) if clauses else ""


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


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


@app.post("/api/auth/login")
def login(payload: LoginPayload) -> dict[str, str]:
    username = payload.username.strip()
    try:
        user = fetch_user(username)
    except Exception:
        user = None
    if not user and username == WEB_ADMIN_USER:
        user = {"username": WEB_ADMIN_USER, "password_hash": password_hash(WEB_ADMIN_PASSWORD), "role": "管理员", "status": "启用"}
    if not user or user.get("status") != "启用" or user.get("password_hash") != password_hash(payload.password):
        raise HTTPException(status_code=401, detail="用户名或密码错误")
    token = secrets.token_urlsafe(32)
    TOKENS[token] = {"username": user["username"], "role": user.get("role", "阅片员")}
    return {"token": token, "username": user["username"], "role": user.get("role", "阅片员")}


@app.get("/api/auth/me")
def me(user: dict[str, str] = Depends(current_user)) -> dict[str, str]:
    return user


@app.get("/api/settings")
def settings(user: dict[str, str] = Depends(admin_user)) -> dict[str, Any]:
    rows = pg_json_rows(
        f"""
        SELECT username, role, status, created_at, updated_at
        FROM public.{USER_TABLE_SQL}
        ORDER BY username
        """,
        timeout=10,
    )
    return {
        "user": user,
        "viewer_url": PACS_VIEWER_URL,
        "registration_url": REGISTRATION_URL,
        "users": rows,
        "database": {"host": PGHOST, "port": PGPORT, "database": PGDATABASE},
        "tables": {
            "dicom": PACS_TABLE,
            "dicom_summary": PACS_SUMMARY_TABLE,
            "dicom_annotations": PACS_ANNOTATION_TABLE,
            "upp_assets": UPP_ASSET_TABLE,
            "upp_stl": UPP_STL_TABLE,
            "registration": REGISTRATION_TABLE,
        },
    }


@app.post("/api/settings/users")
def save_user(payload: UserPayload, user: dict[str, str] = Depends(admin_user)) -> dict[str, str]:
    username = payload.username.strip()
    if not username:
        raise HTTPException(status_code=400, detail="账号不能为空")
    role = payload.role if payload.role in {"管理员", "阅片员", "访客"} else "阅片员"
    status = payload.status if payload.status in {"启用", "停用"} else "启用"
    existing = fetch_user(username)
    if existing and not payload.password:
        run_psql(
            f"""
            UPDATE public.{USER_TABLE_SQL}
            SET role = {sql_literal(role)}, status = {sql_literal(status)}, updated_at = now()
            WHERE username = {sql_literal(username)}
            """,
            timeout=10,
        )
    else:
        if not payload.password:
            raise HTTPException(status_code=400, detail="新账号需要初始密码")
        run_psql(
            f"""
            INSERT INTO public.{USER_TABLE_SQL} (username, password_hash, role, status, updated_at)
            VALUES ({sql_literal(username)}, {sql_literal(password_hash(payload.password))}, {sql_literal(role)}, {sql_literal(status)}, now())
            ON CONFLICT (username) DO UPDATE
            SET password_hash = EXCLUDED.password_hash,
                role = EXCLUDED.role,
                status = EXCLUDED.status,
                updated_at = now()
            """,
            timeout=10,
        )
    return {"status": "ok", "updated_by": user["username"]}


@app.put("/api/settings/users/{username}/password")
def change_password(username: str, payload: PasswordPayload, user: dict[str, str] = Depends(admin_user)) -> dict[str, str]:
    if not payload.password:
        raise HTTPException(status_code=400, detail="密码不能为空")
    run_psql(
        f"""
        UPDATE public.{USER_TABLE_SQL}
        SET password_hash = {sql_literal(password_hash(payload.password))}, updated_at = now()
        WHERE username = {sql_literal(username)}
        """,
        timeout=10,
    )
    return {"status": "ok", "updated_by": user["username"]}


@app.get("/api/status")
def status(user: dict[str, str] = Depends(current_user)) -> dict[str, Any]:
    ok, message = db_available()
    counts: dict[str, Any] = {}
    if ok:
        ensure_registration_table()
        rows = pg_json_rows(
            f"""
            {relation_cte()}
            SELECT
                count(*)::int AS total_ct,
                count(*) FILTER (WHERE pacs_present)::int AS pacs_count,
                count(*) FILTER (WHERE stl_asset_present)::int AS stl_asset_count,
                count(*) FILTER (WHERE list_present)::int AS list_count,
                count(*) FILTER (WHERE stl_present)::int AS stl_count,
                count(*) FILTER (WHERE relation_status = 'complete')::int AS complete_count,
                count(*) FILTER (WHERE relation_status = 'no_stl')::int AS no_stl_count,
                count(*) FILTER (WHERE relation_status = 'pacs_only')::int AS pacs_only_count,
                count(*) FILTER (WHERE relation_status = 'stl_only')::int AS stl_only_count,
                count(*) FILTER (WHERE relation_status = 'list_only')::int AS list_only_count,
                count(*) FILTER (WHERE registration_count > 0)::int AS registered_ct_count,
                count(*) FILTER (WHERE locked_count > 0)::int AS locked_ct_count,
                COALESCE(sum(registration_count), 0)::int AS registration_count,
                COALESCE(sum(locked_count), 0)::int AS locked_registration_count,
                count(*) FILTER (WHERE pacs_present AND COALESCE(completed, false) IS NOT TRUE)::int AS pending_dicom_count,
                COALESCE(sum(CASE WHEN pacs_present THEN COALESCE(undetermined_series, 0) ELSE 0 END), 0)::int AS pending_dicom_series
            FROM relation
            """,
            timeout=24,
        )
        counts = rows[0] if rows else {}
    return {
        "database": {"ok": ok, "message": message, "host": PGHOST, "database": PGDATABASE},
        "viewer_url": PACS_VIEWER_URL,
        "registration_url": REGISTRATION_URL,
        "tables": {
            "dicom": PACS_TABLE,
            "dicom_summary": PACS_SUMMARY_TABLE,
            "dicom_annotations": PACS_ANNOTATION_TABLE,
            "upp_assets": UPP_ASSET_TABLE,
            "upp_stl": UPP_STL_TABLE,
            "registration": REGISTRATION_TABLE,
        },
        "counts": counts,
    }


@app.get("/api/relations")
def relations(
    q: str = "",
    status: str = "all",
    algorithm_model: str = "",
    dicom_part: str = "",
    limit: int = Query(default=300, ge=1, le=1000),
    offset: int = Query(default=0, ge=0),
    user: dict[str, str] = Depends(current_user),
) -> list[dict[str, Any]]:
    ensure_registration_table()
    where = relation_where(q, status, algorithm_model, dicom_part)
    rows = pg_json_rows(
        f"""
        {relation_cte()}
        SELECT
            ct_key, pacs_ct_number, stl_ct_number,
            pacs_present, stl_asset_present, list_present, stl_present, relation_status, match_type,
            pacs_patient_name, upp_patient_name, upp_patient_names, patient_name_match, patient_id, patient_id_masked,
            study_date, study_time, exam_date, task_created_at,
            study_description, exam_description, algorithm_model, upp_status,
            pacs_series_count, dicom_file_count, annotated_series, undetermined_series, completed,
            list_present, list_record_count, stl_file_count, stl_file_count_agg,
            body_parts, dicom_body_parts, dicom_annotation_labels, algorithm_models,
            upp_asset_count, duplicate_model_asset_count, upp_assets, stl_asset_count, stl_assets,
            segment_categories, segment_families,
            registration_count, registered_series, locked_count, registration_updated_at,
            pacs_updated_at, upp_updated_at, stl_updated_at
        FROM relation
        {where}
        ORDER BY
            pacs_present DESC,
            relation_status,
            COALESCE(study_date, '') DESC,
            COALESCE(study_time, '') DESC,
            ct_key
        LIMIT {int(limit)}
        OFFSET {int(offset)}
        """,
        timeout=24,
    )
    return [enrich_patient_match(row) for row in rows]


@app.get("/api/relations/{ct_number}")
def relation_detail(ct_number: str, user: dict[str, str] = Depends(current_user)) -> dict[str, Any]:
    ensure_registration_table()
    ct_key = normalize_ct(ct_number)
    if not ct_key:
        raise HTTPException(status_code=400, detail="invalid CT number")
    rows = pg_json_rows(
        f"""
        {relation_cte()}
        SELECT *
        FROM relation
        WHERE ct_key = {sql_literal(ct_key)}
        LIMIT 1
        """,
        timeout=24,
    )
    if not rows:
        raise HTTPException(status_code=404, detail="CT number not found")
    return enrich_patient_match(rows[0])