Add admin DICOM export controls

PACS_DICOM处理/数据处理网页端/app.py

@@ -8,10 +8,12 @@ import json
 import os
 import secrets
 import subprocess
+import tempfile
 import threading
 import time
 import urllib.error
 import urllib.request
+import zipfile
 from collections import defaultdict
 from datetime import datetime, timedelta
 from pathlib import Path
@@ -24,6 +26,7 @@ from fastapi.responses import FileResponse
 from fastapi.staticfiles import StaticFiles
 from pydantic import BaseModel
 from PIL import Image
+from starlette.background import BackgroundTask
 
 
 APP_DIR = Path(__file__).resolve().parent
@@ -70,7 +73,7 @@ BODY_PARTS = set(BODY_PART_ORDER)
 PHASES = {"plain", "arterial", "portal_venous", "delayed", "unknown", ""}
 CHEST_WINDOWS = {"lung", "mediastinal", "unknown", ""}
 ROLES = {
-    "管理员": ["查看DICOM", "编辑标注", "AI识别", "用户创建", "权限控制", "系统设置"],
+    "管理员": ["查看DICOM", "编辑标注", "AI识别", "影像导出", "批量导出", "用户创建", "权限控制", "系统设置"],
     "阅片员": ["查看DICOM", "编辑标注", "AI识别"],
     "访客": ["查看DICOM"],
 }
@@ -121,6 +124,10 @@ class StudyCompletionIn(BaseModel):
     completed: bool = False
 
 
+class ExportStudiesIn(BaseModel):
+    ct_numbers: list[str] = []
+
+
 class UserIn(BaseModel):
     username: str
     password: str
@@ -619,6 +626,85 @@ def resolve_study_root(study: dict[str, Any]) -> Path:
     return root
 
 
+def safe_archive_name(value: Any, fallback: str = "study") -> str:
+    text = str(value or "").strip()
+    cleaned = "".join("_" if char in '<>:"/\\|?*\0' else char for char in text)
+    cleaned = cleaned.strip().strip(".")
+    return cleaned[:160] or fallback
+
+
+def study_export_folder_name(study: dict[str, Any]) -> str:
+    ct_number = safe_archive_name(study.get("ct_number"), "study")
+    name = safe_archive_name(study.get("source_patient_name") or study.get("patient_name_dicom") or "", "")
+    return f"{ct_number}_{name}" if name else ct_number
+
+
+def add_directory_to_zip(zip_file: zipfile.ZipFile, root: Path, prefix: str) -> int:
+    root = root.resolve()
+    count = 0
+    for path in root.rglob("*"):
+        if not path.is_file():
+            continue
+        try:
+            relative = path.resolve().relative_to(root)
+        except ValueError:
+            continue
+        zip_file.write(path, Path(prefix) / relative)
+        count += 1
+    return count
+
+
+def remove_file(path: str) -> None:
+    try:
+        Path(path).unlink(missing_ok=True)
+    except Exception:
+        pass
+
+
+def build_study_export_zip(ct_numbers: list[str]) -> tuple[str, str]:
+    unique_numbers = []
+    for ct_number in ct_numbers:
+        value = str(ct_number or "").strip()
+        if value and value not in unique_numbers:
+            unique_numbers.append(value)
+    if not unique_numbers:
+        raise HTTPException(status_code=400, detail="请选择需要导出的检查")
+    if len(unique_numbers) > 200:
+        raise HTTPException(status_code=400, detail="单次最多导出 200 个检查")
+
+    temp = tempfile.NamedTemporaryFile(prefix="pacs_dicom_export_", suffix=".zip", delete=False)
+    temp_path = temp.name
+    temp.close()
+    total_files = 0
+    notes = []
+    try:
+        with zipfile.ZipFile(temp_path, mode="w", compression=zipfile.ZIP_STORED, allowZip64=True) as zip_file:
+            for ct_number in unique_numbers:
+                study = get_study_record(ct_number)
+                root = resolve_study_root(study)
+                if not root.exists():
+                    notes.append(f"{ct_number}: DICOM 路径不存在 - {root}")
+                    continue
+                folder = study_export_folder_name(study)
+                file_count = add_directory_to_zip(zip_file, root, folder)
+                total_files += file_count
+                if file_count == 0:
+                    notes.append(f"{ct_number}: 未找到可导出的文件 - {root}")
+            if notes:
+                zip_file.writestr("导出说明.txt", "\n".join(notes))
+        if total_files == 0:
+            remove_file(temp_path)
+            raise HTTPException(status_code=404, detail="没有找到可导出的 DICOM 文件")
+    except HTTPException:
+        raise
+    except Exception as exc:
+        remove_file(temp_path)
+        raise HTTPException(status_code=500, detail=f"导出失败：{exc}") from exc
+
+    filename = f"PACS_DICOM_{safe_archive_name(unique_numbers[0])}.zip" if len(unique_numbers) == 1 else f"PACS_DICOM_batch_{time.strftime('%Y%m%d_%H%M%S')}.zip"
+    return temp_path, filename
+
+
 def read_header(path: Path) -> dict[str, str]:
     tags = [
         "SeriesInstanceUID",
@@ -1129,6 +1215,24 @@ def update_study_completion(ct_number: str, data: StudyCompletionIn, user: str =
     return {"ok": True, "ct_number": ct_number, **study_completion_fields(ct_number)}
 
 
+@app.get("/api/studies/export")
+def export_studies_get(ct_numbers: list[str] = Query(default=[]), _: str = Depends(require_admin)) -> FileResponse:
+    path, filename = build_study_export_zip(ct_numbers)
+    return FileResponse(path, media_type="application/zip", filename=filename, background=BackgroundTask(remove_file, path))
+
+
+@app.get("/api/studies/{ct_number}/export")
+def export_study(ct_number: str, _: str = Depends(require_admin)) -> FileResponse:
+    path, filename = build_study_export_zip([ct_number])
+    return FileResponse(path, media_type="application/zip", filename=filename, background=BackgroundTask(remove_file, path))
+
+
+@app.post("/api/studies/export")
+def export_studies(data: ExportStudiesIn, _: str = Depends(require_admin)) -> FileResponse:
+    path, filename = build_study_export_zip(data.ct_numbers)
+    return FileResponse(path, media_type="application/zip", filename=filename, background=BackgroundTask(remove_file, path))
+
+
 def get_series_files(ct_number: str, series_uid: str) -> list[Path]:
     data = scan_study(ct_number)
     files = data["files"].get(series_uid)