OCR API in Python with requests

pip install requests
export SIGLI_API_KEY=sig_live_...   # from the API Keys page

# OCR one image (PNG, JPEG, WebP, TIFF, BMP or GIF) and print the text.
# pip install requests
import os
import sys

import requests

API = os.environ.get("SIGLI_API_URL", "https://api.sigli.dev")
API_KEY = os.environ["SIGLI_API_KEY"]  # sig_live_... or sig_test_...

path = sys.argv[1] if len(sys.argv) > 1 else "scan-page.png"

with open(path, "rb") as f:
    resp = requests.post(
        f"{API}/v1/process",
        headers={"X-API-Key": API_KEY, "X-Convert-Type": "ocr"},
        files={"file": f},
        data={"language": "eng", "output_format": "text"},
        timeout=120,
    )
resp.raise_for_status()

result = resp.json()["data"]
print(f"pages={result['page_count']} confidence={result['confidence']:.2f}")
print(result["text"])

# OCR a scanned, multi-page PDF. Short PDFs come back inline (HTTP 200);
# longer ones are queued as a job (HTTP 202) that you poll until it finishes.
# pip install requests
import os
import sys
import time

import requests

API = os.environ.get("SIGLI_API_URL", "https://api.sigli.dev")
HEADERS = {"X-API-Key": os.environ["SIGLI_API_KEY"]}

path = sys.argv[1] if len(sys.argv) > 1 else "scanned-letters.pdf"

with open(path, "rb") as f:
    resp = requests.post(
        f"{API}/v1/process",
        headers={**HEADERS, "X-Convert-Type": "ocr"},
        files={"file": (os.path.basename(path), f, "application/pdf")},
        data={"language": "eng"},
        timeout=120,
    )
resp.raise_for_status()
body = resp.json()["data"]

if resp.status_code == 202:
    job_id = body["job_id"]
    print(f"queued job {job_id}")
    while True:
        time.sleep(5)
        job = requests.get(f"{API}/v1/jobs/{job_id}", headers=HEADERS, timeout=30)
        job.raise_for_status()
        body = job.json()["data"]
        if body["status"] == "completed":
            break
        if body["status"] == "failed":
            raise SystemExit(f"OCR failed: {body['code']}: {body['error']}")
    print(f"done in {body['processing_ms'] / 1000:.0f}s, "
          f"{body['credits_charged']} credits")

# Pages are separated by a form feed (\f) in the text.
pages = body["text"].split("\f")
out = os.path.splitext(path)[0] + ".txt"
with open(out, "w", encoding="utf-8") as f:
    f.write(body["text"])
print(f"{len(pages)} pages of text saved to {out}")

# Error handling for the Sigli API. Every error is JSON:
#   {"success": false, "code": "<machine code>", "message": "<human text>"}
# Failed requests are never charged.
# pip install requests
import os
import sys
import time

import requests

API = os.environ.get("SIGLI_API_URL", "https://api.sigli.dev")
HEADERS = {"X-API-Key": os.environ["SIGLI_API_KEY"]}


class SigliError(Exception):
    def __init__(self, status, code, message):
        super().__init__(f"{status} {code}: {message}")
        self.status, self.code, self.message = status, code, message


def process(convert_type, path, retries=3, **fields):
    """POST a file to /v1/process; returns the requests.Response (200 or 202)."""
    for attempt in range(retries + 1):
        with open(path, "rb") as f:
            resp = requests.post(
                f"{API}/v1/process",
                headers={**HEADERS, "X-Convert-Type": convert_type},
                files={"file": f},
                data=fields,
                timeout=120,
            )
        if resp.status_code in (200, 202):
            return resp
        try:
            err = resp.json()
            code, message = err.get("code"), err.get("message")
        except ValueError:
            code, message = f"http_{resp.status_code}", resp.text[:200]
        # 429 rate_limited is the per-minute request limit: wait as told by
        # Retry-After and retry. quota_exceeded / sandbox_limit_reached won't
        # clear by retrying.
        if code == "rate_limited" and attempt < retries:
            time.sleep(int(resp.headers.get("Retry-After", 5 * 2**attempt)))
            continue
        raise SigliError(resp.status_code, code, message)


def show(resp, field):
    data = resp.json()["data"]
    if resp.status_code == 202:  # large input: poll GET /v1/jobs/{job_id}
        print(f"queued as job {data['job_id']}")
    else:
        print(data[field][:500])


path = sys.argv[1] if len(sys.argv) > 1 else "scan.pdf"
try:
    show(process("pdf-to-markdown", path), "markdown")
except SigliError as e:
    if e.code == "no_text_layer":  # 422: scanned PDF, no text to extract
        print("No text layer; running OCR instead")
        show(process("ocr", path, language="eng"), "text")
    elif e.status == 402:  # plan_upgrade_required: feature not on your plan
        print(f"Needs a plan upgrade: {e.message}")
    elif e.status == 413:  # payload_too_large / file_too_large
        print(f"File too large for one request: {e.message}")
    elif e.status == 429:  # quota_exceeded, sandbox_limit_reached, rate_limited
        print(f"Out of credits or rate limited: {e.message}")
    elif e.status == 400:  # bad_request: e.g. OCR files over 20 MB, bad language
        print(f"Bad request: {e.message}")
    else:
        raise