webhook_manager.py

# Generated by Copilot
"""
Programmatic Webhook Management for TTS Project
Creates and manages Hugging Face webhooks automatically
"""

import os
import json
import asyncio
from typing import List, Dict, Optional
from huggingface_hub import HfApi, HfFolder
import requests

class WebhookManager:
    """Manages Hugging Face webhooks programmatically"""
    
    def __init__(self, token: Optional[str] = None):
        self.api = HfApi(token=token)
        self.token = token or HfFolder.get_token()
        self.base_url = "https://huggingface.co/api/webhooks"
        self.space_url = "https://toowired-text2speech-gradio-app.hf.space"
        
    def get_headers(self) -> Dict[str, str]:
        """Get API headers with authentication"""
        return {
            "Authorization": f"Bearer {self.token}",
            "Content-Type": "application/json"
        }
    
    def create_webhook(self, 
                      endpoint: str,
                      name: str,
                      secret: str,
                      events: List[str],
                      target_repos: List[str],
                      description: str = "") -> Dict:
        """Create a new webhook programmatically"""
        
        webhook_data = {
            "url": f"{self.space_url}/webhooks/{endpoint}",
            "name": name,
            "secret": secret,
            "events": events,
            "repos": target_repos,
            "description": description,
            "active": True
        }
        
        try:
            response = requests.post(
                self.base_url,
                headers=self.get_headers(),
                json=webhook_data
            )
            
            if response.status_code == 201:
                print(f"✅ Created webhook: {name}")
                return response.json()
            else:
                print(f"❌ Failed to create webhook {name}: {response.status_code}")
                print(f"Response: {response.text}")
                return {"error": response.text}
                
        except Exception as e:
            print(f"❌ Error creating webhook {name}: {e}")
            return {"error": str(e)}
    
    def list_webhooks(self) -> List[Dict]:
        """List all existing webhooks"""
        try:
            response = requests.get(
                self.base_url,
                headers=self.get_headers()
            )
            
            if response.status_code == 200:
                return response.json()
            else:
                print(f"❌ Failed to list webhooks: {response.status_code}")
                return []
                
        except Exception as e:
            print(f"❌ Error listing webhooks: {e}")
            return []
    
    def delete_webhook(self, webhook_id: str) -> bool:
        """Delete a webhook by ID"""
        try:
            response = requests.delete(
                f"{self.base_url}/{webhook_id}",
                headers=self.get_headers()
            )
            
            if response.status_code == 204:
                print(f"✅ Deleted webhook: {webhook_id}")
                return True
            else:
                print(f"❌ Failed to delete webhook {webhook_id}: {response.status_code}")
                return False
                
        except Exception as e:
            print(f"❌ Error deleting webhook {webhook_id}: {e}")
            return False
    
    def setup_tts_webhooks(self) -> Dict[str, Dict]:
        """Set up all TTS project webhooks automatically"""
        
        webhook_secret = "tts_webhook_secret_2024"
        target_repos = [
            "Toowired/text2speech-gradio-app",
            # Add any model repos you want to monitor
            # "microsoft/speecht5_tts",
            # "suno/bark",
        ]
        
        webhooks_config = {
            "main_automation": {
                "endpoint": "tts_automation",
                "name": "TTS Main Automation",
                "description": "Main automation webhook for TTS project",
                "events": [
                    "repo.content.update",
                    "repo.content.create",
                    "space.runtime.restart",
                    "discussion.create",
                    "discussion.comment.create"
                ]
            },
            "model_sync": {
                "endpoint": "model_sync", 
                "name": "TTS Model Synchronization",
                "description": "Automatically sync new TTS models",
                "events": [
                    "repo.create",
                    "repo.content.update",
                    "model.create"
                ]
            },
            "usage_tracker": {
                "endpoint": "usage_tracker",
                "name": "TTS Usage Analytics",
                "description": "Track usage patterns and performance",
                "events": [
                    "space.runtime.start",
                    "space.runtime.stop",
                    "space.runtime.restart"
                ]
            },
            "error_monitor": {
                "endpoint": "error_monitor", 
                "name": "TTS Error Monitoring",
                "description": "Monitor for deployment errors and issues",
                "events": [
                    "space.runtime.failed",
                    "space.build.failed",
                    "repo.content.failed"
                ]
            }
        }
        
        results = {}
        
        for webhook_key, config in webhooks_config.items():
            result = self.create_webhook(
                endpoint=config["endpoint"],
                name=config["name"],
                secret=webhook_secret,
                events=config["events"],
                target_repos=target_repos,
                description=config["description"]
            )
            results[webhook_key] = result
        
        return results
    
    def cleanup_old_webhooks(self, name_pattern: str = "TTS"):
        """Remove old TTS webhooks to avoid duplicates"""
        webhooks = self.list_webhooks()
        
        for webhook in webhooks:
            if name_pattern in webhook.get("name", ""):
                print(f"🗑️ Removing old webhook: {webhook['name']}")
                self.delete_webhook(webhook["id"])

def setup_webhooks_programmatically():
    """Main function to set up webhooks"""
    print("🔗 Setting up TTS webhooks programmatically...")
    
    manager = WebhookManager()
    
    # Clean up old webhooks first
    print("🗑️ Cleaning up old webhooks...")
    manager.cleanup_old_webhooks("TTS")
    
    # Create new webhooks
    print("🆕 Creating new webhooks...")
    results = manager.setup_tts_webhooks()
    
    # Show results
    print("\n📊 Webhook Setup Results:")
    for webhook_name, result in results.items():
        if "error" in result:
            print(f"❌ {webhook_name}: {result['error']}")
        else:
            print(f"✅ {webhook_name}: Created successfully")
    
    return results

if __name__ == "__main__":
    setup_webhooks_programmatically()