#!/usr/bin/env python3 """ Webhook batch processing logic for NFOGuard """ import asyncio import threading from collections import defaultdict from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timedelta from pathlib import Path from typing import Dict, Set, List, Any, Optional from core.logging import _log from core.nfo_manager import NFOManager from config.settings import config class WebhookBatcher: """Batches webhook events to avoid processing storms""" def __init__(self, nfo_manager: NFOManager): self.pending: Dict[str, Dict] = {} self.timers: Dict[str, threading.Timer] = {} self.processing: Set[str] = set() self.lock = threading.Lock() self.executor = ThreadPoolExecutor(max_workers=config.max_concurrent) self.nfo_manager = nfo_manager def add_webhook(self, key: str, webhook_data: Dict, media_type: str): """Add webhook to batch queue""" with self.lock: if key in self.timers: self.timers[key].cancel() webhook_data['media_type'] = media_type self.pending[key] = webhook_data _log("INFO", f"Batched {media_type} webhook for {key}") _log("DEBUG", f"Batch added - key: {key}, media_type: {media_type}, timer scheduled for {config.batch_delay}s") timer = threading.Timer(config.batch_delay, self._process_item, args=[key]) self.timers[key] = timer timer.start() def _process_item(self, key: str): """Process a batched item""" with self.lock: if key in self.processing or key not in self.pending: return self.processing.add(key) webhook_data = self.pending.pop(key) self.timers.pop(key, None) try: self.executor.submit(self._process_sync, key, webhook_data) except Exception as e: _log("ERROR", f"Error submitting processing for {key}: {e}") with self.lock: self.processing.discard(key) def _process_sync(self, key: str, webhook_data: Dict): """Synchronous processing of webhook data with validation""" try: media_type = webhook_data.get('media_type') path_str = webhook_data.get('path') _log("DEBUG", f"Processing batch item: key={key}, media_type={media_type}, path={path_str}") if not path_str: _log("ERROR", f"No path found in webhook data for key: {key}") return path = Path(path_str) if not path.exists(): _log("WARNING", f"Path does not exist: {path}") return # Import processors here to avoid circular imports from processors.tv_processor import TVProcessor from processors.movie_processor import MovieProcessor from core.database import NFOGuardDatabase from core.path_mapper import PathMapper # Initialize components db = NFOGuardDatabase(config.db_path) path_mapper = PathMapper(config) if media_type == 'tv': processor = TVProcessor(db, self.nfo_manager, path_mapper) # Check if this is a series or episode-specific processing episodes_data = webhook_data.get('episodes', []) if episodes_data: processor.process_webhook_episodes(path, episodes_data) else: processor.process_series(path) elif media_type == 'movie': processor = MovieProcessor(db, self.nfo_manager, path_mapper) processor.process_movie(path, webhook_mode=True) else: _log("ERROR", f"Unknown media type: {media_type}") return _log("INFO", f"Completed processing {media_type} webhook for: {path.name}") except Exception as e: _log("ERROR", f"Error processing batch item {key}: {e}") finally: with self.lock: self.processing.discard(key) async def queue_sonarr_webhook(self, webhook_data: Dict[str, Any]) -> None: """Queue Sonarr webhook for batch processing""" try: event_type = webhook_data.get('eventType') series = webhook_data.get('series', {}) episodes = webhook_data.get('episodes', []) if not series: _log("WARNING", "No series data in Sonarr webhook") return series_title = series.get('title', 'Unknown') imdb_id = series.get('imdbId') series_path = series.get('path') _log("INFO", f"Queuing Sonarr {event_type} webhook: {series_title}") # Create batch key (use IMDb ID if available, otherwise title) batch_key = imdb_id if imdb_id else f"series_{series_title.replace(' ', '_')}" # Find series path from processors.tv_processor import TVProcessor from core.database import NFOGuardDatabase from core.path_mapper import PathMapper db = NFOGuardDatabase(config.db_path) path_mapper = PathMapper(config) tv_processor = TVProcessor(db, self.nfo_manager, path_mapper) found_path = tv_processor.find_series_path(series_title, imdb_id, series_path) if not found_path: _log("WARNING", f"Could not find series path for: {series_title}") return # Prepare webhook data for processing batch_data = { 'path': str(found_path), 'series': series, 'episodes': episodes, 'event_type': event_type } # Add to batch self.add_webhook(batch_key, batch_data, 'tv') except Exception as e: _log("ERROR", f"Error queuing Sonarr webhook: {e}") async def queue_radarr_webhook(self, webhook_data: Dict[str, Any]) -> None: """Queue Radarr webhook for batch processing""" try: event_type = webhook_data.get('eventType') movie = webhook_data.get('movie', {}) if not movie: _log("WARNING", "No movie data in Radarr webhook") return movie_title = movie.get('title', 'Unknown') imdb_id = movie.get('imdbId') movie_path = movie.get('folderPath') _log("INFO", f"Queuing Radarr {event_type} webhook: {movie_title}") # Create batch key (use IMDb ID if available, otherwise title) batch_key = imdb_id if imdb_id else f"movie_{movie_title.replace(' ', '_')}" # Find movie path from processors.movie_processor import MovieProcessor from core.database import NFOGuardDatabase from core.path_mapper import PathMapper db = NFOGuardDatabase(config.db_path) path_mapper = PathMapper(config) movie_processor = MovieProcessor(db, self.nfo_manager, path_mapper) found_path = movie_processor.find_movie_path(movie_title, imdb_id, movie_path) if not found_path: _log("WARNING", f"Could not find movie path for: {movie_title}") return # Prepare webhook data for processing batch_data = { 'path': str(found_path), 'movie': movie, 'event_type': event_type } # Add to batch self.add_webhook(batch_key, batch_data, 'movie') except Exception as e: _log("ERROR", f"Error queuing Radarr webhook: {e}") def get_pending_count(self) -> int: """Get count of pending webhooks""" with self.lock: return len(self.pending) def get_processing_count(self) -> int: """Get count of currently processing webhooks""" with self.lock: return len(self.processing) def shutdown(self): """Shutdown the webhook batcher""" _log("INFO", "Shutting down WebhookBatcher...") # Cancel all pending timers with self.lock: for timer in self.timers.values(): timer.cancel() self.timers.clear() # Shutdown executor self.executor.shutdown(wait=True) _log("INFO", "WebhookBatcher shutdown complete")