Webhooks

import hmac
import hashlib
import time
import json
from flask import Flask, request, abort
 
app = Flask(__name__)
WEBHOOK_SECRET = "your_webhook_secret_here"
 
def verify_signature(payload: bytes, signature: str, timestamp: str) -> bool:
    """Verify the webhook signature from BotEsq."""
    # Check timestamp to prevent replay attacks (5 minute window)
    webhook_time = int(timestamp)
    current_time = int(time.time())
    if abs(current_time - webhook_time) > 300:
        return False
 
    # Compute expected signature
    signed_payload = f"{timestamp}.{payload.decode('utf-8')}"
    expected_signature = hmac.new(
        WEBHOOK_SECRET.encode('utf-8'),
        signed_payload.encode('utf-8'),
        hashlib.sha256
    ).hexdigest()
 
    # Constant-time comparison
    return hmac.compare_digest(signature, f"sha256={expected_signature}")
 
@app.route('/webhooks/botesq', methods=['POST'])
def handle_webhook():
    # Get headers
    signature = request.headers.get('X-BotEsq-Signature', '')
    timestamp = request.headers.get('X-BotEsq-Timestamp', '')
    webhook_id = request.headers.get('X-BotEsq-Webhook-ID', '')
 
    # Verify signature
    if not verify_signature(request.data, signature, timestamp):
        abort(401, 'Invalid signature')
 
    # Parse event
    event = json.loads(request.data)
    event_type = event.get('event')
 
    # Handle different event types
    if event_type == 'consultation.completed':
        handle_consultation_completed(event['data'])
    elif event_type == 'document.analyzed':
        handle_document_analyzed(event['data'])
    elif event_type == 'matter.status_changed':
        handle_matter_status_changed(event['data'])
    elif event_type == 'credits.low':
        handle_credits_low(event['data'])
    elif event_type == 'retainer.expiring':
        handle_retainer_expiring(event['data'])
    else:
        print(f"Unhandled event type: {event_type}")
 
    return 'OK', 200
 
def handle_consultation_completed(data: dict):
    print(f"Consultation {data['consultation_id']} completed")
    # Add your business logic here
 
def handle_document_analyzed(data: dict):
    print(f"Document {data['document_id']} analyzed ({data['page_count']} pages)")
    # Add your business logic here
 
def handle_matter_status_changed(data: dict):
    print(f"Matter {data['matter_id']}: {data['previous_status']} -> {data['new_status']}")
    # Add your business logic here
 
def handle_credits_low(data: dict):
    print(f"Low credits warning: {data['credits_remaining']} remaining")
    # Add your business logic here
 
def handle_retainer_expiring(data: dict):
    print(f"Retainer {data['retainer_id']} expiring at {data['expires_at']}")
    # Add your business logic here
 
if __name__ == '__main__':
    app.run(port=3000)