from __future__ import annotations
import argparse
import os
import re
import statistics
import time
import urllib.request
from pathlib import Path
import fitz
import spectre_rs
CORPUS = {
"f1040.pdf": "https://www.irs.gov/pub/irs-pdf/f1040.pdf",
"f1040sa.pdf": "https://www.irs.gov/pub/irs-pdf/f1040sa.pdf",
"f1099int.pdf": "https://www.irs.gov/pub/irs-pdf/f1099int.pdf",
"w9.pdf": "https://www.irs.gov/pub/irs-pdf/fw9.pdf",
}
def fetch(corpus_dir: Path) -> list[Path]:
corpus_dir.mkdir(parents=True, exist_ok=True)
out = []
for name, url in CORPUS.items():
p = corpus_dir / name
if not p.exists():
print(f"fetch {name}")
urllib.request.urlretrieve(url, p)
out.append(p)
return out
def norm(s: str | None) -> str:
return re.sub(r"\s+", " ", s or "").strip()
def bench(pdf_paths: list[Path]) -> None:
print("=" * 100)
print(f"{'PDF':<32} {'p-cnt':>6} {'s-cnt':>6} {'pmu ms':>8} {'spec ms':>9} "
f"{'speedup':>9} {'name jac':>9} {'val jac':>9}")
print("=" * 100)
total_s_ms = total_p_ms = 0.0
all_name_int = all_name_uni = 0
all_val_int = all_val_uni = 0
for path in pdf_paths:
b = path.read_bytes()
pmu_runs = []
p_widgets: list[tuple[str, str]] = []
for _ in range(3):
d = fitz.open(stream=b)
t0 = time.perf_counter()
p_widgets = [(w.field_name, w.field_value or "")
for pg in d for w in pg.widgets()]
pmu_runs.append((time.perf_counter() - t0) * 1000)
d.close()
spec_runs = []
s_widgets: list[tuple[str, str]] = []
for _ in range(3):
sd = spectre_rs.Document(b)
t0 = time.perf_counter()
s_widgets = [(w.name, w.value) for w in sd.widgets()]
spec_runs.append((time.perf_counter() - t0) * 1000)
pmu_ms = statistics.median(pmu_runs)
spec_ms = statistics.median(spec_runs)
total_p_ms += pmu_ms
total_s_ms += spec_ms
p_names = {n for n, _ in p_widgets if n}
s_names = {n for n, _ in s_widgets if n}
all_name_int += len(p_names & s_names)
all_name_uni += len(p_names | s_names)
name_jacc = (len(p_names & s_names) / len(p_names | s_names)) if (p_names | s_names) else 1.0
p_vals = {(n, norm(v)) for n, v in p_widgets if n}
s_vals = {(n, norm(v)) for n, v in s_widgets if n}
all_val_int += len(p_vals & s_vals)
all_val_uni += len(p_vals | s_vals)
val_jacc = (len(p_vals & s_vals) / len(p_vals | s_vals)) if (p_vals | s_vals) else 1.0
speedup = pmu_ms / spec_ms if spec_ms else float("inf")
print(f"{path.name:<32} {len(p_widgets):>6} {len(s_widgets):>6} "
f"{pmu_ms:>8.2f} {spec_ms:>9.2f} {speedup:>8.2f}x "
f"{name_jacc:>9.3f} {val_jacc:>9.3f}")
print("=" * 100)
print(f"TOTAL speedup {total_p_ms / total_s_ms:.2f}x "
f"({total_p_ms:.1f}ms / {total_s_ms:.1f}ms)")
print(f" field-name jaccard: {all_name_int}/{all_name_uni} "
f"= {all_name_int / max(all_name_uni, 1):.3f}")
print(f" field-value jaccard: {all_val_int}/{all_val_uni} "
f"= {all_val_int / max(all_val_uni, 1):.3f}")
def main() -> None:
p = argparse.ArgumentParser()
p.add_argument("--corpus", default="bench-corpus/widgets",
help="directory to download PDFs into (default ./bench-corpus/widgets)")
args = p.parse_args()
pdfs = fetch(Path(args.corpus))
bench(pdfs)
if __name__ == "__main__":
main()