# Generated by Copilot """ Standalone Webhook Server for TTS Project Deploy this as a separate Space for dedicated webhook handling """ from huggingface_hub import WebhooksServer, WebhookPayload import gradio as gr import json import os from datetime import datetime from typing import Dict, List import requests # Configuration WEBHOOK_SECRET = os.getenv("WEBHOOK_SECRET", "tts_webhook_secret_2024") TTS_SPACE_URL = "https://toowired-text2speech-gradio-app.hf.space" MAIN_REPO = "Toowired/text2speech-gradio-app" # Event storage webhook_events = [] def log_event(event_type: str, repo_name: str, action: str, details: Dict): """Log webhook event""" event = { "timestamp": datetime.now().isoformat(), "type": event_type, "repo": repo_name, "action": action, "details": details } webhook_events.append(event) # Keep last 100 events if len(webhook_events) > 100: webhook_events.pop(0) return event def trigger_space_action(action: str, payload: Dict) -> bool: """Trigger an action on the main TTS Space""" try: # This would communicate with your main TTS Space # For now, we'll log the action print(f"🚀 Triggered action: {action} with payload: {json.dumps(payload, indent=2)}") return True except Exception as e: print(f"❌ Failed to trigger action {action}: {e}") return False # Create webhook server UI def create_webhook_dashboard(): with gr.Blocks(title="🔗 TTS Webhook Server", theme=gr.themes.Soft()) as ui: gr.Markdown(""" # 🔗 TTS Project Webhook Server **Dedicated webhook handler for Text2Speech automation** This server receives and processes webhooks from Hugging Face, triggering automated workflows for your TTS project. ## 🎯 Active Automations: - Auto-redeploy on code changes - Model synchronization - Usage analytics - Error monitoring """) with gr.Tabs(): with gr.Tab("📊 Dashboard"): with gr.Row(): total_events = gr.Number(label="Total Events", value=0, interactive=False) active_automations = gr.Number(label="Active Automations", value=4, interactive=False) last_event = gr.Textbox(label="Last Event", value="No events yet", interactive=False) events_log = gr.JSON(label="Recent Events", value=[]) refresh_btn = gr.Button("🔄 Refresh Dashboard") def refresh_dashboard(): recent_events = webhook_events[-10:] if webhook_events else [] last = webhook_events[-1]["timestamp"] if webhook_events else "No events yet" return len(webhook_events), 4, last, recent_events refresh_btn.click(refresh_dashboard, outputs=[total_events, active_automations, last_event, events_log]) with gr.Tab("⚙️ Configuration"): gr.Markdown(f""" ### Server Configuration **Webhook Secret**: `{WEBHOOK_SECRET}` **Target TTS Space**: `{TTS_SPACE_URL}` **Main Repository**: `{MAIN_REPO}` ### Webhook Endpoints: - `/webhooks/main` - Main automation handler - `/webhooks/models` - Model synchronization - `/webhooks/analytics` - Usage tracking - `/webhooks/monitoring` - Error monitoring """) with gr.Tab("📚 Setup Guide"): gr.Markdown(""" ### How to Use This Webhook Server 1. **Deploy this Space** to get a dedicated webhook URL 2. **Configure Hugging Face webhooks** to point to this server 3. **Set the webhook secret** in your Space environment variables 4. **Monitor events** through the dashboard ### Webhook Configuration: - **URL**: `{your-webhook-space-url}/webhooks/main` - **Secret**: Set as `WEBHOOK_SECRET` environment variable - **Content-Type**: `application/json` ### Events to Monitor: - Repository updates (push, pull request) - Model uploads and updates - Dataset modifications - Space deployments """) return ui # Create webhook server ui = create_webhook_dashboard() server = WebhooksServer(ui=ui, webhook_secret=WEBHOOK_SECRET) @server.add_webhook("/main") async def main_webhook(payload: WebhookPayload): """Main webhook handler for TTS automation""" event = log_event( event_type="main_automation", repo_name=payload.repo.name, action=payload.event.action, details={ "scope": payload.event.scope, "repo_type": payload.repo.type, "private": payload.repo.private } ) response = {"processed": True, "event_id": len(webhook_events), "actions": []} # Handle different event types if payload.event.action == "update" and payload.event.scope.startswith("repo.content"): if payload.repo.name == MAIN_REPO: # Trigger redeploy for main TTS Space success = trigger_space_action("redeploy", {"repo": payload.repo.name}) response["actions"].append({"type": "redeploy", "success": success}) elif payload.repo.type == "model" and payload.event.action in ["create", "update"]: # Handle new/updated models success = trigger_space_action("sync_model", {"model": payload.repo.name}) response["actions"].append({"type": "model_sync", "success": success}) return response @server.add_webhook("/models") async def models_webhook(payload: WebhookPayload): """Dedicated model synchronization webhook""" if payload.repo.type != "model": return {"processed": False, "reason": "not_a_model"} event = log_event( event_type="model_sync", repo_name=payload.repo.name, action=payload.event.action, details={"model_name": payload.repo.name} ) # Evaluate if this model is relevant for TTS is_tts_model = any(tag in payload.repo.name.lower() for tag in ["tts", "speech", "voice", "audio"]) if is_tts_model: success = trigger_space_action("evaluate_model", {"model": payload.repo.name}) return {"processed": True, "model_evaluated": success} return {"processed": True, "skipped": "not_tts_model"} @server.add_webhook("/analytics") async def analytics_webhook(payload: WebhookPayload): """Usage analytics webhook""" event = log_event( event_type="analytics", repo_name=payload.repo.name, action=payload.event.action, details={"tracking": True} ) # Track usage patterns success = trigger_space_action("log_usage", { "repo": payload.repo.name, "action": payload.event.action, "timestamp": event["timestamp"] }) return {"processed": True, "analytics_logged": success} @server.add_webhook("/monitoring") async def monitoring_webhook(payload: WebhookPayload): """Error monitoring webhook""" event = log_event( event_type="monitoring", repo_name=payload.repo.name, action=payload.event.action, details={"monitoring": True} ) # Check for error conditions if payload.event.action == "failed" or "error" in payload.event.scope.lower(): success = trigger_space_action("handle_error", { "repo": payload.repo.name, "error_type": payload.event.scope, "timestamp": event["timestamp"] }) return {"processed": True, "error_handled": success} return {"processed": True, "no_errors": True} if __name__ == "__main__": print("🔗 Starting TTS Webhook Server...") print(f"📡 Webhook Secret: {WEBHOOK_SECRET}") print(f"🎯 Target TTS Space: {TTS_SPACE_URL}") print(f"📦 Main Repository: {MAIN_REPO}") print("🚀 Server launching...") server.launch( server_name="0.0.0.0", server_port=7860, share=False )