#!/usr/bin/env python3 from __future__ import annotations import base64 import hashlib import io import json import os import secrets import subprocess import time import urllib.error import urllib.request from collections import 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 PACS_ROOT = APP_DIR.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") PGTABLE = os.getenv("PGTABLE", "pacs_dicom_files") WEB_USER = os.getenv("PACS_WEB_USER", "admin") WEB_PASSWORD = os.getenv("PACS_WEB_PASSWORD", "123456") PROCESSED_ROOT = Path(os.getenv("PACS_PROCESSED_ROOT", str(PACS_ROOT / "已处理_DICOM数据"))).resolve() KIMI_API_KEY = os.getenv("KIMI_API_KEY", "") KIMI_API_NAME = os.getenv("KIMI_API_NAME", "HIS_Check") KIMI_API_URL = os.getenv("KIMI_API_URL", "https://api.moonshot.cn/v1/chat/completions") KIMI_MODEL = os.getenv("KIMI_MODEL", "kimi-latest") WINDOWS = { "default": None, "bone": (500.0, 1800.0), "soft": (50.0, 360.0), "contrast": (90.0, 140.0), } BODY_PARTS = {"head_neck", "chest", "upper_abdomen", "lower_abdomen", "pelvis"} PHASES = {"arterial", "portal_venous", "delayed", "unknown", ""} CHEST_WINDOWS = {"lung", "mediastinal", "unknown", ""} ROLES = { "管理员": ["查看DICOM", "编辑标注", "AI识别", "用户创建", "权限控制", "系统设置"], "阅片员": ["查看DICOM", "编辑标注", "AI识别"], "访客": ["查看DICOM"], } app = FastAPI(title="DICOM 阅片分类系统") 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 LoginIn(BaseModel): username: str password: str class AnnotationIn(BaseModel): body_parts: list[str] = [] manual_body_parts: list[str] = [] ai_body_parts: list[str] = [] upper_abdomen_phase: str = "" manual_upper_abdomen_phase: str = "" ai_upper_abdomen_phase: str = "" chest_window: str = "" manual_chest_window: str = "" ai_chest_window: str = "" plain_ct: bool = False manual_plain_ct: bool | None = None ai_plain_ct: bool | None = None notes: str = "" skipped: bool = False ai_skipped: bool = False class AIRequest(BaseModel): sample_count: int = 3 class UserIn(BaseModel): username: str password: str role: str = "阅片员" status: str = "启用" def pg_env() -> dict[str, str]: env = os.environ.copy() if os.getenv("PGPASSWORD"): env["PGPASSWORD"] = os.environ["PGPASSWORD"] return env def sql_literal(value: Any) -> str: if value is None: return "NULL" return "'" + str(value).replace("'", "''") + "'" def run_psql(sql: str, timeout: int = 12) -> subprocess.CompletedProcess[str]: return subprocess.run( [ "psql", "-h", PGHOST, "-p", PGPORT, "-U", PGUSER, "-d", PGDATABASE, "-X", "-q", "-t", "-A", "-c", sql, ], text=True, capture_output=True, timeout=timeout, env=pg_env(), ) def pg_scalar(sql: str, timeout: int = 12) -> 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 = 20) -> 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 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: # noqa: BLE001 return False, str(exc) def ensure_annotation_table() -> None: sql = """ CREATE TABLE IF NOT EXISTS public.pacs_dicom_series_annotations ( ct_number text NOT NULL, study_instance_uid text, series_instance_uid text NOT NULL, series_description text, body_parts jsonb NOT NULL DEFAULT '[]'::jsonb, manual_body_parts jsonb NOT NULL DEFAULT '[]'::jsonb, ai_body_parts jsonb NOT NULL DEFAULT '[]'::jsonb, upper_abdomen_phase text NOT NULL DEFAULT '', manual_upper_abdomen_phase text NOT NULL DEFAULT '', ai_upper_abdomen_phase text NOT NULL DEFAULT '', chest_window text NOT NULL DEFAULT '', manual_chest_window text NOT NULL DEFAULT '', ai_chest_window text NOT NULL DEFAULT '', plain_ct boolean NOT NULL DEFAULT false, manual_plain_ct boolean, ai_plain_ct boolean, skipped boolean NOT NULL DEFAULT false, ai_skipped boolean NOT NULL DEFAULT false, notes text NOT NULL DEFAULT '', ai_result jsonb, ai_model text, updated_by text NOT NULL DEFAULT 'admin', created_at timestamptz NOT NULL DEFAULT now(), updated_at timestamptz NOT NULL DEFAULT now(), PRIMARY KEY (ct_number, series_instance_uid) ); ALTER TABLE public.pacs_dicom_series_annotations ADD COLUMN IF NOT EXISTS manual_body_parts jsonb NOT NULL DEFAULT '[]'::jsonb; ALTER TABLE public.pacs_dicom_series_annotations ADD COLUMN IF NOT EXISTS ai_body_parts jsonb NOT NULL DEFAULT '[]'::jsonb; ALTER TABLE public.pacs_dicom_series_annotations ADD COLUMN IF NOT EXISTS manual_upper_abdomen_phase text NOT NULL DEFAULT ''; ALTER TABLE public.pacs_dicom_series_annotations ADD COLUMN IF NOT EXISTS ai_upper_abdomen_phase text NOT NULL DEFAULT ''; ALTER TABLE public.pacs_dicom_series_annotations ADD COLUMN IF NOT EXISTS chest_window text NOT NULL DEFAULT ''; ALTER TABLE public.pacs_dicom_series_annotations ADD COLUMN IF NOT EXISTS manual_chest_window text NOT NULL DEFAULT ''; ALTER TABLE public.pacs_dicom_series_annotations ADD COLUMN IF NOT EXISTS ai_chest_window text NOT NULL DEFAULT ''; ALTER TABLE public.pacs_dicom_series_annotations ADD COLUMN IF NOT EXISTS plain_ct boolean NOT NULL DEFAULT false; ALTER TABLE public.pacs_dicom_series_annotations ADD COLUMN IF NOT EXISTS manual_plain_ct boolean; ALTER TABLE public.pacs_dicom_series_annotations ADD COLUMN IF NOT EXISTS ai_plain_ct boolean; ALTER TABLE public.pacs_dicom_series_annotations ADD COLUMN IF NOT EXISTS skipped boolean NOT NULL DEFAULT false; ALTER TABLE public.pacs_dicom_series_annotations ADD COLUMN IF NOT EXISTS ai_skipped boolean NOT NULL DEFAULT false; ALTER TABLE public.pacs_dicom_series_annotations ADD COLUMN IF NOT EXISTS ai_result jsonb; ALTER TABLE public.pacs_dicom_series_annotations ADD COLUMN IF NOT EXISTS ai_model text; UPDATE public.pacs_dicom_series_annotations SET manual_body_parts = body_parts, manual_upper_abdomen_phase = upper_abdomen_phase WHERE jsonb_array_length(body_parts) > 0 AND jsonb_array_length(manual_body_parts) = 0 AND jsonb_array_length(ai_body_parts) = 0; """ pg_scalar(sql) def password_hash(password: str) -> str: return hashlib.sha256(("pacs-dicom-web:" + password).encode("utf-8")).hexdigest() def ensure_user_table() -> None: sql = f""" CREATE TABLE IF NOT EXISTS public.pacs_web_users ( 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.pacs_web_users (username, password_hash, role, status) VALUES ({sql_literal(WEB_USER)}, {sql_literal(password_hash(WEB_PASSWORD))}, '管理员', '启用') ON CONFLICT (username) DO NOTHING; """ pg_scalar(sql) def web_users() -> list[dict[str, Any]]: try: ensure_user_table() return pg_json_rows( """ SELECT username, role, status, created_at, updated_at FROM public.pacs_web_users ORDER BY username """ ) except Exception: return [{"username": WEB_USER, "role": "管理员", "status": "启用"}] def authenticate_web_user(username: str, password: str) -> bool: try: ok, _ = db_available() if ok: ensure_user_table() rows = pg_json_rows( f""" SELECT username, password_hash, status FROM public.pacs_web_users WHERE username = {sql_literal(username)} LIMIT 1 """ ) if rows: row = rows[0] return row.get("status") == "启用" and row.get("password_hash") == password_hash(password) except Exception: pass return username == WEB_USER and password == WEB_PASSWORD @app.on_event("startup") def startup() -> None: ok, _ = db_available() if ok: ensure_annotation_table() ensure_user_table() 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() if not token: raise HTTPException(status_code=401, detail="unauthorized") user = TOKENS.get(token) if not user: raise HTTPException(status_code=401, detail="unauthorized") return user @app.get("/") def index() -> FileResponse: return FileResponse(STATIC_DIR / "index.html") @app.post("/api/auth/login") def login(data: LoginIn) -> dict[str, str]: if authenticate_web_user(data.username, data.password): token = secrets.token_urlsafe(32) TOKENS[token] = data.username return {"token": token, "username": data.username} raise HTTPException(status_code=401, detail="invalid credentials") @app.get("/api/status") def status() -> dict[str, Any]: db_ok, db_message = db_available() table_count = None if db_ok: try: table_count = int(pg_scalar(f"SELECT count(*) FROM public.{PGTABLE}")) except Exception: table_count = None return { "database": {"ok": db_ok, "message": db_message, "host": PGHOST, "database": PGDATABASE, "table": PGTABLE, "rows": table_count}, "dicom": {"processed_root": str(PROCESSED_ROOT), "exists": PROCESSED_ROOT.exists()}, "ai": {"configured": bool(KIMI_API_KEY), "provider": "Kimi", "name": KIMI_API_NAME, "model": KIMI_MODEL}, "server_time": time.strftime("%Y-%m-%d %H:%M:%S"), } @app.get("/api/studies") def studies(_: str = Depends(require_auth), q: str = "", limit: int = 200) -> list[dict[str, Any]]: where = "" if q: like = "%" + q.replace("%", "").replace("_", "") + "%" where = f"WHERE ct_number ILIKE {sql_literal(like)} OR source_patient_name ILIKE {sql_literal(like)} OR patient_name_dicom ILIKE {sql_literal(like)}" rows = pg_json_rows( f""" SELECT ct_number, batch_name, target_folder_name, source_patient_name, patient_name_dicom, patient_id, study_date, study_time, modality, dicom_file_count, series_count, processed_path, needs_ct_number_fix, status FROM public.{PGTABLE} {where} ORDER BY study_date DESC NULLS LAST, study_time DESC NULLS LAST, ct_number LIMIT {int(limit)} """ ) annotations = pg_json_rows( """ SELECT ct_number, count(*)::int AS annotated_series FROM public.pacs_dicom_series_annotations WHERE skipped IS TRUE OR jsonb_array_length(body_parts) > 0 OR plain_ct IS TRUE OR NULLIF(notes, '') IS NOT NULL OR ai_result IS NOT NULL GROUP BY ct_number """, timeout=8, ) annotation_map = {row["ct_number"]: row["annotated_series"] for row in annotations} for row in rows: total_series = int(row.get("series_count") or 0) row["annotated_series"] = min(total_series, int(annotation_map.get(row["ct_number"], 0))) row["unannotated_series"] = max(0, total_series - int(row["annotated_series"])) return rows def get_study_record(ct_number: str) -> dict[str, Any]: rows = pg_json_rows( f""" SELECT * FROM public.{PGTABLE} WHERE ct_number = {sql_literal(ct_number)} LIMIT 1 """ ) if not rows: raise HTTPException(status_code=404, detail="study not found") 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: direct_matches = list(PROCESSED_ROOT.glob(f"*/{target_folder}")) if direct_matches: return direct_matches[0] recursive = next(PROCESSED_ROOT.rglob(target_folder), None) if recursive: return recursive ct_number = str(study.get("ct_number") or "") recursive = next(PROCESSED_ROOT.rglob(f"{ct_number}-*"), None) if recursive: return recursive return root def read_header(path: Path) -> dict[str, str]: tags = [ "SeriesInstanceUID", "StudyInstanceUID", "AccessionNumber", "PatientName", "PatientID", "PatientBirthDate", "PatientSex", "InstitutionName", "StudyDate", "SeriesNumber", "SeriesDescription", "InstanceNumber", "SliceLocation", "ImagePositionPatient", "AcquisitionTime", "ContentTime", "SeriesTime", "StudyTime", "Modality", "BodyPartExamined", "Manufacturer", "Rows", "Columns", "PixelSpacing", "SliceThickness", "SpacingBetweenSlices", "WindowCenter", "WindowWidth", ] ds = pydicom.dcmread(str(path), stop_before_pixels=True, force=True, specific_tags=tags + ["SpecificCharacterSet"]) return {tag: str(getattr(ds, tag, "")).strip() for tag in tags} def sort_key(item: tuple[Path, dict[str, str]]) -> tuple[float, float, str]: path, meta = item instance = float(meta.get("InstanceNumber") or 0) position = meta.get("ImagePositionPatient", "") z = 0.0 if position: try: z = float(str(position).strip("[]").split(",")[-1]) except Exception: z = 0.0 if meta.get("SliceLocation"): try: z = float(meta["SliceLocation"]) except Exception: pass return (z, instance, str(path)) def valid_parts(parts: Any) -> list[str]: if not isinstance(parts, list): return [] return [part for part in parts if part in BODY_PARTS] def valid_phase(value: Any) -> str: return value if value in PHASES else "" def valid_chest_window(value: Any) -> str: return value if value in CHEST_WINDOWS else "" def bool_or_none(value: Any) -> bool | None: if value is None: return None if isinstance(value, bool): return value if isinstance(value, str): if value.lower() in {"true", "t", "1", "yes", "y"}: return True if value.lower() in {"false", "f", "0", "no", "n"}: return False return bool(value) def merge_parts(manual_parts: list[str], ai_parts: list[str], skipped: bool) -> list[str]: merged: list[str] = [] for part in manual_parts + ai_parts: if part in BODY_PARTS and part not in merged: merged.append(part) return merged def effective_phase(manual_phase: str, ai_phase: str, body_parts: list[str], skipped: bool) -> str: if "upper_abdomen" not in body_parts: return "" return valid_phase(manual_phase) or valid_phase(ai_phase) def effective_chest_window(manual_window: str, ai_window: str, body_parts: list[str], skipped: bool) -> str: if "chest" not in body_parts: return "" return valid_chest_window(manual_window) or valid_chest_window(ai_window) or "unknown" def effective_plain_ct(manual_plain_ct: bool | None, ai_plain_ct: bool | None, fallback: bool, skipped: bool) -> bool: if manual_plain_ct is not None: return manual_plain_ct if ai_plain_ct is not None: return ai_plain_ct return bool(fallback) def normalize_annotation(row: dict[str, Any] | None, default_skipped: bool = False) -> dict[str, Any]: row = row or {} manual_parts = valid_parts(row.get("manual_body_parts") or []) ai_parts = valid_parts(row.get("ai_body_parts") or []) if not manual_parts and not ai_parts: manual_parts = valid_parts(row.get("body_parts") or []) skipped = bool(row.get("skipped", default_skipped)) ai_skipped = bool(row.get("ai_skipped", False)) manual_phase = valid_phase(row.get("manual_upper_abdomen_phase") or "") ai_phase = valid_phase(row.get("ai_upper_abdomen_phase") or "") if not manual_phase and not ai_phase: manual_phase = valid_phase(row.get("upper_abdomen_phase") or "") manual_chest_window = valid_chest_window(row.get("manual_chest_window") or "") ai_chest_window = valid_chest_window(row.get("ai_chest_window") or "") if not manual_chest_window and not ai_chest_window: manual_chest_window = valid_chest_window(row.get("chest_window") or "") body_parts = merge_parts(manual_parts, ai_parts, skipped) plain_ct = effective_plain_ct( bool_or_none(row.get("manual_plain_ct")), bool_or_none(row.get("ai_plain_ct")), bool(row.get("plain_ct", False)), skipped, ) return { "body_parts": body_parts, "manual_body_parts": manual_parts, "ai_body_parts": ai_parts, "upper_abdomen_phase": effective_phase(manual_phase, ai_phase, body_parts, skipped), "manual_upper_abdomen_phase": manual_phase, "ai_upper_abdomen_phase": ai_phase, "chest_window": effective_chest_window(manual_chest_window, ai_chest_window, body_parts, skipped), "manual_chest_window": manual_chest_window, "ai_chest_window": ai_chest_window, "plain_ct": plain_ct, "manual_plain_ct": bool_or_none(row.get("manual_plain_ct")), "ai_plain_ct": bool_or_none(row.get("ai_plain_ct")), "skipped": skipped, "ai_skipped": ai_skipped, "notes": row.get("notes", ""), "updated_at": row.get("updated_at", ""), "ai_model": row.get("ai_model", ""), "updated_by": row.get("updated_by", ""), } def dicom_auto_annotation(meta: dict[str, str]) -> tuple[list[str], str, str]: marker = f"{meta.get('BodyPartExamined', '')} {meta.get('SeriesDescription', '')}".upper() parts: list[str] = [] phase = "" chest_window = "" if "CHEST" in marker and "chest" not in parts: parts.append("chest") chest_window = "unknown" if "ABDOMEN" in marker and "upper_abdomen" not in parts: parts.append("upper_abdomen") phase = "unknown" return parts, phase, chest_window def get_annotations(ct_number: str) -> dict[str, dict[str, Any]]: try: ensure_annotation_table() rows = pg_json_rows( f""" SELECT series_instance_uid, body_parts, manual_body_parts, ai_body_parts, upper_abdomen_phase, manual_upper_abdomen_phase, ai_upper_abdomen_phase, chest_window, manual_chest_window, ai_chest_window, plain_ct, manual_plain_ct, ai_plain_ct, skipped, ai_skipped, notes, updated_at, ai_model, updated_by FROM public.pacs_dicom_series_annotations WHERE ct_number = {sql_literal(ct_number)} """ ) except Exception: return {} return {row["series_instance_uid"]: row for row in rows} def scan_study(ct_number: str) -> dict[str, Any]: cached = STUDY_CACHE.get(ct_number) 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 path not found: {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_list = [] file_map = {} default_skip_rows = [] auto_annotation_rows = [] 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] raw_annotation = annotations.get(uid) annotation = normalize_annotation(raw_annotation) auto_parts, auto_phase, auto_chest_window = dicom_auto_annotation(first) can_apply_auto = bool(auto_parts) and not annotation.get("body_parts") if can_apply_auto: annotation["manual_body_parts"] = auto_parts annotation["body_parts"] = auto_parts if auto_phase: annotation["manual_upper_abdomen_phase"] = auto_phase annotation["upper_abdomen_phase"] = auto_phase if auto_chest_window: annotation["manual_chest_window"] = auto_chest_window annotation["chest_window"] = auto_chest_window auto_annotation_rows.append((uid, first, auto_parts, auto_phase, auto_chest_window)) default_skipped = len(items) < 80 and (not raw_annotation or annotation.get("updated_by") == "system") if default_skipped: annotation["skipped"] = True if default_skipped: default_skip_rows.append((uid, first, len(items))) series_list.append( { "ct_number": ct_number, "series_uid": uid, "study_uid": first.get("StudyInstanceUID", ""), "accession_number": first.get("AccessionNumber", ""), "patient_name": first.get("PatientName", ""), "patient_id": first.get("PatientID", ""), "patient_birth_date": first.get("PatientBirthDate", ""), "patient_sex": first.get("PatientSex", ""), "institution_name": first.get("InstitutionName", ""), "study_date": first.get("StudyDate", "") or study.get("study_date", ""), "series_number": first.get("SeriesNumber", ""), "description": first.get("SeriesDescription", "") or "未命名序列", "count": len(items), "modality": first.get("Modality", ""), "body_part_dicom": first.get("BodyPartExamined", ""), "study_time": first.get("StudyTime", ""), "series_time": first.get("SeriesTime", "") or first.get("AcquisitionTime", "") or first.get("ContentTime", ""), "first_time": first.get("AcquisitionTime", "") or first.get("ContentTime", ""), "last_time": last.get("AcquisitionTime", "") or last.get("ContentTime", ""), "manufacturer": first.get("Manufacturer", ""), "rows": first.get("Rows", ""), "columns": first.get("Columns", ""), "pixel_spacing": first.get("PixelSpacing", ""), "slice_thickness": first.get("SliceThickness", ""), "spacing_between_slices": first.get("SpacingBetweenSlices", ""), "window_center": first.get("WindowCenter", ""), "window_width": first.get("WindowWidth", ""), "default_skipped": default_skipped, "annotation": annotation, } ) for uid, first, auto_parts, auto_phase, auto_chest_window in auto_annotation_rows: try: pg_scalar( f""" INSERT INTO public.pacs_dicom_series_annotations ( ct_number, study_instance_uid, series_instance_uid, series_description, body_parts, manual_body_parts, upper_abdomen_phase, manual_upper_abdomen_phase, chest_window, manual_chest_window, updated_by, updated_at ) VALUES ( {sql_literal(ct_number)}, {sql_literal(first.get('StudyInstanceUID', ''))}, {sql_literal(uid)}, {sql_literal(first.get('SeriesDescription', '') or '未命名序列')}, {sql_literal(json.dumps(auto_parts, ensure_ascii=False))}::jsonb, {sql_literal(json.dumps(auto_parts, ensure_ascii=False))}::jsonb, {sql_literal(auto_phase)}, {sql_literal(auto_phase)}, {sql_literal(auto_chest_window)}, {sql_literal(auto_chest_window)}, 'system', now() ) ON CONFLICT (ct_number, series_instance_uid) DO UPDATE SET body_parts = EXCLUDED.body_parts, manual_body_parts = EXCLUDED.manual_body_parts, upper_abdomen_phase = EXCLUDED.upper_abdomen_phase, manual_upper_abdomen_phase = EXCLUDED.manual_upper_abdomen_phase, chest_window = EXCLUDED.chest_window, manual_chest_window = EXCLUDED.manual_chest_window, updated_by = 'system', updated_at = now() WHERE pacs_dicom_series_annotations.updated_by = 'system' AND jsonb_array_length(pacs_dicom_series_annotations.body_parts) = 0 """, timeout=8, ) except Exception: pass for uid, first, count in default_skip_rows: try: pg_scalar( f""" INSERT INTO public.pacs_dicom_series_annotations ( ct_number, study_instance_uid, series_instance_uid, series_description, body_parts, manual_body_parts, ai_body_parts, upper_abdomen_phase, plain_ct, skipped, notes, updated_by, updated_at ) VALUES ( {sql_literal(ct_number)}, {sql_literal(first.get('StudyInstanceUID', ''))}, {sql_literal(uid)}, {sql_literal(first.get('SeriesDescription', '') or '未命名序列')}, '[]'::jsonb, '[]'::jsonb, '[]'::jsonb, '', false, true, {sql_literal(f'少于80张默认略过/不采用（{count}张）')}, 'system', now() ) ON CONFLICT (ct_number, series_instance_uid) DO UPDATE SET skipped = true, notes = CASE WHEN pacs_dicom_series_annotations.notes = '' THEN EXCLUDED.notes ELSE pacs_dicom_series_annotations.notes END, updated_by = 'system', updated_at = now() WHERE pacs_dicom_series_annotations.updated_by = 'system' """, timeout=8, ) except Exception: pass def series_time_key(row: dict[str, Any]) -> tuple[str, int, str]: return ( row.get("series_time") or row.get("first_time") or row.get("study_time") or "", numeric(row.get("series_number", "")), row.get("description", ""), ) def numeric(value: str) -> int: try: return int(float(value)) except Exception: return 999999 series_list.sort(key=lambda row: (1 if row["annotation"].get("skipped") else 0, *series_time_key(row))) cached = {"cached_at": time.time(), "study": study, "series": series_list, "files": file_map} STUDY_CACHE[ct_number] = cached return cached @app.get("/api/studies/{ct_number}/series") def series(ct_number: str, _: str = Depends(require_auth)) -> dict[str, Any]: data = scan_study(ct_number) return {"study": data["study"], "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="series not found") 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", 50) width = getattr(ds, "WindowWidth", 360) if isinstance(center, pydicom.multival.MultiValue): center = center[0] if isinstance(width, pydicom.multival.MultiValue): width = width[0] try: return float(center), float(width) except Exception: return 50.0, 360.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 as_float(value: Any, default: float = 1.0) -> float: try: return float(value) except Exception: return default def pixel_spacing_from_ds(ds: pydicom.Dataset) -> tuple[float, float]: spacing = getattr(ds, "PixelSpacing", None) if spacing and len(spacing) >= 2: row_spacing = as_float(spacing[0], 1.0) col_spacing = as_float(spacing[1], row_spacing) return max(row_spacing, 0.001), max(col_spacing, 0.001) return 1.0, 1.0 def slice_spacing_from_datasets(datasets: list[pydicom.Dataset]) -> float: distances = [] positions = [] for ds in datasets: position = getattr(ds, "ImagePositionPatient", None) if position and len(position) >= 3: positions.append(np.array([as_float(position[0]), as_float(position[1]), as_float(position[2])], dtype=np.float32)) for first, second in zip(positions, positions[1:]): distance = float(np.linalg.norm(second - first)) if distance > 0.001: distances.append(distance) if distances: return max(float(np.median(distances)), 0.001) locations = [] for ds in datasets: value = getattr(ds, "SliceLocation", None) if value is not None: locations.append(as_float(value, 0.0)) for first, second in zip(locations, locations[1:]): distance = abs(second - first) if distance > 0.001: distances.append(distance) if distances: return max(float(np.median(distances)), 0.001) sample = datasets[0] if datasets else None if sample is not None: spacing = as_float(getattr(sample, "SpacingBetweenSlices", 0), 0.0) if spacing > 0.001: return spacing thickness = as_float(getattr(sample, "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: row_spacing = max(row_spacing, 0.001) col_spacing = max(col_spacing, 0.001) if abs(row_spacing - col_spacing) < 0.01: return pil base = min(row_spacing, col_spacing) target_w = max(1, int(round(pil.width * col_spacing / base))) target_h = max(1, int(round(pil.height * row_spacing / base))) max_edge = 2200 scale = min(1.0, max_edge / max(target_w, target_h)) target_size = (max(1, int(round(target_w * scale))), max(1, int(round(target_h * scale)))) if target_size == pil.size: return pil return pil.resize(target_size, Image.Resampling.BILINEAR) def render_array( arr: np.ndarray, center: float, width: float, invert: bool = False, rotate: int = 0, max_size: int = 900, pixel_spacing: tuple[float, float] = (1.0, 1.0), ) -> 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, pixel_spacing[0], pixel_spacing[1]) if rotate: pil = pil.rotate(-rotate, expand=True) 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"{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, "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/image") def image( ct_number: str, series_uid: str, index: int = 0, plane: str = "axial", window: str = "default", rotate: int = 0, _: str = Depends(require_auth), ) -> Response: files = get_series_files(ct_number, series_uid) index = max(0, index) if plane == "axial" or len(files) < 2: index = min(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, getattr(ds, "PhotometricInterpretation", "") == "MONOCHROME1", rotate, pixel_spacing=pixel_spacing_from_ds(ds), ) return Response(payload, media_type="image/png") stack_data = load_stack_data(ct_number, series_uid) stack = stack_data["stack"] sample_ds = pydicom.dcmread(str(files[min(len(files) - 1, len(files) // 2)]), stop_before_pixels=True, force=True) center, width = window_values(sample_ds, window) if plane == "coronal": index = min(index, stack.shape[1] - 1) arr = stack[:, index, :] spacing = (stack_data["slice_spacing"], stack_data["col_spacing"]) elif plane == "sagittal": index = min(index, stack.shape[2] - 1) arr = stack[:, :, index] spacing = (stack_data["slice_spacing"], stack_data["row_spacing"]) else: raise HTTPException(status_code=400, detail="invalid plane") payload = render_array(np.flipud(arr), center, width, False, rotate, pixel_spacing=spacing) return Response(payload, media_type="image/png") @app.get("/api/dicom-info") def dicom_info(ct_number: str, series_uid: str, index: int = 0, _: str = Depends(require_auth)) -> dict[str, Any]: files = get_series_files(ct_number, series_uid) index = min(max(0, index), len(files) - 1) path = files[index] ds = pydicom.dcmread(str(path), stop_before_pixels=True, force=True) fields = { "patient": { "患者姓名": str(getattr(ds, "PatientName", "")), "患者ID": str(getattr(ds, "PatientID", "")), "检查号": str(getattr(ds, "AccessionNumber", "")), "检查日期": str(getattr(ds, "StudyDate", "")), "检查时间": str(getattr(ds, "StudyTime", "")), "设备厂商": str(getattr(ds, "Manufacturer", "")), }, "series": { "序列描述": str(getattr(ds, "SeriesDescription", "")), "序列号": str(getattr(ds, "SeriesNumber", "")), "文件数量": len(files), "当前文件": path.name, "DICOM路径": str(path), }, "image": { "Rows": str(getattr(ds, "Rows", "")), "Columns": str(getattr(ds, "Columns", "")), "BitsAllocated": str(getattr(ds, "BitsAllocated", "")), "WindowCenter": str(getattr(ds, "WindowCenter", "")), "WindowWidth": str(getattr(ds, "WindowWidth", "")), "Rescale": f"{getattr(ds, 'RescaleSlope', '')} / {getattr(ds, 'RescaleIntercept', '')}", }, "spacing": { "像素间距": str(getattr(ds, "PixelSpacing", "")), "切片厚度": str(getattr(ds, "SliceThickness", "")), "SpacingBetweenSlices": str(getattr(ds, "SpacingBetweenSlices", "")), "ImagePositionPatient": str(getattr(ds, "ImagePositionPatient", "")), }, } return {"path": str(path), "fields": fields} def sql_bool_or_null(value: bool | None) -> str: if value is None: return "NULL" return "true" if value else "false" def save_annotation_payload( ct_number: str, series_uid: str, series_row: dict[str, Any], manual_parts: list[str], ai_parts: list[str], manual_phase: str, ai_phase: str, manual_chest_window: str, ai_chest_window: str, manual_plain_ct: bool | None, ai_plain_ct: bool | None, skipped: bool, ai_skipped: bool, notes: str, user: str, ai_result: dict[str, Any] | None = None, ai_model: str | None = None, ) -> dict[str, Any]: manual_parts = valid_parts(manual_parts) ai_parts = valid_parts(ai_parts) body_parts = merge_parts(manual_parts, ai_parts, skipped) manual_phase = valid_phase(manual_phase) ai_phase = valid_phase(ai_phase) phase = effective_phase(manual_phase, ai_phase, body_parts, skipped) manual_chest_window = valid_chest_window(manual_chest_window) ai_chest_window = valid_chest_window(ai_chest_window) if "chest" not in body_parts: manual_chest_window = "" ai_chest_window = "" chest_window = effective_chest_window(manual_chest_window, ai_chest_window, body_parts, skipped) plain_ct = effective_plain_ct(manual_plain_ct, ai_plain_ct, False, skipped) ensure_annotation_table() pg_scalar( f""" INSERT INTO public.pacs_dicom_series_annotations ( ct_number, study_instance_uid, series_instance_uid, series_description, body_parts, manual_body_parts, ai_body_parts, upper_abdomen_phase, manual_upper_abdomen_phase, ai_upper_abdomen_phase, chest_window, manual_chest_window, ai_chest_window, plain_ct, manual_plain_ct, ai_plain_ct, skipped, ai_skipped, notes, ai_result, ai_model, updated_by, updated_at ) VALUES ( {sql_literal(ct_number)}, {sql_literal(series_row.get('study_uid', ''))}, {sql_literal(series_uid)}, {sql_literal(series_row.get('description', ''))}, {sql_literal(json.dumps(body_parts, ensure_ascii=False))}::jsonb, {sql_literal(json.dumps(manual_parts, ensure_ascii=False))}::jsonb, {sql_literal(json.dumps(ai_parts, ensure_ascii=False))}::jsonb, {sql_literal(phase)}, {sql_literal(manual_phase)}, {sql_literal(ai_phase)}, {sql_literal(chest_window)}, {sql_literal(manual_chest_window)}, {sql_literal(ai_chest_window)}, {'true' if plain_ct else 'false'}, {sql_bool_or_null(manual_plain_ct)}, {sql_bool_or_null(ai_plain_ct)}, {'true' if skipped else 'false'}, {'true' if ai_skipped else 'false'}, {sql_literal(notes)}, {sql_literal(json.dumps(ai_result, ensure_ascii=False)) + '::jsonb' if ai_result is not None else 'NULL'}, {sql_literal(ai_model) if ai_model is not None else 'NULL'}, {sql_literal(user)}, now() ) ON CONFLICT (ct_number, series_instance_uid) DO UPDATE SET study_instance_uid = EXCLUDED.study_instance_uid, series_description = EXCLUDED.series_description, body_parts = EXCLUDED.body_parts, manual_body_parts = EXCLUDED.manual_body_parts, ai_body_parts = EXCLUDED.ai_body_parts, upper_abdomen_phase = EXCLUDED.upper_abdomen_phase, manual_upper_abdomen_phase = EXCLUDED.manual_upper_abdomen_phase, ai_upper_abdomen_phase = EXCLUDED.ai_upper_abdomen_phase, chest_window = EXCLUDED.chest_window, manual_chest_window = EXCLUDED.manual_chest_window, ai_chest_window = EXCLUDED.ai_chest_window, plain_ct = EXCLUDED.plain_ct, manual_plain_ct = EXCLUDED.manual_plain_ct, ai_plain_ct = EXCLUDED.ai_plain_ct, skipped = EXCLUDED.skipped, ai_skipped = EXCLUDED.ai_skipped, notes = EXCLUDED.notes, ai_result = COALESCE(EXCLUDED.ai_result, pacs_dicom_series_annotations.ai_result), ai_model = COALESCE(EXCLUDED.ai_model, pacs_dicom_series_annotations.ai_model), updated_by = EXCLUDED.updated_by, updated_at = now() """ ) return normalize_annotation( { "body_parts": body_parts, "manual_body_parts": manual_parts, "ai_body_parts": ai_parts, "upper_abdomen_phase": phase, "manual_upper_abdomen_phase": manual_phase, "ai_upper_abdomen_phase": ai_phase, "chest_window": chest_window, "manual_chest_window": manual_chest_window, "ai_chest_window": ai_chest_window, "plain_ct": plain_ct, "manual_plain_ct": manual_plain_ct, "ai_plain_ct": ai_plain_ct, "skipped": skipped, "ai_skipped": ai_skipped, "notes": notes, "ai_model": ai_model or series_row.get("annotation", {}).get("ai_model", ""), } ) @app.put("/api/series/{ct_number}/{series_uid}/annotation") def save_annotation(ct_number: str, series_uid: str, data: AnnotationIn, user: str = Depends(require_auth)) -> dict[str, Any]: study = scan_study(ct_number) series_row = next((row for row in study["series"] if row["series_uid"] == series_uid), None) if not series_row: raise HTTPException(status_code=404, detail="series not found") annotation = save_annotation_payload( ct_number=ct_number, series_uid=series_uid, series_row=series_row, manual_parts=data.manual_body_parts or data.body_parts, ai_parts=data.ai_body_parts, manual_phase=data.manual_upper_abdomen_phase or data.upper_abdomen_phase, ai_phase=data.ai_upper_abdomen_phase, manual_chest_window=data.manual_chest_window or data.chest_window, ai_chest_window=data.ai_chest_window, manual_plain_ct=data.manual_plain_ct if data.manual_plain_ct is not None else (data.plain_ct if data.plain_ct and data.ai_plain_ct is None else None), ai_plain_ct=data.ai_plain_ct, skipped=data.skipped, ai_skipped=data.ai_skipped, notes=data.notes, user=user, ) STUDY_CACHE.pop(ct_number, None) return {"ok": True, **annotation} def representative_images(ct_number: str, series_uid: str) -> list[tuple[str, bytes]]: files = get_series_files(ct_number, series_uid) axial_index = max(0, min(len(files) - 1, len(files) // 2)) ds = pydicom.dcmread(str(files[axial_index]), force=True) center, width = window_values(ds, "soft") images = [ ( "轴位原始", render_array( dicom_to_hu(ds), center, width, getattr(ds, "PhotometricInterpretation", "") == "MONOCHROME1", max_size=720, pixel_spacing=pixel_spacing_from_ds(ds), ), ) ] if len(files) >= 3: try: stack_data = load_stack_data(ct_number, series_uid) stack = stack_data["stack"] sample_ds = pydicom.dcmread(str(files[axial_index]), stop_before_pixels=True, force=True) center, width = window_values(sample_ds, "soft") coronal = np.flipud(stack[:, stack.shape[1] // 2, :]) sagittal = np.flipud(stack[:, :, stack.shape[2] // 2]) images.append(("矢状位重建", render_array(sagittal, center, width, False, max_size=720, pixel_spacing=(stack_data["slice_spacing"], stack_data["row_spacing"])))) images.append(("冠状位重建", render_array(coronal, center, width, False, max_size=720, pixel_spacing=(stack_data["slice_spacing"], stack_data["col_spacing"])))) except Exception: pass return images def parse_ai_json(content: str) -> dict[str, Any]: text = content.strip() if text.startswith("```"): text = text.strip("`") if text.startswith("json"): text = text[4:] start = text.find("{") end = text.rfind("}") if start >= 0 and end > start: text = text[start : end + 1] try: return json.loads(text) except Exception: return {"body_parts": [], "upper_abdomen_phase": "", "chest_window": "", "plain_ct": False, "skipped": False, "notes": content[:800]} @app.post("/api/series/{ct_number}/{series_uid}/ai") def ai_classify(ct_number: str, series_uid: str, _: AIRequest, user: str = Depends(require_auth)) -> dict[str, Any]: if not KIMI_API_KEY: raise HTTPException(status_code=400, detail="KIMI_API_KEY 未配置") study = scan_study(ct_number) series_row = next((row for row in study["series"] if row["series_uid"] == series_uid), None) if not series_row: raise HTTPException(status_code=404, detail="series not found") content: list[dict[str, Any]] = [ { "type": "text", "text": ( "请根据这组CT序列的代表图像判断该序列所属部位。" "可选部位键: head_neck(头颈部), chest(胸部), upper_abdomen(上腹部), " "lower_abdomen(下腹部), pelvis(盆腔)。一个序列可包含多个部位。" "如果不是可用于标注的平扫CT影像、定位像、剂量报告或无法判断，请 skipped=true。" "请同时判断 plain_ct 是否为平扫CT。" "如果包含上腹部，请判断期相: arterial(动脉期)、portal_venous(门静脉期)、delayed(延迟期)、unknown(无法判别)。" "如果包含胸部，请判断窗位: lung(肺窗)、mediastinal(纵隔窗)、unknown(无法判别)。" "只返回JSON: {\"body_parts\":[],\"upper_abdomen_phase\":\"\",\"chest_window\":\"\",\"plain_ct\":false,\"skipped\":false,\"notes\":\"\"}。" f"PACS张数: {series_row.get('count', 0)}。" f"序列描述: {series_row.get('description','')}；DICOM部位: {series_row.get('body_part_dicom','')}。" ), } ] for label, png in representative_images(ct_number, series_uid): content.append({"type": "text", "text": label}) content.append({"type": "image_url", "image_url": {"url": "data:image/png;base64," + base64.b64encode(png).decode("ascii")}}) payload = { "model": KIMI_MODEL, "messages": [ {"role": "system", "content": "你是医学影像序列分拣助手，只做部位和期相粗分类，不输出诊断。"}, {"role": "user", "content": content}, ], "temperature": 0.1, } request = urllib.request.Request( KIMI_API_URL, data=json.dumps(payload, ensure_ascii=False).encode("utf-8"), headers={"Authorization": f"Bearer {KIMI_API_KEY}", "Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen(request, timeout=60) as response: raw = response.read().decode("utf-8") except urllib.error.HTTPError as exc: detail = exc.read().decode("utf-8", errors="replace")[:800] raise HTTPException(status_code=502, detail=f"Kimi API 错误: {detail}") from exc except Exception as exc: raise HTTPException(status_code=502, detail=f"Kimi API 调用失败: {exc}") from exc data = json.loads(raw) message = data.get("choices", [{}])[0].get("message", {}).get("content", "") suggestion = parse_ai_json(message) body_parts = [part for part in suggestion.get("body_parts", []) if part in BODY_PARTS] skipped = bool(suggestion.get("skipped", False)) plain_ct = bool(suggestion.get("plain_ct", False)) phase = suggestion.get("upper_abdomen_phase", "") if phase not in PHASES or "upper_abdomen" not in body_parts: phase = "" chest_window = valid_chest_window(suggestion.get("chest_window", "")) if "chest" not in body_parts: chest_window = "" elif not chest_window: chest_window = "unknown" current = normalize_annotation(series_row.get("annotation", {})) annotation = save_annotation_payload( ct_number=ct_number, series_uid=series_uid, series_row=series_row, manual_parts=current.get("manual_body_parts", []), ai_parts=[] if skipped else body_parts, manual_phase=current.get("manual_upper_abdomen_phase", ""), ai_phase="" if skipped else phase, manual_chest_window=current.get("manual_chest_window", ""), ai_chest_window="" if skipped else chest_window, manual_plain_ct=current.get("manual_plain_ct"), ai_plain_ct=None if skipped else plain_ct, skipped=skipped, ai_skipped=skipped, notes=str(suggestion.get("notes", "")) or current.get("notes", ""), user=user, ai_result=suggestion, ai_model=KIMI_MODEL, ) STUDY_CACHE.pop(ct_number, None) return { "ok": True, "provider": "Kimi", "name": KIMI_API_NAME, "model": KIMI_MODEL, **annotation, "raw": suggestion, } @app.get("/api/settings") def settings(_: str = Depends(require_auth)) -> dict[str, Any]: return { "users": web_users(), "roles": [{"name": name, "permissions": permissions} for name, permissions in ROLES.items()], "database": {"host": PGHOST, "port": PGPORT, "database": PGDATABASE, "table": PGTABLE}, "ai": {"provider": "Kimi", "name": KIMI_API_NAME, "model": KIMI_MODEL, "configured": bool(KIMI_API_KEY), "url": KIMI_API_URL}, "dicom": {"processed_root": str(PROCESSED_ROOT)}, } @app.post("/api/settings/users") def create_user(data: UserIn, user: str = Depends(require_auth)) -> dict[str, Any]: if user != WEB_USER: raise HTTPException(status_code=403, detail="仅管理员可创建用户") username = data.username.strip() if not username or len(username) > 64: raise HTTPException(status_code=400, detail="账号格式不正确") if len(data.password) < 6: raise HTTPException(status_code=400, detail="密码至少 6 位") role = data.role if data.role in ROLES else "阅片员" status = data.status if data.status in {"启用", "停用"} else "启用" ensure_user_table() exists = int(pg_scalar(f"SELECT count(*) FROM public.pacs_web_users WHERE username = {sql_literal(username)}") or "0") if exists: raise HTTPException(status_code=409, detail="账号已存在") pg_scalar( f""" INSERT INTO public.pacs_web_users (username, password_hash, role, status, updated_at) VALUES ({sql_literal(username)}, {sql_literal(password_hash(data.password))}, {sql_literal(role)}, {sql_literal(status)}, now()) """ ) return {"ok": True, "username": username, "role": role, "status": status} @app.get("/health") def health() -> Response: return Response("ok", media_type="text/plain")