Cómo integrar Zettana en tu software contable, ERP o CRM

import json
import os
from pathlib import Path
from typing import Iterable

import requests

BASE_URL = "https://backend.zettana.com/v1"
API_KEY = os.environ["ZETTANA_API_KEY"]
INITIAL_SINCE = os.environ.get(
    "ZETTANA_INITIAL_SINCE",
    "2026-04-01T00:00:00.000Z",
)

# Demo simple: este archivo guarda el estado de avance de la integración.
# En producción, normalmente lo guardarías en tu base de datos
# o en la tabla de estado de integraciones de tu sistema.
STATE_FILE = Path("zettana_sync_state.json")
PDF_ROOT = Path("zettana_pdfs")
PAGE_SIZE = 100
XML_BATCH_SIZE = 100  # El endpoint /v1/documents/xml-contents acepta hasta 100 IDs.

session = requests.Session()
session.headers.update({"x-api-key": API_KEY})


def load_state() -> dict:
    """Carga el estado persistido de la integración."""
    if STATE_FILE.exists():
        return json.loads(STATE_FILE.read_text(encoding="utf-8"))

    return {
        "since": INITIAL_SINCE,
        "processed_ids_at_latest_retrieved_at": [],
    }


def save_state(state: dict) -> None:
    """Guarda el punto de reanudación para la siguiente vez que desees obtener documentos."""
    STATE_FILE.write_text(
        json.dumps(state, indent=2, ensure_ascii=False),
        encoding="utf-8",
    )


def chunked(items: list[str], size: int) -> Iterable[list[str]]:
    """Divide una lista en lotes del tamaño indicado."""
    for index in range(0, len(items), size):
        yield items[index : index + size]


def fetch_incremental_page(*, since: str | None, cursor: str | None) -> dict:
    """Obtiene los documentos de forma incremental. Usada para leer los documentos mas nuevos encontrados por Zettana."""
    params = {"limit": PAGE_SIZE}

    if cursor:
        params["cursor"] = cursor
    else:
        params["since"] = since

    response = session.get(
        f"{BASE_URL}/documents/incremental",
        params=params,
        timeout=60,
    )
    response.raise_for_status()
    return response.json()


def fetch_xml_contents(document_ids: list[str]) -> dict[str, str]:
    """Recupera el contenido XML para un lote de documentos."""
    response = session.post(
        f"{BASE_URL}/documents/xml-contents",
        json={"documentIds": document_ids},
        timeout=120,
    )
    response.raise_for_status()

    payload = response.json()
    return {
        item["id"]: item["xmlContent"]
        for item in payload["documents"]
    }


def process_xml_document_in_your_system(
    document: dict,
    xml_content: str,
) -> None:
    """
    Punto de integración para tu sistema.

    Reemplaza esta función por la lógica real de tu producto. Por ejemplo:
    - parsear el XML
    - crear o actualizar un registro contable
    - asociarlo a un cliente, proveedor o transacción
    - etc.
    """
    print(
        f"[XML] Procesado {document['id']} "
        f"({document['documentType']} - {document['numeroDocumento']})"
    )

def process_pdf_document_in_your_system(document: dict) -> Path:
    """
    Punto de integración para PDFs.

    En este demo, la implementación solamente descarga el archivo localmente.
    En tu sistema, podrías cambiarlo por cualquier otra acción. Por ejemplo:
    - subirlo a Sharepoint
    - archivarlo en un file server
    - adjuntarlo a un expediente documental
    - enviarlo a otro servicio interno
    - enviarlo al departamente contable como archivo adjunto de un email
    - etc.
    """
    signed_link_response = session.get(
        f"{BASE_URL}/documents/{document['id']}/signedLink",
        timeout=30,
    )
    signed_link_response.raise_for_status()
    signed_link = signed_link_response.json()["signedLink"]

    # Esta estructura de carpetas es solo un ejemplo. Puedes cambiarla por
    # cualquier destino que tenga sentido para tu sistema.
    target_dir = (
        PDF_ROOT
        / document["rucId"]
        / str(document["year"])
        / f"{document['month']:02d}"
        / document["documentType"]
    )
    target_dir.mkdir(parents=True, exist_ok=True)

    file_name = f"{document['id']}.pdf"
    destination = target_dir / file_name

    with requests.get(signed_link, stream=True, timeout=120) as response:
        response.raise_for_status()
        with destination.open("wb") as output_file:
            for chunk in response.iter_content(chunk_size=8192):
                if chunk:
                    output_file.write(chunk)

    print(f"[PDF] Descargado {destination}")
    return destination


def process_xml_documents(xml_documents: list[dict]) -> None:
    """
    Agrupa IDs en lotes, recupera el contenido XML desde Zettana y delega
    el tratamiento de cada documento a la lógica específica de tu sistema.
    """
    if not xml_documents:
        return

    documents_by_id = {document["id"]: document for document in xml_documents}

    for batch in chunked(list(documents_by_id.keys()), XML_BATCH_SIZE):
        xml_contents = fetch_xml_contents(batch)

        for document_id in batch:
            xml_content = xml_contents[document_id]
            document = documents_by_id[document_id]
            process_xml_document_in_your_system(document, xml_content)


def process_pdf_documents(pdf_documents: list[dict]) -> None:
    """
    Recorre cada documento, obtiene un signed link temporal y delega la acción
    final a la lógica que tu sistema.
    """
    for document in pdf_documents:
        process_pdf_document_in_your_system(document)


def run_daily_sync() -> None:
    """
    Ejecuta un ciclo completo.

    Este flujo asume que ya existe una sincronización recurrente en Zettana
    manteniendo los datos al día. El script arranca desde `since`, procesa
    todas las páginas disponibles y luego persiste el nuevo punto de
    reanudación.
    """
    state = load_state()

    since = state["since"]
    processed_ids_at_latest_retrieved_at = set(
        state.get("processed_ids_at_latest_retrieved_at", [])
    )

    cursor = None
    latest_retrieved_at = since
    latest_timestamp_ids = set(processed_ids_at_latest_retrieved_at)

    while True:
        page = fetch_incremental_page(since=since, cursor=cursor)
        documents = page["documents"]

        fresh_documents = []
        for document in documents:
            if (
                document["retrievedAt"] == since
                and document["id"] in processed_ids_at_latest_retrieved_at
            ):
                continue

            fresh_documents.append(document)

        xml_documents = [
            document
            for document in fresh_documents
            if document["documentFormat"] == "XML"
        ]
        pdf_documents = [
            document
            for document in fresh_documents
            if document["documentFormat"] == "PDF"
        ]

        process_xml_documents(xml_documents)
        process_pdf_documents(pdf_documents)

        for document in fresh_documents:
            document_retrieved_at = document["retrievedAt"]

            # El endpoint incremental ya entrega los documentos ordenados por
            # retrievedAt e id. Por eso, cuando vemos un retrievedAt mayor,
            # sabemos que entramos en un nuevo borde temporal más reciente.
            if document_retrieved_at > latest_retrieved_at:
                latest_retrieved_at = document_retrieved_at
                latest_timestamp_ids = set()

            # Solo guardamos IDs del retrievedAt más reciente visto en esta ejecución.
            # Eso permite reanudar con el mismo `since` en la próxima
            # ejecución y saltarnos únicamente los documentos ya procesados en
            # ese borde temporal.
            if document_retrieved_at == latest_retrieved_at:
                latest_timestamp_ids.add(document["id"])

        if not page["hasMore"]:
            break

        cursor = page["nextCursor"]

    state["since"] = latest_retrieved_at
    state["processed_ids_at_latest_retrieved_at"] = sorted(latest_timestamp_ids)
    save_state(state)

    print(
        "Sync completado. "
        f"Nuevo since: {state['since']}. "
        "IDs procesados en el retrievedAt más reciente: "
        f"{len(state['processed_ids_at_latest_retrieved_at'])}"
    )


if __name__ == "__main__":
    run_daily_sync()