"""Wave-based parallel processor for Spotify API requests""" import random import time import threading from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass, field from typing import Any, Callable, Dict, List, Optional from .spotify_auth import SpotifyAuth from .spotify_client import SpotifyClient, SpotifyRateLimitError from .heavy_rotation_fetcher import HeavyRotationFetcher, TimeRange @dataclass class WaveConfig: """Configuration for wave-based parallel processing""" wave_size: int = 10 # Fans per wave (UI: 5-20) concurrency: int = 3 # Parallel workers within wave (UI: 1-5) inter_wave_delay_ms: int = 500 # Delay between waves (UI: 200-1000) def calculate_wave_count(self, total_fans: int) -> int: """Calculate number of waves needed for given fan count (ceiling division)""" return (total_fans + self.wave_size - 1) // self.wave_size @dataclass class WaveResult: """Results from wave-based processing""" results: List[Dict[str, Any]] = field(default_factory=list) total_time: float = 0.0 wave_count: int = 0 errors_by_wave: Dict[int, int] = field(default_factory=dict) def truncate_token(token: str) -> str: """Truncate token for safe display (first 20 chars + ...)""" return token[:20] + '...' if len(token) > 20 else token def create_error_result(fan_number: int, token: str, error_message: str) -> Dict[str, Any]: """Create standardized error result dictionary""" return { 'fan_number': fan_number, 'token': truncate_token(token), 'success': False, 'artists_count': 0, 'artists': [], 'artist_match': None, 'artist_position': None, 'error': error_message } def process_single_fan( token: str, fan_number: int, client_id: str, client_secret: str, time_range: TimeRange, target_artist_id: Optional[str] = None, limit: int = 50 ) -> Dict[str, Any]: """ Process a single fan's Spotify data (thread-safe, no st.* calls). Args: token: Fan's refresh token fan_number: Fan identifier number client_id: Spotify client ID client_secret: Spotify client secret time_range: Time range for top artists query target_artist_id: Optional artist ID to check for match limit: Number of top artists to fetch Returns: Dict with fan_number, token (truncated), success, artists_count, artists, artist_match, artist_position, error Raises: SpotifyRateLimitError: Re-raised for caller to handle with backoff. Other exceptions are caught and returned as error results. """ truncated_token = truncate_token(token) try: # Create fresh auth/client/fetcher for thread isolation auth = SpotifyAuth( refresh_token=token, client_id=client_id, client_secret=client_secret ) auth.ensure_valid_token() client = SpotifyClient(auth) fetcher = HeavyRotationFetcher(client) # Fetch Heavy Rotation data data = fetcher.get_top_artists(time_range, limit=limit) artists = data.get('items', []) # Check if target artist is in fan's top artists artist_match = None artist_position = None if target_artist_id: for idx, artist in enumerate(artists, 1): if artist['id'] == target_artist_id: artist_match = True artist_position = idx break if artist_match is None: artist_match = False return { 'fan_number': fan_number, 'token': truncated_token, 'success': True, 'artists_count': len(artists), 'artists': artists, 'artist_match': artist_match, 'artist_position': artist_position, 'error': None } except SpotifyRateLimitError: # Re-raise rate limit errors for wave processor to handle with backoff raise except Exception as e: return create_error_result(fan_number, token, str(e)) class WaveProcessor: """Processes fans in waves with configurable parallelism""" def __init__( self, config: WaveConfig, client_id: str, client_secret: str, time_range: TimeRange, target_artist_id: Optional[str] = None ): self.config = config self.client_id = client_id self.client_secret = client_secret self.time_range = time_range self.target_artist_id = target_artist_id self._results: List[Dict[str, Any]] = [] self._results_lock = threading.Lock() self._errors_by_wave: Dict[int, int] = {} def process_all( self, tokens: List[str], progress_callback: Optional[Callable[[int, int, int, int], None]] = None ) -> WaveResult: """ Process all tokens in waves with parallel execution. Args: tokens: List of refresh tokens to process progress_callback: Optional callback(current_fan, total_fans, current_wave, total_waves) Returns: WaveResult with all results and timing info """ start_time = time.time() total_fans = len(tokens) total_waves = self.config.calculate_wave_count(total_fans) self._results = [] self._errors_by_wave = {} fans_processed = 0 for wave_num in range(total_waves): wave_start = wave_num * self.config.wave_size wave_end = min(wave_start + self.config.wave_size, total_fans) wave_tokens = tokens[wave_start:wave_end] # Process this wave with ThreadPoolExecutor wave_errors = 0 with ThreadPoolExecutor(max_workers=self.config.concurrency) as executor: # Submit all tasks in this wave futures = {} for i, token in enumerate(wave_tokens): fan_number = wave_start + i + 1 future = executor.submit( self._process_fan_with_retry, token, fan_number ) futures[future] = fan_number # Collect results as they complete # Note: as_completed() iterates in the main thread, so progress_callback # is safely called from the main thread context (not worker threads) for future in as_completed(futures): fan_number = futures[future] try: result = future.result() with self._results_lock: self._results.append(result) if not result['success']: wave_errors += 1 except Exception as e: # This shouldn't happen as process_single_fan catches exceptions with self._results_lock: self._results.append(create_error_result(fan_number, '(error)', str(e))) wave_errors += 1 fans_processed += 1 if progress_callback: progress_callback(fans_processed, total_fans, wave_num + 1, total_waves) # Track errors by wave if wave_errors > 0: self._errors_by_wave[wave_num + 1] = wave_errors # Inter-wave delay (skip after last wave) if wave_num < total_waves - 1: time.sleep(self.config.inter_wave_delay_ms / 1000.0) # Sort results by fan_number (with fallback for safety) self._results.sort(key=lambda x: x.get('fan_number', 0)) return WaveResult( results=self._results, total_time=time.time() - start_time, wave_count=total_waves, errors_by_wave=self._errors_by_wave ) def _process_fan_with_retry(self, token: str, fan_number: int) -> Dict[str, Any]: """ Process a single fan with rate limit retry handling. On SpotifyRateLimitError, waits for Retry-After duration and retries once. """ try: return process_single_fan( token=token, fan_number=fan_number, client_id=self.client_id, client_secret=self.client_secret, time_range=self.time_range, target_artist_id=self.target_artist_id ) except SpotifyRateLimitError as e: # Handle 429 rate limit with backoff using Retry-After from header # Add random jitter (0-1s) to prevent synchronized retries time.sleep(e.retry_after + random.uniform(0, 1)) # Retry once try: return process_single_fan( token=token, fan_number=fan_number, client_id=self.client_id, client_secret=self.client_secret, time_range=self.time_range, target_artist_id=self.target_artist_id ) except SpotifyRateLimitError as retry_rate_limit: # Still rate limited after retry return create_error_result( fan_number, token, f'Rate limit persisted after {e.retry_after}s backoff (retry_after={retry_rate_limit.retry_after}s)' ) except Exception as retry_error: # Different error on retry error_type = type(retry_error).__name__ return create_error_result( fan_number, token, f'Retry failed with {error_type}: {retry_error}' )