Importing reports from pdf files

[Drmaestro]

Hi,

I am a radiologist/researcher with my own Orthanc PACS system set up as a Docker deployment on my Synology NAS. Now, I am trying to find a solution for organizing/storing/accessing some reports associated with the imaging findings.

I've heard about Radis from ChatGPT, which I use very frequently for IT related issues. It seems to be a very useful solution. Whit GPT's help, I have been able to delpoy RADIS on the NAS, and while it took required a lot of tinkering during setup, it seems to work without issues now.

I have been able to experiment with the included database and file examples but now I want to import my own reports. I have created a new group and now trying to find a way to import reports.

The reports are PDF based. They are text renditions, not images, so they can be searched and read. The importing strucutre of Radis seems to be different though. I haven't seen a way to import the pdf files and extract the relevent information from them using the Radis UI.

How should I proceed? I understand this is a github group, not a forum, so I am sorry if the question is not appropriate, as it isn't related to a development issue. In that case, I'd be grateful if you could lead me to an existing forum/discord/slack server.

While I don't have a strong IT background, I'll be delighted if I also could help with the development process, in case you need help with ultrasound related issues.

Best regards

[medihack]

Hi @Drmaestro! RADIS is completely independent of how the reports are provided. In Heidelberg, we are using a Dagster pipeline that uploads the reports to RADIS (using the RADIS client to the RADIS API of the server). Unfortunately, there is no official standard for where to obtain the reports or how they are stored (unlike radiology images, which use DICOM). So it always depends on the local infrastructure. ADIT and RADIS are not yet officially announced yet, but we are working on the documentation and plan to make an official release soon.

[Drmaestro]

Hello again,

Thank you very much for the information. I'll use Gemini or Chat GPT to write a Python program for this purpose. Gemini actually proposed the following:

"The Missing Piece: An "ETL" Script
Since you have a folder of PDFs on your Synology NAS, you need a small Python script to sit between your files and this form. This script will perform three steps:

Extract: Open each PDF and grab the text (OCR if needed).

Parse: Look for specific keywords (like "Patient ID:" or "Accession No:") to fill in the fields you see in your screenshot.

Load: Push that data into RADIS automatically. "

I'll try this approach to see if it works.

[Drmaestro]

Hi,

I have been able to write some python code using ChatGPT and it seems to be able to extract the required fields from the PDF report and copies them to the appropriate locations on the database. It takes around 20 seconds to porcess 8 pdfs, so 2-3 seconds per pdf. Here's the code:

`from future import annotations

import json
import re
import uuid
import datetime as dt
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor, as_completed

import requests
import fitz # PyMuPDF
from radis_client.client import RadisClient, ReportData

-----------------------

CONFIG

-----------------------

SERVER_URL = "The IP and port of the server"
AUTH_TOKEN = "Auth Token"

GROUP_ID = 4
RADIS_LANGUAGE = "en" # keep "en" even if Turkish text
MODALITIES = ["US"]

Folder behavior: run-in-place (copy .py into folder and run)

INPUT_ROOT = Path(file).resolve().parent
RECURSIVE = True

Concurrency:

5900X sweet spot for small PDFs: 12–16 extract workers; keep upload sequential.

EXTRACT_WORKERS = 12

Batch behavior:

DRY_RUN = False
CONTINUE_ON_ERROR = True
SKIP_ALREADY_DONE = True
SAVE_STATE_EVERY = 50 # save state file every N processed items

State / logging

STATE_FILE = INPUT_ROOT / "batch_state.json"
RUN_LOG_JSONL = INPUT_ROOT / "batch_upload_log.jsonl"

-----------------------

PARSING HELPERS

-----------------------

FILENAME_RE = re.compile(
r"^(?P.+?)-(?P.+?)-(?P\d{11})-(?P.+?)-(?P\d{4}-\d{2}-\d{2})_(?P\d{4}).pdf$",
re.IGNORECASE,
)

def parse_filename(p: Path) -> dict:
m = FILENAME_RE.match(p.name)
if not m:
raise ValueError(
f"Filename doesn't match:\n{p.name}\n"
"Expected: Name-Surname-TC-EXAM-YYYY-MM-DD_HHMM.pdf"
)
d = m.groupdict()
hhmm = d["time"]
study_dt_from_name = dt.datetime.fromisoformat(d["date"]) + dt.timedelta(
hours=int(hhmm[:2]), minutes=int(hhmm[2:])
)
return {
"tc": d["tc"],
"exam_short": d["exam"],
"study_dt_from_name": study_dt_from_name,
"name": d["name"],
"surname": d["surname"],
}

def parse_dt(s: str):
s = (s or "").strip()
for fmt in ("%d.%m.%Y %H:%M", "%d.%m.%Y"):
try:
return dt.datetime.strptime(s, fmt)
except ValueError:
pass
return None

def find_after_label(text: str, label: str):
t = text.replace("\r", "\n")
idx = t.find(label)
if idx == -1:
return None
tail = t[idx + len(label):]
for line in tail.splitlines():
v = line.strip()
if v:
return v
return None

def extract_fields_from_pdf_text(text: str) -> dict:
fields = {}
fields["exam_name"] = find_after_label(text, "Exam Type")
fields["patient_name"] = find_after_label(text, "Name Surname")
fields["protocol_no"] = find_after_label(text, "Protocol Nbr")
fields["patient_no"] = find_after_label(text, "Patient Nbr")

age_sex = find_after_label(text, "Age /Sex")
fields["age_sex_raw"] = age_sex
fields["age"] = None
fields["sex"] = "O"
if age_sex:
    m = re.match(r"(\d+)\s*/\s*(.+)", age_sex)
    if m:
        fields["age"] = int(m.group(1))
        sex_word = m.group(2).strip().lower()
        if "kad" in sex_word:
            fields["sex"] = "F"
        elif "erk" in sex_word:
            fields["sex"] = "M"

order_dt = find_after_label(text, "Order Date")
report_dt = find_after_label(text, "Report Date")
if not report_dt:
    # fallback for line-broken label in your PDFs
    m = re.search(r"Report.*?(\d{2}\.\d{2}\.\d{4}\s+\d{2}:\d{2})", text, re.S)
    if m:
        report_dt = m.group(1)

fields["order_dt_raw"] = order_dt
fields["report_dt_raw"] = report_dt
fields["order_dt"] = parse_dt(order_dt) if order_dt else None
fields["report_dt"] = parse_dt(report_dt) if report_dt else None
return fields

def extract_full_text(pdf_path: Path) -> str:
doc = fitz.open(pdf_path)
try:
parts = []
for page in doc:
parts.append(page.get_text("text"))
return "\n".join(parts).strip()
finally:
doc.close()

def extract_narrative_body(full_text: str) -> str:

t = full_text.replace("\r", "\n")

start_markers = [
    "Dear colleague",
    "Patient",
]

start_idx = None
for sm in start_markers:
    i = t.find(sm)
    if i != -1:
        start_idx = i
        break

if start_idx is None:
    return full_text.strip()

tail = t[start_idx:]

end_markers = [
    "\nRadiology",
    "\nDip.Nbr",
    "* This report",
    "\nForm\n",
    "\nRGK/",
    
]

end_positions = []
for em in end_markers:
    j = tail.find(em)
    if j != -1:
        end_positions.append(j)

if end_positions:
    tail = tail[: min(end_positions)]

lines = [ln.rstrip() for ln in tail.splitlines()]
while lines and not lines[0].strip():
    lines.pop(0)
while lines and not lines[-1].strip():
    lines.pop()

cleaned = "\n".join(lines).strip()
return cleaned if cleaned else full_text.strip()

-----------------------

STATE / LOGGING

-----------------------

def _load_state(path: Path) -> dict:
if not path.exists():
return {"done": {}}
try:
return json.loads(path.read_text(encoding="utf-8"))
except Exception:
backup = path.with_suffix(".corrupt.json")
path.replace(backup)
return {"done": {}}

def _save_state(path: Path, state: dict) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
tmp = path.with_suffix(".tmp")
tmp.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8")
tmp.replace(path)

def _append_jsonl(path: Path, record: dict) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("a", encoding="utf-8") as f:
f.write(json.dumps(record, ensure_ascii=False) + "\n")

def _iter_pdfs(root: Path, recursive: bool) -> list[Path]:
if recursive:
return sorted([p for p in root.rglob(".pdf") if p.is_file()])
return sorted([p for p in root.glob(".pdf") if p.is_file()])

-----------------------

BUILD REPORT (per PDF)

-----------------------

def build_report_from_pdf(pdf_path: Path) -> tuple[ReportData, dict]:
fn = parse_filename(pdf_path)
full_text = extract_full_text(pdf_path)
pdf_fields = extract_fields_from_pdf_text(full_text)
body = extract_narrative_body(full_text)

# Prefer report approval time from PDF; fallback to filename datetime
study_dt = pdf_fields["report_dt"] or fn["study_dt_from_name"]

# Approx DOB (API requirement) from age, if available
approx_dob = dt.date(1900, 1, 1)
if pdf_fields.get("age") is not None and isinstance(study_dt, dt.datetime):
    approx_year = max(1900, study_dt.year - int(pdf_fields["age"]))
    approx_dob = dt.date(approx_year, 1, 1)

report = ReportData(
    document_id=str(uuid.uuid4()),
    language=RADIS_LANGUAGE,
    groups=[GROUP_ID],

    pacs_aet="LOCAL",
    pacs_name="Local",
    pacs_link="",

    patient_id=fn["tc"],
    patient_birth_date=approx_dob,
    patient_sex=pdf_fields.get("sex", "O"),

    study_description=fn["exam_short"] if fn.get("exam_short") else (pdf_fields.get("exam_name") or "US"),
    study_datetime=study_dt,

    study_instance_uid="1.2.826.0.1.3680043.8.498." + uuid.uuid4().hex,
    accession_number=(pdf_fields.get("protocol_no") or ("ACC-" + uuid.uuid4().hex[:10])),

    modalities=MODALITIES,
    metadata={
        "source_filename": pdf_path.name,
        "patient_name_pdf": pdf_fields.get("patient_name") or "",
        "patient_no_pdf": pdf_fields.get("patient_no") or "",
        "protocol_no_pdf": pdf_fields.get("protocol_no") or "",
    },
    body=body,
)

dbg = {
    "pdf": str(pdf_path),
    "tc": fn["tc"],
    "accession": report.accession_number,
    "study_datetime": study_dt.isoformat() if isinstance(study_dt, (dt.datetime, dt.date)) else str(study_dt),
    "body_len": len(body or ""),
    "has_start_marker": ("Dear colleague" in (body or "")) or ("Patient" in (body or "")),
}
return report, dbg

-----------------------

PARALLEL EXTRACTION JOB

-----------------------

def extract_job(pdf_path: Path, done_map: dict, skip_done: bool) -> dict:
"""
Runs in worker threads: parse + extract + build ReportData.
Does NOT upload (upload is done in main thread, sequentially).
Returns a record dict with status: skipped|ready|error
"""
key = str(pdf_path.resolve()).lower()

if skip_done and key in done_map:
    return {
        "status": "skipped",
        "key": key,
        "pdf": str(pdf_path),
        "reason": "already_done",
    }

try:
    report, dbg = build_report_from_pdf(pdf_path)
    return {
        "status": "ready",
        "key": key,
        "report": report,
        "dbg": dbg,
    }
except Exception as e:
    return {
        "status": "error",
        "key": key,
        "pdf": str(pdf_path),
        "error": repr(e),
    }

-----------------------

MAIN

-----------------------

def main():
if not INPUT_ROOT.exists():
raise FileNotFoundError(INPUT_ROOT)

if not AUTH_TOKEN and not DRY_RUN:
    raise RuntimeError('RADIS_TOKEN is not set. In PowerShell:  $env:RADIS_TOKEN="..."')

pdfs = _iter_pdfs(INPUT_ROOT, RECURSIVE)
if not pdfs:
    print("No PDFs found under:", INPUT_ROOT)
    return

state = _load_state(STATE_FILE) if SKIP_ALREADY_DONE else {"done": {}}
done_map = state.get("done", {})

client = None if DRY_RUN else RadisClient(server_url=SERVER_URL, auth_token=AUTH_TOKEN)

run_id = dt.datetime.now().strftime("%Y%m%d_%H%M%S")
total = len(pdfs)

ok = 0
skipped = 0
failed = 0
processed = 0

print(
    f"Batch start: {total} PDFs | root={INPUT_ROOT} | recursive={RECURSIVE} | "
    f"dry_run={DRY_RUN} | extract_workers={EXTRACT_WORKERS}"
)

# Parallel extraction (fast), sequential upload (safe)
with ThreadPoolExecutor(max_workers=EXTRACT_WORKERS) as ex:
    futures = [ex.submit(extract_job, p, done_map, SKIP_ALREADY_DONE) for p in pdfs]

    for fut in as_completed(futures):
        rec = fut.result()

        if rec["status"] == "skipped":
            skipped += 1
            processed += 1
            _append_jsonl(RUN_LOG_JSONL, {
                "run_id": run_id,
                "ts": dt.datetime.now().isoformat(),
                "status": "skipped",
                "pdf": rec["pdf"],
                "reason": rec.get("reason", ""),
            })

        elif rec["status"] == "error":
            failed += 1
            processed += 1
            _append_jsonl(RUN_LOG_JSONL, {
                "run_id": run_id,
                "ts": dt.datetime.now().isoformat(),
                "status": "error",
                "pdf": rec["pdf"],
                "error": rec["error"],
            })
            if not CONTINUE_ON_ERROR:
                raise RuntimeError(f"Extraction failed: {rec['pdf']} -> {rec['error']}")

        else:
            # ready: upload sequentially (or dry-run)
            key = rec["key"]
            report: ReportData = rec["report"]
            dbg = rec["dbg"]

            if DRY_RUN:
                ok += 1
                processed += 1
                done_map[key] = {"status": "dry_run_ok", "ts": dt.datetime.now().isoformat()}
                _append_jsonl(RUN_LOG_JSONL, {
                    "run_id": run_id,
                    "ts": dt.datetime.now().isoformat(),
                    "status": "dry_run_ok",
                    **dbg,
                })
            else:
                try:
                    created = client.create_report(report)
                    ok += 1
                    processed += 1
                    done_map[key] = {
                        "status": "uploaded",
                        "radis_id": created.get("id"),
                        "ts": dt.datetime.now().isoformat(),
                    }
                    _append_jsonl(RUN_LOG_JSONL, {
                        "run_id": run_id,
                        "ts": dt.datetime.now().isoformat(),
                        "status": "uploaded",
                        "radis_id": created.get("id"),
                        **dbg,
                    })
                except requests.HTTPError as e:
                    failed += 1
                    processed += 1
                    resp = e.response
                    _append_jsonl(RUN_LOG_JSONL, {
                        "run_id": run_id,
                        "ts": dt.datetime.now().isoformat(),
                        "status": "http_error",
                        "pdf": dbg.get("pdf", ""),
                        "http_status": resp.status_code if resp else None,
                        "url": resp.url if resp else None,
                        "response_head": (resp.text[:2000] if resp and resp.text else ""),
                        **{k: v for k, v in dbg.items() if k != "pdf"},
                    })
                    if not CONTINUE_ON_ERROR:
                        raise

        # Periodic state save + progress
        if SKIP_ALREADY_DONE and processed % SAVE_STATE_EVERY == 0:
            state["done"] = done_map
            _save_state(STATE_FILE, state)

        if processed % 50 == 0 or processed == total:
            print(f"Progress: {processed}/{total} | ok={ok} skipped={skipped} failed={failed}")

if SKIP_ALREADY_DONE:
    state["done"] = done_map
    _save_state(STATE_FILE, state)

print(f"Batch done. ok={ok} skipped={skipped} failed={failed}")
print("Log:", RUN_LOG_JSONL)
if SKIP_ALREADY_DONE:
    print("State:", STATE_FILE)

if name == "main":
main()
`

This is of course specific to my reports but maybe can be helpful for people who are trying to import thier own.