webhook_integration.py

# Generated by Copilot
"""
Webhook Integration Module for Text2Speech App
Adds webhook capabilities to the existing TTS Gradio app
"""

import json
import os
from datetime import datetime
from typing import Dict, List, Optional
from dataclasses import dataclass, asdict
import gradio as gr

@dataclass
class WebhookConfig:
    """Webhook configuration settings"""
    webhook_secret: str = "tts_webhook_secret_2024"
    auto_redeploy: bool = True
    model_sync: bool = True
    usage_tracking: bool = True
    error_monitoring: bool = True
    notification_email: Optional[str] = None
    slack_webhook_url: Optional[str] = None

@dataclass
class WebhookEvent:
    """Webhook event data structure"""
    timestamp: str
    event_type: str
    repo_name: str
    action: str
    details: Dict
    processed: bool = False

class WebhookIntegration:
    """Webhook integration manager for TTS app"""
    
    def __init__(self, config_file: str = "webhook_config.json"):
        self.config_file = config_file
        self.config = self.load_config()
        self.events: List[WebhookEvent] = []
        self.webhooks_enabled = True
    
    def load_config(self) -> WebhookConfig:
        """Load webhook configuration from file"""
        try:
            if os.path.exists(self.config_file):
                with open(self.config_file, 'r') as f:
                    data = json.load(f)
                return WebhookConfig(**data)
        except Exception as e:
            print(f"Warning: Could not load webhook config: {e}")
        
        return WebhookConfig()
    
    def save_config(self):
        """Save webhook configuration to file"""
        try:
            with open(self.config_file, 'w') as f:
                json.dump(asdict(self.config), f, indent=2)
        except Exception as e:
            print(f"Error saving webhook config: {e}")
    
    def add_event(self, event: WebhookEvent):
        """Add a webhook event to the log"""
        self.events.append(event)
        # Keep only last 50 events
        if len(self.events) > 50:
            self.events = self.events[-50:]
    
    def get_recent_events(self, limit: int = 10) -> List[WebhookEvent]:
        """Get recent webhook events"""
        return self.events[-limit:] if self.events else []
    
    def get_events_summary(self) -> str:
        """Get formatted summary of recent events"""
        if not self.events:
            return "No webhook events recorded yet."
        
        lines = ["📊 Recent Webhook Activity:\n"]
        for event in self.get_recent_events():
            status = "✅" if event.processed else "⏳"
            lines.append(f"{status} {event.timestamp[:19]} | {event.event_type} | {event.repo_name}")
        
        return "\n".join(lines)
    
    def create_webhook_tab(self) -> gr.Tab:
        """Create webhook management tab for the TTS app"""
        
        with gr.Tab("🔗 Webhooks") as webhook_tab:
            gr.Markdown("""
            ## 🔗 Webhook Integration
            
            Automate your TTS workflow with Hugging Face webhooks!
            
            ### 🎯 Available Automations:
            - **Auto-redeploy** when you push code changes
            - **Model sync** when new TTS models are released
            - **Usage tracking** for analytics and optimization
            - **Error monitoring** with instant notifications
            """)
            
            with gr.Row():
                with gr.Column(scale=2):
                    # Configuration Section
                    gr.Markdown("### ⚙️ Configuration")
                    
                    with gr.Row():
                        auto_redeploy_cb = gr.Checkbox(
                            label="🚀 Auto-redeploy on code changes",
                            value=self.config.auto_redeploy
                        )
                        model_sync_cb = gr.Checkbox(
                            label="🔄 Auto-sync new TTS models",
                            value=self.config.model_sync
                        )
                    
                    with gr.Row():
                        usage_tracking_cb = gr.Checkbox(
                            label="📊 Usage tracking",
                            value=self.config.usage_tracking
                        )
                        error_monitoring_cb = gr.Checkbox(
                            label="🚨 Error monitoring",
                            value=self.config.error_monitoring
                        )
                    
                    notification_email = gr.Textbox(
                        label="📧 Notification Email (optional)",
                        placeholder="your@email.com",
                        value=self.config.notification_email or ""
                    )
                    
                    save_config_btn = gr.Button("💾 Save Configuration", variant="primary")
                    config_status = gr.Textbox(label="Status", interactive=False)
                
                with gr.Column(scale=1):
                    # Status Section
                    gr.Markdown("### 📡 Webhook Status")
                    
                    webhook_status = gr.Textbox(
                        label="Connection Status",
                        value="🟢 Ready to receive webhooks" if self.webhooks_enabled else "🔴 Webhooks disabled",
                        interactive=False
                    )
                    
                    webhook_secret = gr.Textbox(
                        label="Webhook Secret",
                        value=self.config.webhook_secret,
                        interactive=False,
                        type="password"
                    )
                    
                    test_webhook_btn = gr.Button("🧪 Test Webhook", variant="secondary")
                    test_result = gr.Textbox(label="Test Result", interactive=False)
            
            # Events Section
            gr.Markdown("### 📋 Recent Events")
            events_display = gr.Textbox(
                label="Event Log",
                value=self.get_events_summary(),
                lines=8,
                interactive=False
            )
            
            refresh_events_btn = gr.Button("🔄 Refresh Events")
            
            # Setup Instructions
            with gr.Accordion("🛠️ Setup Instructions", open=False):
                gr.Markdown(f"""
                ### How to Set Up Webhooks:
                
                1. **Go to [Hugging Face Webhooks Settings](https://huggingface.co/settings/webhooks)**
                
                2. **Click "New webhook"**
                
                3. **Configure the webhook:**
                   - **URL**: `https://toowired-text2speech-gradio-app.hf.space/webhooks/tts_automation`
                   - **Secret**: `{self.config.webhook_secret}`
                   - **Select events**: Repository updates, model uploads, etc.
                
                4. **Target repositories:**
                   - Your TTS Space: `Toowired/text2speech-gradio-app`
                   - Any model repos you want to monitor
                
                5. **Save and test** the webhook connection
                
                ### 🔧 Available Endpoints:
                - `/webhooks/tts_automation` - Main automation
                - `/webhooks/model_sync` - Model synchronization  
                - `/webhooks/usage_tracker` - Usage analytics
                - `/webhooks/error_monitor` - Error notifications
                """)
            
            # Event handlers
            def save_webhook_config(auto_redeploy, model_sync, usage_tracking, error_monitoring, email):
                try:
                    self.config.auto_redeploy = auto_redeploy
                    self.config.model_sync = model_sync
                    self.config.usage_tracking = usage_tracking
                    self.config.error_monitoring = error_monitoring
                    self.config.notification_email = email if email.strip() else None
                    
                    self.save_config()
                    return "✅ Configuration saved successfully!"
                except Exception as e:
                    return f"❌ Error saving configuration: {str(e)}"
            
            def test_webhook():
                try:
                    # Simulate a test event
                    test_event = WebhookEvent(
                        timestamp=datetime.now().isoformat(),
                        event_type="test",
                        repo_name="test-webhook",
                        action="connection_test",
                        details={"test": True},
                        processed=True
                    )
                    self.add_event(test_event)
                    return "✅ Test webhook event logged successfully!"
                except Exception as e:
                    return f"❌ Test failed: {str(e)}"
            
            def refresh_events():
                return self.get_events_summary()
            
            # Connect event handlers
            save_config_btn.click(
                save_webhook_config,
                inputs=[auto_redeploy_cb, model_sync_cb, usage_tracking_cb, error_monitoring_cb, notification_email],
                outputs=[config_status]
            )
            
            test_webhook_btn.click(test_webhook, outputs=[test_result])
            refresh_events_btn.click(refresh_events, outputs=[events_display])
        
        return webhook_tab
    
    def setup_webhook_endpoints(self, app):
        """Set up webhook endpoints on the Gradio app"""
        
        # This would integrate with the FastAPI backend if available
        # For now, we'll document the endpoints that should be created
        webhook_endpoints = {
            "/webhooks/tts_automation": self.handle_automation_webhook,
            "/webhooks/model_sync": self.handle_model_sync_webhook,
            "/webhooks/usage_tracker": self.handle_usage_webhook,
            "/webhooks/error_monitor": self.handle_error_webhook
        }
        
        return webhook_endpoints
    
    def handle_automation_webhook(self, payload):
        """Handle main automation webhook"""
        event = WebhookEvent(
            timestamp=datetime.now().isoformat(),
            event_type="automation",
            repo_name=payload.get("repo", {}).get("name", "unknown"),
            action=payload.get("event", {}).get("action", "unknown"),
            details=payload,
            processed=True
        )
        self.add_event(event)
        
        return {"status": "processed", "event_id": len(self.events)}
    
    def handle_model_sync_webhook(self, payload):
        """Handle model sync webhook"""
        event = WebhookEvent(
            timestamp=datetime.now().isoformat(),
            event_type="model_sync",
            repo_name=payload.get("repo", {}).get("name", "unknown"),
            action=payload.get("event", {}).get("action", "unknown"),
            details=payload,
            processed=True
        )
        self.add_event(event)
        
        return {"status": "processed", "action": "model_sync"}
    
    def handle_usage_webhook(self, payload):
        """Handle usage tracking webhook"""
        event = WebhookEvent(
            timestamp=datetime.now().isoformat(),
            event_type="usage_tracking",
            repo_name=payload.get("repo", {}).get("name", "unknown"),
            action=payload.get("event", {}).get("action", "unknown"),
            details=payload,
            processed=True
        )
        self.add_event(event)
        
        return {"status": "processed", "action": "usage_logged"}
    
    def handle_error_webhook(self, payload):
        """Handle error monitoring webhook"""
        event = WebhookEvent(
            timestamp=datetime.now().isoformat(),
            event_type="error_monitoring",
            repo_name=payload.get("repo", {}).get("name", "unknown"),
            action=payload.get("event", {}).get("action", "unknown"),
            details=payload,
            processed=True
        )
        self.add_event(event)
        
        return {"status": "processed", "action": "error_monitored"}

# Global webhook integration instance
webhook_integration = WebhookIntegration()