Cómo extraer datos de facturas con Claude

mkdir extractor-facturas && cd extractor-facturas
mkdir facturas
python -m venv .venv
source .venv/bin/activate   # En Windows: .venv\Scripts\activate
pip install anthropic

export ANTHROPIC_API_KEY="sk-ant-..."

python -c "import anthropic; r = anthropic.Anthropic().messages.create(model='claude-opus-5-5', max_tokens=50, messages=[{'role':'user','content':'Di hola'}]); print(r.content[0].text)"

# esquema.py
ENTIDAD = {
    "type": "object",
    "properties": {
        "nombre": {"type": ["string", "null"]},
        "id_fiscal": {
            "type": ["string", "null"],
            "description": "NIF/CIF (España), RFC (México), CUIT (Argentina), NIT (Colombia), RUT (Chile), RUC (Perú)... tal cual aparece",
        },
        "direccion": {"type": ["string", "null"]},
    },
    "required": ["nombre", "id_fiscal"],
}

TOOL_FACTURA = {
    "name": "registrar_factura",
    "description": "Registra los datos extraídos de una factura o ticket.",
    "input_schema": {
        "type": "object",
        "properties": {
            "numero_factura": {"type": ["string", "null"]},
            "fecha_emision": {"type": ["string", "null"], "description": "Formato AAAA-MM-DD"},
            "fecha_vencimiento": {"type": ["string", "null"], "description": "Formato AAAA-MM-DD"},
            "emisor": ENTIDAD,
            "receptor": ENTIDAD,
            "moneda": {"type": "string", "description": "Código ISO 4217: EUR, MXN, COP, ARS, CLP, PEN, USD..."},
            "lineas": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "descripcion": {"type": "string"},
                        "cantidad": {"type": ["number", "null"]},
                        "precio_unitario": {"type": ["number", "null"]},
                        "importe": {"type": ["number", "null"], "description": "Importe de la línea sin impuestos, con descuentos aplicados"},
                    },
                    "required": ["descripcion", "importe"],
                },
            },
            "base_imponible": {"type": ["number", "null"]},
            "impuestos": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "tipo": {"type": "string", "description": "Ej: IVA 21%, IVA 16%, IGV 18%"},
                        "porcentaje": {"type": ["number", "null"]},
                        "importe": {"type": "number"},
                    },
                    "required": ["tipo", "importe"],
                },
            },
            "retenciones": {"type": ["number", "null"], "description": "Total retenido (p. ej. IRPF) en positivo"},
            "total": {"type": ["number", "null"], "description": "Total a pagar"},
        },
        "required": ["numero_factura", "fecha_emision", "emisor", "moneda", "lineas", "base_imponible", "impuestos", "total"],
    },
}

# extractor.py
import base64
import sys
import json
from pathlib import Path

import anthropic
from esquema import TOOL_FACTURA

client = anthropic.Anthropic(max_retries=3)
MODELO = "claude-opus-5-5"

SYSTEM = (
    "Eres un asistente contable experto en facturas de España y Latinoamérica. "
    "Extrae SOLO lo que aparece en el documento; nunca inventes ni deduzcas datos ausentes. "
    "Si un campo no aparece, devuelve null. "
    "Convierte importes a número con punto decimal y sin separador de miles (1.234,56 -> 1234.56). "
    "Fechas en formato AAAA-MM-DD; las fechas españolas y latinoamericanas suelen ir como DD/MM/AAAA. "
    "Las retenciones (IRPF, etc.) van en positivo en el campo retenciones, no como impuesto."
)

TIPOS_IMAGEN = {".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".webp": "image/webp", ".gif": "image/gif"}


def _bloque_archivo(ruta: Path) -> dict:
    datos = base64.standard_b64encode(ruta.read_bytes()).decode("utf-8")
    ext = ruta.suffix.lower()
    if ext == ".pdf":
        return {"type": "document", "source": {"type": "base64", "media_type": "application/pdf", "data": datos}}
    if ext in TIPOS_IMAGEN:
        return {"type": "image", "source": {"type": "base64", "media_type": TIPOS_IMAGEN[ext], "data": datos}}
    raise ValueError(f"Formato no soportado: {ext}")


def extraer_factura(ruta, nota: str = "") -> dict:
    ruta = Path(ruta)
    instruccion = "Extrae los datos de esta factura."
    if nota:
        instruccion += "\n\n" + nota
    respuesta = client.messages.create(
        model=MODELO,
        max_tokens=4096,
        system=SYSTEM,
        tools=[TOOL_FACTURA],
        tool_choice={"type": "tool", "name": "registrar_factura"},
        messages=[{"role": "user", "content": [_bloque_archivo(ruta), {"type": "text", "text": instruccion}]}],
    )
    for bloque in respuesta.content:
        if bloque.type == "tool_use":
            return bloque.input
    raise ValueError("Claude no devolvió datos estructurados")


if __name__ == "__main__":
    print(json.dumps(extraer_factura(sys.argv[1]), ensure_ascii=False, indent=2))

# validacion.py
from datetime import date

TOLERANCIA = 0.05


def validar_factura(f: dict) -> list[str]:
    problemas = []

    for campo in ("numero_factura", "fecha_emision", "total"):
        if f.get(campo) in (None, ""):
            problemas.append(f"Falta el campo {campo}")

    base = f.get("base_imponible")
    lineas = f.get("lineas") or []
    if lineas and base is not None:
        suma = round(sum((l.get("importe") or 0) for l in lineas), 2)
        if abs(suma - base) > TOLERANCIA:
            problemas.append(f"Las líneas suman {suma} pero la base imponible es {base}")

    total = f.get("total")
    if base is not None and total is not None:
        impuestos = sum((i.get("importe") or 0) for i in f.get("impuestos") or [])
        retenciones = f.get("retenciones") or 0
        esperado = round(base + impuestos - retenciones, 2)
        if abs(esperado - total) > TOLERANCIA:
            problemas.append(
                f"Base ({base}) + impuestos ({round(impuestos, 2)}) - retenciones ({retenciones}) = {esperado}, pero el total es {total}"
            )

    fecha = f.get("fecha_emision")
    if fecha:
        try:
            date.fromisoformat(fecha)
        except ValueError:
            problemas.append(f"La fecha de emisión '{fecha}' no tiene formato AAAA-MM-DD válido")

    return problemas

# procesar.py
import csv
import json
from pathlib import Path

from extractor import extraer_factura
from validacion import validar_factura


def extraer_con_reintento(ruta, intentos: int = 2):
    datos = extraer_factura(ruta)
    for _ in range(intentos):
        problemas = validar_factura(datos)
        if not problemas:
            return datos, []
        nota = (
            "En una extracción anterior de este mismo documento se detectaron estas incoherencias: "
            + "; ".join(problemas)
            + ". Revisa el documento con cuidado, en especial los importes y las cifras dudosas, "
            "y devuelve los valores tal como aparecen impresos. Si el documento realmente no cuadra, respeta lo impreso."
        )
        datos = extraer_factura(ruta, nota=nota)
    return datos, validar_factura(datos)

# Añade esto al final de procesar.py

EXTENSIONES = {".pdf", ".png", ".jpg", ".jpeg", ".webp", ".gif"}
CAMPOS = ["archivo", "numero_factura", "fecha_emision", "emisor_nombre", "emisor_id_fiscal",
          "receptor_nombre", "receptor_id_fiscal", "moneda", "base_imponible", "total_impuestos",
          "retenciones", "total", "estado"]


def _num(valor):
    return "" if valor is None else f"{valor:.2f}".replace(".", ",")


def procesar_carpeta(carpeta: str = "facturas", salida: str = "facturas.csv"):
    filas = []
    for ruta in sorted(Path(carpeta).iterdir()):
        if ruta.suffix.lower() not in EXTENSIONES:
            continue
        print(f"Procesando {ruta.name}...")
        try:
            datos, problemas = extraer_con_reintento(ruta)
        except Exception as e:
            filas.append({"archivo": ruta.name, "estado": f"ERROR: {e}"})
            continue

        ruta.with_suffix(".json").write_text(json.dumps(datos, ensure_ascii=False, indent=2), encoding="utf-8")
        emisor = datos.get("emisor") or {}
        receptor = datos.get("receptor") or {}
        impuestos = sum((i.get("importe") or 0) for i in datos.get("impuestos") or [])
        filas.append({
            "archivo": ruta.name,
            "numero_factura": datos.get("numero_factura") or "",
            "fecha_emision": datos.get("fecha_emision") or "",
            "emisor_nombre": emisor.get("nombre") or "",
            "emisor_id_fiscal": emisor.get("id_fiscal") or "",
            "receptor_nombre": receptor.get("nombre") or "",
            "receptor_id_fiscal": receptor.get("id_fiscal") or "",
            "moneda": datos.get("moneda") or "",
            "base_imponible": _num(datos.get("base_imponible")),
            "total_impuestos": _num(impuestos),
            "retenciones": _num(datos.get("retenciones")),
            "total": _num(datos.get("total")),
            "estado": "OK" if not problemas else "REVISAR: " + " | ".join(problemas),
        })

    with open(salida, "w", newline="", encoding="utf-8-sig") as f:
        writer = csv.DictWriter(f, fieldnames=CAMPOS, delimiter=";", restval="")
        writer.writeheader()
        writer.writerows(filas)
    print(f"Listo: {len(filas)} facturas en {salida}")


if __name__ == "__main__":
    procesar_carpeta()