Add Docker PACS DICOM web viewer
This commit is contained in:
Codex
866
PACS_DICOM处理/数据处理网页端/app.py
Normal file
866
PACS_DICOM处理/数据处理网页端/app.py
Normal file
@@ -0,0 +1,866 @@
|
||||
#!/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", "unknown", ""}
|
||||
ROLES = {
|
||||
"管理员": ["查看DICOM", "编辑标注", "AI识别", "用户创建", "权限控制", "系统设置"],
|
||||
"阅片员": ["查看DICOM", "编辑标注", "AI识别"],
|
||||
"访客": ["查看DICOM"],
|
||||
}
|
||||
|
||||
app = FastAPI(title="PACS DICOM Viewer")
|
||||
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, np.ndarray]] = {}
|
||||
|
||||
|
||||
class LoginIn(BaseModel):
|
||||
username: str
|
||||
password: str
|
||||
|
||||
|
||||
class AnnotationIn(BaseModel):
|
||||
body_parts: list[str] = []
|
||||
upper_abdomen_phase: str = ""
|
||||
notes: str = ""
|
||||
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,
|
||||
upper_abdomen_phase text NOT NULL DEFAULT '',
|
||||
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 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;
|
||||
"""
|
||||
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,
|
||||
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 NOT TRUE AND jsonb_array_length(body_parts) > 0
|
||||
GROUP BY ct_number
|
||||
""",
|
||||
timeout=8,
|
||||
)
|
||||
annotation_map = {row["ct_number"]: row["annotated_series"] for row in annotations}
|
||||
for row in rows:
|
||||
row["annotated_series"] = annotation_map.get(row["ct_number"], 0)
|
||||
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",
|
||||
"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 get_annotations(ct_number: str) -> dict[str, dict[str, Any]]:
|
||||
try:
|
||||
rows = pg_json_rows(
|
||||
f"""
|
||||
SELECT series_instance_uid, body_parts, upper_abdomen_phase, skipped, notes, updated_at, ai_model
|
||||
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 = {}
|
||||
for uid, items in grouped.items():
|
||||
items.sort(key=sort_key)
|
||||
first = items[0][1]
|
||||
last = items[-1][1]
|
||||
file_map[uid] = [path for path, _ in items]
|
||||
annotation = annotations.get(uid, {})
|
||||
series_list.append(
|
||||
{
|
||||
"ct_number": ct_number,
|
||||
"series_uid": uid,
|
||||
"study_uid": first.get("StudyInstanceUID", ""),
|
||||
"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", ""),
|
||||
"annotation": {
|
||||
"body_parts": annotation.get("body_parts", []),
|
||||
"upper_abdomen_phase": annotation.get("upper_abdomen_phase", ""),
|
||||
"skipped": bool(annotation.get("skipped", False)),
|
||||
"notes": annotation.get("notes", ""),
|
||||
"updated_at": annotation.get("updated_at", ""),
|
||||
"ai_model": annotation.get("ai_model", ""),
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
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, numeric(row["series_number"]), row["description"]))
|
||||
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 render_array(arr: np.ndarray, center: float, width: float, invert: bool = False, rotate: int = 0, max_size: int = 900) -> 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)
|
||||
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(ct_number: str, series_uid: str) -> np.ndarray:
|
||||
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 = []
|
||||
for path in files:
|
||||
ds = pydicom.dcmread(str(path), force=True)
|
||||
arrays.append(dicom_to_hu(ds))
|
||||
stack = np.stack(arrays, axis=0)
|
||||
STACK_CACHE[key] = (time.time(), stack)
|
||||
if len(STACK_CACHE) > 2:
|
||||
oldest = sorted(STACK_CACHE.items(), key=lambda item: item[1][0])[0][0]
|
||||
STACK_CACHE.pop(oldest, None)
|
||||
return stack
|
||||
|
||||
|
||||
@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)
|
||||
return Response(payload, media_type="image/png")
|
||||
|
||||
stack = load_stack(ct_number, series_uid)
|
||||
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, :]
|
||||
elif plane == "sagittal":
|
||||
index = min(index, stack.shape[2] - 1)
|
||||
arr = stack[:, :, index]
|
||||
else:
|
||||
raise HTTPException(status_code=400, detail="invalid plane")
|
||||
payload = render_array(np.flipud(arr), center, width, False, rotate)
|
||||
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}
|
||||
|
||||
|
||||
@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]:
|
||||
body_parts = [] if data.skipped else [part for part in data.body_parts if part in BODY_PARTS]
|
||||
phase = data.upper_abdomen_phase if data.upper_abdomen_phase in PHASES else ""
|
||||
if "upper_abdomen" not in body_parts:
|
||||
phase = ""
|
||||
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")
|
||||
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, upper_abdomen_phase, skipped, notes, 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(phase)},
|
||||
{'true' if data.skipped else 'false'},
|
||||
{sql_literal(data.notes)},
|
||||
{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,
|
||||
upper_abdomen_phase = EXCLUDED.upper_abdomen_phase,
|
||||
skipped = EXCLUDED.skipped,
|
||||
notes = EXCLUDED.notes,
|
||||
updated_by = EXCLUDED.updated_by,
|
||||
updated_at = now()
|
||||
"""
|
||||
)
|
||||
STUDY_CACHE.pop(ct_number, None)
|
||||
return {"ok": True, "body_parts": body_parts, "upper_abdomen_phase": phase, "skipped": data.skipped}
|
||||
|
||||
|
||||
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))]
|
||||
|
||||
if len(files) >= 3:
|
||||
try:
|
||||
stack = load_stack(ct_number, series_uid)
|
||||
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)))
|
||||
images.append(("冠状面", render_array(coronal, center, width, False, max_size=720)))
|
||||
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": "", "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。"
|
||||
"如果包含上腹部，请判断期相: arterial(动脉期)、portal_venous(门静脉期)、unknown(无法判别)。"
|
||||
"只返回JSON: {\"body_parts\":[],\"upper_abdomen_phase\":\"\",\"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))
|
||||
phase = suggestion.get("upper_abdomen_phase", "")
|
||||
if phase not in PHASES or "upper_abdomen" not in body_parts:
|
||||
phase = ""
|
||||
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, upper_abdomen_phase, 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', ''))},
|
||||
'[]'::jsonb,
|
||||
'',
|
||||
false,
|
||||
'',
|
||||
{sql_literal(json.dumps(suggestion, ensure_ascii=False))}::jsonb,
|
||||
{sql_literal(KIMI_MODEL)},
|
||||
{sql_literal(user)},
|
||||
now()
|
||||
)
|
||||
ON CONFLICT (ct_number, series_instance_uid) DO UPDATE SET
|
||||
ai_result = EXCLUDED.ai_result,
|
||||
ai_model = EXCLUDED.ai_model,
|
||||
updated_by = EXCLUDED.updated_by,
|
||||
updated_at = now()
|
||||
"""
|
||||
)
|
||||
STUDY_CACHE.pop(ct_number, None)
|
||||
return {
|
||||
"ok": True,
|
||||
"provider": "Kimi",
|
||||
"name": KIMI_API_NAME,
|
||||
"model": KIMI_MODEL,
|
||||
"body_parts": [] if skipped else body_parts,
|
||||
"upper_abdomen_phase": "" if skipped else phase,
|
||||
"skipped": skipped,
|
||||
"notes": str(suggestion.get("notes", "")),
|
||||
"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")
|
||||
Reference in New Issue
Block a user