Add PACS DICOM preprocessing workflow

.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处理/数据处理结果区/

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` 中排除。

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())