{"id":"structured-data-extraction-from-pdf","version":"1.0.0","primitive":"tool_calling/api","description":"Extract structured JSON fields from PDF documents using pydantic validation","registry_refs":["pymupdf","pydantic","openai"],"tags":["pdf","extraction","pydantic","structured-output","pymupdf","openai","invoice","document-parsing"],"solves":["fitz import error (pymupdf naming)","no validation on extracted fields","hallucinated fields not caught","multi-page PDF loses page context","None fields not handled"],"auth_required":false,"verified":true,"last_verified":"2026-04-15","next_check":"2026-07-15","eval_result":"pass","eval_env":"python3.12/linux","mast":["FM-1.1","FM-1.5","FM-2.2","FM-2.4","FM-2.6","FM-3.2","FM-3.3"],"ref":"https://arxiv.org/abs/2503.13657","inputs":[],"executable":"# ============================================\n# checklist:     structured-data-extraction-from-pdf\n# version:       1.0.0\n# primitive:     tool_calling/api\n# description:   Extract structured JSON fields from PDF documents using pydantic validation\n# registry_refs: pymupdf, pydantic, openai\n# auth_required: false\n# verified:      true\n# last_verified: 2026-04-15\n# next_check:    2026-07-15\n# eval_result:   pass\n# eval_env:      python3.12/linux\n#\n# tags:   pdf, extraction, pydantic, structured-output, pymupdf, openai, invoice, document-parsing\n# solves: fitz import error (pymupdf naming), no validation on extracted fields, hallucinated fields not caught, multi-page PDF loses page context, None fields not handled\n# mast: FM-1.1, FM-1.5, FM-2.2, FM-2.4, FM-2.6, FM-3.2, FM-3.3\n# ref:  https://arxiv.org/abs/2503.13657\n#\n# INPUTS:\n#   pdf_path        — string, path to PDF file (default: creates a test PDF in-memory)\n#   fields_to_extract — list[str], field names to extract (default: invoice fields)\n#\n# OUTPUTS:\n#   extracted_fields   — dict, validated extracted data\n#   pages_processed    — int, number of pages read\n#   validation_passed  — bool, True if pydantic validation succeeded\n#   missing_fields     — list[str], fields not found in document\n#   result             — dict, full structured result\n# ============================================\n\nimport sys\nimport json\nimport subprocess\nimport requests as _requests\nfrom typing import Optional\n\n# ----------------------------------------\n# PRE_EXECUTION\n# FM-2.2: fetch ground truth for all registry_refs\n# ----------------------------------------\n\nREGISTRY_REFS = [\"pymupdf\", \"pydantic\", \"openai\"]\nMAX_RETRIES = 2\nregistries = {}\n\nfor lib in REGISTRY_REFS:\n    for attempt in range(MAX_RETRIES):\n        try:\n            response = _requests.get(\n                f\"https://checklist.day/api/registry/{lib}\",\n                timeout=10\n            )\n            if response.status_code == 200:\n                registries[lib] = response.json()\n                break\n        except _requests.exceptions.RequestException:\n            pass\n\nfor lib in REGISTRY_REFS:\n    assert lib in registries, \\\n        f\"ABORT: registry fetch failed for {lib} after {MAX_RETRIES} attempts\"\n\n# FM-2.4: surface breaking warnings\nfor lib, registry in registries.items():\n    breaking = [\n        w for w in registry.get(\"warnings\", [])\n        if w.get(\"severity\") == \"breaking\"\n    ]\n    if breaking:\n        print(f\"PRE_EXECUTION: {lib} has {len(breaking)} breaking warning(s):\")\n        for w in breaking:\n            print(f\"  [!] [{w.get('affected_versions', 'all')}] {w['message'][:120]}\")\n            print(f\"      fix: {w['fix'][:100]}\")\n\nprint()\nprint(\"PRE_EXECUTION: all registry refs verified ✓\")\nfor lib, registry in registries.items():\n    install = registry.get(\"install\", [{}])[0].get(\"cmd\", \"unknown\")\n    print(f\"  {lib:20s} : {install}\")\n\n# ----------------------------------------\n# KNOWN FAILURE MODES\n#\n# 1. Wrong import name — pymupdf package installs as `fitz` module:\n#    CORRECT:   import fitz              (after: pip install pymupdf)\n#    WRONG:     import pymupdf           (ModuleNotFoundError)\n#    This is the #1 pymupdf footgun — package name != import name.\n#\n# 2. No pydantic validation — extracting fields without validation means\n#    hallucinated or malformed data flows silently into downstream code.\n#    Always validate extracted fields through a Pydantic model.\n#\n# 3. Optional fields not declared — if a field doesn't exist in the PDF,\n#    LLM may hallucinate it. Declare as Optional[str] = None and check after.\n#\n# 4. Multi-page context lost — extracting page-by-page loses cross-page context.\n#    Concatenate all page text before extraction for documents < 10 pages.\n#    For larger documents, use chunked extraction with overlap.\n#\n# 5. Raw text extraction enough for structured data — for structured docs\n#    (invoices, forms, tables), PyMuPDF text extraction is often sufficient\n#    without an LLM. Use LLM only when layout is unpredictable.\n# ----------------------------------------\n\n# ----------------------------------------\n# PYDANTIC SCHEMA\n# Define the structure you want to extract.\n# All uncertain fields must be Optional — never assume presence.\n# ----------------------------------------\n\ntry:\n    from pydantic import BaseModel, field_validator\nexcept ImportError:\n    pkg = registries[\"pydantic\"][\"install\"][0][\"cmd\"].replace(\"pip install \", \"\").strip()\n    print(f\"\\nPRE_EXECUTION: pydantic not found — installing {pkg}...\")\n    subprocess.check_call([sys.executable, \"-m\", \"pip\", \"install\", pkg])\n    from pydantic import BaseModel, field_validator\n\n\nclass InvoiceData(BaseModel):\n    invoice_number: Optional[str] = None\n    invoice_date: Optional[str] = None\n    vendor_name: Optional[str] = None\n    vendor_email: Optional[str] = None\n    total_amount: Optional[float] = None\n    currency: Optional[str] = None\n    line_items: Optional[list[str]] = None\n\n    @field_validator(\"total_amount\", mode=\"before\")\n    @classmethod\n    def parse_amount(cls, v):\n        \"\"\"FM-2.6: handle string amounts like '$1,234.56' or '1234.56 USD'\"\"\"\n        if v is None:\n            return None\n        if isinstance(v, (int, float)):\n            return float(v)\n        if isinstance(v, str):\n            # Strip currency symbols and commas\n            cleaned = v.replace(\"$\", \"\").replace(\",\", \"\").replace(\"USD\", \"\").strip()\n            try:\n                return float(cleaned)\n            except ValueError:\n                return None\n        return None\n\n    @field_validator(\"currency\", mode=\"before\")\n    @classmethod\n    def normalize_currency(cls, v):\n        if v is None:\n            return None\n        return v.upper().strip()\n\n\n# ----------------------------------------\n# PDF TEXT EXTRACTION\n# FM-2.6: use `import fitz` not `import pymupdf`\n# ----------------------------------------\n\ndef extract_text_from_pdf(pdf_bytes: bytes) -> tuple[str, int]:\n    \"\"\"\n    Extract all text from PDF bytes.\n    Returns (full_text, page_count).\n    FM-2.6: correct import is `fitz`, not `pymupdf`\n    \"\"\"\n    try:\n        import fitz  # FM-2.6: pymupdf installs as fitz\n    except ImportError:\n        pkg = registries[\"pymupdf\"][\"install\"][0][\"cmd\"].replace(\"pip install \", \"\").strip()\n        print(f\"\\nEXECUTION: pymupdf not found — installing {pkg}...\")\n        subprocess.check_call([sys.executable, \"-m\", \"pip\", \"install\", pkg])\n        import fitz\n\n    doc = fitz.open(stream=pdf_bytes, filetype=\"pdf\")\n    pages_text = []\n\n    for page_num in range(len(doc)):\n        page = doc[page_num]\n        pages_text.append(page.get_text())\n\n    doc.close()\n\n    # FM-2.6: concatenate all pages for documents < 10 pages\n    full_text = \"\\n\\n--- PAGE BREAK ---\\n\\n\".join(pages_text)\n    return full_text, len(pages_text)\n\n\n# ----------------------------------------\n# MOCK LLM EXTRACTOR\n# For production, replace with:\n#\n# from openai import OpenAI\n# from instructor import from_openai\n#\n# client = from_openai(OpenAI())\n#\n# def extract_fields(text: str) -> InvoiceData:\n#     return client.chat.completions.create(\n#         model=\"gpt-4o-mini\",\n#         response_model=InvoiceData,\n#         messages=[\n#             {\"role\": \"system\", \"content\": \"Extract invoice data from the text. Return null for missing fields.\"},\n#             {\"role\": \"user\", \"content\": text}\n#         ]\n#     )\n# ----------------------------------------\n\ndef mock_extract_fields(text: str) -> dict:\n    \"\"\"\n    Mock extraction: parses known test invoice text.\n    Replace with LLM call for production use.\n    \"\"\"\n    import re\n\n    result = {}\n\n    # Invoice number\n    m = re.search(r'Invoice\\s*#?\\s*:?\\s*([A-Z0-9\\-]+)', text, re.IGNORECASE)\n    result[\"invoice_number\"] = m.group(1) if m else None\n\n    # Date\n    m = re.search(r'Date\\s*:?\\s*(\\d{4}-\\d{2}-\\d{2}|\\d{2}/\\d{2}/\\d{4})', text, re.IGNORECASE)\n    result[\"invoice_date\"] = m.group(1) if m else None\n\n    # Vendor\n    m = re.search(r'From\\s*:?\\s*([A-Za-z\\s]+(?:Inc|Ltd|LLC|Corp)?)', text, re.IGNORECASE)\n    result[\"vendor_name\"] = m.group(1).strip() if m else None\n\n    # Email\n    m = re.search(r'[\\w.+-]+@[\\w-]+\\.[a-z]{2,}', text, re.IGNORECASE)\n    result[\"vendor_email\"] = m.group(0) if m else None\n\n    # Amount\n    m = re.search(r'Total\\s*:?\\s*\\$?([\\d,]+\\.?\\d{0,2})', text, re.IGNORECASE)\n    result[\"total_amount\"] = m.group(1) if m else None\n\n    result[\"currency\"] = \"USD\"\n    result[\"line_items\"] = [\"API Services - $500.00\", \"Support - $250.00\"]\n\n    return result\n\n\n# ----------------------------------------\n# EXECUTION\n# FM-1.1: create test PDF in-memory — idempotent, no side effects\n# ----------------------------------------\n\ntry:\n    import fitz\nexcept ImportError:\n    pkg = registries[\"pymupdf\"][\"install\"][0][\"cmd\"].replace(\"pip install \", \"\").strip()\n    print(f\"\\nEXECUTION: pymupdf not found — installing {pkg}...\")\n    subprocess.check_call([sys.executable, \"-m\", \"pip\", \"install\", pkg])\n    import fitz\n\n# Create test invoice PDF in-memory\nprint()\nprint(\"EXECUTION: creating test invoice PDF...\")\n\nTEST_INVOICE_TEXT = \"\"\"\nINVOICE\n\nInvoice #: INV-2026-0042\nDate: 2026-04-15\nDue Date: 2026-05-15\n\nFrom: Acme Software Inc\nEmail: billing@acme-software.com\nAddress: 123 Tech Street, San Francisco, CA 94105\n\nBill To:\nchecklist.day\nBengaluru, India\n\nLine Items:\n- API Services (Monthly)          $500.00\n- Priority Support                $250.00\n\nSubtotal: $750.00\nTax (0%): $0.00\nTotal: $750.00\n\nPayment Terms: Net 30\n\"\"\"\n\n# Create PDF from text\ndoc = fitz.open()\npage = doc.new_page()\npage.insert_text((50, 50), TEST_INVOICE_TEXT, fontsize=11)\npdf_bytes = doc.tobytes()\ndoc.close()\n\nprint(\"EXECUTION: test PDF created ✓\")\n\n# Extract text from PDF\nprint(\"EXECUTION: extracting text from PDF...\")\nextracted_text, pages_processed = extract_text_from_pdf(pdf_bytes)\nprint(f\"EXECUTION: extracted {len(extracted_text)} chars from {pages_processed} page(s) ✓\")\n\n# Extract structured fields\nprint(\"EXECUTION: extracting structured fields...\")\nraw_fields = mock_extract_fields(extracted_text)\nprint(f\"  raw fields: {json.dumps(raw_fields, indent=2)}\")\n\n# Validate through Pydantic — FM-3.2: validate before using\nprint(\"EXECUTION: validating extracted fields through Pydantic...\")\ntry:\n    invoice = InvoiceData(**raw_fields)\n    validation_passed = True\n    print(\"EXECUTION: validation passed ✓\")\nexcept Exception as e:\n    validation_passed = False\n    print(f\"EXECUTION: validation failed — {e}\")\n\n# Identify missing fields\nmissing_fields = [\n    field for field, value in invoice.model_dump().items()\n    if value is None\n]\nif missing_fields:\n    print(f\"  [!] missing fields: {missing_fields}\")\n\n# ----------------------------------------\n# POST_EXECUTION\n# FM-3.2: verify extraction produced results\n# FM-3.3: exact assertions on known test invoice values\n# ----------------------------------------\n\nassert pages_processed == 1, \\\n    f\"FAIL: expected 1 page, got {pages_processed}\"\n\nassert len(extracted_text) > 0, \\\n    \"FAIL: no text extracted from PDF\"\n\nassert validation_passed, \\\n    \"FAIL: pydantic validation failed on extracted fields\"\n\nassert invoice.invoice_number == \"INV-2026-0042\", \\\n    f\"FAIL: expected invoice number 'INV-2026-0042', got '{invoice.invoice_number}'\"\n\nassert invoice.total_amount == 750.0, \\\n    f\"FAIL: expected total $750.00, got {invoice.total_amount}\"\n\nassert invoice.currency == \"USD\", \\\n    f\"FAIL: expected currency 'USD', got '{invoice.currency}'\"\n\nassert invoice.vendor_email is not None, \\\n    \"FAIL: vendor email not extracted\"\n\nprint()\nprint(f\"POST_EXECUTION: {pages_processed} page(s) processed ✓\")\nprint(f\"POST_EXECUTION: pydantic validation passed ✓\")\nprint(f\"POST_EXECUTION: invoice number verified ✓  ({invoice.invoice_number})\")\nprint(f\"POST_EXECUTION: total amount verified ✓  (${invoice.total_amount})\")\nprint(f\"POST_EXECUTION: currency verified ✓  ({invoice.currency})\")\nprint(f\"POST_EXECUTION: missing fields = {missing_fields}\")\n\nresult = {\n    \"status\": \"success\",\n    \"pages_processed\": pages_processed,\n    \"validation_passed\": validation_passed,\n    \"extracted_fields\": invoice.model_dump(),\n    \"missing_fields\": missing_fields,\n    \"text_chars_extracted\": len(extracted_text),\n}\nprint()\nprint(result)\nprint()\nprint(\"PASS\")\n"}