This commit is contained in:
@@ -138,6 +138,11 @@ BATCH_DELAY=5.0
|
|||||||
# Maximum concurrent series processing
|
# Maximum concurrent series processing
|
||||||
MAX_CONCURRENT_SERIES=3
|
MAX_CONCURRENT_SERIES=3
|
||||||
|
|
||||||
|
# Sequential processing delay for bulk downloads (seconds)
|
||||||
|
# When multiple episodes are downloaded at once, process them one by one with this delay
|
||||||
|
# Set to 0 to disable sequential processing (parallel mode)
|
||||||
|
SEQUENTIAL_DELAY=20.0
|
||||||
|
|
||||||
# API timeout in seconds
|
# API timeout in seconds
|
||||||
TIMEOUT_SECONDS=45
|
TIMEOUT_SECONDS=45
|
||||||
|
|
||||||
|
|||||||
@@ -61,6 +61,7 @@ class NFOGuardConfig:
|
|||||||
# Batching and performance
|
# Batching and performance
|
||||||
self.batch_delay = self._get_float_env("BATCH_DELAY", 5.0, 0.1, 300.0)
|
self.batch_delay = self._get_float_env("BATCH_DELAY", 5.0, 0.1, 300.0)
|
||||||
self.max_concurrent = self._get_int_env("MAX_CONCURRENT_SERIES", 3, 1, 10)
|
self.max_concurrent = self._get_int_env("MAX_CONCURRENT_SERIES", 3, 1, 10)
|
||||||
|
self.sequential_delay = self._get_float_env("SEQUENTIAL_DELAY", 20.0, 0.0, 60.0) # Delay between sequential episodes (default 20s)
|
||||||
|
|
||||||
# Database
|
# Database
|
||||||
self.db_type = os.environ.get("DB_TYPE", "sqlite").lower()
|
self.db_type = os.environ.get("DB_TYPE", "sqlite").lower()
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ Webhook Batching System for NFOGuard
|
|||||||
Handles batching and processing of webhook events to avoid processing storms
|
Handles batching and processing of webhook events to avoid processing storms
|
||||||
"""
|
"""
|
||||||
import threading
|
import threading
|
||||||
|
import time
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Dict, Set
|
from typing import Dict, Set
|
||||||
from concurrent.futures import ThreadPoolExecutor
|
from concurrent.futures import ThreadPoolExecutor
|
||||||
@@ -144,6 +145,12 @@ class WebhookBatcher:
|
|||||||
|
|
||||||
if processing_mode == 'targeted' and episodes_data:
|
if processing_mode == 'targeted' and episodes_data:
|
||||||
_log("INFO", f"Using targeted episode processing for {len(episodes_data)} episodes")
|
_log("INFO", f"Using targeted episode processing for {len(episodes_data)} episodes")
|
||||||
|
|
||||||
|
# Handle sequential processing for bulk downloads
|
||||||
|
if len(episodes_data) > 1 and config.sequential_delay > 0:
|
||||||
|
_log("INFO", f"Processing {len(episodes_data)} episodes sequentially with {config.sequential_delay}s delay")
|
||||||
|
self._process_episodes_sequentially(path_obj, episodes_data)
|
||||||
|
else:
|
||||||
self.tv_processor.process_webhook_episodes(path_obj, episodes_data)
|
self.tv_processor.process_webhook_episodes(path_obj, episodes_data)
|
||||||
else:
|
else:
|
||||||
_log("INFO", f"Using series processing mode (fallback or configured)")
|
_log("INFO", f"Using series processing mode (fallback or configured)")
|
||||||
@@ -164,6 +171,29 @@ class WebhookBatcher:
|
|||||||
with self.lock:
|
with self.lock:
|
||||||
self.processing.discard(key)
|
self.processing.discard(key)
|
||||||
|
|
||||||
|
def _process_episodes_sequentially(self, path_obj: Path, episodes_data: list):
|
||||||
|
"""Process episodes one by one with delays to avoid API spam"""
|
||||||
|
total_episodes = len(episodes_data)
|
||||||
|
for i, episode in enumerate(episodes_data, 1):
|
||||||
|
try:
|
||||||
|
season = episode.get('seasonNumber', '?')
|
||||||
|
episode_num = episode.get('episodeNumber', '?')
|
||||||
|
_log("INFO", f"Processing episode {i}/{total_episodes}: S{season:02d}E{episode_num:02d}")
|
||||||
|
|
||||||
|
# Process single episode
|
||||||
|
self.tv_processor.process_webhook_episodes(path_obj, [episode])
|
||||||
|
|
||||||
|
# Add delay between episodes (except for the last one)
|
||||||
|
if i < total_episodes and config.sequential_delay > 0:
|
||||||
|
_log("INFO", f"Waiting {config.sequential_delay}s before next episode...")
|
||||||
|
time.sleep(config.sequential_delay)
|
||||||
|
|
||||||
|
except Exception as e:
|
||||||
|
_log("ERROR", f"Error processing episode {i}/{total_episodes}: {e}")
|
||||||
|
# Continue with next episode even if one fails
|
||||||
|
|
||||||
|
_log("INFO", f"Completed sequential processing of {total_episodes} episodes")
|
||||||
|
|
||||||
def get_status(self) -> Dict:
|
def get_status(self) -> Dict:
|
||||||
"""Get batch queue status"""
|
"""Get batch queue status"""
|
||||||
with self.lock:
|
with self.lock:
|
||||||
|
|||||||
Reference in New Issue
Block a user