Configure Webhooks

from pulse import Pulse

client = Pulse(api_key="YOUR_API_KEY")

# Get webhook portal link
response = client.webhooks.create_webhook_link()
print(f"Portal URL: {response.link}")

# Open this URL in your browser to configure webhooks

import hmac
import hashlib
import time
import base64

def verify_webhook(payload: str, headers: dict, webhook_secret: str) -> bool:
    """
    Verify webhook authenticity using HMAC signature.
    
    Args:
        payload: Raw request body as string
        headers: Request headers dict
        webhook_secret: Your webhook signing secret from the portal
        
    Returns:
        True if signature is valid, False otherwise
    """
    webhook_id = headers.get('webhook-id')
    webhook_timestamp = headers.get('webhook-timestamp')
    webhook_signature = headers.get('webhook-signature')
    
    if not all([webhook_id, webhook_timestamp, webhook_signature]):
        return False
    
    # Check timestamp to prevent replay attacks (5 minute window)
    current_time = int(time.time())
    if abs(current_time - int(webhook_timestamp)) > 300:
        return False
    
    # Construct signed content
    signed_content = f"{webhook_id}.{webhook_timestamp}.{payload}"
    
    # Extract signature from header (format: v1,signature)
    signature = webhook_signature.split(',')[1] if ',' in webhook_signature else webhook_signature
    
    # Compute expected signature (base64-encoded HMAC-SHA256)
    expected = base64.b64encode(
        hmac.new(
            base64.b64decode(webhook_secret),
            signed_content.encode(),
            hashlib.sha256
        ).digest()
    ).decode()
    
    # Constant-time comparison
    return hmac.compare_digest(signature, expected)

# Example usage with Flask
from flask import Flask, request, abort

app = Flask(__name__)
WEBHOOK_SECRET = "whsec_your_secret_here"

@app.route('/webhook', methods=['POST'])
def handle_webhook():
    payload = request.get_data(as_text=True)
    
    if not verify_webhook(payload, request.headers, WEBHOOK_SECRET):
        abort(401)
    
    # Process the event
    event = request.json
    print(f"Received event: {event['type']}")
    
    return '', 200

from flask import Flask, request, abort
import json

app = Flask(__name__)
WEBHOOK_SECRET = "whsec_your_secret_here"

@app.route('/webhook', methods=['POST'])
def handle_webhook():
    payload = request.get_data(as_text=True)
    
    # Verify webhook signature
    if not verify_webhook(payload, request.headers, WEBHOOK_SECRET):
        abort(401)
    
    event = json.loads(payload)
    
    # Handle different event types
    if event['type'] == 'job.completed':
        job_id = event['data']['job_id']
        print(f"✓ Job {job_id} completed!")
        
        # Fetch the results
        # result = fetch_job_result(job_id)
        # process_extraction(result)
        
    elif event['type'] == 'job.failed':
        job_id = event['data']['job_id']
        error = event['data'].get('error', 'Unknown error')
        print(f"✗ Job {job_id} failed: {error}")
        
        # Handle failure (retry, notify, etc.)
        # handle_job_failure(job_id, error)
        
    elif event['type'] == 'job.cancelled':
        job_id = event['data']['job_id']
        print(f"⊘ Job {job_id} was cancelled")
    
    return '', 200

if __name__ == '__main__':
    app.run(port=3000)

from pulse import Pulse
from flask import Flask, request, abort
import json

# Initialize Pulse client
client = Pulse(api_key="YOUR_API_KEY")

# Flask app for webhook handler
app = Flask(__name__)
WEBHOOK_SECRET = "whsec_your_secret_here"

@app.route('/webhook', methods=['POST'])
def handle_webhook():
    payload = request.get_data(as_text=True)
    
    if not verify_webhook(payload, request.headers, WEBHOOK_SECRET):
        abort(401)
    
    event = json.loads(payload)
    
    if event['type'] == 'job.completed':
        job_id = event['data']['job_id']
        
        # Fetch full results using SDK
        job_result = client.jobs.get_job(job_id=job_id)
        
        if job_result.result:
            print(f"Markdown: {job_result.result.markdown[:100]}...")
            
            # Apply schema post-extraction via /schema endpoint
            schema_result = client.schema(
                extraction_id=job_result.result.extraction_id,
                schema_config={
                    "input_schema": {
                        "type": "object",
                        "properties": {
                            "account_holder": {"type": "string"},
                            "balance": {"type": "number"}
                        }
                    }
                }
            )
            if schema_result.schema_output:
                print(f"Structured data: {schema_result.schema_output}")
    
    return '', 200

# Submit an async job
def submit_job():
    response = client.extract(
        file_url="https://www.impact-bank.com/user/file/dummy_statement.pdf",
        async_=True
    )
    print(f"Job submitted: {response.job_id}")
    print("Waiting for webhook notification...")
    return response.job_id

if __name__ == '__main__':
    # Submit a job, then start webhook handler
    submit_job()
    app.run(port=3000)