Webhooks

from pydantic import BaseModel, Field
from typing import Any
import base64

class MIMEData(BaseModel):
    """Document data in base64 data URL format."""
    filename: str = Field(description="Original filename of the document")
    url: str = Field(description="Base64 data URL of the file content")
    
    @property
    def mime_type(self) -> str:
        """Extract MIME type from the data URL."""
        if self.url.startswith("data:"):
            return self.url.split(";")[0].split(":")[1]
        return "application/octet-stream"
    
    @property
    def content(self) -> bytes:
        """Decode the base64 content."""
        if self.url.startswith("data:"):
            base64_data = self.url.split(",")[1]
            return base64.b64decode(base64_data)
        raise ValueError("URL is not a data URL")

class WebhookPayload(BaseModel):
    document: MIMEData | None = None
    json: dict[str, Any] | None = Field(default=None)

import base64
from fastapi import FastAPI, Request
from pydantic import BaseModel
from typing import Any

app = FastAPI()

class Invoice(BaseModel):
    invoice_number: str
    vendor_name: str
    total_amount: float

@app.post("/webhook")
async def webhook_handler(request: Request):
    payload = await request.json()
    
    # Extract the JSON data
    json_data = payload.get("json")
    document = payload.get("document")
    
    if json_data:
        invoice = Invoice.model_validate(json_data)
        print(f"📬 Received invoice: {invoice.invoice_number} - ${invoice.total_amount}")
    
    if document:
        filename = document.get("filename")
        url = document.get("url", "")
        print(f"📎 Document: {filename}")
        
        # Optionally save the file
        if url.startswith("data:"):
            base64_content = url.split(",")[1]
            file_bytes = base64.b64decode(base64_content)
            with open(f"/tmp/{filename}", "wb") as f:
                f.write(file_bytes)
            print(f"   Saved to /tmp/{filename} ({len(file_bytes)} bytes)")
    
    return {"status": "success"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

from retab import Retab

client = Retab()

# Create a test processor and automation
processor = client.processors.create(
    name="Test Processor",
    model="gpt-4.1",
    json_schema=Invoice.model_json_schema(),
)

# Create an automation with your webhook
automation = client.processors.automations.links.create(
    name="Test Automation",
    processor_id=processor.id,
    webhook_url="https://your-server.com/webhook",
)

# Test with a document
result = client.processors.automations.links.test_document_upload(
    link_id=automation.id,
    document="test_invoice.pdf"
)

print(f"Test result: {result}")

import os
from fastapi import FastAPI, Request, Response, HTTPException
from retab import Retab

client = Retab()
app = FastAPI()

@app.post("/webhook")
async def webhook_handler(request: Request):
    payload = await request.body()
    signature_header = request.headers.get("Retab-Signature")

    if not signature_header:
        raise HTTPException(status_code=400, detail="Missing Retab-Signature header")

    try:
        # Verify the signature
        client.verify_event(
            event_body=payload,
            event_signature=signature_header,
            secret=os.getenv("WEBHOOKS_SECRET"),
        )
    except Exception as e:
        return Response(
            status_code=400, 
            content=f"Signature verification failed: {str(e)}"
        )

    # Signature verified - process the webhook
    import json
    data = json.loads(payload.decode('utf-8'))
    json_data = data.get("json")
    document = data.get("document")
    
    # Process your data...
    
    return Response(status_code=200)