From 594c8c1b935e56900e99f916f81b1d68974898f6 Mon Sep 17 00:00:00 2001 From: Antti Kettunen Date: Sat, 25 Apr 2026 19:19:35 +0300 Subject: [PATCH] Cleanup duplicated code Leftovers from the earlier recovery mission --- web_server.py | 1549 ------------------------------------------------- 1 file changed, 1549 deletions(-) diff --git a/web_server.py b/web_server.py index d564c911..1e036ef6 100644 --- a/web_server.py +++ b/web_server.py @@ -799,218 +799,6 @@ transfer_data_cache = { } -def get_cached_transfer_data(): - """ - Return live Soulseek transfer data with a short cache window. - - The download modal, batch status endpoint, and socket emit loop all call - this helper, so we keep the API hit rate low while still refreshing often. - """ - current_time = time.time() - - with transfer_data_cache['update_lock']: - if (current_time - transfer_data_cache['last_update']) < transfer_data_cache['cache_duration']: - return transfer_data_cache['data'] - - if not soulseek_client: - return {} - - live_transfers_lookup = {} - try: - transfers_data = run_async(soulseek_client._make_request('GET', 'transfers/downloads')) - if transfers_data: - for user_data in transfers_data: - if not isinstance(user_data, dict): - continue - username = user_data.get('username', 'Unknown') - for directory in user_data.get('directories', []) or []: - if not isinstance(directory, dict): - continue - for file_info in directory.get('files', []) or []: - if not isinstance(file_info, dict): - continue - transfer = dict(file_info) - transfer['username'] = username - lookup_key = _make_context_key(username, transfer.get('filename', '')) - live_transfers_lookup[lookup_key] = transfer - - transfer_data_cache['data'] = live_transfers_lookup - transfer_data_cache['last_update'] = current_time - except Exception as e: - logger.warning("Could not fetch live transfers (cached): %s", e) - return {} - - return live_transfers_lookup - - -def _regenerate_batch_m3u(batch, tracks): - """Regenerate an exported playlist M3U after downloads finish.""" - try: - from collections import defaultdict - from difflib import SequenceMatcher - - try: - from unidecode import unidecode as _unidecode - except ImportError: - _unidecode = lambda x: x - - from core.imports.paths import sanitize_filename as _sanitize_filename - - def _norm(text): - return _unidecode(text).lower().strip() if text else '' - - def _track_name(track): - return track.get('name') or track.get('title') or 'Unknown' - - def _track_artist(track): - return track.get('artist') or track.get('artist_name') or 'Unknown' - - playlist_name = batch.get('playlist_name', 'Playlist') - db = get_database() - active_server = config_manager.get_active_media_server() - raw_base = config_manager.get('m3u_export.entry_base_path', '') or '' - entry_base_path = raw_base.rstrip('/\\') - - artist_groups = defaultdict(list) - for idx, track in enumerate(tracks): - artist_groups[_track_artist(track)].append((idx, track)) - - file_path_map = {} - for artist, group in artist_groups.items(): - if not artist or artist == 'Unknown': - for idx, _ in group: - file_path_map[idx] = None - continue - - try: - db_tracks = db.search_tracks(artist=artist, limit=500, server_source=active_server) - except Exception as search_err: - logger.debug("[M3U] Track lookup failed for %s: %s", artist, search_err) - db_tracks = [] - - if not db_tracks: - for idx, _ in group: - file_path_map[idx] = None - continue - - db_entries = [] - for db_track in db_tracks: - db_title = getattr(db_track, 'title', '') or '' - db_entries.append((_norm(db_title), db_track)) - - for idx, track in group: - name = _track_name(track) - if not name: - file_path_map[idx] = None - continue - - s_norm = _norm(name) - matched = None - for db_n, db_track in db_entries: - if not db_n: - continue - if s_norm == db_n or SequenceMatcher(None, s_norm, db_n).ratio() >= 0.7: - matched = db_track - break - - if matched is None: - file_path_map[idx] = None - else: - file_path = getattr(matched, 'file_path', None) - if file_path is None and isinstance(matched, dict): - file_path = matched.get('file_path') - file_path_map[idx] = file_path or None - - lines = ['#EXTM3U', f'#PLAYLIST:{playlist_name}', f'#GENERATED:{datetime.utcnow().isoformat()}Z', ''] - found = 0 - missing = 0 - for idx, track in enumerate(tracks): - duration_ms = track.get('duration_ms', 0) or 0 - artist = _track_artist(track) - name = _track_name(track) - lines.append(f'#EXTINF:{int(duration_ms / 1000)},{artist} - {name}') - file_path = file_path_map.get(idx) - if file_path: - path = f'{entry_base_path}/{file_path}' if entry_base_path else file_path - lines.append(path.replace('\\', '/')) - found += 1 - else: - lines.append(f'# MISSING: {artist} - {name}') - missing += 1 - lines.append('') - - if found == 0: - logger.info("[M3U] Skipping regeneration for %s: no library paths resolved", playlist_name) - return - - transfer_dir = docker_resolve_path(config_manager.get('soulseek.transfer_path', './Transfer')) - playlist_dir = _compute_m3u_folder(transfer_dir, 'playlist', playlist_name, '', '', '') - os.makedirs(playlist_dir, exist_ok=True) - m3u_path = os.path.join(playlist_dir, f'{_sanitize_filename(playlist_name)}.m3u') - - with open(m3u_path, 'w', encoding='utf-8') as f: - f.write('\n'.join(lines)) - - logger.info( - "[M3U] Regenerated M3U on batch complete: %s (%s/%s resolved, %s missing)", - m3u_path, - found, - len(tracks), - missing, - ) - except Exception as e: - logger.error("[M3U] Error in _regenerate_batch_m3u: %s", e) - - -def _sanitize_filename(filename: str) -> str: - """Sanitize a filename for filesystem use.""" - sanitized = re.sub(r'[<>:"/\\|?*]', '_', filename or '') - sanitized = re.sub(r'\s+', ' ', sanitized).strip() - sanitized = sanitized.rstrip('. ') or '_' - if re.match(r'^(CON|PRN|AUX|NUL|COM[1-9]|LPT[1-9])(\.|$)', sanitized, re.IGNORECASE): - sanitized = '_' + sanitized - return sanitized[:200] - - -def _compute_m3u_folder(transfer_dir, context_type, playlist_name, artist_name='', album_name='', year=''): - """Compute the target folder for an M3U file using the template system.""" - from core.imports.paths import get_file_path_from_template - - if context_type == 'album' and artist_name and album_name: - template_context = { - 'artist': artist_name, - 'albumartist': artist_name, - 'album': album_name, - 'title': 'placeholder', - 'track_number': 1, - 'disc_number': 1, - 'year': year, - 'quality': '' - } - folder_path, _ = get_file_path_from_template(template_context, 'album_path') - if folder_path: - return os.path.join(transfer_dir, folder_path) - artist_sanitized = _sanitize_filename(artist_name) - album_sanitized = _sanitize_filename(album_name) - return os.path.join(transfer_dir, artist_sanitized, f"{artist_sanitized} - {album_sanitized}") - - template_context = { - 'artist': 'placeholder', - 'albumartist': 'placeholder', - 'album': 'placeholder', - 'title': 'placeholder', - 'playlist_name': playlist_name, - 'track_number': 1, - 'disc_number': 1, - 'year': '', - 'quality': '' - } - folder_path, _ = get_file_path_from_template(template_context, 'playlist_path') - if folder_path: - return os.path.join(transfer_dir, folder_path) - playlist_sanitized = _sanitize_filename(playlist_name) - return os.path.join(transfer_dir, playlist_sanitized) - # --- Restored Web UI Helper State --- session_completed_downloads = 0 session_stats_lock = threading.Lock() @@ -1039,1179 +827,6 @@ _hydrabase_ws = None _hydrabase_lock = threading.Lock() -def get_cached_beatport_data(section_type, data_key, genre_slug=None): - current_time = time.time() - with beatport_data_cache['cache_lock']: - try: - if section_type == 'homepage': - cache_entry = beatport_data_cache['homepage'].get(data_key) - elif section_type == 'genre' and genre_slug: - cache_entry = beatport_data_cache['genre'].get(genre_slug, {}).get(data_key) - else: - return None - - if not cache_entry: - return None - - age = current_time - cache_entry['timestamp'] - if age < cache_entry['ttl'] and cache_entry['data'] is not None: - return cache_entry['data'] - return None - except Exception as e: - print(f"Cache lookup error for {section_type}/{data_key}: {e}") - return None - - -def set_cached_beatport_data(section_type, data_key, data, genre_slug=None): - current_time = time.time() - with beatport_data_cache['cache_lock']: - try: - if section_type == 'homepage': - if data_key in beatport_data_cache['homepage']: - beatport_data_cache['homepage'][data_key]['data'] = data - beatport_data_cache['homepage'][data_key]['timestamp'] = current_time - elif section_type == 'genre' and genre_slug: - if genre_slug not in beatport_data_cache['genre']: - beatport_data_cache['genre'][genre_slug] = {} - if data_key not in beatport_data_cache['genre'][genre_slug]: - beatport_data_cache['genre'][genre_slug][data_key] = { - 'data': None, - 'timestamp': 0, - 'ttl': 600, - } - beatport_data_cache['genre'][genre_slug][data_key]['data'] = data - beatport_data_cache['genre'][genre_slug][data_key]['timestamp'] = current_time - except Exception as e: - print(f"Cache storage error for {section_type}/{data_key}: {e}") - - -def add_cache_headers(response, cache_duration=300): - response.headers['Cache-Control'] = f'public, max-age={cache_duration}' - response.headers['Pragma'] = 'cache' - return response - - -def _get_max_concurrent(): - return config_manager.get('download_source.max_concurrent', 3) - - -def _get_batch_max_concurrent(is_album=False, source=None): - if is_album and source in ('soulseek', None): - if source == 'soulseek': - return 1 - mode = config_manager.get('download_source.mode', 'soulseek') - if mode == 'soulseek': - return 1 - return _get_max_concurrent() - - -def _update_task_status(task_id, new_status): - with tasks_lock: - if task_id in download_tasks: - download_tasks[task_id]['status'] = new_status - download_tasks[task_id]['status_change_time'] = time.time() - - -class WebUIDownloadMonitor: - def __init__(self): - self.monitoring = False - self.monitor_thread = None - self.monitored_batches = set() - self._lock = threading.Lock() - - def start_monitoring(self, batch_id): - with self._lock: - self.monitored_batches.add(batch_id) - if not self.monitoring: - self.monitoring = True - self.monitor_thread = threading.Thread(target=self._monitor_loop, daemon=True) - self.monitor_thread.start() - - def stop_monitoring(self, batch_id): - with self._lock: - self.monitored_batches.discard(batch_id) - if not self.monitored_batches: - self.monitoring = False - - def shutdown(self): - with self._lock: - self.monitoring = False - self.monitored_batches.clear() - self.monitor_thread = None - - def _monitor_loop(self): - while self.monitoring and self.monitored_batches: - try: - if globals().get('IS_SHUTTING_DOWN', False): - self.monitoring = False - break - self._check_all_downloads() - time.sleep(1) - except Exception as e: - if 'interpreter shutdown' in str(e) or 'cannot schedule new futures' in str(e): - self.monitoring = False - break - print(f"Download monitor error: {e}") - - def _check_all_downloads(self): - current_time = time.time() - live_transfers_lookup = self._get_live_transfers() - exhausted_tasks = [] - completed_tasks = [] - deferred_ops = [] - - with tasks_lock: - for batch_id in list(self.monitored_batches): - if batch_id not in download_batches: - self.monitored_batches.discard(batch_id) - continue - - for task_id in download_batches[batch_id].get('queue', []): - task = download_tasks.get(task_id) - if not task or task['status'] not in ['downloading', 'queued']: - continue - - retry_exhausted = self._should_retry_task(task_id, task, live_transfers_lookup, current_time, deferred_ops) - if retry_exhausted: - exhausted_tasks.append((batch_id, task_id)) - - task_filename = task.get('filename') or task.get('track_info', {}).get('filename') - task_username = task.get('username') or task.get('track_info', {}).get('username') - if task_filename and task_username: - lookup_key = _make_context_key(task_username, task_filename) - live_info = live_transfers_lookup.get(lookup_key) - if live_info: - state = live_info.get('state', '') - has_error = ('Errored' in state or 'Failed' in state or 'Rejected' in state or 'TimedOut' in state) - has_completion = ('Completed' in state or 'Succeeded' in state) - if has_completion and not has_error: - expected_size = live_info.get('size', 0) - transferred = live_info.get('bytesTransferred', 0) - if expected_size > 0 and transferred < expected_size: - if not task.get('_incomplete_warned'): - task['_incomplete_warned'] = True - continue - if has_completion and not has_error and task['status'] == 'downloading': - task.pop('_incomplete_warned', None) - task['status'] = 'post_processing' - task['status_change_time'] = current_time - completed_tasks.append((batch_id, task_id)) - - for op in deferred_ops: - try: - if op[0] == 'cancel_download': - _, download_id, username = op - run_async(soulseek_client.cancel_download(download_id, username, remove=True)) - elif op[0] == 'cleanup_orphan': - _, context_key = op - with matched_context_lock: - matched_downloads_context.pop(context_key, None) - elif op[0] == 'restart_worker': - _, task_id, batch_id = op - missing_download_executor.submit(_download_track_worker, task_id, batch_id) - except Exception as e: - print(f"[Deferred] Error executing deferred operation {op[0]}: {e}") - - for batch_id, task_id in completed_tasks: - try: - missing_download_executor.submit(_run_post_processing_worker, task_id, batch_id) - _on_download_completed(batch_id, task_id, success=True) - except Exception as e: - print(f"[Monitor] Error handling completed task {task_id}: {e}") - - for batch_id, task_id in exhausted_tasks: - try: - _on_download_completed(batch_id, task_id, success=False) - except Exception as e: - print(f"[Monitor] Error handling exhausted task {task_id}: {e}") - - self._validate_worker_counts() - - def _get_live_transfers(self): - try: - if not self.monitoring or not soulseek_client: - return {} - - live_transfers = {} - transfers_data = run_async(soulseek_client._make_request('GET', 'transfers/downloads')) - if transfers_data: - for user_data in transfers_data: - username = user_data.get('username', 'Unknown') - if 'directories' in user_data: - for directory in user_data['directories']: - if 'files' in directory: - for file_info in directory['files']: - key = _make_context_key(username, file_info.get('filename', '')) - live_transfers[key] = file_info - - try: - all_downloads = run_async(soulseek_client.get_all_downloads()) - for download in all_downloads: - if download.username in ('youtube', 'tidal', 'qobuz', 'hifi', 'deezer_dl', 'lidarr'): - key = _make_context_key(download.username, download.filename) - live_transfers[key] = { - 'id': download.id, - 'filename': download.filename, - 'username': download.username, - 'state': download.state, - 'percentComplete': download.progress, - 'size': download.size, - 'bytesTransferred': download.transferred, - 'averageSpeed': download.speed, - } - except Exception as yt_error: - print(f"Monitor: Could not fetch streaming source downloads: {yt_error}") - - return live_transfers - except Exception as e: - if ('interpreter shutdown' in str(e) or 'cannot schedule new futures' in str(e) or 'Event loop is closed' in str(e)): - self.monitoring = False - return {} - print(f"Monitor: Could not fetch live transfers: {e}") - return {} - - def _should_retry_task(self, task_id, task, live_transfers_lookup, current_time, deferred_ops): - ti = task.get('track_info') if isinstance(task.get('track_info'), dict) else {} - task_filename = task.get('filename') or ti.get('filename') - task_username = task.get('username') or ti.get('username') - - if not task_filename or not task_username: - return False - - lookup_key = _make_context_key(task_username, task_filename) - live_info = live_transfers_lookup.get(lookup_key) - - if not live_info: - if current_time - task.get('status_change_time', current_time) > 90: - retry_count = task.get('stuck_retry_count', 0) - last_retry = task.get('last_retry_time', 0) - if retry_count < 3 and (current_time - last_retry) > 30: - task['stuck_retry_count'] = retry_count + 1 - task['last_retry_time'] = current_time - download_id = task.get('download_id') - if task_username and download_id: - deferred_ops.append(('cancel_download', download_id, task_username)) - if task_username and task_filename: - used_sources = task.get('used_sources', set()) - source_key = f"{task_username}_{task_filename}" - used_sources.add(source_key) - task['used_sources'] = used_sources - if task_username and task_filename: - _orphaned_download_keys.add(lookup_key) - deferred_ops.append(('cleanup_orphan', lookup_key)) - task.pop('download_id', None) - task.pop('username', None) - task.pop('filename', None) - task['status'] = 'searching' - task.pop('queued_start_time', None) - task.pop('downloading_start_time', None) - task['status_change_time'] = current_time - batch_id = task.get('batch_id') - if task_id and batch_id: - deferred_ops.append(('restart_worker', task_id, batch_id)) - return False - elif retry_count < 3: - return False - else: - track_label = task.get('track_info', {}).get('name', 'Unknown') - tried_sources = task.get('used_sources', set()) - sources_str = f" (tried {len(tried_sources)} source{'s' if len(tried_sources) != 1 else ''})" if tried_sources else '' - task['status'] = 'failed' - task['error_message'] = f'Download disappeared from transfer list 3 times for "{track_label}"{sources_str} — source may be unavailable' - return bool(task.get('batch_id')) - return False - - state_str = live_info.get('state', '') - progress = live_info.get('percentComplete', 0) - - if any(token in state_str for token in ('Errored', 'Failed', 'Rejected', 'TimedOut')): - retry_count = task.get('error_retry_count', 0) - last_retry = task.get('last_error_retry_time', 0) - if retry_count < 3 and (current_time - last_retry) > 5: - task['error_retry_count'] = retry_count + 1 - task['last_error_retry_time'] = current_time - _ti = task.get('track_info') if isinstance(task.get('track_info'), dict) else {} - username = task.get('username') or _ti.get('username') - filename = task.get('filename') or _ti.get('filename') - download_id = task.get('download_id') - if username and download_id: - deferred_ops.append(('cancel_download', download_id, username)) - if username and filename: - used_sources = task.get('used_sources', set()) - used_sources.add(f"{username}_{filename}") - task['used_sources'] = used_sources - if username and filename: - old_context_key = _make_context_key(username, filename) - _orphaned_download_keys.add(old_context_key) - deferred_ops.append(('cleanup_orphan', old_context_key)) - task.pop('download_id', None) - task.pop('username', None) - task.pop('filename', None) - task['status'] = 'searching' - task.pop('queued_start_time', None) - task.pop('downloading_start_time', None) - task['status_change_time'] = current_time - batch_id = task.get('batch_id') - if task_id and batch_id: - deferred_ops.append(('restart_worker', task_id, batch_id)) - return False - elif retry_count < 3: - return False - task['status'] = 'failed' - task['error_message'] = f'Soulseek transfer errored 3 times for "{task.get("track_info", {}).get("name", "Unknown")}"' - return bool(task.get('batch_id')) - - elif 'Queued' in state_str or task['status'] == 'queued': - if 'queued_start_time' not in task: - task['queued_start_time'] = current_time - return False - queue_time = current_time - task['queued_start_time'] - timeout_threshold = 15.0 if task.get('track_info', {}).get('is_album_download', False) else 90.0 - if queue_time > timeout_threshold: - retry_count = task.get('stuck_retry_count', 0) - last_retry = task.get('last_retry_time', 0) - if retry_count < 3 and (current_time - last_retry) > 30: - task['stuck_retry_count'] = retry_count + 1 - task['last_retry_time'] = current_time - _ti = task.get('track_info') if isinstance(task.get('track_info'), dict) else {} - username = task.get('username') or _ti.get('username') - filename = task.get('filename') or _ti.get('filename') - download_id = task.get('download_id') - if username and download_id: - deferred_ops.append(('cancel_download', download_id, username)) - if username and filename: - used_sources = task.get('used_sources', set()) - used_sources.add(f"{username}_{filename}") - task['used_sources'] = used_sources - if username and filename: - old_context_key = _make_context_key(username, filename) - _orphaned_download_keys.add(old_context_key) - deferred_ops.append(('cleanup_orphan', old_context_key)) - task.pop('download_id', None) - task.pop('username', None) - task.pop('filename', None) - task['status'] = 'searching' - task.pop('queued_start_time', None) - task.pop('downloading_start_time', None) - task['status_change_time'] = current_time - batch_id = task.get('batch_id') - if task_id and batch_id: - deferred_ops.append(('restart_worker', task_id, batch_id)) - return False - elif retry_count < 3: - return False - task['status'] = 'failed' - task['error_message'] = f'Download stayed queued too long 3 times for "{task.get("track_info", {}).get("name", "Unknown")}"' - task.pop('queued_start_time', None) - task.pop('downloading_start_time', None) - return bool(task.get('batch_id')) - - elif 'InProgress' in state_str and progress < 1: - if 'downloading_start_time' not in task: - task['downloading_start_time'] = current_time - return False - download_time = current_time - task['downloading_start_time'] - timeout_threshold = 15.0 if task.get('track_info', {}).get('is_album_download', False) else 90.0 - if download_time > timeout_threshold: - retry_count = task.get('stuck_retry_count', 0) - last_retry = task.get('last_retry_time', 0) - if retry_count < 3 and (current_time - last_retry) > 30: - task['stuck_retry_count'] = retry_count + 1 - task['last_retry_time'] = current_time - _ti = task.get('track_info') if isinstance(task.get('track_info'), dict) else {} - username = task.get('username') or _ti.get('username') - filename = task.get('filename') or _ti.get('filename') - download_id = task.get('download_id') - if username and download_id: - deferred_ops.append(('cancel_download', download_id, username)) - if username and filename: - used_sources = task.get('used_sources', set()) - used_sources.add(f"{username}_{filename}") - task['used_sources'] = used_sources - if username and filename: - old_context_key = _make_context_key(username, filename) - _orphaned_download_keys.add(old_context_key) - deferred_ops.append(('cleanup_orphan', old_context_key)) - task.pop('download_id', None) - task.pop('username', None) - task.pop('filename', None) - task['status'] = 'searching' - task.pop('queued_start_time', None) - task.pop('downloading_start_time', None) - task['status_change_time'] = current_time - batch_id = task.get('batch_id') - if task_id and batch_id: - deferred_ops.append(('restart_worker', task_id, batch_id)) - return False - elif retry_count < 3: - return False - task['status'] = 'failed' - task['error_message'] = f'Download stuck at 0% three times for "{task.get("track_info", {}).get("name", "Unknown")}"' - task.pop('queued_start_time', None) - task.pop('downloading_start_time', None) - return bool(task.get('batch_id')) - else: - bytes_transferred = live_info.get('bytesTransferred', 0) - if progress >= 1 or bytes_transferred > 0: - task.pop('queued_start_time', None) - task.pop('downloading_start_time', None) - task.pop('stuck_retry_count', None) - else: - if 'downloading_start_time' not in task: - task['downloading_start_time'] = current_time - download_time = current_time - task['downloading_start_time'] - timeout_threshold = 15.0 if task.get('track_info', {}).get('is_album_download', False) else 90.0 - if download_time > timeout_threshold: - retry_count = task.get('stuck_retry_count', 0) - last_retry = task.get('last_retry_time', 0) - if retry_count < 3 and (current_time - last_retry) > 30: - task['stuck_retry_count'] = retry_count + 1 - task['last_retry_time'] = current_time - _ti = task.get('track_info') if isinstance(task.get('track_info'), dict) else {} - username = task.get('username') or _ti.get('username') - filename = task.get('filename') or _ti.get('filename') - download_id = task.get('download_id') - if username and download_id: - deferred_ops.append(('cancel_download', download_id, username)) - if username and filename: - used_sources = task.get('used_sources', set()) - used_sources.add(f"{username}_{filename}") - task['used_sources'] = used_sources - if username and filename: - old_context_key = _make_context_key(username, filename) - _orphaned_download_keys.add(old_context_key) - deferred_ops.append(('cleanup_orphan', old_context_key)) - task.pop('download_id', None) - task.pop('username', None) - task.pop('filename', None) - task['status'] = 'searching' - task.pop('queued_start_time', None) - task.pop('downloading_start_time', None) - task['status_change_time'] = current_time - batch_id = task.get('batch_id') - if task_id and batch_id: - deferred_ops.append(('restart_worker', task_id, batch_id)) - return False - elif retry_count >= 3: - task['status'] = 'failed' - task['error_message'] = f'Download stuck in "{state_str}" state 3 times for "{task.get("track_info", {}).get("name", "Unknown")}"' - task.pop('queued_start_time', None) - task.pop('downloading_start_time', None) - return bool(task.get('batch_id')) - - return False - - def _validate_worker_counts(self): - try: - batches_needing_workers = [] - with tasks_lock: - for batch_id in list(self.monitored_batches): - if batch_id not in download_batches: - continue - batch = download_batches[batch_id] - reported_active = batch['active_count'] - max_concurrent = batch['max_concurrent'] - queue = batch.get('queue', []) - queue_index = batch.get('queue_index', 0) - actually_active = 0 - orphaned_tasks = [] - completed_task_ids = batch.get('_completed_task_ids', set()) - for task_id in queue: - if task_id in download_tasks: - task_status = download_tasks[task_id]['status'] - if task_status in ['searching', 'downloading', 'queued', 'post_processing']: - if task_id not in completed_task_ids: - actually_active += 1 - elif task_status in ['failed', 'completed', 'cancelled', 'not_found'] and task_id in queue[queue_index:]: - orphaned_tasks.append(task_id) - if reported_active != actually_active or orphaned_tasks: - if reported_active != actually_active: - batch['active_count'] = actually_active - if actually_active < max_concurrent and queue_index < len(queue): - batches_needing_workers.append(batch_id) - for batch_id in batches_needing_workers: - try: - _start_next_batch_of_downloads(batch_id) - except Exception as e: - print(f"[Worker Validation] Error starting workers for {batch_id}: {e}") - except Exception as validation_error: - print(f"Error in worker count validation: {validation_error}") - - -download_monitor = WebUIDownloadMonitor() - - -def _is_explicit_blocked(track_data): - if config_manager.get('content_filter.allow_explicit', True): - return False - if track_data.get('explicit', False): - return True - sp_data = track_data.get('spotify_data', {}) - if isinstance(sp_data, str): - try: - sp_data = json.loads(sp_data) - except Exception: - sp_data = {} - return sp_data.get('explicit', False) - - -def fix_artist_image_url(thumb_url): - if not thumb_url: - return None - - try: - needs_fixing = ( - thumb_url.startswith('http://localhost:') or - thumb_url.startswith('https://localhost:') or - thumb_url.startswith('/library/') or - thumb_url.startswith('/Items/') or - thumb_url.startswith('/api/') or - thumb_url.startswith('/rest/') - ) - - if needs_fixing: - active_server = config_manager.get_active_media_server() - if active_server == 'plex': - plex_config = config_manager.get_plex_config() - plex_base_url = plex_config.get('base_url', '') - plex_token = plex_config.get('token', '') - if plex_base_url and plex_token: - if thumb_url.startswith('/library/'): - path = thumb_url - else: - from urllib.parse import urlparse - path = urlparse(thumb_url).path - return f"{plex_base_url.rstrip('/')}{path}?X-Plex-Token={plex_token}" - - elif active_server == 'jellyfin': - jellyfin_config = config_manager.get_jellyfin_config() - jellyfin_base_url = jellyfin_config.get('base_url', '') - jellyfin_token = jellyfin_config.get('api_key', '') - if jellyfin_base_url: - if thumb_url.startswith('/Items/') or thumb_url.startswith('/api/'): - path = thumb_url - else: - from urllib.parse import urlparse - path = urlparse(thumb_url).path - if jellyfin_token: - separator = '&' if '?' in path else '?' - return f"{jellyfin_base_url.rstrip('/')}{path}{separator}X-Emby-Token={jellyfin_token}" - return f"{jellyfin_base_url.rstrip('/')}{path}" - - elif active_server == 'navidrome': - navidrome_config = config_manager.get_navidrome_config() - navidrome_base_url = navidrome_config.get('base_url', '') - navidrome_username = navidrome_config.get('username', '') - navidrome_password = navidrome_config.get('password', '') - if navidrome_base_url and navidrome_username and navidrome_password: - if thumb_url.startswith('/rest/'): - path = thumb_url - else: - from urllib.parse import urlparse - path = urlparse(thumb_url).path - import hashlib - import secrets - salt = secrets.token_hex(6) - token = hashlib.md5((navidrome_password + salt).encode()).hexdigest() - separator = '&' if '?' in path else '?' - auth_params = f"u={navidrome_username}&t={token}&s={salt}&v=1.16.1&c=SoulSync&f=json" - return f"{navidrome_base_url.rstrip('/')}{path}{separator}{auth_params}" - - return thumb_url - except Exception as e: - print(f"Error fixing image URL '{thumb_url}': {e}") - return thumb_url - - -def _find_completed_file_robust(download_dir, api_filename, transfer_dir=None): - import os - import re - from difflib import SequenceMatcher - from unidecode import unidecode - - if '||' in api_filename: - _, title = api_filename.split('||', 1) - api_filename = title - - def normalize_for_finding(text: str) -> str: - if not text: - return "" - text = unidecode(text).lower() - text = re.sub(r'[._/]', ' ', text) - text = re.sub(r'[\[\(].*?[\]\)]', '', text) - text = re.sub(r'[^a-z0-9\s-]', '', text) - return ' '.join(text.split()).strip() - - def _path_matches_api_dirs(file_path): - path_parts = set(p.lower() for p in file_path.replace('\\', '/').split('/')) - return all(d in path_parts for d in api_dir_parts) - - def search_in_directory(search_dir, location_name): - best_fuzzy_path = None - highest_fuzzy_similarity = 0.0 - exact_matches = [] - for root, dirs, files in os.walk(search_dir): - dirs[:] = [d for d in dirs if d != 'ss_quarantine'] - for file in files: - if os.path.basename(file) == target_basename: - file_path = os.path.join(root, file) - if api_dir_parts and _path_matches_api_dirs(file_path): - return file_path, 1.0 - if not api_dir_parts: - return file_path, 1.0 - exact_matches.append(file_path) - continue - - file_stem, file_ext_part = os.path.splitext(file) - stripped_stem = re.sub(r'_\d{10,}$', '', file_stem) - if stripped_stem != file_stem and stripped_stem + file_ext_part == target_basename: - file_path = os.path.join(root, file) - if api_dir_parts and _path_matches_api_dirs(file_path): - return file_path, 1.0 - if not api_dir_parts: - return file_path, 1.0 - exact_matches.append(file_path) - continue - - normalized_file = normalize_for_finding(file) - similarity = SequenceMatcher(None, normalized_target, normalized_file).ratio() - if similarity > highest_fuzzy_similarity: - highest_fuzzy_similarity = similarity - best_fuzzy_path = os.path.join(root, file) - - if exact_matches: - if len(exact_matches) == 1: - return exact_matches[0], 1.0 - best = exact_matches[0] - best_score = -1 - for m in exact_matches: - m_parts = set(p.lower() for p in m.replace('\\', '/').split('/')) - score = sum(1 for d in api_dir_parts if d in m_parts) - if score > best_score: - best_score = score - best = m - return best, 1.0 - - return best_fuzzy_path, highest_fuzzy_similarity - - target_basename = extract_filename(api_filename) - normalized_target = normalize_for_finding(target_basename) - api_path_normalized = api_filename.replace('\\', '/') if api_filename else '' - api_dir_parts = [p.lower() for p in api_path_normalized.split('/')[:-1] if p] - - best_downloads_path, downloads_similarity = search_in_directory(download_dir, 'downloads') - if downloads_similarity > 0.85: - return (best_downloads_path, 'downloads') - - transfer_similarity = 0.0 - if transfer_dir and os.path.exists(transfer_dir): - best_transfer_path, transfer_similarity = search_in_directory(transfer_dir, 'transfer') - if transfer_similarity > 0.85: - return (best_transfer_path, 'transfer') - - return (None, None) - - -def _cleanup_empty_directories(download_path, moved_file_path): - try: - current_dir = os.path.dirname(moved_file_path) - while current_dir != download_path and current_dir.startswith(download_path): - is_empty = not any(not f.startswith('.') for f in os.listdir(current_dir)) - if is_empty: - os.rmdir(current_dir) - current_dir = os.path.dirname(current_dir) - else: - break - except Exception as e: - print(f"Warning: An error occurred during directory cleanup: {e}") - - -def _sweep_empty_download_directories(): - try: - download_path = docker_resolve_path(config_manager.get('soulseek.download_path', './downloads')) - if not os.path.isdir(download_path): - return 0 - - removed = 0 - for dirpath, _dirnames, _filenames in os.walk(download_path, topdown=False): - if os.path.normpath(dirpath) == os.path.normpath(download_path): - continue - try: - entries = os.listdir(dirpath) - except OSError: - continue - visible = [e for e in entries if not e.startswith('.')] - if not visible: - try: - for hidden in entries: - try: - os.remove(os.path.join(dirpath, hidden)) - except Exception: - pass - os.rmdir(dirpath) - removed += 1 - except OSError: - pass - - return removed - except Exception as e: - print(f"[Folder Cleanup] Error sweeping empty directories: {e}") - return 0 - - -def _get_audio_quality_string(file_path): - try: - ext = os.path.splitext(file_path)[1].lower() - if ext == '.flac': - audio = FLAC(file_path) - return f"FLAC {audio.info.bits_per_sample}bit" - if ext == '.mp3': - from mutagen.mp3 import MP3, BitrateMode - audio = MP3(file_path) - bitrate_kbps = audio.info.bitrate // 1000 - if audio.info.bitrate_mode == BitrateMode.VBR: - return "MP3-VBR" - return f"MP3-{bitrate_kbps}" - if ext in ('.m4a', '.aac', '.mp4'): - audio = MP4(file_path) - return f"M4A-{audio.info.bitrate // 1000}" - if ext == '.ogg': - audio = OggVorbis(file_path) - return f"OGG-{audio.info.bitrate // 1000}" - if ext == '.opus': - from mutagen.oggopus import OggOpus - audio = OggOpus(file_path) - return f"OPUS-{audio.info.bitrate // 1000}" - return '' - except Exception as e: - logger.debug(f"Could not determine audio quality for {file_path}: {e}") - return '' - - -def parse_youtube_playlist(url): - try: - import re - - def clean_youtube_artist(artist_string): - if not artist_string: - return artist_string - original_artist = artist_string - artist_string = artist_string.replace('"', '').replace("'", '') - artist_string = re.sub(r'\s*\([^)]*\)', '', artist_string).strip() - artist_string = re.sub(r'\s*\[[^\]]*\]', '', artist_string).strip() - for suffix in [ - r'\s*-\s*Topic\s*$', - r'\s*VEVO\s*$', - r'\s*Music\s*$', - r'\s*Official\s*$', - r'\s*Records\s*$', - r'\s*Entertainment\s*$', - r'\s*TV\s*$', - r'\s*Channel\s*$', - ]: - artist_string = re.sub(suffix, '', artist_string, flags=re.IGNORECASE).strip() - for sep in [',', '&', ' and ', ' x ', ' X ', ' feat.', ' ft.', ' featuring', ' with', ' vs ', ' vs.']: - if sep in artist_string: - artist_string = artist_string.split(sep)[0].strip() - break - artist_string = re.sub(r'\s+', ' ', artist_string).strip() - artist_string = re.sub(r'^\-\s*|\s*\-$', '', artist_string).strip() - artist_string = re.sub(r'^,\s*|\s*,$', '', artist_string).strip() - return artist_string or original_artist - - def clean_youtube_track_title(title, artist_name=None): - if not title: - return title - original_title = title - artist_removed = False - if artist_name and '-' in title: - artist_pattern = r'^' + re.escape(artist_name.strip()) + r'(?:\s*[&,x]\s*[^-]+)?\s*[-–—]\s*' - cleaned_title = re.sub(artist_pattern, '', title, flags=re.IGNORECASE).strip() - if cleaned_title != title: - title = cleaned_title - artist_removed = True - else: - artist_end_pattern = r'\s*[-–—]\s*' + re.escape(artist_name.strip()) + r'(?:\s*[&,x]\s*[^-]+)?\s*$' - cleaned_title = re.sub(artist_end_pattern, '', title, flags=re.IGNORECASE).strip() - if cleaned_title != title: - title = cleaned_title - artist_removed = True - title = re.sub(r'【[^】]*】', '', title) - title = re.sub(r'\s*\([^)]*\)', '', title) - title = re.sub(r'\s*\(.*$', '', title) - title = re.sub(r'\[[^\]]*\]', '', title) - title = re.sub(r'\{[^}]*\}', '', title) - title = re.sub(r'<[^>]*>', '', title) - if artist_removed: - title = re.sub(r'\s*-\s*.*$', '', title) - title = re.split(r'\s*\|\s*', title)[0].strip() - for pattern in [ - r'\bapple\s+music\b', - r'\bfull\s+video\b', - r'\bmusic\s+video\b', - r'\bofficial\s+video\b', - r'\bofficial\s+music\s+video\b', - r'\bofficial\b', - r'\bcensored\s+version\b', - r'\buncensored\s+version\b', - r'\bexplicit\s+version\b', - r'\blive\s+version\b', - r'\bversion\b', - r'\btopic\b', - r'\baudio\b', - r'\blyrics?\b', - r'\blyric\s+video\b', - r'\bwith\s+lyrics?\b', - r'\bvisuali[sz]er\b', - r'\bmv\b', - r'\bdirectors?\s+cut\b', - r'\bremaster(ed)?\b', - r'\bremix\b', - ]: - title = re.sub(pattern, '', title, flags=re.IGNORECASE) - if artist_name: - collab_pattern = rf'\b{re.escape(artist_name)}\s*[&,]\s*\w+|[\w\s]+[&,]\s*{re.escape(artist_name)}\b' - if not re.search(collab_pattern, title, flags=re.IGNORECASE): - title = re.sub(rf'\b{re.escape(artist_name)}\b', '', title, flags=re.IGNORECASE) - title = re.sub(rf'\b{re.escape(artist_name)}\s*[-–—:]\s*', '', title, flags=re.IGNORECASE) - title = re.sub(rf'^{re.escape(artist_name)}\s*[-–—:]\s*', '', title, flags=re.IGNORECASE) - title = re.sub(r'\s+prod\.?\s+\S+', '', title, flags=re.IGNORECASE) - title = re.sub(r'["\']', '', title) - for pattern in [ - r'\s+feat\.?\s+.+$', - r'\s+ft\.?\s+.+$', - r'\s+featuring\s+.+$', - r'\s+with\s+.+$', - ]: - title = re.sub(pattern, '', title, flags=re.IGNORECASE).strip() - title = re.sub(r'\s+', ' ', title).strip() - title = re.sub(r'^[-–—:,.\s]+|[-–—:,.\s]+$', '', title).strip() - return title or original_title - - ydl_opts = { - 'quiet': True, - 'no_warnings': True, - 'extract_flat': 'in_playlist', - 'skip_download': True, - 'lazy_playlist': False, - } - - with yt_dlp.YoutubeDL(ydl_opts) as ydl: - playlist_info = ydl.extract_info(url, download=False) - if not playlist_info: - return None - - playlist_name = playlist_info.get('title', 'Unknown Playlist') - playlist_id = playlist_info.get('id', 'unknown_id') - entries = list(playlist_info.get('entries', []) or []) - tracks = [] - - for entry in entries: - if not entry: - continue - raw_title = entry.get('title', 'Unknown Track') - raw_uploader = entry.get('uploader', 'Unknown Artist') - duration = entry.get('duration', 0) - video_id = entry.get('id', '') - cleaned_artist = clean_youtube_artist(raw_uploader) - cleaned_title = clean_youtube_track_title(raw_title, cleaned_artist) - tracks.append({ - 'id': video_id, - 'name': cleaned_title, - 'artists': [cleaned_artist], - 'duration_ms': duration * 1000 if duration else 0, - 'raw_title': raw_title, - 'raw_artist': raw_uploader, - 'url': f"https://www.youtube.com/watch?v={video_id}", - }) - - return { - 'id': playlist_id, - 'name': playlist_name, - 'tracks': tracks, - 'track_count': len(tracks), - 'url': url, - 'source': 'youtube', - } - except Exception as e: - print(f"Error parsing YouTube playlist: {e}") - return None - - -def get_download_status(): - if not soulseek_client: - return jsonify({"transfers": []}) - - try: - global processed_download_ids - transfers_data = run_async(soulseek_client._make_request('GET', 'transfers/downloads')) - all_transfers = [] - completed_matched_downloads = [] - _files_claimed_this_cycle = set() - - if transfers_data: - for user_data in transfers_data: - username = user_data.get('username', 'Unknown') - if 'directories' in user_data: - for directory in user_data['directories']: - if 'files' in directory: - for file_info in directory['files']: - file_info['username'] = username - all_transfers.append(file_info) - state = file_info.get('state', '').lower() - if ('succeeded' in state or 'completed' in state) and 'errored' not in state and 'rejected' not in state: - _fi_size = file_info.get('size', 0) - _fi_transferred = file_info.get('bytesTransferred', 0) - if _fi_size > 0 and _fi_transferred < _fi_size: - continue - filename_from_api = file_info.get('filename') - if not filename_from_api: - continue - - context_key = _make_context_key(username, filename_from_api) - if context_key in _orphaned_download_keys: - with matched_context_lock: - has_active_context = context_key in matched_downloads_context - if has_active_context: - _orphaned_download_keys.discard(context_key) - else: - download_dir = docker_resolve_path(config_manager.get('soulseek.download_path', './downloads')) - found_result = _find_completed_file_robust(download_dir, filename_from_api) - found_path = found_result[0] if found_result and found_result[0] else None - orphan_cleaned = False - if found_path: - try: - os.remove(found_path) - orphan_cleaned = True - except Exception as e: - print(f"Failed to delete orphaned file (will retry next poll): {e}") - else: - orphan_cleaned = True - if orphan_cleaned: - transfer_id = file_info.get('id') - if transfer_id: - try: - run_async(soulseek_client.cancel_download(str(transfer_id), username, remove=True)) - except Exception: - pass - _orphaned_download_keys.discard(context_key) - continue - - if context_key in processed_download_ids: - continue - - with matched_context_lock: - context = matched_downloads_context.get(context_key) - available_keys = list(matched_downloads_context.keys())[:5] if not context else None - - if context and context_key not in _stale_transfer_keys: - pass - - if context: - download_dir = docker_resolve_path(config_manager.get('soulseek.download_path', './downloads')) - found_result = _find_completed_file_robust(download_dir, filename_from_api) - found_path = found_result[0] if found_result and found_result[0] else None - - if found_path: - _norm_path = os.path.normpath(found_path) - if _norm_path not in _files_claimed_this_cycle: - _files_claimed_this_cycle.add(_norm_path) - completed_matched_downloads.append((context_key, context, found_path)) - with _download_retry_lock: - if context_key in _download_retry_attempts: - del _download_retry_attempts[context_key] - else: - with _download_retry_lock: - if context_key not in _download_retry_attempts: - _download_retry_attempts[context_key] = {'count': 1, 'first_attempt': time.time()} - else: - _download_retry_attempts[context_key]['count'] += 1 - retry_count = _download_retry_attempts[context_key]['count'] - if retry_count >= _download_retry_max: - processed_download_ids.add(context_key) - del _download_retry_attempts[context_key] - if completed_matched_downloads: - def process_completed_downloads(): - for context_key, context, found_path in completed_matched_downloads: - try: - _pp_task_id = context.get('task_id') - _pp_batch_id = context.get('batch_id') - if _pp_task_id and _pp_batch_id: - _pp_target = _post_process_matched_download_with_verification - _pp_args = (context_key, context, found_path, _pp_task_id, _pp_batch_id) - else: - _pp_target = _post_process_matched_download - _pp_args = (context_key, context, found_path) - thread = threading.Thread(target=_pp_target, args=_pp_args) - thread.daemon = True - thread.start() - processed_download_ids.add(context_key) - except Exception as e: - print(f"Error starting post-processing thread for {context_key}: {e}") - processing_thread = threading.Thread(target=process_completed_downloads) - processing_thread.daemon = True - processing_thread.start() - - return jsonify({"transfers": all_transfers}) - except Exception as e: - logger.error(f"Error building download status: {e}") - return jsonify({"error": str(e)}), 500 - - -def _get_windowed_calls(key, current_total): - now = time.time() - history = _enrichment_activity_log.setdefault(key, collections.deque(maxlen=17300)) - history.append((now, current_total)) - - cutoff_1h = now - 3600 - cutoff_24h = now - 86400 - oldest_1h_total = current_total - oldest_24h_total = current_total - found_24h = False - for ts, total in history: - if not found_24h and ts >= cutoff_24h: - oldest_24h_total = total - found_24h = True - if ts >= cutoff_1h: - oldest_1h_total = total - break - - return max(0, current_total - oldest_1h_total), max(0, current_total - oldest_24h_total) - - -def _get_enrichment_status(): - services = {} - workers_info = [ - ('musicbrainz', 'MusicBrainz', lambda: mb_worker), - ('spotify_enrichment', 'Spotify', lambda: spotify_enrichment_worker), - ('itunes_enrichment', 'iTunes', lambda: itunes_enrichment_worker), - ('deezer_enrichment', 'Deezer', lambda: deezer_worker), - ('tidal_enrichment', 'Tidal', lambda: tidal_enrichment_worker), - ('qobuz_enrichment', 'Qobuz', lambda: qobuz_enrichment_worker), - ('lastfm', 'Last.fm', lambda: lastfm_worker), - ('genius', 'Genius', lambda: genius_worker), - ('audiodb', 'AudioDB', lambda: audiodb_worker), - ('discogs', 'Discogs', lambda: discogs_worker), - ] - configured_checks = { - 'spotify_enrichment': lambda: bool(config_manager.get('spotify.client_id') and config_manager.get('spotify.client_secret')), - 'tidal_enrichment': lambda: bool(tidal_client and getattr(tidal_client, 'access_token', None)), - 'qobuz_enrichment': lambda: bool(qobuz_enrichment_worker and qobuz_enrichment_worker.client and qobuz_enrichment_worker.client.user_auth_token), - 'lastfm': lambda: bool(config_manager.get('lastfm.api_key', '')), - 'genius': lambda: bool(config_manager.get('genius.access_token', '')), - } - - for key, name, get_worker in workers_info: - worker = get_worker() - if worker is not None: - is_alive = worker.thread is not None and worker.thread.is_alive() - try: - configured = configured_checks.get(key, lambda: True)() - except Exception: - configured = False - - stats = worker.stats - total_processed = stats.get('matched', 0) + stats.get('not_found', 0) + stats.get('errors', 0) - calls_1h, calls_24h = _get_windowed_calls(key, total_processed) - - has_item = getattr(worker, 'current_item', None) is not None - if has_item: - _idle_since.pop(key, None) - is_idle = False - else: - if key not in _idle_since: - _idle_since[key] = time.time() - is_idle = (time.time() - _idle_since[key]) >= _IDLE_GRACE_SECONDS - - svc_data = { - 'name': name, - 'configured': configured, - 'running': worker.running and is_alive and not worker.paused, - 'paused': worker.paused, - 'idle': is_alive and not worker.paused and is_idle, - 'calls_1h': calls_1h, - 'calls_24h': calls_24h, - } - if key == 'spotify_enrichment': - try: - svc_data['daily_budget'] = worker._get_daily_budget_info() - except Exception: - pass - services[key] = svc_data - else: - services[key] = { - 'name': name, - 'configured': False, - 'running': False, - 'paused': False, - 'idle': False, - 'calls_1h': 0, - 'calls_24h': 0, - } - - services['acoustid'] = { - 'name': 'AcoustID', - 'configured': bool(config_manager.get('acoustid.api_key', '')), - } - services['listenbrainz'] = { - 'name': 'ListenBrainz', - 'configured': bool(config_manager.get('listenbrainz.token', '')), - } - return services - - -def _build_system_stats(): - try: - import psutil - except Exception: - psutil = None - - from datetime import timedelta - - start_time = getattr(app, 'start_time', time.time()) - uptime = str(timedelta(seconds=int(time.time() - start_time))) - memory_usage = '0%' - if psutil is not None: - memory_usage = f"{psutil.virtual_memory().percent}%" - - active_downloads = len([batch_id for batch_id, batch_data in download_batches.items() if batch_data.get('phase') == 'downloading']) - with session_stats_lock: - finished_downloads = session_completed_downloads - - total_download_speed = 0.0 - try: - transfers_data = run_async(soulseek_client._make_request('GET', 'transfers/downloads')) - if transfers_data: - for user_data in transfers_data: - if 'directories' in user_data: - for directory in user_data['directories']: - if 'files' in directory: - for file_info in directory['files']: - state = file_info.get('state', '').lower() - if 'inprogress' in state or 'downloading' in state or 'transferring' in state: - speed = file_info.get('averageSpeed', 0) - if isinstance(speed, (int, float)) and speed > 0: - total_download_speed += float(speed) - except Exception: - pass - - return { - 'uptime': uptime, - 'memory_usage': memory_usage, - 'active_downloads': active_downloads, - 'finished_downloads': finished_downloads, - 'download_speed': f"{total_download_speed / (1024 * 1024):.1f} MB/s" if total_download_speed > 1024 * 1024 else f"{total_download_speed / 1024:.1f} KB/s", - } - - -def _is_hydrabase_active(): - try: - if hydrabase_client is None or not hydrabase_client.is_connected(): - return False - return dev_mode_enabled - except (NameError, Exception): - return False - # --- Automation Engine --- try: automation_engine = AutomationEngine(get_database()) @@ -2221,46 +836,8 @@ except Exception as e: automation_engine = None # --- Automation Progress Tracking --- -automation_progress_states = {} # automation_id (int) -> state dict -automation_progress_lock = threading.Lock() _scan_library_automation_id = None -def _init_automation_progress(automation_id, automation_name, action_type): - """Initialize progress state when an automation starts running.""" - with automation_progress_lock: - automation_progress_states[automation_id] = { - 'status': 'running', - 'action_type': action_type, - 'progress': 0, 'phase': 'Starting...', 'current_item': '', - 'processed': 0, 'total': 0, - 'log': [{'type': 'info', 'text': f'Starting {automation_name}'}], - 'started_at': datetime.now().isoformat(), - 'finished_at': None, - } - - -def _update_automation_progress(automation_id, **kwargs): - """Update progress state from handler threads. Thread-safe.""" - if automation_id is None: - return - with automation_progress_lock: - state = automation_progress_states.get(automation_id) - if not state: - return - for k, v in kwargs.items(): - if k == 'log_line': - state['log'].append({'type': kwargs.get('log_type', 'info'), 'text': v}) - if len(state['log']) > 50: - state['log'] = state['log'][-50:] - elif k != 'log_type': - state[k] = v - # Immediate emit on finish so frontend gets final state without waiting for loop - if kwargs.get('status') in ('finished', 'error'): - state['finished_at'] = datetime.now().isoformat() - try: - socketio.emit('automation:progress', {str(automation_id): dict(state)}) - except Exception: - pass def _register_automation_handlers(): """Register real SoulSync action handlers with the automation engine.""" @@ -3842,92 +2419,18 @@ except Exception as e: # --- Global Streaming State Management --- # Thread-safe state tracking for streaming functionality -stream_state = { - "status": "stopped", # States: stopped, loading, queued, ready, error - "progress": 0, - "track_info": None, - "file_path": None, # Path to the audio file in the 'Stream' folder - "error_message": None -} -stream_lock = threading.Lock() # Prevent race conditions -stream_background_task = None -stream_executor = ThreadPoolExecutor(max_workers=1) # Only one stream at a time # --- Global OAuth State Management --- # Store PKCE values for Tidal OAuth flow -tidal_oauth_state = { - "code_verifier": None, - "code_challenge": None -} -tidal_oauth_lock = threading.Lock() -db_update_executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="DBUpdate") -db_update_worker = None -db_update_state = { - "status": "idle", # idle, running, finished, error - "phase": "Idle", - "progress": 0, - "current_item": "", - "processed": 0, - "total": 0, - "error_message": "", - "removed_artists": 0, - "removed_albums": 0, - "removed_tracks": 0 -} -_db_update_automation_id = None # Set when automation triggers DB update, used by callbacks -_scan_library_automation_id = None # Set when automation triggers library scan, used for tracking # Quality Scanner state -quality_scanner_state = { - "status": "idle", # idle, running, finished, error - "phase": "Ready to scan", - "progress": 0, - "processed": 0, - "total": 0, - "quality_met": 0, - "low_quality": 0, - "matched": 0, - "error_message": "", - "results": [] # List of low quality tracks with match status -} -quality_scanner_lock = threading.Lock() -quality_scanner_executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="QualityScanner") # Duplicate Cleaner state -duplicate_cleaner_state = { - "status": "idle", # idle, running, finished, error - "phase": "Ready to scan", - "progress": 0, - "files_scanned": 0, - "total_files": 0, - "duplicates_found": 0, - "deleted": 0, - "space_freed": 0, # in bytes - "error_message": "" -} -duplicate_cleaner_lock = threading.Lock() -duplicate_cleaner_executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="DuplicateCleaner") # --- Retag Tool Globals --- -retag_state = { - "status": "idle", - "phase": "Ready", - "progress": 0, - "current_track": "", - "total_tracks": 0, - "processed": 0, - "error_message": "" -} -retag_lock = threading.Lock() -retag_executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="RetagWorker") # --- Sync Page Globals --- -sync_executor = ThreadPoolExecutor(max_workers=3, thread_name_prefix="SyncWorker") -active_sync_workers = {} # Key: playlist_id, Value: Future object -sync_states = {} # Key: playlist_id, Value: dict with progress info -sync_lock = threading.Lock() -db_update_lock = threading.Lock() # --- Automation Progress Tracking --- automation_progress_states = {} # automation_id (int) -> state dict @@ -3972,12 +2475,9 @@ def _update_automation_progress(automation_id, **kwargs): # --- Global Matched Downloads Context Management --- # Shared with core.runtime_state so the refactored pipeline and web # server operate on the same context registry. -_orphaned_download_keys = set() # Context keys of downloads abandoned during retry # --- Download Missing Tracks Modal State Management --- # Thread-safe state tracking for modal download functionality with batch management -missing_download_executor = ThreadPoolExecutor(max_workers=3, thread_name_prefix="MissingTrackWorker") -batch_locks = {} # batch_id -> Lock() for atomic batch operations def _get_max_concurrent(): """Get configured max concurrent downloads. Default 3.""" @@ -3998,8 +2498,6 @@ def _get_batch_max_concurrent(is_album=False, source=None): # --- Session Download Statistics --- # Track individual download completions (matches dashboard.py behavior) -session_completed_downloads = 0 -session_stats_lock = threading.Lock() def _mark_task_completed(task_id, track_info=None): """ @@ -4018,22 +2516,10 @@ def _mark_task_completed(task_id, track_info=None): # Server-side timer system for automatic wishlist processing (replaces client-side JavaScript timers) # --- Automatic Wishlist/Watchlist Processing Flags --- # Processing state flags (guards/recovery — timers are now managed by AutomationEngine) -wishlist_auto_processing = False # Flag to prevent concurrent auto-processing -wishlist_auto_processing_timestamp = 0 # Timestamp when processing started (for stuck detection) -wishlist_timer_lock = threading.Lock() # Thread safety for flag operations -watchlist_auto_scanning = False # Flag to prevent concurrent auto-scanning -watchlist_auto_scanning_timestamp = 0 # Timestamp when scanning started (for stuck detection) -watchlist_timer_lock = threading.Lock() # Thread safety for flag operations # --- Shared Transfer Data Cache --- # Cache transfer data to avoid hammering the Soulseek API with multiple concurrent modals -transfer_data_cache = { - 'data': {}, - 'last_update': 0, - 'update_lock': threading.Lock(), - 'cache_duration': 0.75 # Cache for 0.75 seconds for faster updates -} def get_cached_transfer_data(): """ @@ -4115,23 +2601,6 @@ def get_cached_transfer_data(): # --- Beatport Data Cache --- # Cache Beatport scraping data to reduce load times and avoid hammering Beatport.com -beatport_data_cache = { - 'homepage': { - 'hero_tracks': {'data': None, 'timestamp': 0, 'ttl': 86400}, # 24 hours - 'top_10_lists': {'data': None, 'timestamp': 0, 'ttl': 86400}, # 24 hours - 'top_10_releases': {'data': None, 'timestamp': 0, 'ttl': 86400}, # 24 hours - 'new_releases': {'data': None, 'timestamp': 0, 'ttl': 86400}, # 24 hours - 'hype_picks': {'data': None, 'timestamp': 0, 'ttl': 86400}, # 24 hours - 'featured_charts': {'data': None, 'timestamp': 0, 'ttl': 86400}, # 24 hours - 'dj_charts': {'data': None, 'timestamp': 0, 'ttl': 86400} # 24 hours - }, - 'genre': { - # Future expansion for genre-specific caching - # 'house': {'top_10': {...}, 'releases': {...}}, - # 'techno': {'top_10': {...}, 'releases': {...}} - }, - 'cache_lock': threading.Lock() -} @@ -5242,7 +3711,6 @@ def cleanup_monitor(): logger.error(f"Error cleaning up batch locks: {e}") # Global shutdown flag -IS_SHUTTING_DOWN = False def _shutdown_executor(executor, name): """Shut down a ThreadPoolExecutor without waiting for long-running tasks.""" @@ -6536,7 +5004,6 @@ def spa_catch_all(page): # Tracks cumulative item-processed totals over time for windowed counting. # Each entry: (timestamp, cumulative_total). Polled every ~5s, 24h = ~17280 entries. -_enrichment_activity_log = {} # key -> deque of (timestamp, total) def _get_windowed_calls(key, current_total): """Record current cumulative total and return (calls_1h, calls_24h). @@ -6564,8 +5031,6 @@ def _get_windowed_calls(key, current_total): return max(0, current_total - oldest_1h_total), max(0, current_total - oldest_24h_total) -_idle_since = {} # worker_key -> timestamp when current_item first became None -_IDLE_GRACE_SECONDS = 5 # Don't report idle until current_item has been None this long def _get_enrichment_status(): """Get lightweight status for all enrichment services (no DB queries). @@ -6662,17 +5127,6 @@ def _get_enrichment_status(): # Status check caching to reduce unnecessary API calls -_status_cache = { - 'spotify': {'connected': False, 'response_time': 0, 'source': 'itunes'}, - 'media_server': {'connected': False, 'response_time': 0, 'type': None}, - 'soulseek': {'connected': False, 'response_time': 0} -} -_status_cache_timestamps = { - 'spotify': 0, - 'media_server': 0, - 'soulseek': 0 -} -STATUS_CACHE_TTL = 120 # Cache for 2 minutes (reduces API calls while staying fresh) @app.route('/status') def get_status(): @@ -7698,7 +6152,6 @@ def handle_settings(): except Exception as e: return jsonify({"error": str(e)}), 500 -dev_mode_enabled = False @app.route('/api/dev-mode', methods=['GET', 'POST']) def handle_dev_mode(): @@ -7713,8 +6166,6 @@ def handle_dev_mode(): return jsonify({"enabled": dev_mode_enabled}) # ── Hydrabase WebSocket Connection ── -_hydrabase_ws = None -_hydrabase_lock = threading.Lock() # ── Hydrabase Comparison Store ── import collections as _collections