Best Docling Alternative for RAG Pipeline Ingestion

import os
import json
import time

import requests
from docling.chunking import HybridChunker
from docling.document_converter import DocumentConverter

FILE_PATH = "/path/to/your/report.pdf"  # a local PDF, DOCX, PPTX, XLSX, CSV, HTML, or Markdown file
HEADERS = {"X-API-Key": os.environ["ENCONVERT_API_KEY"]}

# Docling: convert and chunk locally
# (pip install docling; needs Python 3.10+ and compute you provide)
converter = DocumentConverter()
result = converter.convert(FILE_PATH)

chunker = HybridChunker()
docling_chunks = [chunk.text for chunk in chunker.chunk(dl_doc=result.document)]

# EnConvert: upload the file, poll for completion, download JSONL chunks
# Step 1: Upload the file (the multipart field is named "files")
with open(FILE_PATH, "rb") as f:
    upload = requests.post(
        "https://api.enconvert.com/v2/ingest/files",
        headers=HEADERS,
        files={"files": (os.path.basename(FILE_PATH), f)},
        timeout=120,
    )
upload.raise_for_status()
job_id = upload.json()["job_id"]

# Step 2: Poll until the job completes (give up after 5 minutes)
deadline = time.monotonic() + 300
while True:
    poll = requests.get(
        f"https://api.enconvert.com/v2/ingest/{job_id}",
        headers=HEADERS,
        timeout=30,
    )
    poll.raise_for_status()
    job = poll.json()
    if job["status"] == "completed":
        break
    if job["status"] in ("failed", "canceled"):
        raise RuntimeError(f"Ingest {job['status']}: {job.get('error_message')}")
    if time.monotonic() > deadline:
        raise TimeoutError(f"Ingest job {job_id} is still {job['status']} after 5 minutes")
    time.sleep(2)

# Step 3: Download the JSONL, one chunk per line. The output_url is a signed URL
# (valid 15 minutes): plain GET, no API key.
jsonl = requests.get(job["output_url"], timeout=60)
jsonl.raise_for_status()
chunks = [json.loads(line) for line in jsonl.iter_lines() if line]

print(f"Docling: {len(docling_chunks)} chunks, EnConvert: {len(chunks)} chunks")

# Each chunk has "content" (the text to embed) plus "metadata": source_url (the
# filename), title, headings_path (a list of headings), section, word_count, chunk_index
if chunks:
    first = chunks[0]
    print(first["metadata"]["headings_path"], first["content"][:200])