> For clean Markdown of any page, append .md to the page URL.
> For a complete documentation index, see https://docs.emtelligent.com/llms.txt.
> For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.emtelligent.com/_mcp/server.

# Example Python Client

Below is an example Python CLI that submits documents to an OCR server, polls for status, retrieves full document text, and collects timing stats.

Please refer to the [end-user tutorial](/ocr/end-user-tutorial) for detailed usage instructions regarding individual API endpoints.

## Usage

The client requires the `requests` and `tqdm` Python packages.

```bash
OCR_API_KEY=my-secret-api-key python example_client.py \
    --server https://ocr.emtelligent.com \
    --concurrency 8 \
    pdf_dir/*.pdf
```

Document text for each successfully processed PDF will be placed in the same directory as the original file, with a `.txt` extension instead of `.pdf`
(e.g. `dir/doc.pdf` will produce `dir/doc.txt`)

Passing `-r full` instead saves each document's complete result JSON (including per-page timings) with a `.json` extension.

The client sends the SHA-256 hash of each file as its `Idempotency-Key`. When you run the client again on the same files, the server returns the existing jobs, including failed ones, and does not process the documents again until their jobs are deleted or expire. Two files with the same content share one job.

The CLI also provides usage instructions when invoked with `-h`.

## Code

```python title={"example_client.py"}
#!/usr/bin/env python3
"""Bulk submit PDFs to an OCR server, download text, and report timings."""

import argparse
import hashlib
import json
import os
import statistics
import sys
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional
from urllib.parse import quote

import requests
from requests.adapters import HTTPAdapter
from tqdm import tqdm
from urllib3.util.retry import Retry


DEFAULT_SERVER = "https://ocr.emtelligent.com"
DEFAULT_CONCURRENCY = 8
DEFAULT_JOB_TIMEOUT = 3600
POLL_INTERVAL = 2.0


@dataclass
class DocResult:
    pdf: Path
    state: str = "pending"            # "ok" | "failed"
    error: Optional[str] = None
    duration_s: Optional[float] = None  # client wall-clock: submit -> result saved
    total_pages: int = 0
    failed_pages: int = 0
    render_ms: list = field(default_factory=list)  # per-page, server-side
    ocr_ms: list = field(default_factory=list)     # per-page, server-side


@dataclass
class Stats:
    submitted: int = 0
    completed: int = 0
    failed: int = 0
    pages_done: int = 0
    lock: threading.Lock = field(default_factory=threading.Lock)

    def snapshot(self) -> dict:
        with self.lock:
            return {
                "ok": self.completed,
                "fail": self.failed,
                "pages": self.pages_done,
                "in_flight": self.submitted - self.completed - self.failed,
            }


def submit_pdf(session, server, pdf_path):
    with open(pdf_path, "rb") as f:
        body = f.read()
    r = session.post(
        f"{server}/jobs",
        data=body,
        headers={
            "Content-Type": "application/pdf",
            "X-Filename": quote(pdf_path.name),
            "Idempotency-Key": hashlib.sha256(body).hexdigest(),
        },
        timeout=120,
    )
    r.raise_for_status()
    return r.json()["job_id"]


def wait_for_job(session, server, job_id, job_timeout):
    deadline = time.monotonic() + job_timeout
    while True:
        r = session.get(f"{server}/jobs/{job_id}", timeout=60)
        r.raise_for_status()
        status = r.json()
        if status.get("state") in ("done", "failed"):
            return status
        if time.monotonic() > deadline:
            raise TimeoutError(f"job {job_id} not finished after {job_timeout} s")
        time.sleep(POLL_INTERVAL)


def fetch_result(session, server, job_id):
    r = session.get(f"{server}/jobs/{job_id}/result",
                    params={"timings": "true"}, timeout=120)
    r.raise_for_status()
    return r.json()


def save_text(result, out_path):
    pages = sorted(result.get("pages", []), key=lambda p: p["page_number"])
    text = "\n\n".join(p["text"] for p in pages if p.get("text"))
    out_path.write_text(text, encoding="utf-8")


def process_one(pdf, server, session, stats, bar, results_type, job_timeout,
                results):
    rec = DocResult(pdf=pdf)
    t0 = time.monotonic()
    try:
        with stats.lock:
            stats.submitted += 1
        bar.set_postfix(stats.snapshot(), refresh=False)
        job_id = submit_pdf(session, server, pdf)

        status = wait_for_job(session, server, job_id, job_timeout)
        if status.get("state") != "done":
            raise RuntimeError(status.get("error") or "job failed")

        result = fetch_result(session, server, job_id)
        if results_type == "full":
            with open(pdf.with_suffix(".json"), "w") as f:
                json.dump(result, f)
        else:
            save_text(result, pdf.with_suffix(".txt"))

        rec.duration_s = time.monotonic() - t0
        rec.total_pages = result.get("total_pages", 0)
        rec.failed_pages = len(result.get("failed_pages") or [])

        for p in (result.get("page_timings") or {}).get("pages", []):
            if p.get("render_ms") is not None:
                rec.render_ms.append(p["render_ms"])
            if p.get("ocr_ms") is not None:
                rec.ocr_ms.append(p["ocr_ms"])

        rec.state = "ok"
        with stats.lock:
            stats.completed += 1
            stats.pages_done += rec.total_pages
    except Exception as e:
        rec.state = "failed"
        rec.error = f"{type(e).__name__}: {e}"
        rec.duration_s = time.monotonic() - t0
        with stats.lock:
            stats.failed += 1
    finally:
        results.append(rec)
        bar.set_postfix(stats.snapshot(), refresh=False)
        bar.update(1)


# --- summary helpers ---------------------------------------------------------

def pct(values, p):
    if not values:
        return None
    s = sorted(values)
    k = (len(s) - 1) * p
    lo = int(k)
    hi = min(lo + 1, len(s) - 1)
    return s[lo] + (s[hi] - s[lo]) * (k - lo) if lo != hi else s[lo]


def fmt_dur(seconds):
    if seconds is None:
        return "n/a"
    if seconds < 1:
        return f"{seconds * 1000:.0f} ms"
    if seconds < 60:
        return f"{seconds:.2f} s"
    m, s = divmod(seconds, 60)
    if m < 60:
        return f"{int(m)}m {s:.1f}s"
    h, m = divmod(m, 60)
    return f"{int(h)}h {int(m)}m {s:.0f}s"


def summarize_dist(values, unit="ms"):
    if not values:
        return "  (no samples)"
    return (
        f"  n={len(values)}  "
        f"min={min(values):.1f}  "
        f"mean={statistics.fmean(values):.1f}  "
        f"p50={pct(values, 0.5):.1f}  "
        f"p95={pct(values, 0.95):.1f}  "
        f"max={max(values):.1f} {unit}"
    )


def print_summary(results, wall_seconds):
    ok = [r for r in results if r.state == "ok"]
    bad = [r for r in results if r.state == "failed"]

    total_pages = sum(r.total_pages for r in ok)
    failed_subpages = sum(r.failed_pages for r in ok)

    doc_durations_s = [r.duration_s for r in ok if r.duration_s is not None]
    all_render_ms = [v for r in ok for v in r.render_ms]
    all_ocr_ms    = [v for r in ok for v in r.ocr_ms]

    sum_doc_s = sum(doc_durations_s)
    eff_parallel = (sum_doc_s / wall_seconds) if wall_seconds > 0 else 0.0
    docs_per_min = (len(ok) / wall_seconds * 60) if wall_seconds > 0 else 0.0
    pages_per_s  = (total_pages / wall_seconds) if wall_seconds > 0 else 0.0

    slowest_doc = max(ok, key=lambda r: r.duration_s) if ok else None
    largest_doc = max(ok, key=lambda r: r.total_pages) if ok else None

    print()
    print("=" * 72)
    print("Final statistics")
    print("=" * 72)
    print(f"Wall-clock total      : {fmt_dur(wall_seconds)}")
    print(f"Files processed       : {len(results)}  "
          f"(ok={len(ok)}, failed={len(bad)})")
    print(f"Pages processed       : {total_pages}  "
          f"(failed sub-pages: {failed_subpages})")
    print(f"Throughput            : {docs_per_min:.1f} docs/min, "
          f"{pages_per_s:.2f} pages/s")
    print(f"Effective parallelism : {eff_parallel:.2f}x  "
          f"(sum doc time / wall time)")
    print()

    print("Per-document wall time (client-side, submit -> saved):")
    if doc_durations_s:
        ms = [s * 1000 for s in doc_durations_s]
        print(summarize_dist(ms, unit="ms"))
        print(f"  total={fmt_dur(sum_doc_s)}")
    else:
        print("  (no samples)")
    print()

    print("Per-page render time (server-side):")
    print(summarize_dist(all_render_ms))
    print()

    print("Per-page OCR time (server-side):")
    print(summarize_dist(all_ocr_ms))
    print()

    if all_render_ms or all_ocr_ms:
        r_sum = sum(all_render_ms)
        o_sum = sum(all_ocr_ms)
        tot = r_sum + o_sum
        if tot:
            print(f"Stage split (page-time) : render {r_sum / tot:.0%}  "
                  f"ocr {o_sum / tot:.0%}")
            print()

    if slowest_doc:
        print(f"Slowest doc           : {slowest_doc.pdf.name}  "
              f"({fmt_dur(slowest_doc.duration_s)}, "
              f"{slowest_doc.total_pages} pages)")
    if largest_doc:
        print(f"Largest doc           : {largest_doc.pdf.name}  "
              f"({largest_doc.total_pages} pages, "
              f"{fmt_dur(largest_doc.duration_s)})")


# --- entry point -------------------------------------------------------------

def main():
    parser = argparse.ArgumentParser(
        description="Bulk-process PDFs through an OCR server.")
    parser.add_argument("--key", default=os.environ.get("OCR_API_KEY"),
                        help="API key (or set OCR_API_KEY in env)")
    parser.add_argument("--server", default=DEFAULT_SERVER,
                        help=f"OCR server URL (default: {DEFAULT_SERVER})")
    parser.add_argument("-j", "--concurrency", type=int,
                        default=DEFAULT_CONCURRENCY,
                        help=f"max in-flight jobs (default: {DEFAULT_CONCURRENCY})")
    parser.add_argument("--job-timeout", type=float,
                        default=DEFAULT_JOB_TIMEOUT,
                        help="seconds to wait for each job before giving up "
                             f"(default: {DEFAULT_JOB_TIMEOUT})")
    parser.add_argument("-v", "--verbose", action="store_true",
                        help="print full timing statistics on completion")
    parser.add_argument("-r", "--results_type", choices=["text", "full"],
                        default="text",
                        help="type of result to download per document "
                             "(default: text)")
    parser.add_argument("pdfs", nargs="+", type=Path, help="PDF files to process")
    args = parser.parse_args()
    if not args.key:
        parser.error("no API key (use --key or set OCR_API_KEY)")

    server = args.server.rstrip("/")
    pdfs = [p for p in args.pdfs if p.is_file()]
    for p in args.pdfs:
        if not p.is_file():
            print(f"skipping (not a file): {p}", file=sys.stderr)
    if not pdfs:
        print("no input files", file=sys.stderr)
        return 1

    results: list[DocResult] = []
    stats = Stats()

    session = requests.Session()
    session.headers["Authorization"] = f"Bearer {args.key}"
    retry = Retry(
        total=5,
        connect=5,
        read=5,
        status=5,
        backoff_factor=0.5,
        status_forcelist=(502, 503, 504),
        allowed_methods=frozenset({"GET", "POST"}),
        raise_on_status=False,
    )
    adapter = HTTPAdapter(
        pool_connections=args.concurrency,
        pool_maxsize=args.concurrency,
        max_retries=retry,
    )
    session.mount("https://", adapter)
    session.mount("http://", adapter)

    wall_start = time.monotonic()
    with tqdm(total=len(pdfs), unit="file", desc="processing",
              smoothing=0.05) as bar:
        with ThreadPoolExecutor(max_workers=args.concurrency) as pool:
            futures = [
                pool.submit(process_one, pdf, server, session, stats, bar,
                            args.results_type, args.job_timeout, results)
                for pdf in pdfs
            ]
            for f in futures:
                f.result()
    wall_seconds = time.monotonic() - wall_start

    ok = sum(1 for r in results if r.state == "ok")
    bad = [r for r in results if r.state == "failed"]

    if args.verbose:
        print_summary(results, wall_seconds)
    else:
        print(
            f"\nDone in {fmt_dur(wall_seconds)}: "
            f"ok={ok} failed={len(bad)} "
            f"(use --verbose for full statistics)",
            file=sys.stderr,
        )
    if bad:
        print(f"\nFailures ({len(bad)}):", file=sys.stderr)
        for r in bad[:20]:
            print(f"  {r.pdf}: {r.error}", file=sys.stderr)
        if len(bad) > 20:
            print(f"  ... ({len(bad) - 20} more)", file=sys.stderr)
    return 1 if bad else 0


if __name__ == "__main__":
    sys.exit(main())

```