lukas.pupkalipinski/Aniworld
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

cleanup 2

Browse Source
This commit is contained in:
Lukas Pupka-Lipinski
2025-10-05 22:22:04 +02:00
parent fe2df1514c
commit 85f2d2c6f7
32 changed files with 10140 additions and 11824 deletions
Download Patch File Download Diff File Expand all files Collapse all files

537
src/infrastructure/external/api_client.py vendored
View File

@@ -1,537 +0,0 @@
"""
REST API & Integration Module for AniWorld App
This module provides comprehensive REST API endpoints for external integrations,
webhook support, API authentication, and export functionality.
"""
import json
import csv
import io
import uuid
import hmac
import hashlib
import time
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Any, Callable
from functools import wraps
import logging
import requests
import threading
from dataclasses import dataclass, field
from flask import Blueprint, request, jsonify, make_response, current_app
from werkzeug.security import generate_password_hash, check_password_hash
from auth import require_auth, optional_auth
from error_handler import handle_api_errors, RetryableError, NonRetryableError
@dataclass
class APIKey:
"""Represents an API key for external integrations."""
key_id: str
name: str
key_hash: str
permissions: List[str]
rate_limit_per_hour: int = 1000
created_at: datetime = field(default_factory=datetime.now)
last_used: Optional[datetime] = None
is_active: bool = True
@dataclass
class WebhookEndpoint:
"""Represents a webhook endpoint configuration."""
webhook_id: str
name: str
url: str
events: List[str]
secret: Optional[str] = None
is_active: bool = True
retry_attempts: int = 3
created_at: datetime = field(default_factory=datetime.now)
last_triggered: Optional[datetime] = None
class APIKeyManager:
"""Manage API keys for external integrations."""
def __init__(self):
self.api_keys: Dict[str, APIKey] = {}
self.rate_limits: Dict[str, Dict[str, int]] = {} # key_id -> {hour: count}
self.lock = threading.Lock()
self.logger = logging.getLogger(__name__)
def create_api_key(self, name: str, permissions: List[str], rate_limit: int = 1000) -> tuple:
"""Create a new API key and return the key and key_id."""
key_id = str(uuid.uuid4())
raw_key = f"aniworld_{uuid.uuid4().hex}"
key_hash = generate_password_hash(raw_key)
api_key = APIKey(
key_id=key_id,
name=name,
key_hash=key_hash,
permissions=permissions,
rate_limit_per_hour=rate_limit
)
with self.lock:
self.api_keys[key_id] = api_key
self.logger.info(f"Created API key: {name} ({key_id})")
return raw_key, key_id
def validate_api_key(self, raw_key: str) -> Optional[APIKey]:
"""Validate an API key and return the associated APIKey object."""
with self.lock:
for api_key in self.api_keys.values():
if api_key.is_active and check_password_hash(api_key.key_hash, raw_key):
api_key.last_used = datetime.now()
return api_key
return None
def check_rate_limit(self, key_id: str) -> bool:
"""Check if API key is within rate limits."""
current_hour = datetime.now().replace(minute=0, second=0, microsecond=0)
with self.lock:
if key_id not in self.api_keys:
return False
api_key = self.api_keys[key_id]
if key_id not in self.rate_limits:
self.rate_limits[key_id] = {}
hour_key = current_hour.isoformat()
current_count = self.rate_limits[key_id].get(hour_key, 0)
if current_count >= api_key.rate_limit_per_hour:
return False
self.rate_limits[key_id][hour_key] = current_count + 1
# Clean old entries (keep only last 24 hours)
cutoff = current_hour - timedelta(hours=24)
for hour_key in list(self.rate_limits[key_id].keys()):
if datetime.fromisoformat(hour_key) < cutoff:
del self.rate_limits[key_id][hour_key]
return True
def revoke_api_key(self, key_id: str) -> bool:
"""Revoke an API key."""
with self.lock:
if key_id in self.api_keys:
self.api_keys[key_id].is_active = False
self.logger.info(f"Revoked API key: {key_id}")
return True
return False
def list_api_keys(self) -> List[Dict[str, Any]]:
"""List all API keys (without sensitive data)."""
with self.lock:
return [
{
'key_id': key.key_id,
'name': key.name,
'permissions': key.permissions,
'rate_limit_per_hour': key.rate_limit_per_hour,
'created_at': key.created_at.isoformat(),
'last_used': key.last_used.isoformat() if key.last_used else None,
'is_active': key.is_active
}
for key in self.api_keys.values()
]
class WebhookManager:
"""Manage webhook endpoints and delivery."""
def __init__(self):
self.webhooks: Dict[str, WebhookEndpoint] = {}
self.delivery_queue = []
self.delivery_thread = None
self.running = False
self.lock = threading.Lock()
self.logger = logging.getLogger(__name__)
def start(self):
"""Start webhook delivery service."""
if self.running:
return
self.running = True
self.delivery_thread = threading.Thread(target=self._delivery_loop, daemon=True)
self.delivery_thread.start()
self.logger.info("Webhook delivery service started")
def stop(self):
"""Stop webhook delivery service."""
self.running = False
if self.delivery_thread:
self.delivery_thread.join(timeout=5)
self.logger.info("Webhook delivery service stopped")
def create_webhook(self, name: str, url: str, events: List[str], secret: Optional[str] = None) -> str:
"""Create a new webhook endpoint."""
webhook_id = str(uuid.uuid4())
webhook = WebhookEndpoint(
webhook_id=webhook_id,
name=name,
url=url,
events=events,
secret=secret
)
with self.lock:
self.webhooks[webhook_id] = webhook
self.logger.info(f"Created webhook: {name} ({webhook_id})")
return webhook_id
def delete_webhook(self, webhook_id: str) -> bool:
"""Delete a webhook endpoint."""
with self.lock:
if webhook_id in self.webhooks:
del self.webhooks[webhook_id]
self.logger.info(f"Deleted webhook: {webhook_id}")
return True
return False
def trigger_event(self, event_type: str, data: Dict[str, Any]):
"""Trigger webhook event for all subscribed endpoints."""
event_data = {
'event': event_type,
'timestamp': datetime.now().isoformat(),
'data': data
}
with self.lock:
for webhook in self.webhooks.values():
if webhook.is_active and event_type in webhook.events:
self.delivery_queue.append((webhook, event_data))
self.logger.debug(f"Triggered webhook event: {event_type}")
def _delivery_loop(self):
"""Main delivery loop for webhook events."""
while self.running:
try:
if self.delivery_queue:
with self.lock:
webhook, event_data = self.delivery_queue.pop(0)
self._deliver_webhook(webhook, event_data)
else:
time.sleep(1)
except Exception as e:
self.logger.error(f"Error in webhook delivery loop: {e}")
time.sleep(1)
def _deliver_webhook(self, webhook: WebhookEndpoint, event_data: Dict[str, Any]):
"""Deliver webhook event to endpoint."""
for attempt in range(webhook.retry_attempts):
try:
headers = {'Content-Type': 'application/json'}
# Add signature if secret is provided
if webhook.secret:
payload = json.dumps(event_data)
signature = hmac.new(
webhook.secret.encode(),
payload.encode(),
hashlib.sha256
).hexdigest()
headers['X-Webhook-Signature'] = f"sha256={signature}"
response = requests.post(
webhook.url,
json=event_data,
headers=headers,
timeout=30
)
if response.status_code < 400:
webhook.last_triggered = datetime.now()
self.logger.debug(f"Webhook delivered successfully: {webhook.webhook_id}")
break
else:
self.logger.warning(f"Webhook delivery failed (HTTP {response.status_code}): {webhook.webhook_id}")
except Exception as e:
self.logger.error(f"Webhook delivery error (attempt {attempt + 1}): {e}")
if attempt < webhook.retry_attempts - 1:
time.sleep(2 ** attempt) # Exponential backoff
def list_webhooks(self) -> List[Dict[str, Any]]:
"""List all webhook endpoints."""
with self.lock:
return [
{
'webhook_id': webhook.webhook_id,
'name': webhook.name,
'url': webhook.url,
'events': webhook.events,
'is_active': webhook.is_active,
'created_at': webhook.created_at.isoformat(),
'last_triggered': webhook.last_triggered.isoformat() if webhook.last_triggered else None
}
for webhook in self.webhooks.values()
]
class ExportManager:
"""Manage data export functionality."""
def __init__(self, series_app=None):
self.series_app = series_app
self.logger = logging.getLogger(__name__)
def export_anime_list_json(self, include_missing_only: bool = False) -> Dict[str, Any]:
"""Export anime list as JSON."""
try:
if not self.series_app or not self.series_app.List:
return {'anime_list': [], 'metadata': {'count': 0}}
anime_list = []
series_list = self.series_app.List.GetList()
for serie in series_list:
# Skip series without missing episodes if filter is enabled
if include_missing_only and not serie.episodeDict:
continue
anime_data = {
'name': serie.name or serie.folder,
'folder': serie.folder,
'key': getattr(serie, 'key', None),
'missing_episodes': {}
}
if hasattr(serie, 'episodeDict') and serie.episodeDict:
for season, episodes in serie.episodeDict.items():
if episodes:
anime_data['missing_episodes'][str(season)] = list(episodes)
anime_list.append(anime_data)
return {
'anime_list': anime_list,
'metadata': {
'count': len(anime_list),
'exported_at': datetime.now().isoformat(),
'include_missing_only': include_missing_only
}
}
except Exception as e:
self.logger.error(f"Failed to export anime list as JSON: {e}")
raise RetryableError(f"JSON export failed: {e}")
def export_anime_list_csv(self, include_missing_only: bool = False) -> str:
"""Export anime list as CSV."""
try:
output = io.StringIO()
writer = csv.writer(output)
# Write header
writer.writerow(['Name', 'Folder', 'Key', 'Season', 'Episode', 'Missing'])
if not self.series_app or not self.series_app.List:
return output.getvalue()
series_list = self.series_app.List.GetList()
for serie in series_list:
# Skip series without missing episodes if filter is enabled
if include_missing_only and not serie.episodeDict:
continue
name = serie.name or serie.folder
folder = serie.folder
key = getattr(serie, 'key', '')
if hasattr(serie, 'episodeDict') and serie.episodeDict:
for season, episodes in serie.episodeDict.items():
for episode in episodes:
writer.writerow([name, folder, key, season, episode, 'Yes'])
else:
writer.writerow([name, folder, key, '', '', 'No'])
return output.getvalue()
except Exception as e:
self.logger.error(f"Failed to export anime list as CSV: {e}")
raise RetryableError(f"CSV export failed: {e}")
def export_download_statistics(self) -> Dict[str, Any]:
"""Export download statistics and metrics."""
try:
# This would integrate with download manager statistics
from performance_optimizer import download_manager
stats = download_manager.get_statistics()
return {
'download_statistics': stats,
'metadata': {
'exported_at': datetime.now().isoformat()
}
}
except Exception as e:
self.logger.error(f"Failed to export download statistics: {e}")
raise RetryableError(f"Statistics export failed: {e}")
class NotificationService:
"""External notification service integration."""
def __init__(self):
self.services = {}
self.logger = logging.getLogger(__name__)
def register_discord_webhook(self, webhook_url: str, name: str = "discord"):
"""Register Discord webhook for notifications."""
self.services[name] = {
'type': 'discord',
'webhook_url': webhook_url
}
self.logger.info(f"Registered Discord webhook: {name}")
def register_telegram_bot(self, bot_token: str, chat_id: str, name: str = "telegram"):
"""Register Telegram bot for notifications."""
self.services[name] = {
'type': 'telegram',
'bot_token': bot_token,
'chat_id': chat_id
}
self.logger.info(f"Registered Telegram bot: {name}")
def send_notification(self, message: str, title: str = None, service_name: str = None):
"""Send notification to all or specific services."""
services_to_use = [service_name] if service_name else list(self.services.keys())
for name in services_to_use:
if name in self.services:
try:
service = self.services[name]
if service['type'] == 'discord':
self._send_discord_notification(service, message, title)
elif service['type'] == 'telegram':
self._send_telegram_notification(service, message, title)
except Exception as e:
self.logger.error(f"Failed to send notification via {name}: {e}")
def _send_discord_notification(self, service: Dict, message: str, title: str = None):
"""Send Discord webhook notification."""
payload = {
'embeds': [{
'title': title or 'AniWorld Notification',
'description': message,
'color': 0x00ff00,
'timestamp': datetime.now().isoformat()
}]
}
response = requests.post(service['webhook_url'], json=payload, timeout=10)
response.raise_for_status()
def _send_telegram_notification(self, service: Dict, message: str, title: str = None):
"""Send Telegram bot notification."""
text = f"*{title}*\n\n{message}" if title else message
payload = {
'chat_id': service['chat_id'],
'text': text,
'parse_mode': 'Markdown'
}
url = f"https://api.telegram.org/bot{service['bot_token']}/sendMessage"
response = requests.post(url, json=payload, timeout=10)
response.raise_for_status()
# Global instances
api_key_manager = APIKeyManager()
webhook_manager = WebhookManager()
export_manager = ExportManager()
notification_service = NotificationService()
def require_api_key(permissions: List[str] = None):
"""Decorator to require valid API key with optional permissions."""
def decorator(f):
@wraps(f)
def decorated_function(*args, **kwargs):
auth_header = request.headers.get('Authorization', '')
if not auth_header.startswith('Bearer '):
return jsonify({
'status': 'error',
'message': 'Invalid authorization header format'
}), 401
api_key = auth_header[7:] # Remove 'Bearer ' prefix
validated_key = api_key_manager.validate_api_key(api_key)
if not validated_key:
return jsonify({
'status': 'error',
'message': 'Invalid API key'
}), 401
# Check rate limits
if not api_key_manager.check_rate_limit(validated_key.key_id):
return jsonify({
'status': 'error',
'message': 'Rate limit exceeded'
}), 429
# Check permissions
if permissions:
missing_permissions = set(permissions) - set(validated_key.permissions)
if missing_permissions:
return jsonify({
'status': 'error',
'message': f'Missing permissions: {", ".join(missing_permissions)}'
}), 403
# Store API key info in request context
request.api_key = validated_key
return f(*args, **kwargs)
return decorated_function
return decorator
def init_api_integrations():
"""Initialize API integration services."""
webhook_manager.start()
def cleanup_api_integrations():
"""Clean up API integration services."""
webhook_manager.stop()
# Export main components
__all__ = [
'APIKeyManager',
'WebhookManager',
'ExportManager',
'NotificationService',
'api_key_manager',
'webhook_manager',
'export_manager',
'notification_service',
'require_api_key',
'init_api_integrations',
'cleanup_api_integrations'
]