File size: 3,752 Bytes
53ef5b1 3cbf9f8 53ef5b1 3cbf9f8 53ef5b1 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 |
import json
import hmac
import hashlib
from http.server import BaseHTTPRequestHandler, HTTPServer
import logging
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# Webhook secret for verification (set this when creating the webhook)
WEBHOOK_SECRET = "your-base64-encoded-secret" # Replace with your secret
class WebhookHandler(BaseHTTPRequestHandler):
def do_POST(self):
try:
# Read the request body
content_length = int(self.headers['Content-Length'])
post_data = self.rfile.read(content_length).decode('utf-8')
payload = json.loads(post_data)
# Verify webhook signature
signature = self.headers.get('X-Daily-Signature')
if not self.verify_signature(post_data, signature):
logger.warning("Invalid webhook signature")
self.send_response(401)
self.end_headers()
return
# Handle Daily.co dial-in events
event_type = payload.get('event')
if event_type in ['dialin.connected', 'dialin.stopped']:
logger.info(f"Received {event_type} event: {json.dumps(payload, indent=2)}")
# Example: Process dial-in event
if event_type == 'dialin.connected':
call_id = payload.get('callId')
caller = payload.get('From')
logger.info(f"Dial-in connected: Call ID {call_id} from {caller}")
# Trigger bot startup (e.g., call /start endpoint)
self.handle_dialin_connected(payload)
elif event_type == 'dialin.stopped':
call_id = payload.get('callId')
logger.info(f"Dial-in stopped: Call ID {call_id}")
# Send 200 OK response
self.send_response(200)
self.send_header('Content-type', 'application/json')
self.end_headers()
self.wfile.write(b'{"status": "Event received"}')
else:
logger.info(f"Ignored event type: {event_type}")
self.send_response(200)
self.end_headers()
except Exception as e:
logger.error(f"Error processing webhook: {str(e)}")
self.send_response(500)
self.end_headers()
def verify_signature(self, payload, signature):
"""Verify the webhook signature using HMAC-SHA256."""
if not signature or not WEBHOOK_SECRET:
return False
computed = hmac.new(
WEBHOOK_SECRET.encode('utf-8'),
payload.encode('utf-8'),
hashlib.sha256
).hexdigest()
return hmac.compare_digest(computed, signature)
def handle_dialin_connected(self, payload):
"""Handle dialin.connected event (e.g., trigger bot startup)."""
# Example: Send payload to /start endpoint (modify URL as needed)
import requests
start_url = "https://<your-space>.hf.space/start" # Replace with your endpoint
try:
response = requests.post(start_url, json=payload, timeout=5)
logger.info(f"Sent to /start: {response.status_code} - {response.text}")
except requests.RequestException as e:
logger.error(f"Failed to send to /start: {str(e)}")
def run(server_class=HTTPServer, handler_class=WebhookHandler, port=8000):
server_address = ('', port)
httpd = server_class(server_address, handler_class)
logger.info(f"Starting webhook server on port {port}...")
httpd.serve_forever()
if __name__ == '__main__':
run() |