Analyze 10,000 documents overnight.

# schemas.py — what we want from every document.
SCHEMA = {
    "type": "object",
    "properties": {
        "title":       {"type": "string"},
        "doc_type":    {"type": "string", "enum": ["invoice", "contract", "receipt", "report", "other"]},
        "date":        {"type": "string", "description": "ISO YYYY-MM-DD if visible, else null"},
        "amounts":     {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "label":  {"type": "string"},
                    "value":  {"type": "number"},
                    "currency": {"type": "string"},
                },
                "required": ["value", "currency"],
            },
        },
        "parties":     {"type": "array", "items": {"type": "string"}},
        "key_clauses": {"type": "array", "items": {"type": "string"}},
        "language":    {"type": "string"},
    },
    "required": ["title", "doc_type", "language"],
}

# enqueue.py — fire one batch request per document.
import os
import json
import sqlite3
import httpx
from pathlib import Path
from schemas import SCHEMA
 
API = "https://api.anoman.io"
KEY = os.environ["ANOMAN_API_KEY"]
 
# Local SQLite for progress + idempotency. Reuse this script — already-
# queued docs are skipped on re-run.
db = sqlite3.connect("batch_run.db")
db.execute("""
    CREATE TABLE IF NOT EXISTS jobs (
        doc_path TEXT PRIMARY KEY,
        job_id   TEXT,
        status   TEXT DEFAULT 'queued',
        result   TEXT,
        error    TEXT,
        enqueued_at REAL DEFAULT (strftime('%s', 'now'))
    )
""")
db.commit()
 
SYSTEM = """Extract structured metadata from the document. Return JSON
matching the provided schema. If a field is not present in the doc,
omit it. Don't hallucinate values."""
 
def enqueue_doc(doc_path: Path):
    # Skip if already queued
    cur = db.execute("SELECT job_id FROM jobs WHERE doc_path = ?", (str(doc_path),))
    if row := cur.fetchone():
        return row[0]
 
    content = doc_path.read_text()  # or PDF→text extraction
    response = httpx.post(
        f"{API}/v1/chat/completions",
        headers={
            "Authorization": f"Bearer {KEY}",
            "x-anoman-prefer-batch": "true",
        },
        json={
            "model": "deepseek-v3",  # Long context, batch-friendly, cheap
            "messages": [
                {"role": "system", "content": SYSTEM},
                {"role": "user",   "content": f"Schema:\n{json.dumps(SCHEMA)}\n\nDoc:\n{content[:50_000]}"},
            ],
            "response_format": {"type": "json_object"},
            "metadata": {"doc_path": str(doc_path)},
        },
    )
    response.raise_for_status()
    job = response.json()
 
    db.execute(
        "INSERT INTO jobs (doc_path, job_id) VALUES (?, ?)",
        (str(doc_path), job["id"]),
    )
    db.commit()
    return job["id"]
 
# Walk a directory of docs
DOCS = Path("./docs_to_analyze")
for doc in DOCS.glob("*.txt"):  # or *.pdf
    enqueue_doc(doc)
    print(f"queued {doc.name}")

# poll_results.py — run in a separate process. Polls every queued
# job; persists results + errors. Idempotent — re-running picks up
# where it left off.
import json, time, sqlite3, httpx
 
API, KEY = "https://api.anoman.io", os.environ["ANOMAN_API_KEY"]
 
db = sqlite3.connect("batch_run.db")
 
def poll_one(job_id: str) -> tuple[str, dict | None, str | None]:
    """Returns (status, result, error_message). status ∈ {queued, complete, failed}."""
    r = httpx.get(
        f"{API}/anoman/v1/batch/{job_id}",
        headers={"Authorization": f"Bearer {KEY}"},
        timeout=30,
    )
    if r.status_code == 200:
        body = r.json()
        try:
            payload = json.loads(body["choices"][0]["message"]["content"])
            return "complete", payload, None
        except (KeyError, json.JSONDecodeError) as e:
            return "failed", None, f"bad_payload: {e}"
    if r.status_code == 202:
        return "queued", None, None
    return "failed", None, f"http_{r.status_code}: {r.text[:200]}"
 
while True:
    rows = db.execute(
        "SELECT doc_path, job_id FROM jobs WHERE status = 'queued'"
    ).fetchall()
    if not rows:
        print("all jobs complete or failed — exit")
        break
 
    for doc_path, job_id in rows:
        status, result, err = poll_one(job_id)
        if status == "queued":
            continue
        if status == "complete":
            db.execute(
                "UPDATE jobs SET status = ?, result = ? WHERE doc_path = ?",
                ("complete", json.dumps(result), doc_path),
            )
            print(f"✓ {doc_path}")
        else:
            db.execute(
                "UPDATE jobs SET status = ?, error = ? WHERE doc_path = ?",
                ("failed", err, doc_path),
            )
            print(f"✗ {doc_path}: {err}")
        db.commit()
 
    time.sleep(30)

# progress.py — run anytime to see how far along the batch is.
import sqlite3
 
db = sqlite3.connect("batch_run.db")
total, queued, complete, failed = db.execute("""
    SELECT
        COUNT(*),
        SUM(CASE WHEN status = 'queued'   THEN 1 ELSE 0 END),
        SUM(CASE WHEN status = 'complete' THEN 1 ELSE 0 END),
        SUM(CASE WHEN status = 'failed'   THEN 1 ELSE 0 END)
    FROM jobs
""").fetchone()
 
print(f"Total:    {total}")
print(f"Queued:   {queued}    ({100*queued/total:.1f}%)")
print(f"Complete: {complete}  ({100*complete/total:.1f}%)")
print(f"Failed:   {failed}    ({100*failed/total:.1f}%)")
 
# Total savings from batch routing — _anoman.savings_usd on each result
import json
total_saved = 0
for (result_json,) in db.execute("SELECT result FROM jobs WHERE status = 'complete'"):
    body = json.loads(result_json) if result_json else {}
    saved = body.get("_anoman", {}).get("savings_usd", "0")
    total_saved += float(saved)
 
print(f"Total saved vs realtime: ${total_saved:.2f}")