#!/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())