Spaces:
Runtime error
Runtime error
| # 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) | |
| 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 | |
| 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"} | |
| 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} | |
| 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 | |
| ) |