From f3e3cfff1dc16ed6bf6ee3aba36dd95aa2a4707a Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 27 May 2026 00:54:14 +0800 Subject: [PATCH] Add PACS DICOM preprocessing workflow --- .gitignore | 7 + PACS_DICOM处理/README.md | 19 + .../preprocess_pacs_dicom_batch.py | 403 ++++++++++++++++++ 3 files changed, 429 insertions(+) create mode 100644 PACS_DICOM处理/README.md create mode 100644 PACS_DICOM处理/数据处理工作区/preprocess_pacs_dicom_batch.py diff --git a/.gitignore b/.gitignore index 669b7c8..a228445 100644 --- a/.gitignore +++ b/.gitignore @@ -21,3 +21,10 @@ UPP列表处理/数据处理工作区/06_PostgreSQL建表结构.sql UPP_STL处理/ UPP_数据库构建/UPP_STL资产同步报告.json UPP_数据库构建/UPP_STL文件family顺序明细.csv + +# PACS DICOM 实数据、软链接数据目录和批次处理结果不提交 +PACS_DICOM处理/待处理_DICOM数据 +PACS_DICOM处理/待处理_DICOM数据/ +PACS_DICOM处理/已处理_DICOM数据 +PACS_DICOM处理/已处理_DICOM数据/ +PACS_DICOM处理/数据处理结果区/ diff --git a/PACS_DICOM处理/README.md b/PACS_DICOM处理/README.md new file mode 100644 index 0000000..80a9c15 --- /dev/null +++ b/PACS_DICOM处理/README.md @@ -0,0 +1,19 @@ +# PACS DICOM 数据处理 + +本目录用于 PACS DICOM 批次数据的本地预处理和归档脚本管理。 + +## 目录约定 + +- `待处理_DICOM数据/`：本地待处理 DICOM 数据目录，通常为数据盘软链接，不提交到 Git。 +- `已处理_DICOM数据/`：本地已处理 DICOM 数据目录，通常为数据盘软链接，不提交到 Git。 +- `数据处理工作区/`：可提交的处理脚本、模板和说明。 +- `数据处理结果区/`：批次处理产生的清单、报告、SQL 输出等结果数据，不提交到 Git。 +- `数据处理网页端/`：预留给后续网页端工具。 + +## 当前预处理逻辑 + +`数据处理工作区/preprocess_pacs_dicom_batch.py` 会读取 DICOM 元数据中的 `AccessionNumber` 作为真实 `ct_number`，用于修正 PACS 导出顶层目录名中的检查号。 + +脚本会生成检查级清单、文件级清单和改名计划。实际落地已处理数据时，优先使用硬链接，避免在同一数据盘上重复占用整份 DICOM 空间。 + +数据目录和处理结果包含 DICOM 或患者相关信息，已在 `.gitignore` 中排除。 diff --git a/PACS_DICOM处理/数据处理工作区/preprocess_pacs_dicom_batch.py b/PACS_DICOM处理/数据处理工作区/preprocess_pacs_dicom_batch.py new file mode 100644 index 0000000..a5ba21f --- /dev/null +++ b/PACS_DICOM处理/数据处理工作区/preprocess_pacs_dicom_batch.py @@ -0,0 +1,403 @@ +#!/usr/bin/env python3 +"""Preprocess a PACS DICOM batch by normalizing study folders. + +The top-level PACS export folder is assumed to contain one folder per study. +The exported folder name may contain the wrong CT number, so the canonical +ct_number is read from DICOM AccessionNumber (0008,0050). +""" + +from __future__ import annotations + +import argparse +import csv +import json +import os +import re +import shutil +import sys +from collections import Counter, defaultdict +from dataclasses import asdict, dataclass +from datetime import datetime +from pathlib import Path +from typing import Iterable + +import pydicom + + +DICOM_TAGS = [ + "AccessionNumber", + "PatientName", + "PatientID", + "PatientBirthDate", + "PatientSex", + "StudyInstanceUID", + "StudyDate", + "StudyTime", + "StudyID", + "Modality", + "BodyPartExamined", + "ProtocolName", + "StudyDescription", + "SeriesInstanceUID", + "SOPInstanceUID", +] + + +@dataclass +class StudyRow: + batch_name: str + source_folder_name: str + source_ct_number: str + source_patient_name: str + ct_number: str + target_folder_name: str + needs_ct_number_fix: bool + patient_name_dicom: str + patient_id: str + patient_birth_date: str + patient_sex: str + study_date: str + study_time: str + study_id: str + modality: str + body_part_examined: str + protocol_name: str + study_description: str + accession_numbers: str + raw_accession_numbers: str + study_instance_uids: str + series_count: int + dicom_file_count: int + total_file_count: int + total_bytes: int + source_path: str + processed_path: str + status: str + notes: str + + +def text(value: object) -> str: + if value is None: + return "" + return str(value).strip() + + +def most_common(counter: Counter[str]) -> str: + if not counter: + return "" + return counter.most_common(1)[0][0] + + +def sanitize_component(name: str) -> str: + cleaned = name.replace("/", "_").replace("\x00", "_").strip() + return cleaned or "UNKNOWN" + + +def parse_export_folder_name(name: str) -> tuple[str, str]: + if "-" not in name: + return name, "" + ct_number, patient_name = name.split("-", 1) + return ct_number.strip(), patient_name.strip() + + +def canonical_ct_number(accession_number: str) -> str: + accession_number = accession_number.strip() + match = re.fullmatch(r"((?:D)?CT\d+)-\d+", accession_number) + if match: + return match.group(1) + return accession_number + + +def iter_files(folder: Path) -> Iterable[Path]: + for path in folder.rglob("*"): + if path.is_file(): + yield path + + +def read_meta(path: Path) -> dict[str, str]: + ds = pydicom.dcmread( + str(path), + stop_before_pixels=True, + force=True, + specific_tags=DICOM_TAGS + ["SpecificCharacterSet"], + ) + return {tag: text(getattr(ds, tag, "")) for tag in DICOM_TAGS} + + +def scan_study_folder(batch_name: str, folder: Path, processed_batch_root: Path) -> tuple[StudyRow, list[dict[str, str]]]: + source_ct_number, source_patient_name = parse_export_folder_name(folder.name) + files = sorted(iter_files(folder)) + + counters: dict[str, Counter[str]] = defaultdict(Counter) + file_rows: list[dict[str, str]] = [] + dicom_file_count = 0 + total_bytes = 0 + errors = [] + + for path in files: + try: + size = path.stat().st_size + total_bytes += size + meta = read_meta(path) + dicom_file_count += 1 + except Exception as exc: # noqa: BLE001 - keep scanning and report the file. + errors.append(f"{path}: {exc}") + continue + + for key in DICOM_TAGS: + value = meta.get(key, "") + if value: + counters[key][value] += 1 + + file_rows.append( + { + "ct_number": "", + "source_folder_name": folder.name, + "source_relative_path": str(path.relative_to(folder)), + "processed_relative_path": "", + "sop_instance_uid": meta.get("SOPInstanceUID", ""), + "series_instance_uid": meta.get("SeriesInstanceUID", ""), + "study_instance_uid": meta.get("StudyInstanceUID", ""), + "bytes": str(size), + } + ) + + raw_accession_numbers = sorted(counters["AccessionNumber"]) + accession_numbers = sorted({canonical_ct_number(value) for value in raw_accession_numbers if value}) + study_instance_uids = sorted(counters["StudyInstanceUID"]) + + status = "ok" + notes: list[str] = [] + if errors: + status = "error" + notes.append(f"metadata_read_errors={len(errors)}") + if raw_accession_numbers and raw_accession_numbers != accession_numbers: + notes.append("normalized_accession_suffixes") + if not accession_numbers: + status = "error" + notes.append("missing_accession_number") + ct_number = source_ct_number + elif len(accession_numbers) > 1: + status = "error" + notes.append("multiple_accession_numbers") + ct_number = most_common(counters["AccessionNumber"]) + else: + ct_number = accession_numbers[0] + + target_folder_name = sanitize_component(ct_number) + if source_patient_name: + target_folder_name = f"{target_folder_name}-{sanitize_component(source_patient_name)}" + processed_path = processed_batch_root / target_folder_name + + needs_ct_number_fix = source_ct_number != ct_number + + for row in file_rows: + row["ct_number"] = ct_number + row["processed_relative_path"] = str(Path(target_folder_name) / row["source_relative_path"]) + + row = StudyRow( + batch_name=batch_name, + source_folder_name=folder.name, + source_ct_number=source_ct_number, + source_patient_name=source_patient_name, + ct_number=ct_number, + target_folder_name=target_folder_name, + needs_ct_number_fix=needs_ct_number_fix, + patient_name_dicom=most_common(counters["PatientName"]), + patient_id=most_common(counters["PatientID"]), + patient_birth_date=most_common(counters["PatientBirthDate"]), + patient_sex=most_common(counters["PatientSex"]), + study_date=most_common(counters["StudyDate"]), + study_time=most_common(counters["StudyTime"]), + study_id=most_common(counters["StudyID"]), + modality=most_common(counters["Modality"]), + body_part_examined=most_common(counters["BodyPartExamined"]), + protocol_name=most_common(counters["ProtocolName"]), + study_description=most_common(counters["StudyDescription"]), + accession_numbers="|".join(accession_numbers), + raw_accession_numbers="|".join(raw_accession_numbers), + study_instance_uids="|".join(study_instance_uids), + series_count=len(counters["SeriesInstanceUID"]), + dicom_file_count=dicom_file_count, + total_file_count=len(files), + total_bytes=total_bytes, + source_path=str(folder), + processed_path=str(processed_path), + status=status, + notes=";".join(notes), + ) + return row, file_rows + + +def write_csv(path: Path, rows: list[dict[str, object]], fieldnames: list[str]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("w", newline="", encoding="utf-8") as fh: + writer = csv.DictWriter(fh, fieldnames=fieldnames) + writer.writeheader() + writer.writerows(rows) + + +def hardlink_tree(source_folder: Path, target_folder: Path) -> tuple[int, int]: + linked = 0 + already_ok = 0 + for source in iter_files(source_folder): + rel = source.relative_to(source_folder) + target = target_folder / rel + target.parent.mkdir(parents=True, exist_ok=True) + if target.exists(): + try: + if os.path.samefile(source, target): + already_ok += 1 + continue + except OSError: + pass + if target.stat().st_size != source.stat().st_size: + raise RuntimeError(f"target exists with different size: {target}") + target.unlink() + try: + os.link(source, target) + except OSError as exc: + if exc.errno != 18: # EXDEV + raise + shutil.copy2(source, target) + linked += 1 + return linked, already_ok + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--source", type=Path, required=True) + parser.add_argument("--processed-root", type=Path, required=True) + parser.add_argument("--results-root", type=Path, required=True) + parser.add_argument("--batch-name", default="") + parser.add_argument("--apply", action="store_true", help="create processed hardlink tree") + parser.add_argument("--write-file-manifest", action="store_true") + args = parser.parse_args() + + source_root = args.source.resolve() + batch_name = args.batch_name or source_root.name + processed_batch_root = (args.processed_root / batch_name).resolve() + result_dir = (args.results_root / batch_name).resolve() + result_dir.mkdir(parents=True, exist_ok=True) + + started_at = datetime.now().isoformat(timespec="seconds") + study_folders = sorted([p for p in source_root.iterdir() if p.is_dir()]) + study_rows: list[StudyRow] = [] + file_rows_all: list[dict[str, str]] = [] + + for index, folder in enumerate(study_folders, start=1): + print(f"[{index}/{len(study_folders)}] scan {folder.name}", flush=True) + study_row, file_rows = scan_study_folder(batch_name, folder, processed_batch_root) + study_rows.append(study_row) + if args.write_file_manifest: + file_rows_all.extend(file_rows) + + ct_counts = Counter(row.ct_number for row in study_rows if row.ct_number) + target_counts = Counter(row.target_folder_name for row in study_rows if row.target_folder_name) + duplicate_ct_numbers = sorted([ct for ct, count in ct_counts.items() if count > 1]) + duplicate_target_folders = sorted([name for name, count in target_counts.items() if count > 1]) + + for row in study_rows: + notes = [row.notes] if row.notes else [] + if row.ct_number in duplicate_ct_numbers: + row.status = "error" + notes.append("duplicate_ct_number") + if row.target_folder_name in duplicate_target_folders: + row.status = "error" + notes.append("duplicate_target_folder") + row.notes = ";".join([note for note in notes if note]) + + study_dicts = [asdict(row) for row in study_rows] + study_fieldnames = list(asdict(study_rows[0]).keys()) if study_rows else list(StudyRow.__dataclass_fields__) + + write_csv(result_dir / "study_manifest.csv", study_dicts, study_fieldnames) + write_csv( + result_dir / "rename_plan.csv", + [ + { + "source_folder_name": row.source_folder_name, + "source_ct_number": row.source_ct_number, + "ct_number": row.ct_number, + "target_folder_name": row.target_folder_name, + "needs_ct_number_fix": row.needs_ct_number_fix, + "status": row.status, + "notes": row.notes, + } + for row in study_rows + ], + [ + "source_folder_name", + "source_ct_number", + "ct_number", + "target_folder_name", + "needs_ct_number_fix", + "status", + "notes", + ], + ) + if args.write_file_manifest: + write_csv( + result_dir / "file_manifest.csv", + file_rows_all, + [ + "ct_number", + "source_folder_name", + "source_relative_path", + "processed_relative_path", + "sop_instance_uid", + "series_instance_uid", + "study_instance_uid", + "bytes", + ], + ) + + ok_rows = [row for row in study_rows if row.status == "ok"] + error_rows = [row for row in study_rows if row.status != "ok"] + summary = { + "batch_name": batch_name, + "source_root": str(source_root), + "processed_batch_root": str(processed_batch_root), + "result_dir": str(result_dir), + "started_at": started_at, + "finished_scan_at": datetime.now().isoformat(timespec="seconds"), + "apply": args.apply, + "source_study_folder_count": len(study_folders), + "ok_study_count": len(ok_rows), + "error_study_count": len(error_rows), + "needs_ct_number_fix_count": sum(1 for row in study_rows if row.needs_ct_number_fix), + "duplicate_ct_numbers": duplicate_ct_numbers, + "duplicate_target_folders": duplicate_target_folders, + "dicom_file_count": sum(row.dicom_file_count for row in study_rows), + "total_file_count": sum(row.total_file_count for row in study_rows), + "total_bytes": sum(row.total_bytes for row in study_rows), + "linked_file_count": 0, + "already_linked_file_count": 0, + } + + if error_rows: + summary_path = result_dir / "summary.json" + summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8") + print(f"ERROR: found {len(error_rows)} invalid study rows; see {summary_path}", file=sys.stderr) + return 2 + + if args.apply: + processed_batch_root.mkdir(parents=True, exist_ok=True) + linked_total = 0 + already_total = 0 + for index, row in enumerate(ok_rows, start=1): + print(f"[{index}/{len(ok_rows)}] link {row.source_folder_name} -> {row.target_folder_name}", flush=True) + linked, already_ok = hardlink_tree(Path(row.source_path), Path(row.processed_path)) + linked_total += linked + already_total += already_ok + summary["linked_file_count"] = linked_total + summary["already_linked_file_count"] = already_total + summary["finished_apply_at"] = datetime.now().isoformat(timespec="seconds") + + summary_path = result_dir / "summary.json" + summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8") + print(json.dumps(summary, ensure_ascii=False, indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())