#!/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 errno 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", ] CT_NUMBER_RE = re.compile(r"\bD?CT\d{6,}\b", re.IGNORECASE) @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 normalize_ct_text(value: str) -> str: return value.strip().upper() def extract_ct_numbers(*values: object) -> list[str]: candidates: list[str] = [] for value in values: for match in CT_NUMBER_RE.findall(text(value)): normalized = normalize_ct_text(match) if normalized not in candidates: candidates.append(normalized) return candidates def canonical_ct_number(accession_number: str) -> str: accession_number = normalize_ct_text(accession_number) 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}) fallback_ct_numbers = extract_ct_numbers( source_ct_number, folder.name, *counters["StudyID"].keys(), *counters["AccessionNumber"].keys(), *counters["PatientID"].keys(), ) 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: if fallback_ct_numbers: notes.append("missing_accession_number_used_folder_or_dicom_fallback") ct_number = fallback_ct_numbers[0] else: 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: already_ok += 1 continue else: raise RuntimeError(f"target exists with different size: {target}") try: os.link(source, target) except OSError as exc: if exc.errno not in {errno.EXDEV, errno.EPERM, errno.EOPNOTSUPP}: raise shutil.copyfile(source, target) try: shutil.copystat(source, target) except OSError as stat_exc: if stat_exc.errno not in {errno.EPERM, errno.EOPNOTSUPP}: raise linked += 1 return linked, already_ok def write_sync_sql(path: Path, manifest_path: Path) -> None: sql = f"""BEGIN; CREATE TABLE IF NOT EXISTS public.pacs_dicom_files ( ct_number text PRIMARY KEY, batch_name text NOT NULL, source_folder_name text, source_ct_number text, source_patient_name text, target_folder_name text, needs_ct_number_fix boolean NOT NULL DEFAULT false, patient_name_dicom text, patient_id text, patient_birth_date text, patient_sex text, study_date text, study_time text, study_id text, modality text, body_part_examined text, protocol_name text, study_description text, accession_numbers text, raw_accession_numbers text, study_instance_uids text, series_count integer, dicom_file_count integer, total_file_count integer, total_bytes bigint, source_path text, processed_path text, status text, notes text, created_at timestamptz NOT NULL DEFAULT now(), updated_at timestamptz NOT NULL DEFAULT now() ); CREATE TEMP TABLE pacs_dicom_files_stage ( batch_name text, source_folder_name text, source_ct_number text, source_patient_name text, ct_number text, target_folder_name text, needs_ct_number_fix text, patient_name_dicom text, patient_id text, patient_birth_date text, patient_sex text, study_date text, study_time text, study_id text, modality text, body_part_examined text, protocol_name text, study_description text, accession_numbers text, raw_accession_numbers text, study_instance_uids text, series_count text, dicom_file_count text, total_file_count text, total_bytes text, source_path text, processed_path text, status text, notes text ) ON COMMIT DROP; \\copy pacs_dicom_files_stage FROM '{manifest_path}' WITH (FORMAT csv, HEADER true, ENCODING 'UTF8') INSERT INTO public.pacs_dicom_files ( ct_number, batch_name, source_folder_name, source_ct_number, source_patient_name, target_folder_name, needs_ct_number_fix, patient_name_dicom, patient_id, patient_birth_date, patient_sex, study_date, study_time, study_id, modality, body_part_examined, protocol_name, study_description, accession_numbers, raw_accession_numbers, study_instance_uids, series_count, dicom_file_count, total_file_count, total_bytes, source_path, processed_path, status, notes, updated_at ) SELECT ct_number, batch_name, source_folder_name, source_ct_number, source_patient_name, target_folder_name, COALESCE(NULLIF(needs_ct_number_fix, '')::boolean, false), patient_name_dicom, patient_id, patient_birth_date, patient_sex, study_date, study_time, study_id, modality, body_part_examined, protocol_name, study_description, accession_numbers, raw_accession_numbers, study_instance_uids, NULLIF(series_count, '')::integer, NULLIF(dicom_file_count, '')::integer, NULLIF(total_file_count, '')::integer, NULLIF(total_bytes, '')::bigint, source_path, processed_path, status, notes, now() FROM pacs_dicom_files_stage ON CONFLICT (ct_number) DO UPDATE SET batch_name = EXCLUDED.batch_name, source_folder_name = EXCLUDED.source_folder_name, source_ct_number = EXCLUDED.source_ct_number, source_patient_name = EXCLUDED.source_patient_name, target_folder_name = EXCLUDED.target_folder_name, needs_ct_number_fix = EXCLUDED.needs_ct_number_fix, patient_name_dicom = EXCLUDED.patient_name_dicom, patient_id = EXCLUDED.patient_id, patient_birth_date = EXCLUDED.patient_birth_date, patient_sex = EXCLUDED.patient_sex, study_date = EXCLUDED.study_date, study_time = EXCLUDED.study_time, study_id = EXCLUDED.study_id, modality = EXCLUDED.modality, body_part_examined = EXCLUDED.body_part_examined, protocol_name = EXCLUDED.protocol_name, study_description = EXCLUDED.study_description, accession_numbers = EXCLUDED.accession_numbers, raw_accession_numbers = EXCLUDED.raw_accession_numbers, study_instance_uids = EXCLUDED.study_instance_uids, series_count = EXCLUDED.series_count, dicom_file_count = EXCLUDED.dicom_file_count, total_file_count = EXCLUDED.total_file_count, total_bytes = EXCLUDED.total_bytes, source_path = EXCLUDED.source_path, processed_path = EXCLUDED.processed_path, status = EXCLUDED.status, notes = EXCLUDED.notes, updated_at = now(); COMMIT; """ path.write_text(sql, encoding="utf-8") 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") parser.add_argument("--legacy-batch-layout", action="store_true", help="store processed studies under processed-root/batch-name") 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 if args.legacy_batch_layout else args.processed_root).absolute() 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_sync_sql(result_dir / "sync_pacs_dicom_files.sql", (result_dir / "study_manifest.csv").resolve()) 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())