admin/PACS
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Export selected DICOM series

Browse Source
This commit is contained in:
Codex
2026-05-27 14:43:17 +08:00
parent 223e853bad
commit 2a858be166
4 changed files with 305 additions and 162 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -72,8 +72,17 @@ BODY_PART_ORDER = ["head_neck", "chest", "upper_abdomen", "lower_abdomen", "pelv
BODY_PARTS = set(BODY_PART_ORDER)
PHASES = {"plain", "arterial", "portal_venous", "delayed", "unknown", ""}
CHEST_WINDOWS = {"lung", "mediastinal", "unknown", ""}
BODY_PART_LABELS = {
"head_neck": "头颈部",
"chest": "胸部",
"upper_abdomen": "上腹部",
"lower_abdomen": "下腹部",
"pelvis": "盆腔",
}
PHASE_LABELS = {"plain": "平扫", "arterial": "动脉期", "portal_venous": "门静脉期", "delayed": "延迟期", "unknown": "无法判别"}
CHEST_WINDOW_LABELS = {"lung": "肺窗", "mediastinal": "纵隔窗", "unknown": "无法判别"}
ROLES = {
"管理员": ["查看DICOM", "编辑标注", "AI识别", "影像导出", "批量导出", "用户创建", "权限控制", "系统设置"],
"管理员": ["查看DICOM", "编辑标注", "AI识别", "影像导出", "用户创建", "权限控制", "系统设置"],
"阅片员": ["查看DICOM", "编辑标注", "AI识别"],
"访客": ["查看DICOM"],
}
@@ -124,10 +133,6 @@ class StudyCompletionIn(BaseModel):
completed: bool = False
class ExportStudiesIn(BaseModel):
ct_numbers: list[str] = []
class UserIn(BaseModel):
username: str
password: str
@@ -639,19 +644,28 @@ def study_export_folder_name(study: dict[str, Any]) -> str:
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 series_export_labels(annotation: dict[str, Any]) -> list[str]:
labels: list[str] = []
body_parts = valid_parts(annotation.get("body_parts") or [])
phase = valid_phase(annotation.get("upper_abdomen_phase") or "")
chest_window = valid_chest_window(annotation.get("chest_window") or "")
for part in body_parts:
if part == "upper_abdomen" and phase:
labels.append(f"{BODY_PART_LABELS[part]}-{PHASE_LABELS.get(phase, phase)}")
elif part == "chest" and chest_window:
labels.append(f"{BODY_PART_LABELS[part]}-{CHEST_WINDOW_LABELS.get(chest_window, chest_window)}")
else:
labels.append(BODY_PART_LABELS.get(part, part))
if annotation.get("skipped"):
labels.append("略过_不采用")
return labels or ["未标注"]
def series_export_folder_name(series: dict[str, Any], index: int) -> str:
number = safe_archive_name(series.get("series_number") or index, str(index))
description = safe_archive_name(series.get("description") or "未命名序列", "未命名序列")
labels = "_".join(safe_archive_name(label, "未标注") for label in series_export_labels(series.get("annotation") or {}))
return safe_archive_name(f"序列{number}_{description}_{labels}", f"序列{index}_{labels}")
def remove_file(path: str) -> None:
@@ -661,35 +675,42 @@ def remove_file(path: str) -> None:
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 个检查")
def build_series_export_zip(ct_number: str, series_uids: list[str]) -> tuple[str, str]:
selected_uids = []
for uid in series_uids:
value = str(uid or "").strip()
if value and value not in selected_uids:
selected_uids.append(value)
if not selected_uids:
raise HTTPException(status_code=400, detail="请选择要导出的序列")
if len(selected_uids) > 200:
raise HTTPException(status_code=400, detail="单次最多导出 200 个序列")
temp = tempfile.NamedTemporaryFile(prefix="pacs_dicom_export_", suffix=".zip", delete=False)
data = scan_study(ct_number)
study_folder = study_export_folder_name(data["study"])
series_by_uid = {row["series_uid"]: row for row in data["series"]}
temp = tempfile.NamedTemporaryFile(prefix="pacs_dicom_series_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}")
for order, uid in enumerate(selected_uids, start=1):
series = series_by_uid.get(uid)
files = data["files"].get(uid) or []
if not series or not files:
notes.append(f"{uid}: 序列不存在或没有文件")
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}")
sequence_folder = series_export_folder_name(series, order)
prefix = Path(study_folder) / sequence_folder
for file_index, path in enumerate(files, start=1):
if not path.exists() or not path.is_file():
notes.append(f"{uid}: 文件不存在 - {path}")
continue
archive_name = Path(prefix) / f"{file_index:05d}_{safe_archive_name(path.name, 'image.dcm')}"
zip_file.write(path, str(archive_name))
total_files += 1
if notes:
zip_file.writestr("导出说明.txt", "\n".join(notes))
if total_files == 0:
@@ -701,7 +722,7 @@ def build_study_export_zip(ct_numbers: list[str]) -> tuple[str, str]:
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"
filename = f"PACS_DICOM_{safe_archive_name(ct_number)}_{time.strftime('%Y%m%d_%H%M%S')}.zip"
return temp_path, filename
@@ -1215,21 +1236,9 @@ 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)
def export_study(ct_number: str, series_uids: list[str] = Query(default=[]), _: str = Depends(require_admin)) -> FileResponse:
path, filename = build_series_export_zip(ct_number, series_uids)
return FileResponse(path, media_type="application/zip", filename=filename, background=BackgroundTask(remove_file, path))