c = get_http_client()

r = await safe_fetch(c, url, timeout=60)

r.raise_for_status()

dst.write_bytes(r.content)

fmt: str = "markdown") -> tuple[str, int]:

"""Fast text-layer extraction. Returns (content, page_count); text empty => scanned."""

doc = _pymupdf_open(path, password)

try:

total = doc.page_count

wanted: list[int] = []

if pages:

sep = pages.replace(" ", "")

for rng in sep.split(","):

if not rng:

continue

if "-" in rng:

a, _, b = rng.partition("-")

lo = max(1, int(a))

hi = min(total, int(b)) if b else total

wanted.extend(range(lo, hi + 1))

else:

p = int(rng)

if 1 <= p <= total:

wanted.append(p)

else:

wanted = list(range(1, total + 1))

if fmt == "markdown":

parts = [doc[p - 1].get_text("text") for p in wanted]

return "\n\n".join(parts).strip(), total

if fmt == "json":

obj = [{"page": p, "text": doc[p - 1].get_text("text")} for p in wanted]

return json.dumps(obj, ensure_ascii=False), total

if fmt == "html":

parts = [doc[p - 1].get_text("html") for p in wanted]

return "\n".join(parts).strip(), total

raise ValueError(f"unsupported format: {fmt}")

finally:

"""Fraction of chars in the Private Use Area (broken ToUnicode maps).

if not text:

return 0.0

bad = sum(1 for ch in text if 0xE000 <= ord(ch) <= 0xF8FF or 0xF0000 <= ord(ch) <= 0xFFFFD)

"""Import docling + build the converter (slow; 30s init guard)."""

os.environ.setdefault("CUDA_VISIBLE_DEVICES", "")

from docling.document_converter import DocumentConverter, PdfFormatOption

from docling.datamodel.base_models import InputFormat

from docling.datamodel.pipeline_options import (

AcceleratorDevice,

AcceleratorOptions,

PdfPipelineOptions,

RapidOcrOptions,

)

opts = PdfPipelineOptions()

opts.accelerator_options = AcceleratorOptions(device=AcceleratorDevice.CPU, num_threads=2)

opts.do_table_structure = False

opts.do_formula_enrichment = False

opts.do_code_enrichment = False

opts.do_picture_classification = False

opts.do_picture_description = False

opts.ocr_options = RapidOcrOptions()

"""Full OCR pipeline (CPU, rapidocr backend). Called only when no usable

src = path

tmp_subset = ""

if max_pages:

doc = _pymupdf_open(path)

try:

if doc.page_count > max_pages:

doc.select(list(range(max_pages)))

tmp_subset = path + ".subset.pdf"

doc.save(tmp_subset, garbage=3, deflate=True)

src = tmp_subset

finally:

doc.close()

try:

res = conv.convert(src)

doc = res.document

out: dict[str, str] = {}

if "markdown" in fmts:

out["markdown"] = (doc.export_to_markdown() or "").strip()

if "json" in fmts:

out["json"] = json.dumps(doc.export_to_dict(), ensure_ascii=False)

if "html" in fmts:

out["html"] = (doc.export_to_html() or "").strip()

if len(fmts) == 1:

return out[fmts[0]]

return out

finally:

if tmp_subset and os.path.isfile(tmp_subset):

try:

os.remove(tmp_subset)

except OSError:

fmts = [f.strip() for f in raw.split(",") if f.strip()]

bad = [f for f in fmts if f not in _FORMATS]

if bad:

raise ValueError(f"unsupported format(s): {', '.join(bad)}; supported: {', '.join(_FORMATS)}")

input_path: str | list[str],

output_dir: str = "",

format: str = "markdown",

password: str = "",

pages: str = "",

hybrid: str = "",

hybrid_mode: str = "",

) -> dict[str, Any]:

sources = [input_path] if isinstance(input_path, str) else input_path

local_files: list[str] = []

skipped: list[str] = []

tmpdir = ""

out_tmpdir = ""

force_ocr = bool(hybrid) or bool(hybrid_mode)

results: dict[str, Any] = {}

fmts = _formats_arg(format)

try:

for src in sources:

src = src.strip()

if not src:

continue

if os.path.isfile(src):

local_files.append(src)

continue

if src.startswith("file://"):

skipped.append(f"{src} (file:// URLs not supported — pass a local path instead)")

continue

if src.startswith(("http://", "https://")):

if not tmpdir:

tmpdir = tempfile.mkdtemp(prefix="pdfx_")

fname = os.path.basename(urlsplit(src).path) or f"download_{len(local_files)}.pdf"

if not fname.lower().endswith(".pdf"):

fname += ".pdf"

dl = Path(tmpdir) / fname

await _download_pdf(src, dl)

local_files.append(str(dl))

continue

local_files.append(src)

if not local_files:

if skipped:

return {"success": False,

"error": "No valid PDF files provided. Skipped: " + "; ".join(skipped)}

return {"success": False, "error": "No valid PDF files provided"}

for fp in local_files:

path = Path(fp)

t0 = time.monotonic()

method = "pymupdf"

try:

text, page_count = _pymupdf_extract(str(path), pages, password, "markdown")

# No usable text layer (empty, PUA-garbage) or forced OCR:

# donsetch fusion-trust audit — a PUA-heavy glyph stream is a

# broken encoding (garbage), not text, so it must be OCR'd.

if force_ocr or len(text) < 30 or _pua_ratio(text) > _PUA_FLOOR:

method = "docling-ocr"

# Init (import + converter build) has its own guard: ONNX

# Runtime C++ constructors can hang, and init failure should

# not burn the whole OCR budget (donsetch ocr.rs).

try:

conv = await asyncio.wait_for(

asyncio.to_thread(_docling_build), timeout=_OCR_INIT_BUDGET)

except asyncio.TimeoutError:

return {"success": False,

"error": f"OCR engine init timed out after {_OCR_INIT_BUDGET}s: {path.name}"}

content = await asyncio.wait_for(

asyncio.to_thread(

_docling_extract, conv, str(path), fmts, _OCR_MAX_PAGES),

timeout=_OCR_PAGE_BUDGET - _OCR_INIT_BUDGET)

elif len(fmts) == 1:

if fmts[0] != "markdown":

text = (await asyncio.to_thread(

_pymupdf_extract, str(path), pages, password, fmts[0]))[0]

content = text

else:

got: dict[str, str] = {}

for f in fmts:

got[f] = (await asyncio.to_thread(

_pymupdf_extract, str(path), pages, password, f))[0]

content = got

except asyncio.TimeoutError:

return {"success": False, "error": f"OCR timed out after {_OCR_PAGE_BUDGET}s: {path.name}"}

except Exception as e:

return {"success": False, "error": f"PDF extraction failed: {path.name}: {e}"}

if isinstance(content, str):

if len(content) > 50000:

content = content[:50000] + "\n\n[... truncated at 50000 chars ...]"

else:

for f, c in content.items():

if len(c) > 50000:

content[f] = c[:50000] + "\n\n[... truncated at 50000 chars ...]"

results[path.name] = {

"content": content,

"method": method,

"pages": page_count,

"elapsed_s": round(time.monotonic() - t0, 1),

}

if output_dir.strip():

out = Path(output_dir.strip())

out.mkdir(parents=True, exist_ok=True)

else:

out_tmpdir = tempfile.mkdtemp(prefix="pdfx_out_")

out = Path(out_tmpdir)

for name, r in results.items():

# H1-para: contain writes inside `out` — `name` derives from a remote

# URL path basename which can carry `..` and escape output_dir.

safe_stem = Path(name).stem

if ".." in safe_stem or "" == safe_stem or safe_stem in (".", ".."):

safe_stem = "extract"

if isinstance(r["content"], str):

(out / f"{safe_stem}.{_EXT[fmts[0]]}").write_text(r["content"], encoding="utf-8")

else:

for f, c in r["content"].items():

(out / f"{safe_stem}.{_EXT[f]}").write_text(c, encoding="utf-8")

return {

"success": True,

"files": str(out),

"results": results,

"input_count": len(local_files),

}

except Exception as e:

return {"success": False, "error": f"PDF extraction failed: {e}"}

finally:

if tmpdir and os.path.isdir(tmpdir):

shutil.rmtree(tmpdir, ignore_errors=True)

if out_tmpdir and os.path.isdir(out_tmpdir):