Example Python Client

#!/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())