Best Unstructured Alternative for PDF Ingestion

import os
import json
import time

import requests

# Step 1: The Unstructured Partition Endpoint partitions the PDF and returns typed elements by default
# Chunking is opt-in: add chunking_strategy "by_title" and tune size limits
with open("/path/to/your/report.pdf", "rb") as f:
    partition = requests.post(
        "https://api.unstructuredapp.io/general/v0/general",
        headers={"unstructured-api-key": os.environ["UNSTRUCTURED_API_KEY"]},
        files={"files": ("report.pdf", f, "application/pdf")},
        data={
            "strategy": "hi_res",
            # Add "chunking_strategy": "by_title" to receive chunks instead of elements
            # Add "max_characters" and "overlap" to tune chunk size
        },
        timeout=300,
    )
partition.raise_for_status()
elements = partition.json()

for element in elements:
    print(element["type"], element["text"])

# Output: typed elements such as Title, NarrativeText, Table, ListItem
# Map element types and metadata to your vector store schema before loading

# Step 2: EnConvert uploads the PDF once and queues an ingest job
with open("/path/to/your/report.pdf", "rb") as f:
    upload = requests.post(
        "https://api.enconvert.com/v2/ingest/files",
        headers={"X-API-Key": os.environ["ENCONVERT_API_KEY"]},
        files={"files": ("report.pdf", f, "application/pdf")},
        timeout=120,
    )
upload.raise_for_status()
job_id = upload.json()["job_id"]

# Step 3: Poll until the job completes (queued, processing, then completed)
while True:
    status_response = requests.get(
        f"https://api.enconvert.com/v2/ingest/{job_id}",
        headers={"X-API-Key": os.environ["ENCONVERT_API_KEY"]},
        timeout=30,
    )
    status_response.raise_for_status()
    job = status_response.json()
    if job["status"] == "completed":
        break
    if job["status"] in ("failed", "canceled"):
        raise RuntimeError(job.get("error_message") or f"Ingest job {job['status']}")
    time.sleep(2)

# Step 4: Download the JSONL, one chunk per line
# The output_url is signed (valid 15 minutes). Plain GET, no API key.
jsonl_response = requests.get(job["output_url"], timeout=60)
jsonl_response.raise_for_status()

# Split on "\n" only: str.splitlines() would also split on unicode line separators inside chunk text
chunks = [
    json.loads(line)
    for line in jsonl_response.content.decode("utf-8").split("\n")
    if line.strip()
]
print(f"Loaded {len(chunks)} chunks ready for embedding")

# Each chunk has "content" (the text to embed) plus "metadata" with source_url,
# title, headings_path (a list), section, chunk_index and word_count.
# Embed chunk["content"] and store chunk["metadata"] alongside it. If your vector
# store only accepts scalar metadata, join the list: " > ".join(headings_path)
for chunk in chunks[:3]:
    print(" > ".join(chunk["metadata"]["headings_path"]), "|", chunk["content"][:80])