standalone_webhook_server.py

# 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
    )