Spaces:
Sleeping
Sleeping
File size: 4,531 Bytes
fb1970b | 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 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 | import os
import uvicorn
import requests
import hashlib
import hmac
import time
import logging
from fastapi import FastAPI, Request, HTTPException
# from dotenv import load_dotenv # DEV
# load_dotenv() # DEV
log = logging.getLogger('uvicorn.error')
app = FastAPI()
@app.get("/")
def home():
return {"app": "ok"}
@app.post("/synthesia_webhook")
async def synthesia_webhook(request: Request):
"""
Webhook to handle Synthesia events. Verifies signature, processes payload,
and forwards data to app API while logging issues for failures.
"""
try:
# Extract and validate headers
request_timestamp = request.headers.get("Synthesia-Timestamp")
request_signature = request.headers.get("Synthesia-Signature")
if not request_timestamp or not request_signature:
raise HTTPException(status_code=401, detail="Unauthorized: Missing required headers")
# Validate the signature
request_body = (await request.body()).decode("utf-8")
message = f"{request_timestamp}.{request_body}"
if not validate_signature(message, request_signature):
raise HTTPException(status_code=401, detail="Unauthorized: Invalid signature")
# Parse and validate payload
request_data = await request.json()
data = request_data.get('data', {})
title = data.get('title', "Unknown")
video_status = request_data.get('type', 'video.failed')
if not data or video_status not in ['video.failed', 'video.completed']:
log.error(f"Unexpected payload received: {request_data}")
raise HTTPException(status_code=400, detail="Bad request: Invalid payload structure")
# Forward data to app's API
payload = {'data': data, 'video_status': video_status}
status = await forward_video_data(payload)
if status >= 400:
log.warning(f"Failed to forward data for video '{title}' with status {status}")
except HTTPException as http_exc:
log.error(f"HTTP exception occurred: {http_exc.detail}")
raise http_exc
except Exception as e:
log.error(f"Unexpected error: {e}")
raise HTTPException(status_code=500, detail="Internal server error")
return {"message": "ok"}
def get_signature(message):
key = os.getenv('SYNTHESIA_WEBHOOK_SECRET', '')
signature = None
try:
signature = hmac.new(
key.encode("utf-8"),
message.encode("utf-8"),
hashlib.sha256,
).hexdigest()
except Exception as e:
log.error(f"Error getting signature: {e}")
finally:
return signature
def validate_signature(message, request_signature):
valid = False
try:
system_signature = get_signature(message)
valid = system_signature == request_signature
if not valid:
log.warning("Invalid Synthesia signature!")
except Exception as e:
log.error(f"Error validating Synthesia signature: {e}")
finally:
return valid
async def forward_video_data(payload:dict) -> int:
""" Sends POST requests to create video from template in Synthesia.
:param dict payload: Request payload (title, templateId, templateData)
:returns int status
"""
status = 500
app_key = os.getenv('APP_KEY', None)
url = os.getenv('APP_URL', None)
headers = {
'Authorization': f"Bearer {app_key}",
'Content-Type': 'application/json'
}
title = payload.get('title', '')
try:
response = requests.post(url, json=payload, headers=headers)
status = response.status_code
if response.ok:
data = response.json()
video_id = data.get('id', None)
else:
log.error(f"Failed to create {title}. Status: {status}, Reason: {response.reason}")
except requests.exceptions.ConnectionError as conn_err:
log.error(f"Error creating {title}. Connection error occurred: {conn_err}")
except requests.exceptions.Timeout as timeout_err:
log.error(f"Error creating {title}. Timeout error occurred: {timeout_err}")
except ValueError as json_err:
log.error(f"Error creating {title}. JSON decoding error: {json_err}")
except Exception as e:
log.error(f"Error creating {title}. Error: {e}")
finally:
return status
|