Zero-Shot Native Doc Parsing with Gemma 4: Treating PDFs as Pictures

# Python 3.10+ required
python --version

# Create a digital setting
python -m venv gemma4-env
supply gemma4-env/bin/activate       # macOS / Linux
gemma4-envScriptsactivate          # Home windows

# Set up packages
pip set up 
  "transformers>=4.51.0" 
  "torch>=2.3.0" 
  "speed up>=0.30.0" 
  "pymupdf>=1.24.0" 
  "Pillow>=10.0.0" 
  "bitsandbytes>=0.43.0"

# Log into Hugging Face (paste your learn token when prompted)
pip set up huggingface_hub
huggingface-cli login

# device_check.py
# Run this earlier than loading Gemma 4 to verify your compute setting.
# Save as device_check.py and run: python device_check.py

def detect_device():
    """
    Detect the most effective obtainable compute gadget.
    Returns (device_str, dtype, load_kwargs) to be used with from_pretrained.
    """
    attempt:
        import torch
    besides ImportError:
        elevate RuntimeError("PyTorch not discovered. Set up: pip set up torch")

    if torch.cuda.is_available():
        identify = torch.cuda.get_device_name(0)
        vram = torch.cuda.get_device_properties(0).total_memory / 1e9
        print(f"CUDA GPU: {identify} ({vram:.1f} GB VRAM)")
        return "cuda", torch.bfloat16, {"device_map": "auto", "torch_dtype": torch.bfloat16}

    elif hasattr(torch.backends, "mps") and torch.backends.mps.is_available():
        print("Apple Silicon MPS detected")
        return "mps", torch.float16, {"device_map": "mps", "torch_dtype": torch.float16}

    else:
        print("No GPU discovered -- CPU fallback (gradual however useful)")
        return "cpu", None, {"device_map": "cpu"}


if __name__ == "__main__":
    gadget, dtype, kwargs = detect_device()
    print(f"Machine  : {gadget}")
    print(f"Dtype   : {dtype}")
    print(f"Prepared for Gemma 4 loading with: {kwargs}")

# pdf_renderer.py
# Conditions: pip set up pymupdf Pillow
# Utilization: import and instantiate PDFRenderer; name render_page() or render_all()

import pymupdf
from PIL import Picture
from pathlib import Path


class PDFRenderer:
    """
    Converts PDF pages to PIL Pictures for downstream VLM inference.

    No exterior dependencies past PyMuPDF -- no Poppler, no Ghostscript.
    Output photos are in RGB mode, prepared for direct use with Gemma 4's AutoProcessor.
    """

    def __init__(self, dpi: int = 200):
        """
        Args:
            dpi: Render decision.
                 150 -- quick classification move (fewer tokens, decrease high quality)
                 200 -- manufacturing commonplace for typed textual content and printed paperwork
                 300 -- high-fidelity, advisable for handwriting or small glyphs
        """
        self.dpi = dpi
        # PyMuPDF makes use of a zoom issue relative to the 72 DPI PDF baseline.
        # zoom=1.0 = 72 DPI, zoom=2.78 = 200 DPI, zoom=4.17 = 300 DPI.
        self._zoom = dpi / 72.0
        self._matrix = pymupdf.Matrix(self._zoom, self._zoom)

    def render_page(self, pdf_path: str, page_index: int = 0) -> Picture.Picture:
        """
        Render a single PDF web page to a PIL Picture.

        Args:
            pdf_path:   Path to the PDF file
            page_index: Zero-based web page index (0 = first web page)

        Returns:
            PIL.Picture.Picture in RGB mode, prepared for Gemma 4's processor

        Raises:
            IndexError: If page_index is out of vary for this PDF
            FileNotFoundError: If the PDF path doesn't exist
        """
        path = Path(pdf_path)
        if not path.exists():
            elevate FileNotFoundError(f"PDF not discovered: {pdf_path}")

        doc = pymupdf.open(str(path))

        if page_index >= len(doc):
            doc.shut()
            elevate IndexError(
                f"Web page index {page_index} out of vary -- "
                f"this PDF has {len(doc)} web page(s)"
            )

        web page = doc[page_index]

        # get_pixmap renders the web page on the zoom matrix outlined in __init__.
        # The ensuing pixmap accommodates uncooked RGB bytes at self.dpi decision.
        pix = web page.get_pixmap(matrix=self._matrix)
        doc.shut()

        # Convert uncooked bytes to PIL Picture -- that is the format Gemma 4 expects
        return Picture.frombytes("RGB", [pix.width, pix.height], pix.samples)

    def render_all(self, pdf_path: str) -> listing[Image.Image]:
        """
        Render each web page of a PDF to a listing of PIL Pictures.

        Returns pages so as: index 0 = first web page, index -1 = final web page.
        The listing is absolutely materialized -- for very giant PDFs, use render_range
        to course of in chunks.
        """
        doc = pymupdf.open(pdf_path)
        photos = []

        for i in vary(len(doc)):
            pix = doc[i].get_pixmap(matrix=self._matrix)
            photos.append(Picture.frombytes("RGB", [pix.width, pix.height], pix.samples))

        doc.shut()
        return photos

    def render_range(
        self, pdf_path: str, begin: int, finish: int
    ) -> listing[Image.Image]:
        """
        Render a particular web page vary (begin inclusive, finish unique).

        Use this for the extraction move within the two-pass pipeline -- solely pages
        that the classification move recognized as content-bearing have to be
        rendered at excessive decision.
        """
        doc = pymupdf.open(pdf_path)
        photos = []
        finish = min(finish, len(doc))  # Clamp to precise web page rely

        for i in vary(begin, finish):
            pix = doc[i].get_pixmap(matrix=self._matrix)
            photos.append(Picture.frombytes("RGB", [pix.width, pix.height], pix.samples))

        doc.shut()
        return photos

    def page_count(self, pdf_path: str) -> int:
        """Return the variety of pages in a PDF with out rendering any of them."""
        doc = pymupdf.open(pdf_path)
        rely = len(doc)
        doc.shut()
        return rely

# Fast check -- add this after the category definition and run the file instantly
if __name__ == "__main__":
    import sys
    if len(sys.argv) < 2:
        print("Utilization: python pdf_renderer.py ")
        sys.exit(1)

    pdf_path = sys.argv[1]
    renderer = PDFRenderer(dpi=200)
    pages = renderer.render_all(pdf_path)
    print(f"Rendered {len(pages)} web page(s)")
    for i, img in enumerate(pages):
        print(f"  Web page {i}: {img.width} x {img.top} px")
    # Save web page 0 as a PNG for visible inspection
    pages[0].save("page_0_preview.png")
    print("Saved page_0_preview.png -- confirm it seems to be appropriate earlier than operating inference")

python pdf_renderer.py your_invoice.pdf

# gemma4_loader.py
# Conditions: pip set up transformers>=4.51.0 torch speed up
# Run: python gemma4_loader.py
# First run downloads ~10 GB of weights -- subsequent runs load from cache

import re
import torch
from PIL import Picture
from transformers import AutoProcessor, Gemma4ForConditionalGeneration

MODEL_ID = "google/gemma-4-E4B-it"
# Use "google/gemma-4-E2B-it" for six GB VRAM machines


def load_model(model_id: str = MODEL_ID):
    """
    Load Gemma 4 and its processor.
    device_map="auto" distributes throughout all obtainable GPUs,
    or falls again to CPU if none are discovered.
    """
    print(f"Loading {model_id}...")

    processor = AutoProcessor.from_pretrained(model_id)

    mannequin = Gemma4ForConditionalGeneration.from_pretrained(
        model_id,
        torch_dtype=torch.bfloat16,   # Coaching dtype -- highest quality/reminiscence steadiness
        device_map="auto",            # Auto-distributes throughout GPUs or falls again to CPU
    )
    mannequin.eval()
    print(f"Mannequin prepared on: {mannequin.gadget}")
    return mannequin, processor


def query_document_page(
    mannequin,
    processor,
    page_image: Picture.Picture,
    immediate: str,
    token_budget: int = 1120,
    enable_thinking: bool = False,
    max_new_tokens: int = 1024,
) -> str:
    """
    Ship a single doc web page picture + textual content immediate to Gemma 4.

    Args:
        page_image:     PIL Picture of the PDF web page (from PDFRenderer)
        immediate:         What you need extracted or answered about this web page
        token_budget:   Visible token price range -- 70/140/280/560/1120.
                        Increased = extra element, extra VRAM, slower inference.
                        Use 1120 for dense invoices, 280 for fast classification.
        enable_thinking: If True, the mannequin causes step-by-step earlier than answering.
                         Improves accuracy on advanced layouts at the price of latency.
        max_new_tokens:  Most tokens to generate within the response

    Returns:
        Mannequin response as a plain string, with  block stripped if current
    """
    messages = [
        {
            "role": "user",
            "content": [
                # Image comes first -- this is required for optimal Gemma 4 performance
                {"type": "image", "image": page_image},
                {"type": "text",  "text": prompt},
            ],
        }
    ]

    # apply_chat_template codecs the messages and injects the visible token price range
    inputs = processor.apply_chat_template(
        messages,
        add_generation_prompt=True,
        tokenize=True,
        return_dict=True,
        return_tensors="pt",
        # token_budget controls what number of visible tokens symbolize the picture
        # Increased price range = extra spatial element preserved = higher for dense docs
        num_image_tokens=token_budget,
        # Toggles the mannequin's chain-of-thought reasoning move earlier than the ultimate reply
        enable_thinking=enable_thinking,
    ).to(mannequin.gadget)

    with torch.no_grad():
        output_ids = mannequin.generate(
            **inputs,
            max_new_tokens=max_new_tokens,
            temperature=0.1,   # Low temperature for structured extraction -- deterministic
            top_p=0.95,
            do_sample=True,
        )

    # Decode solely the newly generated tokens, not the enter immediate
    new_tokens = output_ids[0][inputs["input_ids"].form[-1]:]
    uncooked = processor.decode(new_tokens, skip_special_tokens=True).strip()

    # Strip the chain-of-thought block when pondering mode was enabled.
    # The ... content material is the mannequin's reasoning course of,
    # not a part of the structured output. Callers solely want the ultimate reply.
    return re.sub(r".*?", "", uncooked, flags=re.DOTALL).strip()


# ── Fast first check ──────────────────────────────────────────────────────────

if __name__ == "__main__":
    from pdf_renderer import PDFRenderer
    import sys

    if len(sys.argv) < 2:
        print("Utilization: python gemma4_loader.py ")
        sys.exit(1)

    mannequin, processor = load_model()
    renderer = PDFRenderer(dpi=200)

    # Render the primary web page
    page_img = renderer.render_page(sys.argv[1], page_index=0)
    print(f"Web page rendered: {page_img.width}x{page_img.top} px")

    # Plain description immediate -- good sanity examine earlier than structured extraction
    description = query_document_page(
        mannequin, processor,
        page_image=page_img,
        immediate="Describe what you see on this doc. What sort of doc is it and what data does it include?",
        token_budget=560,
        enable_thinking=False,
    )
    print("n── Doc description ──")
    print(description)

python gemma4_loader.py your_invoice.pdf

# invoice_parser.py
# Conditions: pdf_renderer.py and gemma4_loader.py in the identical listing
# Run: python invoice_parser.py 

import re
import json
import torch
from dataclasses import dataclass, area
from pathlib import Path
from typing import Elective
from PIL import Picture
from transformers import AutoProcessor, Gemma4ForConditionalGeneration

from pdf_renderer import PDFRenderer
from gemma4_loader import load_model, query_document_page

MODEL_ID = "google/gemma-4-E4B-it"

# ── Knowledge mannequin ────────────────────────────────────────────────────────────────

@dataclass
class LineItem:
    description: str
    amount: Elective[float]
    unit_price: Elective[str]
    whole: Elective[str]

@dataclass
class ParsedInvoice:
    vendor_name: Elective[str]
    invoice_number: Elective[str]
    invoice_date: Elective[str]
    due_date: Elective[str]
    line_items: listing[LineItem] = area(default_factory=listing)
    subtotal: Elective[str] = None
    tax: Elective[str] = None
    total_due: Elective[str] = None
    forex: Elective[str] = None
    # Fields the place the mannequin returned None, empty, or "unknown"
    # -- route these for human overview
    low_confidence_fields: listing[str] = area(default_factory=listing)
    raw_output: str = ""

# ── Extraction immediate ─────────────────────────────────────────────────────────

EXTRACTION_PROMPT = """This can be a web page from a provider bill. Extract all obtainable data and return it as a single JSON object.

Required JSON format:
{
  "vendor_name": "string or null",
  "invoice_number": "string or null",
  "invoice_date": "string or null",
  "due_date": "string or null",
  "line_items": [
    {
      "description": "string",
      "quantity": number or null,
      "unit_price": "string or null",
      "total": "string or null"
    }
  ],
  "subtotal": "string or null",
  "tax": "string or null",
  "total_due": "string or null",
  "forex": "string or null"
}

Guidelines:
- Return ONLY the JSON object -- no preamble, no rationalization, no markdown fences.
- If a area isn't seen on this web page, set it to null.
- Don't invent values. When you can't learn a quantity clearly, set it to null.
- Protect the unique forex image (e.g. $, €, £, ₦) in financial fields.
- Embody ALL line gadgets you may see, even when the desk runs off the seen space."""

# ── Output parser ─────────────────────────────────────────────────────────────

def extract_json_block(textual content: str) -> Elective[dict]:
    """
    Discover the primary JSON object within the mannequin's output.
    Handles each naked JSON and markdown-fenced ```json ... ``` blocks,
    since some mannequin outputs embrace fences regardless of being informed to not.
    """
    # Attempt markdown fence first
    fence = re.search(r"```(?:json)?s*({.*?})s*```", textual content, re.DOTALL)
    if fence:
        attempt:
            return json.masses(fence.group(1))
        besides json.JSONDecodeError:
            move
    # Fall again to any naked JSON object within the output
    naked = re.search(r"({.*})", textual content, re.DOTALL)
    if naked:
        attempt:
            return json.masses(naked.group(1))
        besides json.JSONDecodeError:
            move
    return None

def build_parsed_invoice(raw_output: str) -> ParsedInvoice:
    """
    Convert uncooked mannequin output to a typed ParsedInvoice.
    By no means raises -- fields that can not be parsed default to None
    and are added to low_confidence_fields for downstream dealing with.
    """
    information = extract_json_block(raw_output)

    if information is None:
        return ParsedInvoice(
            vendor_name=None, invoice_number=None,
            invoice_date=None, due_date=None,
            low_confidence_fields=["all_fields -- JSON parse failed"],
            raw_output=raw_output,
        )

    low_conf = []

    def safe_str(key: str) -> Elective[str]:
        """Extract a string area; add to low_confidence if lacking or placeholder."""
        val = information.get(key)
        if val is None or str(val).strip().decrease() in ("", "null", "unknown", "n/a"):
            low_conf.append(key)
            return None
        return str(val).strip()

    # Parse line gadgets from the JSON array
    gadgets = []
    for merchandise in information.get("line_items", []):
        if not isinstance(merchandise, dict):
            proceed
        qty = merchandise.get("amount")
        gadgets.append(LineItem(
            description=str(merchandise.get("description", "")).strip(),
            amount=float(qty) if qty isn't None else None,
            unit_price=merchandise.get("unit_price"),
            whole=merchandise.get("whole"),
        ))

    return ParsedInvoice(
        vendor_name=safe_str("vendor_name"),
        invoice_number=safe_str("invoice_number"),
        invoice_date=safe_str("invoice_date"),
        due_date=safe_str("due_date"),
        line_items=gadgets,
        subtotal=safe_str("subtotal"),
        tax=safe_str("tax"),
        total_due=safe_str("total_due"),
        forex=safe_str("forex"),
        low_confidence_fields=low_conf,
        raw_output=raw_output,
    )

# ── Bill Parser ────────────────────────────────────────────────────────────

class InvoiceParser:
    """
    Finish-to-end native bill extraction utilizing Gemma 4.
    Processes every PDF web page as a picture -- works on scanned and digital PDFs alike.
    """

    def __init__(self, model_id: str = MODEL_ID, dpi: int = 200):
        self.mannequin, self.processor = load_model(model_id)
        self.renderer = PDFRenderer(dpi=dpi)

    def parse(self, pdf_path: str, token_budget: int = 1120) -> ParsedInvoice:
        """
        Parse a single bill PDF.
        For multi-page invoices, extracts from all pages and merges outcomes,
        with later pages filling in fields not discovered on earlier pages.

        Args:
            pdf_path:     Path to the bill PDF
            token_budget: Visible token price range per web page (560 or 1120 advisable)

        Returns:
            ParsedInvoice with all extracted fields and low_confidence_fields listing
        """
        path = Path(pdf_path)
        if not path.exists():
            elevate FileNotFoundError(f"File not discovered: {pdf_path}")

        pages = self.renderer.render_all(pdf_path)
        print(f"Processing {len(pages)} web page(s) from {path.identify}...")

        merged: Elective[ParsedInvoice] = None

        for i, page_img in enumerate(pages):
            print(f"  Extracting web page {i + 1}/{len(pages)}...")
            uncooked = query_document_page(
                self.mannequin, self.processor,
                page_image=page_img,
                immediate=EXTRACTION_PROMPT,
                token_budget=token_budget,
                enable_thinking=False,   # Use pondering=True for advanced multi-column layouts
            )
            page_result = build_parsed_invoice(uncooked)

            if merged is None:
                merged = page_result
            else:
                # Merge: later pages fill in fields that had been null on earlier pages.
                # Line gadgets are all the time collected throughout pages (handles multi-page tables).
                merged = self._merge(merged, page_result)

        return merged or ParsedInvoice(
            vendor_name=None, invoice_number=None,
            invoice_date=None, due_date=None,
            low_confidence_fields=["no_pages_processed"],
        )

    def _merge(self, base: ParsedInvoice, replace: ParsedInvoice) -> ParsedInvoice:
        """
        Merge two ParsedInvoice outcomes.
        Scalar fields: maintain base worth until it is None (replace fills in).
        line_items: accumulate from each pages.
        low_confidence_fields: intersection of each -- a area is just "assured"
        if it was discovered on at the least one web page.
        """
        def decide(a, b):
            return a if a isn't None else b

        all_low_conf = listing(set(base.low_confidence_fields) & set(replace.low_confidence_fields))

        return ParsedInvoice(
            vendor_name=decide(base.vendor_name, replace.vendor_name),
            invoice_number=decide(base.invoice_number, replace.invoice_number),
            invoice_date=decide(base.invoice_date, replace.invoice_date),
            due_date=decide(base.due_date, replace.due_date),
            line_items=base.line_items + replace.line_items,
            subtotal=decide(base.subtotal, replace.subtotal),
            tax=decide(base.tax, replace.tax),
            total_due=decide(base.total_due, replace.total_due),
            forex=decide(base.forex, replace.forex),
            low_confidence_fields=all_low_conf,
            raw_output=base.raw_output + "n---page---n" + replace.raw_output,
        )

    def parse_directory(self, dir_path: str, token_budget: int = 1120) -> dict[str, ParsedInvoice]:
        """
        Batch-process all PDFs in a listing.
        Returns a dict mapping filename -> ParsedInvoice.
        Failed recordsdata are logged and skipped somewhat than elevating.
        """
        outcomes = {}
        pdfs = listing(Path(dir_path).glob("*.pdf"))
        print(f"Discovered {len(pdfs)} PDF(s) in {dir_path}")

        for pdf_path in pdfs:
            attempt:
                outcomes[pdf_path.name] = self.parse(str(pdf_path), token_budget)
                standing = "OK" if not outcomes[pdf_path.name].low_confidence_fields else 
                         f"LOW CONF: {outcomes[pdf_path.name].low_confidence_fields}"
                print(f"  {pdf_path.identify}: {standing}")
            besides Exception as e:
                print(f"  {pdf_path.identify}: FAILED -- {e}")

        return outcomes


# ── Run it ────────────────────────────────────────────────────────────────────

if __name__ == "__main__":
    import sys

    if len(sys.argv) < 2:
        print("Utilization: python invoice_parser.py ")
        sys.exit(1)

    parser = InvoiceParser()
    outcome = parser.parse(sys.argv[1])

    print("n── Extracted Bill ──")
    print(f"Vendor        : {outcome.vendor_name}")
    print(f"Bill #     : {outcome.invoice_number}")
    print(f"Bill Date  : {outcome.invoice_date}")
    print(f"Due Date      : {outcome.due_date}")
    print(f"Foreign money      : {outcome.forex}")
    print(f"Whole Due     : {outcome.total_due}")
    print(f"Line Objects    : {len(outcome.line_items)}")
    for merchandise in outcome.line_items:
        print(f"  - {merchandise.description}: qty={merchandise.amount}, unit={merchandise.unit_price}, whole={merchandise.whole}")
    if outcome.low_confidence_fields:
        print(f"n⚠ Low-confidence fields (overview manually): {outcome.low_confidence_fields}")

python invoice_parser.py supplier_invoice.pdf

parser = InvoiceParser()
outcomes = parser.parse_directory("./invoices/")
# outcomes is a dict: {"invoice_001.pdf": ParsedInvoice, "invoice_002.pdf": ParsedInvoice, ...}

# two_pass_pipeline.py
# Conditions: pdf_renderer.py, gemma4_loader.py, and invoice_parser.py in the identical listing

from pdf_renderer import PDFRenderer
from gemma4_loader import load_model, query_document_page
from invoice_parser import EXTRACTION_PROMPT

CLASSIFICATION_PROMPT = """Take a look at this doc web page and classify it with a single label.

Select precisely one:
- invoice_header    (firm logos, vendor handle, bill quantity, date)
- line_items        (desk of merchandise/companies with portions and costs)
- totals            (subtotals, taxes, grand whole, cost directions)
- phrases             (phrases and situations, authorized textual content, return coverage)
- cowl             (title web page, desk of contents, doc cowl)
- clean             (empty or practically empty web page)
- different             (anything)

Reply with ONLY the label -- no rationalization, no punctuation."""

EXTRACTABLE_CLASSES = {"invoice_header", "line_items", "totals"}


def two_pass_parse(pdf_path: str, mannequin, processor) -> dict:
    """
    Two-pass bill extraction pipeline.

    Go 1 (token_budget=280): Classify every web page cheaply.
    Go 2 (token_budget=1120): Full extraction solely on content material pages.

    Returns a dict with extracted fields and page-level classification metadata.
    """
    renderer_fast = PDFRenderer(dpi=150)   # Decrease DPI for classification -- sooner
    renderer_full = PDFRenderer(dpi=200)   # Full DPI for extraction

    # ── Go 1: Classify all pages at 280-token price range ─────────────────────
    print("Go 1: Web page classification...")
    fast_pages = renderer_fast.render_all(pdf_path)
    page_labels = {}

    for i, page_img in enumerate(fast_pages):
        label = query_document_page(
            mannequin, processor,
            page_image=page_img,
            immediate=CLASSIFICATION_PROMPT,
            token_budget=280,         # Low-cost move -- simply must learn the web page sort
            enable_thinking=False,
            max_new_tokens=16,        # Classification response is all the time brief
        ).strip().decrease()
        page_labels[i] = label
        standing = "EXTRACT" if label in EXTRACTABLE_CLASSES else "skip"
        print(f"  Web page {i}: '{label}' -> {standing}")

    extraction_pages = [i for i, l in page_labels.items() if l in EXTRACTABLE_CLASSES]

    if not extraction_pages:
        print("No extractable pages discovered -- falling again to full-document extraction")
        extraction_pages = listing(vary(len(fast_pages)))

    skipped = len(fast_pages) - len(extraction_pages)
    print(f"nPass 1 full: {len(extraction_pages)} pages to extract, "
          f"{skipped} skipped ({skipped/len(fast_pages)*100:.0f}% token saving)")

    # ── Go 2: Full extraction at 1120-token price range ────────────────────────
    print("nPass 2: Full extraction...")
    full_pages = renderer_full.render_all(pdf_path)
    extracted_outputs = []

    for i in extraction_pages:
        print(f"  Extracting web page {i} ({page_labels[i]})...")
        uncooked = query_document_page(
            mannequin, processor,
            page_image=full_pages[i],
            immediate=EXTRACTION_PROMPT,
            token_budget=1120,
            enable_thinking=False,
            max_new_tokens=1024,
        )
        extracted_outputs.append({"web page": i, "label": page_labels[i], "output": uncooked})

    return {
        "page_labels": page_labels,
        "extraction_pages": extraction_pages,
        "outputs": extracted_outputs,
    }

mannequin, processor = load_model()
outcome = two_pass_parse("supplier_invoice_5pages.pdf", mannequin, processor)

# Allow pondering mode for advanced paperwork
outcome = query_document_page(
    mannequin, processor,
    page_image=page_img,
    immediate=EXTRACTION_PROMPT,
    token_budget=1120,
    enable_thinking=True,     # Provides reasoning hint earlier than structured output
    max_new_tokens=2048,      # Considering outputs are longer -- improve the price range
)
# The ... block is already stripped by query_document_page
# outcome accommodates solely the ultimate JSON reply

# validation.py
# pip set up pydantic>=2.0

from pydantic import BaseModel, field_validator
from typing import Elective
import re

class InvoiceValidator(BaseModel):
    """
    Enterprise-rule validation on prime of uncooked ParsedInvoice output.
    Use this earlier than committing to an accounting system.
    """
    vendor_name: Elective[str]
    invoice_number: Elective[str]
    invoice_date: Elective[str]
    due_date: Elective[str]
    total_due: Elective[str]
    forex: Elective[str]

    @field_validator("invoice_number")
    @classmethod
    def invoice_number_format(cls, v):
        """Flag bill numbers that appear to be they had been hallucinated."""
        if v and never re.match(r"^[A-Z0-9-/.]+$", v.strip()):
            elevate ValueError(f"Surprising bill quantity format: {v!r}")
        return v

    @field_validator("total_due")
    @classmethod
    def total_due_has_value(cls, v):
        """Invoices with out a total_due ought to by no means auto-commit."""
        if not v:
            elevate ValueError("total_due is required for automated processing")
        return v

    @field_validator("forex")
    @classmethod
    def currency_is_known(cls, v):
        KNOWN = {"USD", "EUR", "GBP", "NGN", "CAD", "AUD", "JPY", "CNY"}
        if v and v.higher() not in KNOWN:
            elevate ValueError(f"Unrecognized forex: {v!r}")
        return v.higher() if v else v


def validate_for_commit(bill) -> tuple[bool, list[str]]:
    """
    Validate a ParsedInvoice earlier than committing to accounting system.
    Returns (can_commit, list_of_errors).
    """
    errors = []
    attempt:
        InvoiceValidator(
            vendor_name=bill.vendor_name,
            invoice_number=bill.invoice_number,
            invoice_date=bill.invoice_date,
            due_date=bill.due_date,
            total_due=bill.total_due,
            forex=bill.forex,
        )
    besides Exception as e:
        errors = [str(err) for err in e.errors()] if hasattr(e, "errors") else [str(e)]

    # Additionally block commit if any vital fields are low-confidence
    vital = {"vendor_name", "invoice_number", "total_due"}
    low_critical = vital & set(bill.low_confidence_fields)
    if low_critical:
        errors.append(f"Low-confidence on vital fields: {low_critical}")

    return len(errors) == 0, errors