Paperless-ngx OCR speed: 100 scanned pages on 4 vCPU

#!/usr/bin/env python3
"""Runs in the VM. bench.py <scans_dir> <outdir>
Uploads every PDF in <scans_dir>/pdf through POST /api/documents/post_document/ one after another (queued, as a scanner or a script would),
waits for the consumption tasks, then compares the OCR text Paperless stored with the exact text that was drawn on the pages (truth/),
and runs phrase searches through GET /api/documents/?query=."""
import sys, os, json, time, glob, re, urllib.request, urllib.error, urllib.parse, uuid, datetime, statistics as st
from rapidfuzz.distance import Levenshtein
scans, out = sys.argv[1:3]; os.makedirs(out, exist_ok=True); U = "http://127.0.0.1:8080"
def call(method, path, body=None, headers=None, raw=None):
    h = dict(headers or {}); data = raw
    if body is not None: data = json.dumps(body).encode(); h["Content-Type"] = "application/json"
    r = urllib.request.Request(U + path, data=data, method=method, headers=h)
    try:
        with urllib.request.urlopen(r, timeout=300) as x: t = x.read(); return x.status, (json.loads(t) if t else None)
    except urllib.error.HTTPError as e: return e.code, e.read().decode()[:400]
st_, r = call("POST", "/api/token/", {"username": "admin", "password": "lab-only-not-a-secret"}); TOK = {"Authorization": "Token " + r["token"]}
def upload(path, title):
    b = uuid.uuid4().hex; name = os.path.basename(path); data = open(path, "rb").read()
    def f(k, v): return f'--{b}\r\nContent-Disposition: form-data; name="{k}"\r\n\r\n{v}\r\n'.encode()
    body = f(  "title", title) + f'--{b}\r\nContent-Disposition: form-data; name="document"; filename="{name}"\r\nContent-Type: application/pdf\r\n\r\n'.encode() + data + f"\r\n--{b}--\r\n".encode()
    return call("POST", "/api/documents/post_document/", headers={**TOK, "Content-Type": f"multipart/form-data; boundary={b}"}, raw=body)
meta = {m["doc"]: m for m in json.load(open(f"{scans}/meta.json"))}; pdfs = sorted(glob.glob(f"{scans}/pdf/*.pdf")); tasks = {}
res = {"docs": len(pdfs), "pages": sum(m["pages"] for m in meta.values())}
t0 = time.time()
for p in pdfs:
    slug = os.path.basename(p)[:-4]; s, r = upload(p, slug); assert s == 200, (s, r); tasks[slug] = r.strip('"') if isinstance(r, str) else r
res["upload_secs"] = time.time() - t0; print("uploaded", res["upload_secs"], flush=True)
def task(tid):
    s, r = call("GET", f"/api/tasks/?task_id={tid}", headers=TOK)
    res_ = r.get("results") if isinstance(r, dict) else r
    return res_[0] if res_ else None
done = {}; res["poll_log"] = []
while len(done) < len(tasks) and time.time() - t0 < 5400:
    for slug, tid in tasks.items():
        if slug in done: continue
        t = task(tid)
        if t and t.get("status") in ("success", "failure"): done[slug] = (t, time.time())
    res["poll_log"].append([round(time.time() - t0, 1), len(done)]); time.sleep(3)
t_end = time.time(); res["total_secs"] = t_end - t0; res["tasks"] = {}
for slug, (t, _) in done.items():
    res["tasks"][slug] = {"status": t["status"], "pages": meta[slug]["pages"], "started_to_done": t.get("duration_seconds"), "wait_secs": t.get("wait_time_seconds"), "result": str(t.get("result_data"))[:200]}
res["finished_ok"] = sum(1 for v in res["tasks"].values() if v["status"] == "success")
# accuracy: stored OCR text vs the text drawn on the pages
s, r = call("GET", "/api/documents/?page_size=100&fields=id,title,content,page_count", headers=TOK); docs = {d["title"]: d for d in r["results"]}
norm = lambda t: " ".join(t.replace("\f", " ").split())
acc = []
for slug in meta:
    if slug not in docs: acc.append({"doc": slug, "missing": True}); continue
    truth = norm(open(f"{scans}/truth/{slug}.txt").read()); got = norm(docs[slug]["content"])
    cer = Levenshtein.distance(truth, got) / len(truth); tw, gw = truth.split(), got.split(); wer = Levenshtein.distance(tw, gw) / len(tw)
    acc.append({"doc": slug, "chars": len(truth), "words": len(tw), "cer": cer, "wer": wer, "page_count": docs[slug].get("page_count")})
res["accuracy"] = acc
os.makedirs(f"{out}/ocr", exist_ok=True); os.makedirs(f"{out}/truth", exist_ok=True)
for slug in meta:
    if slug in docs: open(f"{out}/ocr/{slug}.txt", "w").write(docs[slug]["content"])
    open(f"{out}/truth/{slug}.txt", "w").write(open(f"{scans}/truth/{slug}.txt").read())
# search: 6 consecutive words from page 5 of each document, as an exact phrase, 3 passes for timing
searches = []
for slug in meta:
    pages = open(f"{scans}/truth/{slug}.txt").read().split("\f"); words = [w for w in " ".join(pages[min(4, len(pages) - 1)].split()).split()]
    phrase = " ".join(words[40:46]); q = urllib.parse.quote(f'"{phrase}"'); times = []; ids = []
    for _ in range(3):
        t = time.perf_counter(); s, r = call("GET", f"/api/documents/?query={q}&fields=id,title", headers=TOK); times.append((time.perf_counter() - t) * 1000)
        ids = [d["title"] for d in r["results"]] if s == 200 else [f"HTTP {s}"]
    searches.append({"doc": slug, "phrase": phrase, "hits": ids, "found_in_expected": slug in ids, "ms": times})
res["search"] = searches
json.dump(res, open(f"{out}/bench.json", "w"), indent=1); print("done", res["total_secs"], res["finished_ok"])