from __future__ import annotations

import json
import hashlib
import os
import re
import secrets
import subprocess
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


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")
UPP_ASSET_TABLE = os.getenv("UPP_ASSET_TABLE", "upp_exam_assets")
UPP_STL_TABLE = os.getenv("UPP_STL_TABLE", "upp_stl_files")
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")

IDENT_RE = re.compile(r"^[A-Za-z_][A-Za-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)
UPP_ASSET_TABLE_SQL = ident(UPP_ASSET_TABLE)
UPP_STL_TABLE_SQL = ident(UPP_STL_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 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 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()
    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}
        ),
        u_raw AS (
            SELECT
                u.*,
                upper(u.ct_number) AS ct_key
            FROM public.{UPP_ASSET_TABLE_SQL} u
        ),
        u AS (
            SELECT DISTINCT ON (ct_key) *
            FROM u_raw
            ORDER BY ct_key, stl_present DESC, list_present DESC, updated_at DESC NULLS LAST
        ),
        s_raw AS (
            SELECT
                s.*,
                upper(s.ct_number) AS ct_key
            FROM public.{UPP_STL_TABLE_SQL} s
        ),
        s AS (
            SELECT DISTINCT ON (ct_key) *
            FROM s_raw
            ORDER BY ct_key, file_count DESC, updated_at DESC NULLS LAST
        ),
        keys AS (
            SELECT ct_key FROM p
            UNION
            SELECT ct_key FROM u
        ),
        relation AS (
            SELECT
                k.ct_key,
                p.ct_number AS pacs_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 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,
                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 ? 'lower_abdomen' THEN '下腹部' END,
                    CASE WHEN ps.body_parts ? 'pelvis' THEN '盆腔' END
                ], NULL)) AS dicom_body_parts,
                u.patient_name AS upp_patient_name,
                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,
                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,
                u.selected_list_record,
                u.list_records,
                u.stl_candidates,
                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.file_names,
                s.files AS stl_files,
                s.updated_at AS stl_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 u ON u.ct_key = k.ct_key
            LEFT JOIN s ON s.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", "lower_abdomen", "pelvis"}
        if dicom_part not in valid_parts:
            raise HTTPException(status_code=400, detail="invalid DICOM part filter")
        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,
        "users": rows,
        "database": {"host": PGHOST, "port": PGPORT, "database": PGDATABASE},
        "tables": {
            "dicom": PACS_TABLE,
            "dicom_summary": PACS_SUMMARY_TABLE,
            "upp_assets": UPP_ASSET_TABLE,
            "upp_stl": UPP_STL_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:
        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 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,
        "tables": {
            "dicom": PACS_TABLE,
            "dicom_summary": PACS_SUMMARY_TABLE,
            "upp_assets": UPP_ASSET_TABLE,
            "upp_stl": UPP_STL_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),
    user: dict[str, str] = Depends(current_user),
) -> list[dict[str, Any]]:
    where = relation_where(q, status, algorithm_model, dicom_part)
    return 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, 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, segment_categories, segment_families,
            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)}
        """,
        timeout=24,
    )


@app.get("/api/relations/{ct_number}")
def relation_detail(ct_number: str, user: dict[str, str] = Depends(current_user)) -> dict[str, Any]:
    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 rows[0]