Bank statement to Excel with Python and Nutrient

{
  "parts": [{ "file": "document" }],
  "output": { "type": "xlsx" }
}

# schema.py.

BANK_STATEMENT_SCHEMA = {
    "type": "object",
    "properties": {
        "bank_name": {
            "type": "string",
            "description": "Name of the financial institution that issued the statement",
        },
        "account_holder_name": {
            "type": "string",
            "description": "Name of the account holder the statement belongs to",
        },
        "account_number": {
            "type": "string",
            "description": "Account number or IBAN, however it's labeled on the statement",
        },
        "currency": {
            "type": "string",
            "description": "ISO 4217 currency code, e.g. USD or EUR",
        },
        "statement_start_date": {
            "type": "string",
            "format": "date",
            "description": "First date of the statement period",
        },
        "statement_end_date": {
            "type": "string",
            "format": "date",
            "description": "Last date of the statement period",
        },
        "opening_balance": {
            "type": "number",
            "description": "Account balance at the start of the statement period",
        },
        "closing_balance": {
            "type": "number",
            "description": "Account balance at the end of the statement period",
        },
        "transactions": {
            "type": "array",
            "description": "One entry per row in the transaction table, in statement order",
            "items": {
                "type": "object",
                "properties": {
                    "date": {"type": "string", "format": "date"},
                    "description": {
                        "type": "string",
                        "description": "Transaction description or memo as printed",
                    },
                    "amount": {
                        "type": "number",
                        "description": "Signed amount — negative for debits, positive for credits",
                    },
                    "running_balance": {
                        "type": "number",
                        "description": "Account balance after this transaction, if the statement shows one",
                    },
                },
                "required": ["date", "amount"],
            },
        },
    },
    "required": ["account_number", "currency", "opening_balance", "closing_balance", "transactions"],
}

# extract.py.
import json
import os

import requests
from dotenv import load_dotenv

from schema import BANK_STATEMENT_SCHEMA

load_dotenv()
ENDPOINT = "https://api.nutrient.io/extraction/extract"


def extract_statement(pdf_path: str) -> dict:
    """Extract fields, retaining the full API response for review."""
    api_key = os.getenv("NUTRIENT_API_KEY")
    if not api_key:
        raise RuntimeError("Set NUTRIENT_API_KEY in your environment or .env file.")
    instructions = {
        "schema": BANK_STATEMENT_SCHEMA,
        "instructions": "Extract all transactions in statement order. Include fees and interest as separate rows. Use negative amounts for debits and positive amounts for credits. Do not invent missing transactions.",
        "parseConfig": {"mode": "understand"},
        "options": {"includeCitations": True},
    }
    try:
        with open(pdf_path, "rb") as source:
            response = requests.post(
                ENDPOINT,
                headers={"Authorization": f"Bearer {api_key}"},
                files={"file": source},
                data={"instructions": json.dumps(instructions)},
                timeout=(10, 330),
            )
    except requests.RequestException as exc:
        raise RuntimeError("Extraction request failed; check connectivity and retry deliberately.") from exc

    try:
        result = response.json()
    except ValueError as exc:
        raise RuntimeError(f"API returned HTTP {response.status_code} with a non-JSON body.") from exc

    if not response.ok:
        details = result if isinstance(result, dict) else {}
        message = details.get("errorMessage") or details.get("message") or "Request failed"
        raise RuntimeError(f"API HTTP {response.status_code}: {message}")
    if not isinstance(result, dict) or result.get("status") != 200:
        raise RuntimeError("Unexpected extraction response; expected status 200.")
    if not isinstance(result.get("output"), dict) or not isinstance(result["output"].get("data"), dict):
        raise RuntimeError("Extraction response is missing output.data.")
    return result

{
  "output": {
    "data": {
      "transactions": [
        {
          "amount": -42.18
        }
      ]
    },
    "metadata": {
      "transactions": [
        {
          "amount": {
            "match": "id_match",
            "citation": {
              "confidence": {
                "groundedness": 0.9997683167457581,
                "schemaValid": true
              },
              "regions": [
                {
                  "bbox": {
                    "height": 31.70635986328125,
                    "width": 77.5469970703125,
                    "x": 1091.9189453125,
                    "y": 902.34375
                  },
                  "pageIndex": 0,
                  "pageNumber": 1
                }
              ],
              "status": "cited"
            },
            "pageNumber": 1,
            "bbox": {
              "height": 31.70635986328125,
              "width": 77.5469970703125,
              "x": 1091.9189453125,
              "y": 902.34375
            }
          }
        }
      ]
    }
  }
}

# review.py.
from decimal import Decimal

from jsonschema import FormatChecker, validate

from schema import BANK_STATEMENT_SCHEMA


def review_fields(data, metadata, path=""):
    """Yield each scalar value with its citation and any review reason."""
    if isinstance(data, dict):
        for key, value in data.items():
            citation = metadata.get(key, {}) if isinstance(metadata, dict) else {}
            yield from review_fields(value, citation, f"{path}.{key}" if path else key)
    elif isinstance(data, list):
        for index, value in enumerate(data):
            citation = metadata[index] if isinstance(metadata, list) and index < len(metadata) else {}
            yield from review_fields(value, citation, f"{path}[{index}]")
    else:
        citation = metadata if isinstance(metadata, dict) else {}
        match = citation.get("match")
        if not citation:
            reason = "Missing citation; check source"
        elif match not in ("id_match", "id_match_multiblock"):
            reason = f"Grounding needs review: {match or 'missing match'}"
        else:
            reason = ""
        if not reason and (citation.get("pageNumber") is None or not citation.get("bbox")):
            reason = "Missing source location; check source"
        yield {"field": path, "value": data, "reason": reason, "citation": citation}


def check_balances(data):
    """Validate schema and compare balances for a single-currency statement."""
    validate(data, BANK_STATEMENT_SCHEMA, format_checker=FormatChecker())

    def money(value):
        amount = Decimal(str(value))
        if not amount.is_finite():
            raise ValueError("Balances and amounts must be finite numbers.")
        return amount

    checks = []
    balance = money(data["opening_balance"])
    if not data["transactions"]:
        checks.append({"field": "transactions", "reason": "No transactions extracted; confirm against the statement"})
    for index, transaction in enumerate(data["transactions"]):
        balance += money(transaction["amount"])
        if transaction.get("running_balance") is not None and money(transaction["running_balance"]) != balance:
            checks.append({"field": f"transactions[{index}].running_balance", "reason": f"Balance mismatch: expected {balance}"})
    if balance != money(data["closing_balance"]):
        checks.append({"field": "closing_balance", "reason": f"Opening balance plus transactions is {balance}; extracted closing balance is {data['closing_balance']}"})
    return checks

# to_excel.py.
import json
from datetime import date

from openpyxl import Workbook
from openpyxl.styles import Font, PatternFill
from openpyxl.utils import get_column_letter

FLAG_FILL = PatternFill(start_color="FFF2CC", end_color="FFF2CC", fill_type="solid")


def put(sheet, row, column, value):
    """Write source text literally, even when it starts with an equals sign."""
    cell = sheet.cell(row=row, column=column, value=value)
    if isinstance(value, str):
        cell.data_type = "s"
    return cell


def write_statement_to_excel(data, reviews, checks, output_path):
    wb = Workbook()
    summary = wb.active
    summary.title = "Summary"
    flagged = {item["field"] for item in reviews if item["reason"]}
    flagged.update(check["field"] for check in checks)
    fields = [
        ("Bank", "bank_name"), ("Account holder", "account_holder_name"),
        ("Account number", "account_number"), ("Currency", "currency"),
        ("Statement start", "statement_start_date"), ("Statement end", "statement_end_date"),
        ("Opening balance", "opening_balance"), ("Closing balance", "closing_balance"),
    ]
    summary.append(["Field", "Value"])
    for row, (label, key) in enumerate(fields, start=2):
        put(summary, row, 1, label)
        value = data.get(key)
        if key.endswith("_date") and value is not None:
            value = date.fromisoformat(value)
        cell = put(summary, row, 2, value)
        if key in flagged:
            cell.fill = FLAG_FILL
    summary.append(["Review status", "Pending human review"])

    transactions = wb.create_sheet("Transactions")
    transactions.append(["Date", "Description", "Amount", "Running balance"])
    for index, transaction in enumerate(data["transactions"]):
        for column, key in enumerate(["date", "description", "amount", "running_balance"], start=1):
            value = transaction.get(key)
            if key == "date":
                value = date.fromisoformat(value)
            cell = put(transactions, index + 2, column, value)
            if f"transactions[{index}].{key}" in flagged:
                cell.fill = FLAG_FILL

    evidence = wb.create_sheet("Review")
    evidence.append(["Field", "Value", "Review reason", "Match", "Confidence (if returned)",
                     "Citation confidence metadata", "Page", "Bounding box"])
    for row, item in enumerate(reviews, start=2):
        citation = item["citation"]
        details = citation.get("citation")
        signals = details.get("confidence") if isinstance(details, dict) else None
        values = [item["field"], item["value"], item["reason"] or "Source matched; verify before use",
                  citation.get("match"), citation.get("confidence"),
                  json.dumps(signals) if signals is not None else None, citation.get("pageNumber"),
                  json.dumps(citation.get("bbox")) if citation.get("bbox") else None]
        for column, value in enumerate(values, start=1):
            cell = put(evidence, row, column, value)
            if item["reason"]:
                cell.fill = FLAG_FILL

    balances = wb.create_sheet("Checks")
    balances.append(["Field", "Result"])
    rows = checks or [{"field": "Balances", "reason": "Arithmetic matches; this does not prove extraction accuracy"}]
    for row, check in enumerate(rows, start=2):
        put(balances, row, 1, check["field"])
        put(balances, row, 2, check["reason"])
        if checks:
            balances.cell(row, 2).fill = FLAG_FILL

    for sheet in wb:
        sheet.freeze_panes = "A2"
        sheet.auto_filter.ref = sheet.dimensions
        for cell in sheet[1]:
            cell.font = Font(bold=True)
        for column in sheet.columns:
            width = min(65, max(14, max(len(str(cell.value or "")) for cell in column) + 2))
            sheet.column_dimensions[get_column_letter(column[0].column)].width = width
        for row in sheet:
            for cell in row:
                if isinstance(cell.value, date):
                    cell.number_format = "yyyy-mm-dd"
    wb.save(output_path)

# main.py.
import json
import sys
from pathlib import Path

from jsonschema import ValidationError

from extract import extract_statement
from review import check_balances, review_fields
from to_excel import write_statement_to_excel


def process_statement(pdf_path: str, output_path: str) -> None:
    source = Path(pdf_path)
    target = Path(output_path)
    evidence_path = target.with_suffix(".json")
    if target.suffix.lower() != ".xlsx":
        raise ValueError("Use an .xlsx output path.")
    if source.resolve() in (target.resolve(), evidence_path.resolve()):
        raise ValueError("Output files must not overwrite the source.")
    if target.exists() or evidence_path.exists():
        raise ValueError("Choose a new output name; the workbook or evidence file already exists.")

    result = extract_statement(str(source))
    evidence_path.write_text(json.dumps(result, indent=2, ensure_ascii=False), encoding="utf-8")
    data = result["output"]["data"]
    checks = check_balances(data)
    reviews = list(review_fields(data, result["output"].get("metadata", {})))
    write_statement_to_excel(data, reviews, checks, target)
    print(f"Wrote {target}; status: pending human review")
    print(f"{sum(bool(item['reason']) for item in reviews)} citation review flag(s); {len(checks)} balance issue(s)")
    print(f"Full response and citations: {evidence_path}")


if __name__ == "__main__":
    try:
        if len(sys.argv) != 3:
            raise ValueError("Usage: python main.py statement.pdf statement.xlsx")
        process_statement(sys.argv[1], sys.argv[2])
    except (OSError, RuntimeError, ValueError, ValidationError) as exc:
        print(f"Error: {exc}", file=sys.stderr)
        sys.exit(1)