This commit is contained in:
+206
-24
@@ -3,45 +3,227 @@
|
||||
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 processing to reduce duplicate work"""
|
||||
"""Batches webhook events to avoid processing storms"""
|
||||
|
||||
def __init__(self, nfo_manager: NFOManager):
|
||||
self.nfo_manager = nfo_manager
|
||||
self.pending: Dict[str, Dict] = {}
|
||||
self.batch_timeout = 30
|
||||
self.processing_lock = asyncio.Lock()
|
||||
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
|
||||
|
||||
# NOTE: This is a placeholder extraction - the full 139-line class implementation
|
||||
# needs to be copied from the original nfoguard.py file (lines 1562-1700)
|
||||
#
|
||||
# Key methods that need to be extracted include:
|
||||
# - queue_sonarr_webhook()
|
||||
# - queue_radarr_webhook()
|
||||
# - _start_batch_timer()
|
||||
# - _process_batched_webhooks()
|
||||
# - Various helper methods for webhook processing
|
||||
#
|
||||
# This placeholder allows the import structure to work while we continue
|
||||
# the refactoring process.
|
||||
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 - PLACEHOLDER"""
|
||||
_log("INFO", f"Queuing Sonarr webhook: {webhook_data.get('eventType', 'Unknown')}")
|
||||
# Full implementation needs to be copied from original file
|
||||
pass
|
||||
"""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 - PLACEHOLDER"""
|
||||
_log("INFO", f"Queuing Radarr webhook: {webhook_data.get('eventType', 'Unknown')}")
|
||||
# Full implementation needs to be copied from original file
|
||||
pass
|
||||
"""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")
|
||||
Reference in New Issue
Block a user