add timestamp in logging uvicorn

.env

@@ -0,0 +1,8 @@
+# Copy this file to .env and adjust as needed.
+# The indexer reads these to connect to MongoDB, so you don't have to
+# pass --mongo-uri/--db/--collection on every command.
+
+MONGO_URI=mongodb://root:AtHEntRyPrOchite@172.16.44.35:27017/
+MONGO_DB=rsabhk_simrs
+MONGO_COLLECTION=dicom-index
+PATH_MAPPINGS_FILE=/app/config.json

api.py

@@ -0,0 +1,355 @@
+#!/usr/bin/env python3
+"""
+DICOM Sender API Server
+=======================
+A small REST API that wraps dicom_sender.py so a client can trigger sends by
+Accession Number (single or as a group) without needing DICOM tooling itself.
+
+Destination PACS/router details and the MongoDB index connection are fixed
+in config.json on the server. Clients only ever supply accession numbers.
+
+Sends run as background jobs (a single C-STORE association can take a while
+for large studies), and the client polls for status/results using the
+returned job_id.
+
+Run:
+    pip install fastapi uvicorn pydicom pynetdicom
+    python3 api_server.py
+    # or: uvicorn api_server:app --host 0.0.0.0 --port 8000
+
+Endpoints:
+    GET  /health                    liveness check
+    GET  /config                    sanitized view of active server config
+    POST /echo                      synchronous C-ECHO connectivity test
+    POST /jobs                      queue a send job -> {job_id}
+    GET  /jobs                      list recent jobs (most recent first)
+    GET  /jobs/{job_id}             full status/results for one job
+"""
+
+import json
+import os
+import sys
+import time
+import uuid
+import threading
+from collections import OrderedDict
+from concurrent.futures import ThreadPoolExecutor
+from datetime import datetime, timezone
+from typing import List, Optional
+
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel, Field, model_validator
+from pymongo import MongoClient
+
+import sender as sender
+
+LOGGING_CONFIG = {
+    "version": 1,
+    "disable_existing_loggers": False,
+    "formatters": {
+        "default": {
+            "()": "uvicorn.logging.DefaultFormatter",
+            "fmt": "%(asctime)s %(levelprefix)s %(message)s",
+            "datefmt": "%Y-%m-%d %H:%M:%S",
+        },
+        "access": {
+            "()": "uvicorn.logging.AccessFormatter",
+            "fmt": '%(asctime)s %(levelprefix)s %(client_addr)s - "%(request_line)s" %(status_code)s',
+            "datefmt": "%Y-%m-%d %H:%M:%S",
+        },
+    },
+    "handlers": {
+        "default": {"formatter": "default", "class": "logging.StreamHandler", "stream": "ext://sys.stderr"},
+        "access": {"formatter": "access", "class": "logging.StreamHandler", "stream": "ext://sys.stdout"},
+    },
+    "loggers": {
+        "uvicorn": {"handlers": ["default"], "level": "INFO"},
+        "uvicorn.error": {"level": "INFO"},
+        "uvicorn.access": {"handlers": ["access"], "level": "INFO", "propagate": False},
+    },
+}
+
+CONFIG_PATH = os.environ.get("DICOM_API_CONFIG", os.path.join(os.path.dirname(__file__), "config.json"))
+
+with open(CONFIG_PATH, "r", encoding="utf-8") as f:
+    CONFIG = json.load(f)
+
+DEST = CONFIG["destination"]
+SEND_DELAY = CONFIG.get("send_delay", 0.0)
+MAX_CONCURRENT_JOBS = CONFIG.get("max_concurrent_jobs", 1)
+JOB_HISTORY_LIMIT = CONFIG.get("job_history_limit", 500)
+
+INDEX_SOURCE = CONFIG["index_source"]
+if INDEX_SOURCE.get("type") != "mongodb":
+    raise RuntimeError(
+        f"Unsupported index_source.type: {INDEX_SOURCE.get('type')!r}. This server build only supports 'mongodb'."
+    )
+
+# The URI can carry credentials, so prefer an env var over the checked-in
+# config.json when both are present (handy for docker-compose + .env).
+_MONGO_URI = os.environ.get("MONGO_URI") or INDEX_SOURCE.get("uri")
+if not _MONGO_URI:
+    raise RuntimeError("No MongoDB URI configured: set index_source.uri in config.json or the MONGO_URI env var.")
+
+_mongo_client = MongoClient(_MONGO_URI)
+_mongo_collection = _mongo_client[INDEX_SOURCE["database"]][INDEX_SOURCE["collection"]]
+_FILEPATH_FIELD = INDEX_SOURCE.get("filepath_field", "FilePath")
+_ACCESSION_FIELD = INDEX_SOURCE.get("accession_field", "AccessionNumber")
+
+# Host path -> container path translation, e.g. Mongo stores
+# "/mnt/pacs_data/study1/img.dcm" (the path on the machine that indexed the
+# files) but this container has that same folder mounted at "/data".
+_PATH_PREFIX_MAP = [(m["from"], m["to"]) for m in CONFIG.get("path_mappings", [])]
+
+
+def _mongo_lookup(accession_numbers):
+    return sender.find_matching_files_mongo(
+        _mongo_collection,
+        accession_numbers,
+        filepath_field=_FILEPATH_FIELD,
+        accession_field=_ACCESSION_FIELD,
+        path_prefix_map=_PATH_PREFIX_MAP,
+    )
+
+
+app = FastAPI(title="DICOM Sender API", version="1.0")
+
+_executor = ThreadPoolExecutor(max_workers=MAX_CONCURRENT_JOBS)
+_jobs_lock = threading.Lock()
+_jobs = OrderedDict()  # job_id -> job dict, insertion order = creation order
+
+
+def _now():
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _trim_history():
+    # Only drop finished jobs (completed/error), oldest first, keep queued/running always.
+    while len(_jobs) > JOB_HISTORY_LIMIT:
+        for jid, job in _jobs.items():
+            if job["status"] in ("completed", "error"):
+                del _jobs[jid]
+                break
+        else:
+            break  # nothing removable
+
+
+def _serializable_result(result):
+    """Convert send_accessions()'s result dict (which contains sets) to JSON-safe types."""
+    if result is None:
+        return None
+    out = dict(result)
+    out["found_accessions"] = sorted(result.get("found_accessions", set()))
+    out["missing_accessions"] = sorted(result.get("missing_accessions", set()))
+    out["results"] = [
+        {
+            "filepath": fp,
+            "accession": acc,
+            "status_code": (f"0x{code:04X}" if code is not None else None),
+            "status_label": label,
+            "success": bool(code is not None and sender.is_success(code)),
+        }
+        for fp, acc, code, label in result.get("results", [])
+    ]
+    return out
+
+
+def _run_job(job_id):
+    with _jobs_lock:
+        job = _jobs[job_id]
+        job["status"] = "running"
+        job["started_at"] = _now()
+        accessions = job["requested_accessions"]
+        dry_run = job["dry_run"]
+
+    def progress_cb(current, total, filepath, accession):
+        with _jobs_lock:
+            _jobs[job_id]["progress"] = {"current": current, "total": total}
+
+    try:
+        result = sender.send_accessions(
+            accession_numbers=accessions,
+            lookup_fn=_mongo_lookup,
+            host=DEST["host"],
+            port=DEST["port"],
+            ae_title=DEST["ae_title"],
+            calling_ae_title=DEST.get("calling_ae_title", sender.DEFAULT_CALLING_AE_TITLE),
+            timeout=DEST.get("timeout", 30),
+            delay=SEND_DELAY,
+            dry_run=dry_run,
+            verbose=False,
+            progress_callback=progress_cb,
+        )
+    except Exception as exc:
+        with _jobs_lock:
+            job = _jobs[job_id]
+            job["status"] = "error"
+            job["error"] = f"Unhandled exception: {exc}"
+            job["finished_at"] = _now()
+        return
+
+    with _jobs_lock:
+        job = _jobs[job_id]
+        job["result"] = _serializable_result(result)
+        job["error"] = result.get("error")
+        job["status"] = "error" if result.get("error") else "completed"
+        job["finished_at"] = _now()
+        _trim_history()
+
+
+class SendRequest(BaseModel):
+    accessions: Optional[List[str]] = Field(
+        default=None, description="List of accession numbers to send as a group."
+    )
+    accession: Optional[str] = Field(
+        default=None, description="A single accession number to send (alternative to 'accessions')."
+    )
+    dry_run: bool = Field(
+        default=False, description="If true, resolve and report matching files without sending."
+    )
+
+    @model_validator(mode="after")
+    def _require_at_least_one(self):
+        if not self.accessions and not self.accession:
+            raise ValueError("Provide 'accession' (single) or 'accessions' (list).")
+        return self
+
+    def all_accessions(self) -> List[str]:
+        out = list(self.accessions or [])
+        if self.accession:
+            out.append(self.accession)
+        # de-dupe, strip, drop empties, preserve order
+        seen = set()
+        cleaned = []
+        for a in out:
+            a = (a or "").strip()
+            if a and a not in seen:
+                seen.add(a)
+                cleaned.append(a)
+        return cleaned
+
+
+@app.get("/health")
+def health():
+    return {"status": "ok", "time": _now()}
+
+
+@app.get("/config")
+def get_config():
+    try:
+        _mongo_client.admin.command("ping")
+        mongo_reachable = True
+    except Exception:
+        mongo_reachable = False
+
+    return {
+        "index_source_reachable": mongo_reachable,
+        "destination": {
+            "host": DEST["host"],
+            "port": DEST["port"],
+            "ae_title": DEST["ae_title"],
+            "calling_ae_title": DEST.get("calling_ae_title", sender.DEFAULT_CALLING_AE_TITLE),
+            "timeout": DEST.get("timeout", 30),
+        },
+        "index_source": {
+            "type": "mongodb",
+            "database": INDEX_SOURCE["database"],
+            "collection": INDEX_SOURCE["collection"],
+        },
+        "path_mappings": [{"from": f, "to": t} for f, t in _PATH_PREFIX_MAP],
+        "max_concurrent_jobs": MAX_CONCURRENT_JOBS,
+    }
+
+
+@app.post("/echo")
+def echo():
+    """Synchronous C-ECHO connectivity test against the configured destination."""
+    from pynetdicom import AE
+    from pynetdicom.sop_class import Verification
+
+    ae = AE(ae_title=DEST.get("calling_ae_title", sender.DEFAULT_CALLING_AE_TITLE))
+    ae.add_requested_context(Verification)
+    timeout = DEST.get("timeout", 30)
+    ae.acse_timeout = timeout
+    ae.dimse_timeout = timeout
+    ae.network_timeout = timeout
+
+    try:
+        assoc = ae.associate(DEST["host"], DEST["port"], ae_title=DEST["ae_title"])
+    except Exception as exc:
+        raise HTTPException(status_code=502, detail=f"Association error: {exc}")
+
+    if not assoc.is_established:
+        raise HTTPException(status_code=502, detail="Could not establish association.")
+
+    status = assoc.send_c_echo()
+    code, label = sender.describe_status(status)
+    assoc.release()
+
+    success = code is not None and sender.is_success(code)
+    return {
+        "success": success,
+        "status_code": (f"0x{code:04X}" if code is not None else None),
+        "status_label": label,
+    }
+
+
+@app.post("/jobs", status_code=202)
+def create_job(req: SendRequest):
+    try:
+        _mongo_client.admin.command("ping")
+    except Exception as exc:
+        raise HTTPException(status_code=500, detail=f"Cannot reach MongoDB index source: {exc}")
+
+    accessions = req.all_accessions()
+    job_id = str(uuid.uuid4())
+    job = {
+        "job_id": job_id,
+        "status": "queued",
+        "dry_run": req.dry_run,
+        "requested_accessions": accessions,
+        "created_at": _now(),
+        "started_at": None,
+        "finished_at": None,
+        "progress": {"current": 0, "total": 0},
+        "result": None,
+        "error": None,
+    }
+    with _jobs_lock:
+        _jobs[job_id] = job
+
+    _executor.submit(_run_job, job_id)
+
+    return {"job_id": job_id, "status": "queued", "status_url": f"/jobs/{job_id}"}
+
+
+@app.get("/jobs")
+def list_jobs(limit: int = 50):
+    with _jobs_lock:
+        items = list(_jobs.values())[::-1][:limit]
+        return [
+            {
+                "job_id": j["job_id"],
+                "status": j["status"],
+                "dry_run": j["dry_run"],
+                "requested_count": len(j["requested_accessions"]),
+                "created_at": j["created_at"],
+                "finished_at": j["finished_at"],
+            }
+            for j in items
+        ]
+
+
+@app.get("/jobs/{job_id}")
+def get_job(job_id: str):
+    with _jobs_lock:
+        job = _jobs.get(job_id)
+        if job is None:
+            raise HTTPException(status_code=404, detail="Job not found.")
+        return dict(job)
+
+
+if __name__ == "__main__":
+    import uvicorn
+
+    uvicorn.run(app, host=CONFIG["server"]["host"], port=CONFIG["server"]["port"], log_config=LOGGING_CONFIG)

app.py

@@ -1,599 +0,0 @@
-#!/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()

config.json

@@ -0,0 +1,29 @@
+{
+    "destination": {
+      "host": "172.16.44.35",
+      "port": 11112,
+      "ae_title": "DCMROUTER",
+      "calling_ae_title": "DCM-RSABHK",
+      "timeout": 30
+    },
+    "index_source": {
+      "type": "mongodb",
+      "uri": "mongodb://root:AtHEntRyPrOchite@172.16.44.35:27017/",
+      "database": "rsabhk_simrs",
+      "collection": "dicom-index",
+      "filepath_field": "FilePath",
+      "accession_field": "AccessionNumber"
+    },  
+    "path_mappings": [
+        {"from": "/mnt/pacs_data", "to": "/data"},
+        {"from": "Z:/2026", "to": "Z:/2026"},
+        {"from": "/mnt/data/2026", "to": "Z:/2026"}
+      ],
+    "send_delay": 0.0,
+    "max_concurrent_jobs": 1000,
+    "job_history_limit": 500,
+    "server": {
+      "host": "0.0.0.0",
+      "port": 8000
+    }
+  }

requirements.txt

@@ -11,3 +11,6 @@
 pydicom>=3.0.2
 openpyxl>=3.1.5
 pynetdicom>=3.0.4
+fastapi
+uvicorn[standard]
+pymongo

sender.py

@@ -155,6 +155,79 @@ def load_accession_list(path):
     return accessions
 
 
+def translate_path(path, prefix_map):
+    """
+    Rewrite `path` using the first matching (from_prefix, to_prefix) pair in
+    prefix_map, e.g. translating a host path stored in an index/database to
+    the path it's actually mounted at inside a container.
+
+    Handles the common cross-platform case where indexing ran on Windows
+    (paths like "Z:\\some folder\\study1\\img.dcm" or "Z:/some folder/...")
+    but the sender runs on Linux: separators are normalized to "/" and the
+    prefix match is case-insensitive (Windows paths/drive letters are
+    case-insensitive; this has no meaningful downside on Linux paths).
+
+    prefix_map: list of (from_prefix, to_prefix) tuples, checked in order.
+        from_prefix may use "/" or "\\" — both are accepted.
+    Returns path unchanged if it's falsy, prefix_map is empty, or no prefix
+    matches. Output always uses "/" separators (Linux-style), matching
+    to_prefix's own style.
+    """
+    if not path or not prefix_map:
+        return path
+
+    normalized_path = path.replace("\\", "/")
+
+    for from_prefix, to_prefix in prefix_map:
+        norm_from = from_prefix.replace("\\", "/").rstrip("/")
+        if not norm_from:
+            continue
+        lower_path = normalized_path.lower()
+        lower_from = norm_from.lower()
+        if lower_path == lower_from:
+            return to_prefix.rstrip("/")
+        if lower_path.startswith(lower_from + "/"):
+            remainder = normalized_path[len(norm_from):]  # starts with "/"
+            return to_prefix.rstrip("/") + remainder
+
+    return path
+
+
+def find_matching_files_mongo(
+    collection,
+    accession_numbers,
+    filepath_field="FilePath",
+    accession_field="AccessionNumber",
+    path_prefix_map=None,
+):
+    """
+    Query a pymongo Collection for documents whose accession_field is in
+    accession_numbers. Yields dicts shaped like the CSV path's rows: at least
+    {"FilePath": ..., "AccessionNumber": ...} (normalized to those exact keys
+    regardless of what the underlying document's field names are), plus
+    whatever other fields the document has.
+
+    accession_numbers: iterable of accession number strings to match against.
+    path_prefix_map: optional list of (from_prefix, to_prefix) tuples used to
+        rewrite FilePath, e.g. when the path stored in Mongo is a host path
+        but files are mounted at a different path inside a container. See
+        translate_path().
+    """
+    accession_numbers = [a for a in accession_numbers if a]
+    if not accession_numbers:
+        return
+
+    # $in has practical size limits on very large lists; chunk defensively.
+    CHUNK = 1000
+    for i in range(0, len(accession_numbers), CHUNK):
+        chunk = accession_numbers[i:i + CHUNK]
+        for doc in collection.find({accession_field: {"$in": chunk}}):
+            row = dict(doc)
+            row["FilePath"] = translate_path(doc.get(filepath_field), path_prefix_map)
+            row["AccessionNumber"] = doc.get(accession_field)
+            yield row
+
+
 def find_matching_files(index_path, accession_numbers):
     """
     Stream the index CSV and yield (filepath, accession_number, sop_class_uid,
@@ -223,7 +296,307 @@ def cmd_echo(args):
 
 
 # ---------------------------------------------------------------------------
-# Subcommand: send
+# Core send logic (shared by the CLI 'send' subcommand and any programmatic
+# caller, e.g. an API server). Returns a result dict rather than printing +
+# sys.exit()-ing, so it is safe to call from long-running processes.
+# ---------------------------------------------------------------------------
+def send_accessions(
+    accession_numbers,
+    host,
+    port,
+    ae_title,
+    source_index=None,
+    lookup_fn=None,
+    calling_ae_title=DEFAULT_CALLING_AE_TITLE,
+    timeout=30,
+    delay=0.0,
+    dry_run=False,
+    verbose=True,
+    progress_callback=None,
+):
+    """
+    Find files matching accession_numbers and send them via C-STORE to
+    (host, port, ae_title). This is the reusable core used by both the CLI
+    'send' subcommand and any programmatic caller (e.g. a web API).
+
+    Exactly one of these must be given to say *where* to look up files:
+      source_index  -> path to a CSV index (dicom_indexer.py 'scan' output)
+      lookup_fn      -> callable(accession_numbers) -> iterable of row dicts,
+                        each with at least "FilePath" and "AccessionNumber"
+                        keys. Use this to back the lookup with MongoDB or
+                        anything else instead of a CSV file — see
+                        find_matching_files_mongo() for a ready-made one.
+
+    accession_numbers: iterable of accession number strings.
+    progress_callback: optional callable(current, total, filepath, accession)
+        invoked right before each file is sent (real sends only).
+    verbose: if True, mirrors the original CLI's stdout/stderr messages.
+
+    Returns a dict:
+        {
+            "found_accessions": set,
+            "missing_accessions": set,
+            "missing_files": [filepath, ...],
+            "unreadable": [filepath, ...],
+            "dry_run": bool,
+            "sendable_count": int,
+            "results": [(filepath, accession, code_or_None, label), ...],
+            "error": str or None,   # set on hard failures (association, etc.)
+        }
+    """
+    def log(msg, err=False):
+        if verbose:
+            print(msg, file=sys.stderr if err else sys.stdout)
+
+    if (source_index is None) == (lookup_fn is None):
+        raise ValueError("send_accessions() requires exactly one of source_index or lookup_fn.")
+
+    accession_numbers = set(accession_numbers)
+    result = {
+        "found_accessions": set(),
+        "missing_accessions": set(),
+        "missing_files": [],
+        "unreadable": [],
+        "dry_run": dry_run,
+        "sendable_count": 0,
+        "results": [],
+        "error": None,
+    }
+
+    index_desc = f"'{source_index}'" if source_index else "MongoDB"
+    log(f"Looking up {len(accession_numbers)} accession number(s) in index {index_desc}...")
+
+    # ---- Step 2: find matching files in the index ----
+    if lookup_fn is not None:
+        matched_rows = list(lookup_fn(accession_numbers))
+    else:
+        matched_rows = list(find_matching_files(source_index, accession_numbers))
+
+    if not matched_rows:
+        log("No matching files found in the index for the given accession number(s).")
+        result["missing_accessions"] = set(accession_numbers)
+        return result
+
+    found_accessions = {r["AccessionNumber"] for r in matched_rows}
+    missing_accessions = accession_numbers - found_accessions
+    result["found_accessions"] = found_accessions
+    result["missing_accessions"] = missing_accessions
+    log(f"Found {len(matched_rows)} file(s) across {len(found_accessions)} accession number(s).")
+    if missing_accessions:
+        log(
+            f"Warning: {len(missing_accessions)} accession number(s) had no matching "
+            f"file in the index: {', '.join(sorted(missing_accessions)[:10])}"
+            + (" ..." if len(missing_accessions) > 10 else ""),
+            err=True,
+        )
+
+    # ---- Step 3: verify files exist on disk ----
+    sendable = []
+    missing_files = []
+    for row in matched_rows:
+        filepath = row["FilePath"]
+        if os.path.isfile(filepath):
+            sendable.append(row)
+        else:
+            missing_files.append(filepath)
+    result["missing_files"] = missing_files
+
+    if missing_files:
+        log(
+            f"Warning: {len(missing_files)} file(s) listed in the index no longer "
+            f"exist on disk and will be skipped:",
+            err=True,
+        )
+        for fp in missing_files[:10]:
+            log(f"  - {fp}", err=True)
+        if len(missing_files) > 10:
+            log(f"  ... and {len(missing_files) - 10} more", err=True)
+
+    result["sendable_count"] = len(sendable)
+    if not sendable:
+        log("No sendable files remain after checking disk. Aborting.")
+        return result
+
+    if dry_run:
+        log("\n--- DRY RUN: the following files would be sent ---")
+        for row in sendable:
+            log(f"  [{row.get('AccessionNumber','')}] {row['FilePath']}")
+        log(f"\nTotal: {len(sendable)} file(s). No network connection was made.")
+        result["results"] = [
+            (row["FilePath"], row.get("AccessionNumber", ""), None, "Dry run - not sent")
+            for row in sendable
+        ]
+        return result
+
+    # ---- Step 4: read each file's SOP Class UID / Transfer Syntax UID to
+    #      build presentation contexts (max 128 per association) ----
+    log("Reading SOP Class / Transfer Syntax info from files to build presentation contexts...")
+    file_info = []  # (filepath, row, sop_class_uid, transfer_syntax_uid)
+    unreadable = []
+    for row in sendable:
+        filepath = row["FilePath"]
+        try:
+            ds_meta = pydicom.dcmread(filepath, stop_before_pixels=True, force=True)
+            sop_class = getattr(ds_meta, "SOPClassUID", None)
+            ts = ds_meta.file_meta.TransferSyntaxUID if hasattr(ds_meta, "file_meta") else None
+            if sop_class is None:
+                unreadable.append(filepath)
+                continue
+            file_info.append((filepath, row, str(sop_class), str(ts) if ts else None))
+        except Exception:
+            unreadable.append(filepath)
+    result["unreadable"] = unreadable
+
+    if unreadable:
+        log(
+            f"Warning: {len(unreadable)} file(s) could not be read as DICOM and "
+            f"will be skipped:",
+            err=True,
+        )
+        for fp in unreadable[:10]:
+            log(f"  - {fp}", err=True)
+
+    if not file_info:
+        log("No readable DICOM files remain. Aborting.")
+        return result
+
+    # Build the minimal set of presentation contexts needed: one context per
+    # DISTINCT (SOP Class UID, Transfer Syntax UID) pair actually used by the
+    # files being sent. We deliberately do NOT bundle multiple transfer
+    # syntaxes into a single context, because the peer only accepts ONE
+    # transfer syntax per context — if a SOP Class has some files in (say)
+    # uncompressed Explicit VR LE and others in JPEG Lossless, bundling them
+    # into one context risks the peer accepting only one of the two, and
+    # files using the other transfer syntax would still fail to send. A
+    # separate context per pair guarantees every transfer syntax actually
+    # present gets its own negotiation slot.
+    #
+    # This also fixes the root cause of "No presentation context ... has
+    # been accepted by the peer with '<transfer syntax>'" errors: previously
+    # contexts were requested using pynetdicom's DEFAULT transfer syntax
+    # list (uncompressed only), so compressed files (JPEG Lossless, JPEG
+    # 2000, RLE, etc.) never had a matching context at all.
+    seen_pairs = set()
+    context_pairs = []  # list of (sop_class, transfer_syntax_or_None)
+    for _, _, sop_class, ts in file_info:
+        pair = (sop_class, ts)
+        if pair not in seen_pairs:
+            seen_pairs.add(pair)
+            context_pairs.append(pair)
+
+    if len(context_pairs) > 128:
+        log(
+            f"Error: files span {len(context_pairs)} distinct SOP "
+            f"Class/Transfer Syntax combinations, which exceeds the 128 "
+            f"presentation contexts allowed per association. Split the "
+            f"send into smaller batches.",
+            err=True,
+        )
+        result["error"] = "too_many_presentation_contexts"
+        return result
+
+    # ---- Step 5: build AE, request contexts, associate ----
+    ae = AE(ae_title=calling_ae_title)
+    for sop_class, ts in context_pairs:
+        if ts:
+            ae.add_requested_context(sop_class, ts)
+        else:
+            # Transfer syntax couldn't be determined for this file
+            # (unexpected/corrupt file_meta) — fall back to pynetdicom's
+            # default uncompressed transfer syntax list rather than failing.
+            ae.add_requested_context(sop_class)
+
+    ae.acse_timeout = timeout
+    ae.dimse_timeout = timeout
+    ae.network_timeout = timeout
+
+    distinct_sop_classes = {p[0] for p in context_pairs}
+    log(
+        f"\nConnecting to {ae_title}@{host}:{port} "
+        f"(calling AE: {calling_ae_title}) requesting "
+        f"{len(context_pairs)} presentation context(s) covering "
+        f"{len(distinct_sop_classes)} SOP Class(es)..."
+    )
+
+    try:
+        assoc = ae.associate(host, port, ae_title=ae_title)
+    except Exception as exc:
+        log(f"Association error: {exc}", err=True)
+        result["error"] = f"association_error: {exc}"
+        return result
+
+    if not assoc.is_established:
+        log(
+            "FAILED: could not establish association. Check host/port/AE title, "
+            "that the destination accepts the SOP classes being sent, and that "
+            "it allows associations from this calling AE title.",
+            err=True,
+        )
+        result["error"] = "association_not_established"
+        return result
+
+    # ---- Step 6: send each file via C-STORE ----
+    send_results = []  # (filepath, accession, code, label)
+    total = len(file_info)
+    try:
+        for i, (filepath, row, sop_class, ts) in enumerate(file_info, start=1):
+            accession = row.get("AccessionNumber", "")
+            if progress_callback:
+                progress_callback(i, total, filepath, accession)
+            if verbose:
+                sys.stdout.write(f"\rSending {i}/{total} (accession {accession})...")
+                sys.stdout.flush()
+
+            try:
+                status = assoc.send_c_store(filepath)
+            except Exception as exc:
+                send_results.append((filepath, accession, None, f"Exception: {exc}"))
+                continue
+
+            code, label = describe_status(status)
+            send_results.append((filepath, accession, code, label))
+
+            if delay:
+                time.sleep(delay)
+    finally:
+        assoc.release()
+
+    if verbose:
+        print()  # newline after progress line
+
+    result["results"] = send_results
+
+    # ---- Step 7: summarize ----
+    succeeded = [r for r in send_results if r[2] is not None and is_success(r[2])]
+    failed = [r for r in send_results if r not in succeeded]
+
+    log(f"\n--- Send Summary ---")
+    log(f"Total attempted: {len(send_results)}")
+    log(f"Succeeded:       {len(succeeded)}")
+    log(f"Failed:          {len(failed)}")
+
+    if failed:
+        log("\nFailed files:")
+        for filepath, accession, code, label in failed:
+            code_str = f"0x{code:04X}" if code is not None else "N/A"
+            log(f"  [{accession}] {filepath}")
+            log(f"      -> status {code_str}: {label}")
+
+    return result
+
+
+def write_report_csv(path, results):
+    """Write the (filepath, accession, code, label) results list to a CSV report."""
+    with open(path, "w", newline="", encoding="utf-8") as f:
+        writer = csv.writer(f)
+        writer.writerow(["FilePath", "AccessionNumber", "StatusCode", "StatusLabel"])
+        for filepath, accession, code, label in results:
+            code_str = f"0x{code:04X}" if code is not None else ""
+            writer.writerow([filepath, accession, code_str, label])
+
+
+# ---------------------------------------------------------------------------
+# Subcommand: send (thin CLI wrapper around send_accessions())
 # ---------------------------------------------------------------------------
 def cmd_send(args):
     # ---- Step 1: resolve the set of accession numbers to send ----
@@ -251,212 +624,30 @@ def cmd_send(args):
         print(f"Error: index source file not found: {args.source_index}", file=sys.stderr)
         sys.exit(1)
 
-    print(f"Looking up {len(accession_numbers)} accession number(s) in index '{args.source_index}'...")
+    result = send_accessions(
-
+        source_index=args.source_index,
-    # ---- Step 2: find matching files in the index ----
+        accession_numbers=accession_numbers,
-    matched_rows = list(find_matching_files(args.source_index, accession_numbers))
+        host=args.host,
-
+        port=args.port,
-    if not matched_rows:
+        ae_title=args.ae_title,
-        print("No matching files found in the index for the given accession number(s).")
+        calling_ae_title=args.calling_ae_title,
-        sys.exit(0)
+        timeout=args.timeout,
-
+        delay=args.delay,
-    found_accessions = {r["AccessionNumber"] for r in matched_rows}
+        dry_run=args.dry_run,
-    missing_accessions = accession_numbers - found_accessions
+        verbose=True,
-    print(f"Found {len(matched_rows)} file(s) across {len(found_accessions)} accession number(s).")
-    if missing_accessions:
-        print(
-            f"Warning: {len(missing_accessions)} accession number(s) had no matching "
-            f"file in the index: {', '.join(sorted(missing_accessions)[:10])}"
-            + (" ..." if len(missing_accessions) > 10 else ""),
-            file=sys.stderr,
-        )
-
-    # ---- Step 3: verify files exist on disk, and (for real sends) preload
-    #      each file's SOP Class UID + Transfer Syntax UID so we can build
-    #      the minimal set of presentation contexts to request ----
-    sendable = []
-    missing_files = []
-    for row in matched_rows:
-        filepath = row["FilePath"]
-        if os.path.isfile(filepath):
-            sendable.append(row)
-        else:
-            missing_files.append(filepath)
-
-    if missing_files:
-        print(
-            f"Warning: {len(missing_files)} file(s) listed in the index no longer "
-            f"exist on disk and will be skipped:",
-            file=sys.stderr,
-        )
-        for fp in missing_files[:10]:
-            print(f"  - {fp}", file=sys.stderr)
-        if len(missing_files) > 10:
-            print(f"  ... and {len(missing_files) - 10} more", file=sys.stderr)
-
-    if not sendable:
-        print("No sendable files remain after checking disk. Aborting.")
-        sys.exit(1)
-
-    if args.dry_run:
-        print("\n--- DRY RUN: the following files would be sent ---")
-        for row in sendable:
-            print(f"  [{row.get('AccessionNumber','')}] {row['FilePath']}")
-        print(f"\nTotal: {len(sendable)} file(s). No network connection was made.")
-        return
-
-    # ---- Step 4: read each file's SOP Class UID / Transfer Syntax UID to
-    #      build presentation contexts (max 128 per association) ----
-    print("Reading SOP Class / Transfer Syntax info from files to build presentation contexts...")
-    file_info = []  # (filepath, row, sop_class_uid, transfer_syntax_uid) or None on read failure
-    unreadable = []
-    for row in sendable:
-        filepath = row["FilePath"]
-        try:
-            ds_meta = pydicom.dcmread(filepath, stop_before_pixels=True, force=True)
-            sop_class = getattr(ds_meta, "SOPClassUID", None)
-            ts = ds_meta.file_meta.TransferSyntaxUID if hasattr(ds_meta, "file_meta") else None
-            if sop_class is None:
-                unreadable.append(filepath)
-                continue
-            file_info.append((filepath, row, str(sop_class), str(ts) if ts else None))
-        except Exception as exc:
-            unreadable.append(filepath)
-
-    if unreadable:
-        print(
-            f"Warning: {len(unreadable)} file(s) could not be read as DICOM and "
-            f"will be skipped:",
-            file=sys.stderr,
-        )
-        for fp in unreadable[:10]:
-            print(f"  - {fp}", file=sys.stderr)
-
-    if not file_info:
-        print("No readable DICOM files remain. Aborting.")
-        sys.exit(1)
-
-    # Build the minimal set of presentation contexts needed: one context per
-    # DISTINCT (SOP Class UID, Transfer Syntax UID) pair actually used by the
-    # files being sent. We deliberately do NOT bundle multiple transfer
-    # syntaxes into a single context, because the peer only accepts ONE
-    # transfer syntax per context — if a SOP Class has some files in (say)
-    # uncompressed Explicit VR LE and others in JPEG Lossless, bundling them
-    # into one context risks the peer accepting only one of the two, and
-    # files using the other transfer syntax would still fail to send. A
-    # separate context per pair guarantees every transfer syntax actually
-    # present gets its own negotiation slot.
-    #
-    # This also fixes the root cause of "No presentation context ... has
-    # been accepted by the peer with '<transfer syntax>'" errors: previously
-    # contexts were requested using pynetdicom's DEFAULT transfer syntax
-    # list (uncompressed only), so compressed files (JPEG Lossless, JPEG
-    # 2000, RLE, etc.) never had a matching context at all.
-    seen_pairs = set()
-    context_pairs = []  # list of (sop_class, transfer_syntax_or_None)
-    for _, _, sop_class, ts in file_info:
-        pair = (sop_class, ts)
-        if pair not in seen_pairs:
-            seen_pairs.add(pair)
-            context_pairs.append(pair)
-
-    if len(context_pairs) > 128:
-        print(
-            f"Error: files span {len(context_pairs)} distinct SOP "
-            f"Class/Transfer Syntax combinations, which exceeds the 128 "
-            f"presentation contexts allowed per association. Split the "
-            f"send into smaller batches.",
-            file=sys.stderr,
-        )
-        sys.exit(1)
-
-    # ---- Step 5: build AE, request contexts, associate ----
-    ae = AE(ae_title=args.calling_ae_title)
-    for sop_class, ts in context_pairs:
-        if ts:
-            ae.add_requested_context(sop_class, ts)
-        else:
-            # Transfer syntax couldn't be determined for this file
-            # (unexpected/corrupt file_meta) — fall back to pynetdicom's
-            # default uncompressed transfer syntax list rather than failing.
-            ae.add_requested_context(sop_class)
-
-    ae.acse_timeout = args.timeout
-    ae.dimse_timeout = args.timeout
-    ae.network_timeout = args.timeout
-
-    distinct_sop_classes = {p[0] for p in context_pairs}
-    print(
-        f"\nConnecting to {args.ae_title}@{args.host}:{args.port} "
-        f"(calling AE: {args.calling_ae_title}) requesting "
-        f"{len(context_pairs)} presentation context(s) covering "
-        f"{len(distinct_sop_classes)} SOP Class(es)..."
     )
 
-    try:
+    if result.get("error"):
-        assoc = ae.associate(args.host, args.port, ae_title=args.ae_title)
-    except Exception as exc:
-        print(f"Association error: {exc}", file=sys.stderr)
         sys.exit(1)
 
-    if not assoc.is_established:
+    if result["dry_run"] or result["sendable_count"] == 0:
-        print(
+        return
-            "FAILED: could not establish association. Check host/port/AE title, "
-            "that the destination accepts the SOP classes being sent, and that "
-            "it allows associations from this calling AE title.",
-            file=sys.stderr,
-        )
-        sys.exit(1)
 
-    # ---- Step 6: send each file via C-STORE ----
+    send_results = result["results"]
-    results = []  # (filepath, accession, code, label)
+    failed = [r for r in send_results if r[2] is None or not is_success(r[2])]
-    try:
-        for i, (filepath, row, sop_class, ts) in enumerate(file_info, start=1):
-            accession = row.get("AccessionNumber", "")
-            sys.stdout.write(
-                f"\rSending {i}/{len(file_info)} (accession {accession})..."
-            )
-            sys.stdout.flush()
-
-            try:
-                status = assoc.send_c_store(filepath)
-            except Exception as exc:
-                results.append((filepath, accession, None, f"Exception: {exc}"))
-                continue
-
-            code, label = describe_status(status)
-            results.append((filepath, accession, code, label))
-
-            if args.delay:
-                time.sleep(args.delay)
-    finally:
-        assoc.release()
-
-    print()  # newline after progress line
-
-    # ---- Step 7: report results ----
-    succeeded = [r for r in results if r[2] is not None and is_success(r[2])]
-    failed = [r for r in results if r not in succeeded]
-
-    print(f"\n--- Send Summary ---")
-    print(f"Total attempted: {len(results)}")
-    print(f"Succeeded:       {len(succeeded)}")
-    print(f"Failed:          {len(failed)}")
-
-    if failed:
-        print("\nFailed files:")
-        for filepath, accession, code, label in failed:
-            code_str = f"0x{code:04X}" if code is not None else "N/A"
-            print(f"  [{accession}] {filepath}")
-            print(f"      -> status {code_str}: {label}")
 
     if args.report:
-        with open(args.report, "w", newline="", encoding="utf-8") as f:
+        write_report_csv(args.report, send_results)
-            writer = csv.writer(f)
-            writer.writerow(["FilePath", "AccessionNumber", "StatusCode", "StatusLabel"])
-            for filepath, accession, code, label in results:
-                code_str = f"0x{code:04X}" if code is not None else ""
-                writer.writerow([filepath, accession, code_str, label])
         print(f"\nDetailed report saved to: {args.report}")
 
     if failed: