2026-06-24 17:24:11 +07:00

599 lines
23 KiB
Python

#!/usr/bin/env python3
"""
DICOM Folder Indexer (CLI)
============================
Recursively scans a folder for DICOM files, extracts key metadata
(patient, study, series, instance level), builds an index, lets you
search/filter that index, and export it to CSV or Excel.
Dependencies:
pip install pydicom
pip install openpyxl # only needed if you use --xlsx or an .xlsx index
Usage:
# Build an index from a folder and save it
python dicom_indexer.py scan /path/to/dicom_folder -o index.csv
# Build an index and also save as Excel
python dicom_indexer.py scan /path/to/dicom_folder -o index.csv --xlsx index.xlsx
# Search an existing index
python dicom_indexer.py search index.csv --patient "Doe" --modality CT
# Search across all fields for a keyword
python dicom_indexer.py search index.csv -q "chest"
# List unique patients/studies/series in an index
python dicom_indexer.py summary index.csv
"""
import os
import sys
import csv
import argparse
import traceback
from datetime import datetime
# ---------------------------------------------------------------------------
# Dependency check with a friendly message.
# Note: pandas is no longer required — all CSV/Excel handling below streams
# rows directly via the csv module and openpyxl, which keeps memory usage
# low regardless of folder/index size. openpyxl is only needed if you use
# --xlsx (scan) or an .xlsx index file (search/summary); it's checked
# lazily where it's actually used, not at startup.
# ---------------------------------------------------------------------------
MISSING = []
try:
import pydicom
except ImportError:
MISSING.append("pydicom")
if MISSING:
print("Missing required packages: " + ", ".join(MISSING))
print("Install them with:")
print(f" pip install {' '.join(MISSING)}")
sys.exit(1)
# ---------------------------------------------------------------------------
# Metadata fields to extract from each DICOM file.
# Format: (DataFrame column name, DICOM keyword)
# Add/remove entries here to customize what gets indexed.
# ---------------------------------------------------------------------------
DICOM_FIELDS = [
("PatientName", "PatientName"),
("PatientID", "PatientID"),
("PatientBirthDate", "PatientBirthDate"),
("PatientSex", "PatientSex"),
("PatientAge", "PatientAge"),
("StudyDate", "StudyDate"),
("StudyTime", "StudyTime"),
("StudyDescription", "StudyDescription"),
("StudyInstanceUID", "StudyInstanceUID"),
("AccessionNumber", "AccessionNumber"),
("Modality", "Modality"),
("SeriesDescription", "SeriesDescription"),
("SeriesNumber", "SeriesNumber"),
("SeriesInstanceUID", "SeriesInstanceUID"),
("InstanceNumber", "InstanceNumber"),
("SOPInstanceUID", "SOPInstanceUID"),
("Manufacturer", "Manufacturer"),
("ManufacturerModelName", "ManufacturerModelName"),
("InstitutionName", "InstitutionName"),
("BodyPartExamined", "BodyPartExamined"),
("Rows", "Rows"),
("Columns", "Columns"),
("SliceThickness", "SliceThickness"),
]
# Maps CLI search flags -> DataFrame column name
SEARCH_FLAG_TO_COLUMN = {
"patient": "PatientName",
"patient_id": "PatientID",
"study_date": "StudyDate",
"study": "StudyDescription",
"modality": "Modality",
"series": "SeriesDescription",
"body_part": "BodyPartExamined",
"accession": "AccessionNumber",
}
# ---------------------------------------------------------------------------
# Extension-based fast skip.
#
# DICOM files often have NO extension at all (common with PACS/modality
# exports), or use .dcm/.dicom/.ima/.img. Because a missing extension is
# normal and valid, we do NOT filter to an allow-list of DICOM extensions
# only — that would silently skip real DICOM files.
#
# Instead we skip files whose extension is UNAMBIGUOUSLY something else
# (images, documents, archives, executables, etc.) without even opening
# them. Everything else (no extension, .dcm, or anything unrecognized)
# still goes through extract_metadata() as before, so correctness is
# preserved while common junk/sidecar files are skipped cheaply.
# ---------------------------------------------------------------------------
SKIP_EXTENSIONS = {
# common non-DICOM image formats
".jpg", ".jpeg", ".png", ".gif", ".bmp", ".tif", ".tiff", ".webp", ".ico", ".svg",
# documents / text / data
".txt", ".md", ".csv", ".json", ".xml", ".html", ".htm", ".pdf", ".doc", ".docx",
".xls", ".xlsx", ".ppt", ".pptx", ".log", ".ini", ".yaml", ".yml",
# archives
".zip", ".rar", ".7z", ".tar", ".gz", ".bz2", ".xz",
# executables / scripts / libraries
".exe", ".dll", ".so", ".bat", ".sh", ".py", ".js", ".jar",
# media
".mp3", ".mp4", ".avi", ".mov", ".wav",
# misc OS/editor cruft
".ds_store", ".db", ".tmp", ".bak", ".lnk", ".url",
}
# Used only when --strict-extension is passed: files must have one of these
# extensions (or no extension) to be opened at all.
DICOM_EXTENSIONS = {".dcm", ".dicom", ".dic", ".ima", ".img", ".dcm30"}
def format_dicom_date(value):
"""Convert DICOM DA format (YYYYMMDD) to YYYY-MM-DD for readability."""
if not value:
return ""
value = str(value)
if len(value) == 8 and value.isdigit():
try:
return datetime.strptime(value, "%Y%m%d").strftime("%Y-%m-%d")
except ValueError:
return value
return value
def extract_metadata(filepath):
"""
Read a single file with pydicom and return a dict of extracted fields,
or None if the file is not a valid DICOM file.
"""
try:
ds = pydicom.dcmread(filepath, stop_before_pixels=True, force=False)
except Exception:
# Try once more with force=True in case the file is missing the
# standard preamble but is still valid DICOM (common with exported
# files that strip the 128-byte header).
try:
ds = pydicom.dcmread(filepath, stop_before_pixels=True, force=True)
if "SOPClassUID" not in ds and "Modality" not in ds:
return None
except Exception:
return None
row = {"FilePath": filepath, "FileName": os.path.basename(filepath)}
for col_name, keyword in DICOM_FIELDS:
try:
value = getattr(ds, keyword, "")
except Exception:
value = ""
if value is None:
value = ""
value = str(value).strip()
if keyword in ("StudyDate", "PatientBirthDate"):
value = format_dicom_date(value)
row[col_name] = value
try:
size_bytes = os.path.getsize(filepath)
except OSError:
size_bytes = 0
row["FileSizeKB"] = round(size_bytes / 1024, 1)
return row
def scan_folder(root_folder, show_progress=True, strict_extension=False):
"""
Walk root_folder recursively and yield a metadata dict for every valid
DICOM file found, one at a time.
This is a generator (not a list-returning function) so that callers can
write each row to disk as it's produced, instead of holding every row
in memory at once. Memory use stays roughly constant regardless of how
many files are in the folder.
Extension filtering (speed optimization):
- Default: files with an extension that's unambiguously NOT DICOM
(.jpg, .txt, .zip, etc. — see SKIP_EXTENSIONS) are skipped without
being opened. Files with no extension, a .dcm-style extension, or
anything unrecognized are still opened and checked, since DICOM
files commonly have no extension at all.
- strict_extension=True: only files with no extension or a
.dcm/.dicom/.ima/.img-style extension (see DICOM_EXTENSIONS) are
opened. Faster, but will silently skip real DICOM files saved
with an unusual extension.
Note: total file count for progress percentage isn't known up front
(we never pre-list the whole tree into memory), so progress is reported
as a running count instead of a percentage.
"""
count_seen = 0
count_found = 0
count_skipped = 0
for dirpath, _dirnames, filenames in os.walk(root_folder):
for fn in filenames:
filepath = os.path.join(dirpath, fn)
count_seen += 1
ext = os.path.splitext(fn)[1].lower()
if strict_extension:
should_skip = ext != "" and ext not in DICOM_EXTENSIONS
else:
should_skip = ext in SKIP_EXTENSIONS
if should_skip:
count_skipped += 1
else:
row = extract_metadata(filepath)
if row is not None:
count_found += 1
yield row
if show_progress and count_seen % 25 == 0:
sys.stdout.write(
f"\rScanning... {count_seen} files checked — {count_found} DICOM found "
f"({count_skipped} skipped by extension)"
)
sys.stdout.flush()
if show_progress and count_seen:
sys.stdout.write(
f"\rScanning... {count_seen} files checked — {count_found} DICOM found "
f"({count_skipped} skipped by extension)\n"
)
sys.stdout.flush()
# ---------------------------------------------------------------------------
# Subcommand: scan
# ---------------------------------------------------------------------------
def cmd_scan(args):
folder = args.folder
if not os.path.isdir(folder):
print(f"Error: '{folder}' is not a valid folder.", file=sys.stderr)
sys.exit(1)
print(f"Scanning '{folder}' for DICOM files...")
# Column order for the CSV (FilePath/FileName first, then the
# standard DICOM_FIELDS, then file size last).
fieldnames = ["FilePath", "FileName"] + [c for c, _ in DICOM_FIELDS] + ["FileSizeKB"]
out_csv = args.output
row_count = 0
modality_counts = {}
patient_ids = set()
study_uids = set()
series_uids = set()
try:
with open(out_csv, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
for row in scan_folder(folder, show_progress=not args.quiet, strict_extension=args.strict_extension):
writer.writerow(row)
row_count += 1
# Track summary stats incrementally (O(1) memory) instead
# of loading everything back into a DataFrame afterward.
modality_counts[row.get("Modality", "")] = modality_counts.get(row.get("Modality", ""), 0) + 1
if row.get("PatientID"):
patient_ids.add(row["PatientID"])
if row.get("StudyInstanceUID"):
study_uids.add(row["StudyInstanceUID"])
if row.get("SeriesInstanceUID"):
series_uids.add(row["SeriesInstanceUID"])
except Exception as exc:
print(f"Error during scan: {exc}", file=sys.stderr)
if args.verbose:
traceback.print_exc()
sys.exit(1)
if row_count == 0:
os.remove(out_csv) # don't leave a header-only file behind
print("No valid DICOM files found in this folder.")
sys.exit(0)
print(f"Indexed {row_count} DICOM files.")
print(f"CSV index saved to: {out_csv}")
if args.xlsx:
# Excel format can't be streamed row-by-row as easily as CSV, so we
# read the CSV back in chunks and write it out to .xlsx. This keeps
# peak memory bounded by chunk size rather than the full dataset,
# at the cost of a second pass over the data.
_csv_to_xlsx(out_csv, args.xlsx)
print(f"Excel index saved to: {args.xlsx}")
if not args.quiet:
print("\n--- Index Summary ---")
print(f"Total files indexed: {row_count}")
print(f"Unique patients: {len(patient_ids)}")
print(f"Unique studies: {len(study_uids)}")
print(f"Unique series: {len(series_uids)}")
print("Modalities:")
for modality, count in sorted(modality_counts.items(), key=lambda kv: -kv[1]):
label = modality if modality else "(unknown)"
print(f" {label:<10} {count}")
print()
def _csv_to_xlsx(csv_path, xlsx_path, chunksize=20000):
"""
Convert a CSV file to .xlsx without ever loading the entire CSV into
memory at once. Uses openpyxl's write-only mode, which streams rows
directly to disk instead of building the whole workbook in RAM.
"""
try:
from openpyxl import Workbook
except ImportError:
print("Error: openpyxl is required for --xlsx export. Install it with:", file=sys.stderr)
print(" pip install openpyxl", file=sys.stderr)
sys.exit(1)
wb = Workbook(write_only=True)
ws = wb.create_sheet("DICOM Index")
with open(csv_path, "r", newline="", encoding="utf-8") as f:
reader = csv.reader(f)
for row in reader:
ws.append(row)
wb.save(xlsx_path)
# ---------------------------------------------------------------------------
# Subcommand: search
# ---------------------------------------------------------------------------
DISPLAY_LIMIT = 200 # cap rows kept in memory for screen display
def cmd_search(args):
field_filters = {}
for flag, column in SEARCH_FLAG_TO_COLUMN.items():
value = getattr(args, flag, None)
if value:
field_filters[column] = value.lower()
query = args.query.lower() if args.query else None
match_count = 0
display_rows = [] # bounded to DISPLAY_LIMIT regardless of result size
fieldnames = None
writer = None
out_f = None
try:
if args.output:
out_f = open(args.output, "w", newline="", encoding="utf-8")
for row in _iter_index_rows(args.index):
if not _row_matches(row, field_filters, query):
continue
match_count += 1
if out_f is not None:
if writer is None:
fieldnames = list(row.keys())
writer = csv.DictWriter(out_f, fieldnames=fieldnames)
writer.writeheader()
writer.writerow(row)
if len(display_rows) < DISPLAY_LIMIT:
display_rows.append(row)
finally:
if out_f is not None:
out_f.close()
print(f"{match_count} matching record(s) found.\n")
if match_count == 0:
if args.output and os.path.exists(args.output):
os.remove(args.output) # no header-only leftover file
return
display_columns = args.columns.split(",") if args.columns else [
"PatientName", "PatientID", "StudyDate", "StudyDescription",
"Modality", "SeriesDescription", "SeriesNumber", "InstanceNumber",
"FileName",
]
display_columns = [c for c in display_columns if c in display_rows[0]]
_print_table(display_rows, display_columns)
if match_count > len(display_rows):
print(f"\n... showing first {len(display_rows)} of {match_count} matches. Use -o to save all results.")
if args.output:
print(f"\nFiltered results ({match_count} rows) saved to: {args.output}")
def _row_matches(row, field_filters, query):
"""Check a single CSV row (dict) against field filters and free-text query."""
for column, needle in field_filters.items():
haystack = str(row.get(column, "")).lower()
if needle not in haystack:
return False
if query:
if not any(query in str(v).lower() for v in row.values()):
return False
return True
# ---------------------------------------------------------------------------
# Subcommand: summary
# ---------------------------------------------------------------------------
def cmd_summary(args):
total = 0
patient_ids = set()
study_uids = set()
series_uids = set()
modality_counts = {}
group_counts = {} # (PatientName, PatientID, StudyDescription, SeriesDescription) -> count
for row in _iter_index_rows(args.index):
total += 1
if row.get("PatientID"):
patient_ids.add(row["PatientID"])
if row.get("StudyInstanceUID"):
study_uids.add(row["StudyInstanceUID"])
if row.get("SeriesInstanceUID"):
series_uids.add(row["SeriesInstanceUID"])
modality = row.get("Modality", "")
modality_counts[modality] = modality_counts.get(modality, 0) + 1
key = (
row.get("PatientName", ""), row.get("PatientID", ""),
row.get("StudyDescription", ""), row.get("SeriesDescription", ""),
)
group_counts[key] = group_counts.get(key, 0) + 1
if total == 0:
print("Index is empty.")
return
print("\n--- Index Summary ---")
print(f"Total files indexed: {total}")
print(f"Unique patients: {len(patient_ids)}")
print(f"Unique studies: {len(study_uids)}")
print(f"Unique series: {len(series_uids)}")
print("Modalities:")
for modality, count in sorted(modality_counts.items(), key=lambda kv: -kv[1]):
label = modality if modality else "(unknown)"
print(f" {label:<10} {count}")
print("\nPatients / Studies / Series:")
for (pname, pid, study, series), count in group_counts.items():
print(f" {pname} | {pid} | {study} | {series} -> {count} file(s)")
print()
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _iter_index_rows(path):
"""
Yield rows (as dicts) from an index file one at a time, without ever
loading the whole file into memory. Supports both .csv and .xlsx.
"""
if not os.path.isfile(path):
print(f"Error: index file '{path}' not found. Run 'scan' first.", file=sys.stderr)
sys.exit(1)
try:
if path.lower().endswith(".xlsx"):
try:
from openpyxl import load_workbook
except ImportError:
print("Error: openpyxl is required to read .xlsx index files. Install it with:", file=sys.stderr)
print(" pip install openpyxl", file=sys.stderr)
sys.exit(1)
# read_only=True streams rows from disk instead of loading the
# whole sheet into memory.
wb = load_workbook(path, read_only=True, data_only=True)
ws = wb.active
rows_iter = ws.iter_rows(values_only=True)
header = [str(h) if h is not None else "" for h in next(rows_iter)]
for raw_row in rows_iter:
yield {
header[i]: ("" if raw_row[i] is None else str(raw_row[i]))
for i in range(len(header))
}
wb.close()
else:
with open(path, "r", newline="", encoding="utf-8") as f:
reader = csv.DictReader(f)
for row in reader:
yield {k: (v if v is not None else "") for k, v in row.items()}
except SystemExit:
raise
except Exception as exc:
print(f"Error reading index file: {exc}", file=sys.stderr)
sys.exit(1)
def _print_table(rows, columns):
"""Print a list of dicts as a simple aligned text table for the given columns."""
if not rows:
return
col_widths = {
col: min(max(len(col), max(len(str(r.get(col, ""))) for r in rows)), 40)
for col in columns
}
header = " ".join(col.ljust(col_widths[col]) for col in columns)
print(header)
print("-" * len(header))
for row in rows:
line = " ".join(
str(row.get(col, ""))[: col_widths[col]].ljust(col_widths[col])
for col in columns
)
print(line)
# ---------------------------------------------------------------------------
# Argument parsing
# ---------------------------------------------------------------------------
def build_parser():
parser = argparse.ArgumentParser(
prog="dicom_indexer.py",
description="Index DICOM folders by metadata and search the resulting index.",
)
subparsers = parser.add_subparsers(dest="command", required=True)
# scan
p_scan = subparsers.add_parser("scan", help="Scan a folder and build a DICOM metadata index.")
p_scan.add_argument("folder", help="Path to the folder containing DICOM files (scanned recursively).")
p_scan.add_argument("-o", "--output", default="dicom_index.csv", help="Output CSV path (default: dicom_index.csv).")
p_scan.add_argument("--xlsx", help="Also save the index as an Excel (.xlsx) file at this path.")
p_scan.add_argument("-q", "--quiet", action="store_true", help="Suppress progress output and summary.")
p_scan.add_argument("--verbose", action="store_true", help="Show full error tracebacks on failure.")
p_scan.add_argument(
"--strict-extension", action="store_true",
help=(
"Only attempt to read files with a .dcm/.dicom/.ima/.img extension "
"(or no extension). Fastest option, but will silently skip DICOM "
"files saved with an unusual extension. Off by default."
),
)
p_scan.set_defaults(func=cmd_scan)
# search
p_search = subparsers.add_parser("search", help="Search/filter an existing DICOM index.")
p_search.add_argument("index", help="Path to the index CSV/XLSX file (created by 'scan').")
p_search.add_argument("-q", "--query", help="Free-text search across all fields.")
p_search.add_argument("--patient", help="Filter by patient name (substring match).")
p_search.add_argument("--patient-id", dest="patient_id", help="Filter by patient ID (substring match).")
p_search.add_argument("--study", help="Filter by study description (substring match).")
p_search.add_argument("--study-date", dest="study_date", help="Filter by study date, e.g. 2024-01-15 (substring match).")
p_search.add_argument("--modality", help="Filter by modality, e.g. CT, MR, US.")
p_search.add_argument("--series", help="Filter by series description (substring match).")
p_search.add_argument("--body-part", dest="body_part", help="Filter by body part examined.")
p_search.add_argument("--accession", help="Filter by accession number.")
p_search.add_argument("--columns", help="Comma-separated list of columns to display (default: a sensible subset).")
p_search.add_argument("-o", "--output", help="Save filtered results to this CSV path.")
p_search.set_defaults(func=cmd_search)
# summary
p_summary = subparsers.add_parser("summary", help="Show summary statistics for an existing index.")
p_summary.add_argument("index", help="Path to the index CSV/XLSX file (created by 'scan').")
p_summary.set_defaults(func=cmd_summary)
return parser
def main():
parser = build_parser()
args = parser.parse_args()
args.func(args)
if __name__ == "__main__":
main()