#!/usr/bin/env python3 """ PDF table extraction — IRS 2025 Tax Table (Publication 1040, i1040tt). Turns 11 pages of three-up printed tables into one flat CSV, then proves the result is complete rather than asking anyone to take it on trust. Why this document is not a trivial extraction: * Each page carries three independent column blocks side by side, so a row of printed text holds three unrelated table rows. * Column headers repeat on every page and must not become data. * Grey "band marker" lines (a lone "30,000") sit between groups of rows. * Page 2 contains a worked "Sample Table" example whose numbers look exactly like real rows. A naive extractor silently swallows them. * The sub-$3,000 section on page 2 has a broken text layer: single rows are split across several lines by the PDF's internal layout. The last two are handled by scoping extraction to pages 3-13 and reporting the excluded range explicitly, rather than emitting numbers that cannot be trusted. Standard library plus `pdftotext` (poppler), which ships on every platform. Usage: python3 scripts/extract_tax_table.py data/raw/irs-tax-table-2025.pdf data/clean """ import csv import json import re import subprocess import sys from pathlib import Path FIRST_PAGE, LAST_PAGE = 3, 13 # the uniform $50-band section BAND_WIDTH = 50 COLUMNS = ["single", "married_filing_jointly", "married_filing_separately", "head_of_household"] # The section this run deliberately does not touch, and why. EXCLUDED = { "range": "$0 - $2,999", "page": 2, "reason": "variable band widths ($5/$10/$25) and a broken PDF text layer that " "splits single rows across multiple lines; also shares the page with a " "worked 'Sample Table' example whose numbers mimic real rows", "recommendation": "manual entry, ~120 rows, or a second pass using positional " "extraction rather than text flow", } NUMBER = re.compile(r"\b\d{1,3}(?:,\d{3})*\b") def page_text(pdf_path, page): """Render one page preserving its printed column layout.""" result = subprocess.run( ["pdftotext", "-layout", "-f", str(page), "-l", str(page), str(pdf_path), "-"], capture_output=True, text=True, check=True) return result.stdout def parse_line(line): """ Pull every complete table row out of one printed line. A line holds up to three rows, each six numbers wide. A six-number group is accepted only when its first two values form a valid $50 band — which is what separates a real row from a band marker, a page number, or a stray footnote. """ tokens = [int(match.group().replace(",", "")) for match in NUMBER.finditer(line)] rows, index = [], 0 while index + 6 <= len(tokens): group = tokens[index:index + 6] if group[1] - group[0] == BAND_WIDTH: rows.append(group) index += 6 else: index += 1 # not a row start; slide along and retry return rows def verify(rows): """ Check the extraction against the structure the document guarantees, so the output can be signed off without hand-checking 1,940 rows. """ problems = [] bands = sorted(row[0] for row in rows) duplicates = {value for value in bands if bands.count(value) > 1} if len(bands) < 5000 else set() seen, dupes = set(), set() for value in bands: if value in seen: dupes.add(value) seen.add(value) if dupes: problems.append(f"{len(dupes)} duplicated income bands") expected = list(range(bands[0], bands[-1] + BAND_WIDTH, BAND_WIDTH)) missing = sorted(set(expected) - set(bands)) if missing: problems.append(f"{len(missing)} missing bands, first at ${missing[0]:,}") by_band = {row[0]: row for row in rows} for position, column in enumerate(COLUMNS, start=2): previous = None breaks = 0 for band in expected: if band not in by_band: continue value = by_band[band][position] if previous is not None and value < previous: breaks += 1 previous = value if breaks: problems.append(f"{column}: tax decreases as income rises at {breaks} points") return { "rows": len(rows), "expected_rows": len(expected), "income_from": bands[0], "income_to": bands[-1] + BAND_WIDTH, "duplicate_bands": len(dupes), "missing_bands": len(missing), "monotonicity_breaks": sum(1 for p in problems if "decreases" in p), "problems": problems, "complete": not problems and len(bands) == len(expected), } def main(pdf_path, output_dir): output_dir = Path(output_dir) output_dir.mkdir(parents=True, exist_ok=True) rows, per_page, skipped_lines = [], {}, 0 for page in range(FIRST_PAGE, LAST_PAGE + 1): found = 0 for line in page_text(pdf_path, page).splitlines(): if not line.strip(): continue parsed = parse_line(line) if parsed: for row in parsed: rows.append(row + [page]) found += len(parsed) elif NUMBER.search(line) and not re.search(r"[A-Za-z]{4}", line): skipped_lines += 1 # numeric line that formed no valid row per_page[page] = found rows.sort(key=lambda row: row[0]) checks = verify([row[:6] for row in rows]) with open(output_dir / "irs-tax-table-2025.csv", "w", newline="", encoding="utf-8") as handle: writer = csv.writer(handle) writer.writerow(["income_at_least", "income_but_less_than", *COLUMNS, "source_page"]) writer.writerows(rows) report = { "source": { "document": "IRS Publication 1040 (2025) — Tax and Earned Income Credit Tables", "file": str(pdf_path), "url": "https://www.irs.gov/pub/irs-pdf/i1040tt.pdf", "pages_processed": f"{FIRST_PAGE}-{LAST_PAGE}", }, "extracted": { "rows": len(rows), "values": len(rows) * len(COLUMNS), "rows_per_page": per_page, "numeric_lines_skipped": skipped_lines, }, "verification": checks, "excluded_section": EXCLUDED, } (output_dir / "irs-extraction-report.json").write_text(json.dumps(report, indent=2) + "\n") print(f"rows extracted {len(rows):>7,} (expected {checks['expected_rows']:,})") print(f"values extracted {len(rows) * len(COLUMNS):>7,}") print(f"income range ${checks['income_from']:,} - ${checks['income_to']:,}") print(f"missing bands {checks['missing_bands']:>7,}") print(f"duplicate bands {checks['duplicate_bands']:>7,}") print(f"skipped num lines {skipped_lines:>7,}") print(f"verification {'PASS' if checks['complete'] else 'FAIL — ' + '; '.join(checks['problems'])}") if __name__ == "__main__": if len(sys.argv) != 3: sys.exit(__doc__.strip().splitlines()[-1]) main(sys.argv[1], sys.argv[2])