Webhooks

from flask import Flask, request, abort
import hmac
import hashlib
import time
import json

app = Flask(__name__)

# Your webhook secret from the portal
WEBHOOK_SECRET = "your_webhook_secret_here"

def verify_webhook(payload, headers, webhook_secret):
    """Verify webhook authenticity using HMAC signature."""
    
    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
    expected = hmac.new(
        webhook_secret.encode(),
        signed_content.encode(),
        hashlib.sha256
    ).hexdigest()
    
    # Constant-time comparison
    return hmac.compare_digest(signature, expected)

@app.route('/webhook', methods=['POST'])
def handle_webhook():
    payload = request.get_data(as_text=True)
    
    # Verify webhook
    if not verify_webhook(payload, request.headers, WEBHOOK_SECRET):
        abort(401)
    
    event = json.loads(payload)
    
    # Handle different event types
    event_type = event.get('type')
    event_data = event.get('data')
    
    if event_type == 'job.completed':
        handle_job_completed(event_data)
    elif event_type == 'job.failed':
        handle_job_failed(event_data)
    elif event_type == 'job.cancelled':
        handle_job_cancelled(event_data)
    
    return '', 200

def handle_job_completed(data):
    job_id = data['job_id']
    pages_processed = data.get('pages_processed', 0)
    processing_time = data.get('processing_time', 0)
    
    print(f"Job {job_id} completed successfully")
    print(f"Processed {pages_processed} pages in {processing_time} seconds")
    
    # Fetch the full results using the job ID
    # Process results, update database, notify users, etc.

def handle_job_failed(data):
    job_id = data['job_id']
    error = data.get('error', 'Unknown error')
    
    print(f"Job {job_id} failed: {error}")
    # Handle failure - retry, notify user, etc.

def handle_job_cancelled(data):
    job_id = data['job_id']
    print(f"Job {job_id} was cancelled")
    # Clean up any resources

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

import requests
import json
from typing import Optional, Dict, Any
from dataclasses import dataclass
from datetime import datetime

@dataclass
class PulseJob:
    job_id: str
    file_path: str
    schema: Optional[Dict]
    created_at: datetime
    status: str = "pending"

class PulseWebhookClient:
    """Complete Pulse API client with webhook support."""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://dev.api.runpulse.com"
        self.headers = {"x-api-key": api_key}
        self.jobs = {}  # In production, use a proper database
    
    def setup_webhooks(self) -> str:
        """Get webhook portal link for configuration."""
        response = requests.post(
            f"{self.base_url}/webhook",
            headers=self.headers
        )
        if response.status_code == 200:
            return response.json()['link']
        raise Exception(f"Failed to get webhook portal: {response.text}")
    
    def extract_async_with_webhook(self, 
                                  file_path: str,
                                  schema: Optional[Dict] = None,
                                  pages: Optional[str] = None) -> str:
        """
        Start async extraction that will notify via webhook when complete.
        
        Returns:
            job_id for tracking
        """
        url = f"{self.base_url}/extract_async"
        
        with open(file_path, 'rb') as f:
            files = {'file': f}
            data = {}
            
            if schema:
                data['schema'] = json.dumps(schema)
            if pages:
                data['pages'] = pages
            
            response = requests.post(
                url, 
                files=files, 
                data=data, 
                headers=self.headers
            )
        
        if response.status_code == 200:
            job_info = response.json()
            job_id = job_info['job_id']
            
            # Store job info for webhook correlation
            job = PulseJob(
                job_id=job_id,
                file_path=file_path,
                schema=schema,
                created_at=datetime.now()
            )
            self.jobs[job_id] = job
            
            return job_id
        
        raise Exception(f"Failed to start extraction: {response.text}")
    
    def get_job_result(self, job_id: str) -> Dict[str, Any]:
        """
        Fetch job results after webhook notification.
        For large documents, this handles the S3 URL response.
        """
        url = f"{self.base_url}/job/{job_id}"
        response = requests.get(url, headers=self.headers)
        
        if response.status_code == 200:
            job_data = response.json()
            
            if job_data['status'] == 'completed':
                result = job_data['result']
                
                # Handle large document URL response
                if result.get('is_url'):
                    content_response = requests.get(result['url'])
                    return content_response.json()
                
                return result
            
            raise Exception(f"Job not completed: {job_data['status']}")
        
        raise Exception(f"Failed to get job result: {response.text}")

# Usage example
if __name__ == "__main__":
    client = PulseWebhookClient(api_key="YOUR_API_KEY")
    
    # One-time setup: Configure webhooks
    portal_link = client.setup_webhooks()
    print(f"Configure webhooks at: {portal_link}")
    print("Add your webhook endpoint URL in the portal")
    
    # Extract document - webhook will notify when complete
    job_id = client.extract_async_with_webhook(
        "large_document.pdf",
        schema={
            "title": "string",
            "sections": [{
                "heading": "string",
                "content": "string"
            }]
        },
        pages="1-100"
    )
    
    print(f"Extraction started: {job_id}")
    print("You'll receive a webhook notification when complete!")