Python

CONSTAIA_API_KEY=ck_test_...
CONSTAIA_WEBHOOK_SECRET=whsec_...

from constaia import Constaia

client = Constaia()  # lee CONSTAIA_API_KEY

analysis = client.analyze(
    "dni_valid.jpg",
    expect="es_dni",
    checks={"not_expired": True},
    storage="none",
    language="es",
)

print(analysis["verdict"]["status"])                    # valid
print(analysis["fields"]["document_number"]["value"])   # 12345678Z
for reason in analysis["verdict"]["reasons"]:
    print(reason["code"], reason["severity"], reason["message"])

{
  "id": "an_01J...",
  "object": "analysis",
  "status": "completed",
  "livemode": false,
  "document": { "type": "es_dni", "label": "DNI (España)", "confidence": 0.97, "side": "both", "country": "ESP" },
  "verdict": {
    "expected": ["es_dni"],
    "match": true,
    "status": "valid",
    "reasons": [
      { "code": "type_match", "severity": "info", "message": "El documento es DNI (España)." },
      { "code": "not_expired", "severity": "info", "message": "Vigente hasta el 12/03/2031." }
    ]
  },
  "fields": {
    "document_number": { "value": "12345678Z", "confidence": 0.99, "validated": true, "source": { "page": 1, "bbox": [0.61, 0.12, 0.83, 0.16] } }
  },
  "checks": [
    { "code": "nif_check_digit", "passed": true, "message": "La letra del documento 12345678Z es correcta." }
  ],
  "warnings": [],
  "usage": { "credits": 0, "pages": 1 }
}

from constaia.types import Analysis


def decide(analysis: Analysis) -> str:
    verdict = analysis["verdict"]
    if verdict is None:
        return "sin veredicto"
    if verdict["status"] == "valid":
        return "aceptado"
    if verdict["status"] == "review":
        return "revisión manual"
    errors = [r["message"] for r in verdict["reasons"] if r["severity"] == "error"]
    return "rechazado: " + "; ".join(errors)

from constaia import Constaia

client = Constaia()

receipt = client.analyze(
    file_url="https://files.example.com/receipts/payment_receipt.pdf",
    expect="payment_receipt",
    checks={
        "expected_amount": 45,
        "expected_iban": "ES7921000813610123456789",
        "expected_reference": "INSCRIPCION 123",
    },
)

with open("invoice.pdf", "rb") as fh:
    invoice = client.analyze(fh, expect="invoice", export=["xlsx"], metadata={"supplier": "acme"})
print(invoice["exports"].get("xlsx"))  # URL firmada, válida 24 h

kind = client.classify("unknown.pdf", expect=["invoice", "payment_receipt"])  # 0,2 créditos
print(kind["document"], kind["candidates"])

from constaia import (
    Constaia,
    ConstaiaError,
    InsufficientCreditsError,
    InvalidRequestError,
    RateLimitError,
)

client = Constaia()

USER_MESSAGES = {
    "file_too_large": "El archivo supera los 20 MB.",
    "unsupported_file_type": "Sube una imagen (JPG, PNG, WEBP, HEIC) o un PDF.",
    "unreadable_image": "No se puede leer la imagen. Haz otra foto con buena luz.",
    "too_many_pages": "El PDF tiene demasiadas páginas.",
}


def check_document(path: str) -> dict:
    try:
        analysis = client.analyze(path, expect="es_dni", checks={"not_expired": True})
    except InvalidRequestError as exc:
        return {"error": USER_MESSAGES.get(exc.code or "", "El documento no se ha podido procesar.")}
    except (InsufficientCreditsError, RateLimitError) as exc:
        print(f"Constaia {exc.status} {exc.code} request_id={exc.request_id}")
        return {"error": "El servicio está ocupado. Inténtalo en unos minutos."}
    except ConstaiaError as exc:
        print(f"Constaia {exc.status} {exc.code} request_id={exc.request_id}: {exc.message}")
        return {"error": "No hemos podido comprobar el documento. Inténtalo más tarde."}
    return {"id": analysis["id"], "status": analysis["verdict"]["status"]}

from constaia import Constaia

client = Constaia(
    timeout=60.0,        # segundos por intento
    max_retries=3,
    max_concurrency=4,
)

analysis = client.analyze("dni_valid.jpg", expect="es_dni", idempotency_key="user-42-dni", timeout=90)

print(client.last_request_id)   # "req_…": inclúyelo si contactas con soporte
print(client.last_rate_limit)   # RateLimit(limit=10, remaining=9, reset=1, policy='10;w=1', retry_after=None)

import asyncio
from pathlib import Path

from constaia import AsyncConstaia, ConstaiaError


async def main() -> None:
    files = sorted(Path("inbox").glob("*.pdf"))
    async with AsyncConstaia(max_concurrency=4) as client:
        results = await asyncio.gather(
            *(client.analyze(path, expect="invoice") for path in files),
            return_exceptions=True,
        )
        for path, result in zip(files, results):
            if isinstance(result, ConstaiaError):
                print(path.name, "error", result.code, result.request_id)
            else:
                print(path.name, result["verdict"]["status"], result["fields"].get("total", {}).get("value"))

        async for analysis in client.analyses.list(status="completed", type="invoice"):
            print(analysis["id"])


asyncio.run(main())

import time

from constaia import Constaia

client = Constaia()
analysis = client.analyze("contract.pdf", async_=True)
while analysis["status"] in ("queued", "processing"):
    time.sleep(3)
    analysis = client.analyses.get(analysis["id"])
print(analysis["status"], analysis.get("error"))

import os

import constaia

SECRET = os.environ["CONSTAIA_WEBHOOK_SECRET"]


def handle(raw_body: bytes, headers) -> int:
    try:
        event = constaia.webhooks.verify(raw_body, headers, SECRET)
    except constaia.WebhookVerificationError:
        return 400
    # Deduplica por headers["webhook-id"]: es el mismo en todos los reintentos.
    match event["type"]:
        case "analysis.completed" | "analysis.review_required" | "analysis.failed":
            analysis = event["data"]
            print(event["type"], analysis["id"], (analysis.get("verdict") or {}).get("status"))
        case "batch.completed":
            print("batch", event["data"]["id"], event["data"]["counts"])
        case "credits.low":
            print("credits", event["data"]["credits_available"])
    return 204

import json
import os
from pathlib import Path

import pytest

import constaia
from constaia import Constaia

FIXTURES = Path(__file__).parent / "fixtures"
needs_test_key = pytest.mark.skipif(
    not os.environ.get("CONSTAIA_API_KEY", "").startswith("ck_test_"),
    reason="CONSTAIA_API_KEY=ck_test_... required",
)


@pytest.fixture(scope="module")
def client():
    with Constaia() as c:
        yield c


@needs_test_key
def test_valid_dni(client):
    analysis = client.analyze(FIXTURES / "dni_valid.jpg", expect="es_dni")
    assert analysis["livemode"] is False
    assert analysis["usage"]["credits"] == 0
    assert analysis["verdict"]["status"] == "valid"
    assert analysis["fields"]["document_number"]["value"] == "12345678Z"


@needs_test_key
def test_expired_dni_is_invalid(client):
    analysis = client.analyze(FIXTURES / "dni_expired.jpg", expect="es_dni")
    assert analysis["verdict"]["status"] == "invalid"
    assert {"code": "not_expired", "severity": "error", "message": "Caducado el 15/06/2020."} in analysis["verdict"]["reasons"]


@needs_test_key
def test_blurry_needs_review(client):
    analysis = client.analyze(FIXTURES / "blurry.jpg", expect="es_dni")
    assert analysis["verdict"]["status"] == "review"
    assert "blurry" in analysis["warnings"]


def test_webhook_signature_roundtrip():
    secret = "whsec_c2VjcmV0LWRlLXBydWViYXMtMzItYnl0ZXMtbGFyZ28="
    body = json.dumps({"type": "analysis.completed", "data": {"id": "an_test"}})
    headers = constaia.webhooks.sign(body, secret)
    assert constaia.webhooks.verify(body.encode(), headers, secret)["data"]["id"] == "an_test"
    with pytest.raises(constaia.WebhookVerificationError):
        constaia.webhooks.verify(body.encode() + b" ", headers, secret)

CONSTAIA_API_KEY=ck_test_... pytest -q

import base64
import hashlib
import hmac
import json
import time


def verify_webhook(payload: bytes, headers, secret: str, tolerance: int = 300) -> dict:
    msg_id = headers.get("webhook-id", "")
    timestamp = headers.get("webhook-timestamp", "")
    signatures = headers.get("webhook-signature", "")
    if not (msg_id and timestamp.isdigit() and signatures) or abs(time.time() - int(timestamp)) > tolerance:
        raise ValueError("invalid webhook headers")
    key = base64.b64decode(secret.removeprefix("whsec_"))
    expected = "v1," + base64.b64encode(
        hmac.new(key, f"{msg_id}.{timestamp}.".encode() + payload, hashlib.sha256).digest()
    ).decode()
    if not any(hmac.compare_digest(candidate, expected) for candidate in signatures.split()):
        raise ValueError("invalid webhook signature")
    return json.loads(payload)