spectre_pdf 1.0.0

Native Rust PDF extraction engine: text, markdown for RAG, AcroForm widgets, image decoding, and encrypted PDFs. Lazy parser, persistent Document handle, no C dependencies.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150

"""End-to-end sanity sweep of the Python API across a real PDF corpus.

For every PDF in the chosen corpus, calls every Document surface and
every free function. Counts successes, failures (with a one-line tag for
each), and reports totals. Intended as a smoke test to confirm the wheel
installs and the surface contract still holds across a wide variety of
real-world PDFs after a refactor.

Usage:
    python scripts/corpus_sweep_python.py [--limit N] [--corpus PATH]
"""

from __future__ import annotations

import argparse
import sys
import time
import traceback
from pathlib import Path

import spectre_rs


def attempt(label: str, fn):
    try:
        result = fn()
        return ("ok", label, len(result) if hasattr(result, "__len__") else result)
    except Exception as e:
        return ("fail", label, f"{type(e).__name__}: {str(e)[:100]}")


def sweep_one(path: Path) -> dict:
    bytes_ = path.read_bytes()
    results = {}

    # Free functions
    for label, fn in [
        ("free.extract_text", lambda: spectre_rs.extract_text(bytes_)),
        ("free.extract_metadata", lambda: spectre_rs.extract_metadata(bytes_)),
        ("free.extract_pages", lambda: spectre_rs.extract_pages(bytes_)),
        ("free.extract_tables", lambda: spectre_rs.extract_tables(bytes_)),
        ("free.extract_words", lambda: spectre_rs.extract_words(bytes_)),
        ("free.extract_blocks", lambda: spectre_rs.extract_blocks(bytes_)),
        ("free.extract_toc", lambda: spectre_rs.extract_toc(bytes_)),
        ("free.extract_links", lambda: spectre_rs.extract_links(bytes_)),
        ("free.extract_annotations", lambda: spectre_rs.extract_annotations(bytes_)),
        ("free.extract_images", lambda: spectre_rs.extract_images(bytes_)),
        ("free.extract_page_info", lambda: spectre_rs.extract_page_info(bytes_)),
        ("free.extract_document_info", lambda: spectre_rs.extract_document_info(bytes_)),
    ]:
        results[label] = attempt(label, fn)

    # Document handle
    try:
        doc = spectre_rs.Document(bytes_)
    except Exception as e:
        results["Document.__init__"] = ("fail", "Document.__init__", f"{type(e).__name__}: {e}")
        return results
    results["Document.__init__"] = ("ok", "Document.__init__", doc.page_count)

    for label, fn in [
        ("doc.text", lambda: doc.text()),
        ("doc.text_lenient", lambda: doc.text_lenient()),
        ("doc.words", lambda: doc.words()),
        ("doc.blocks", lambda: doc.blocks()),
        ("doc.toc", lambda: doc.toc()),
        ("doc.links", lambda: doc.links()),
        ("doc.annotations", lambda: doc.annotations()),
        ("doc.images", lambda: doc.images()),
        ("doc.pages", lambda: doc.pages()),
        ("doc.widgets", lambda: doc.widgets()),
        ("doc.info", lambda: doc.info()),
        ("doc.dict", lambda: doc.dict()),
        ("doc.markdown", lambda: doc.markdown()),
        ("doc.search", lambda: doc.search("the")),
        ("doc.text_positioned", lambda: doc.text_positioned()),
    ]:
        results[label] = attempt(label, fn)

    return results


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--limit", type=int, default=100)
    parser.add_argument(
        "--corpus",
        default="bench-corpus/ICDAR_2013_table_evaluate-master/"
        "icdar2013-competition-dataset-with-gt/competition-dataset-eu",
    )
    args = parser.parse_args()

    print(f"spectre_rs version: {spectre_rs.__version__}")
    print()

    pdfs = sorted(Path(args.corpus).rglob("*.pdf"))[: args.limit]
    print(f"sweeping {len(pdfs)} PDFs from {args.corpus}")

    # tally[surface] = (ok_count, fail_count)
    tally: dict[str, tuple[int, int]] = {}
    fail_examples: dict[str, list[tuple[str, str]]] = {}
    pdf_total_failures = 0

    t0 = time.perf_counter()
    for i, pdf in enumerate(pdfs, 1):
        results = sweep_one(pdf)
        pdf_had_fail = False
        for label, (status, _, info) in results.items():
            ok, fail = tally.get(label, (0, 0))
            if status == "ok":
                tally[label] = (ok + 1, fail)
            else:
                tally[label] = (ok, fail + 1)
                pdf_had_fail = True
                fail_examples.setdefault(label, []).append((pdf.name, str(info)))
        if pdf_had_fail:
            pdf_total_failures += 1
        if i % 25 == 0:
            print(f"  ... {i}/{len(pdfs)}")
    elapsed = time.perf_counter() - t0

    print()
    print("=" * 78)
    print(f"{'surface':<32} {'ok':>6} {'fail':>6} {'rate':>8}")
    print("-" * 78)
    for label in sorted(tally):
        ok, fail = tally[label]
        total = ok + fail
        rate = 100.0 * ok / total if total else 0
        print(f"{label:<32} {ok:>6d} {fail:>6d} {rate:>7.1f}%")
    print("-" * 78)
    print(f"{len(pdfs)} PDFs total in {elapsed:.2f}s "
          f"({len(pdfs) * len(tally) / elapsed:.0f} surface-calls/sec)")
    print(f"{pdf_total_failures} PDFs had at least one surface failure")
    print()
    if fail_examples:
        print("=" * 78)
        print("Failure breakdown (first 3 per surface):")
        for label, fails in sorted(fail_examples.items()):
            print(f"  {label} — {len(fails)} fails")
            for pdf_name, msg in fails[:3]:
                print(f"    · {pdf_name}: {msg}")


if __name__ == "__main__":
    try:
        main()
    except Exception:
        traceback.print_exc()
        sys.exit(1)