"""DICOM intake: PHI de-identification at rest, safe metadata, PNG preview. Uploaded studies are rewritten in place with every denylisted PHI tag blanked before anything else reads them; only a small allowlisted metadata dict (with the study date truncated to a year) is ever persisted alongside the document. """ from pathlib import Path import pydicom from PIL import Image from pydicom.datadict import tag_for_keyword PHI_DENYLIST: tuple[str, ...] = ( "PatientName", "PatientID", "PatientBirthDate", "OtherPatientIDs", "OtherPatientNames", "PatientAddress", "PatientTelephoneNumbers", "InstitutionName", "InstitutionAddress", "ReferringPhysicianName", "PerformingPhysicianName", "OperatorsName", "AccessionNumber", ) _SAFE_STR_FIELDS = ("Modality", "Manufacturer", "ManufacturerModelName", "SoftwareVersions") _SAFE_INT_FIELDS = ("Rows", "Columns") _MAGIC_PNG = b"\x89PNG\r\n\x1a\n" _MAGIC_JPEG = b"\xff\xd8\xff" _MAGIC_PDF = b"%PDF-" _MAGIC_DICM = b"DICM" def sniff_kind(path: Path, mime: str) -> str: """Classify a stored upload by magic bytes. The client extension and mime (passed for logging/diagnostics only) are never trusted alone. Returns 'dicom' | 'png' | 'jpeg' | 'pdf'. """ with path.open("rb") as fh: head = fh.read(132) if len(head) >= 132 and head[128:132] == _MAGIC_DICM: return "dicom" if head.startswith(_MAGIC_PNG): return "png" if head.startswith(_MAGIC_JPEG): return "jpeg" if head.startswith(_MAGIC_PDF): return "pdf" raise ValueError("unsupported file type") def process_dicom(path: Path, *, rewrite: bool = True) -> tuple[dict, str | None]: """De-identify a DICOM file *in place*, extract safe metadata, render a preview. Every denylisted PHI tag present is blanked and the file rewritten on disk before returning, so the study is de-identified at rest. Returns (meta, preview_path); preview_path is None when pixel data is missing or unreadable. Raises ValueError for non-DICOM input. ``rewrite=False`` skips the at-rest rewrite for callers whose input is synthetic and PHI-free by construction (the demo seeder); metadata and preview extraction behave identically. """ try: ds = pydicom.dcmread(path, force=True) except Exception as exc: raise ValueError("not a valid DICOM file") from exc if len(ds) == 0 or "TransferSyntaxUID" not in ds.file_meta: raise ValueError("not a valid DICOM file") for keyword in PHI_DENYLIST: tag = tag_for_keyword(keyword) if tag is not None and tag in ds: ds[tag].value = "" meta: dict[str, object] = {} for keyword in _SAFE_STR_FIELDS: value = ds.get(keyword) if value is not None and str(value): meta[keyword] = str(value) for keyword in _SAFE_INT_FIELDS: value = ds.get(keyword) if value is not None: meta[keyword] = int(value) study_date = str(ds.get("StudyDate") or "") if study_date: meta["StudyDate"] = study_date[:4] # year only — never persist the full date pixel_spacing = ds.get("PixelSpacing") if pixel_spacing is not None: meta["PixelSpacing"] = str(pixel_spacing) if rewrite: ds.save_as(path, enforce_file_format=True) # rewrite: de-identified at rest with path.open("rb") as fh: head = fh.read(132) if len(head) < 132 or head[128:132] != _MAGIC_DICM: import logging logging.getLogger("claimflow.dicom").error( "post-rewrite verification failed for %s: size=%d head=%s", path, path.stat().st_size, head[:16].hex(), ) preview = path.with_suffix(".png") try: arr = ds.pixel_array.astype("float32") except Exception: return meta, None lo, hi = float(arr.min()), float(arr.max()) arr = (arr - lo) / (hi - lo) * 255.0 if hi > lo else arr * 0.0 arr8 = arr.astype("uint8") if arr8.ndim == 3: # color or multi-frame: preview the first plane arr8 = arr8[..., 0] if arr8.shape[-1] in (3, 4) else arr8[0] Image.fromarray(arr8).save(preview) return meta, str(preview)