Spaces:
Runtime error
Runtime error
File size: 7,145 Bytes
f76cef0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 | # 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() |