diff --git a/api/downloads.py b/api/downloads.py index 88fd690c..2f087874 100644 --- a/api/downloads.py +++ b/api/downloads.py @@ -5,7 +5,7 @@ Download management endpoints — list, cancel active downloads. from flask import request, current_app from .auth import require_api_key from .helpers import api_success, api_error -from core.imports.runtime_state import download_tasks, tasks_lock +from core.runtime_state import download_tasks, tasks_lock def _serialize_download(task_id, task): diff --git a/api/system.py b/api/system.py index 9bdf107b..290da76e 100644 --- a/api/system.py +++ b/api/system.py @@ -55,7 +55,7 @@ def register_routes(bp): def system_activity(): """Recent activity feed.""" try: - from core.imports.runtime_state import activity_feed + from core.runtime_state import activity_feed items = list(activity_feed) if activity_feed else [] return api_success({"activities": items}) except Exception as e: @@ -74,7 +74,7 @@ def register_routes(bp): # Active download count download_count = 0 try: - from core.imports.runtime_state import download_tasks, tasks_lock + from core.runtime_state import download_tasks, tasks_lock with tasks_lock: download_count = sum( 1 for t in download_tasks.values() diff --git a/core/imports/context.py b/core/imports/context.py index cf9504ff..c19aa180 100644 --- a/core/imports/context.py +++ b/core/imports/context.py @@ -324,3 +324,49 @@ def build_import_album_info( "album_type": album_type, "total_tracks": int(total_tracks) if str(total_tracks).isdigit() else total_tracks, } + + +def detect_album_info_web(context, artist_context=None): + """Best-effort album detection for single-track downloads.""" + context = normalize_import_context(context) + if artist_context is None: + artist_context = context.get("artist") or {} + + album_info = build_import_album_info(context) + if album_info.get("is_album"): + return album_info + + album_ctx = get_import_context_album(context) + track_info = get_import_track_info(context) + original_search = get_import_original_search(context) + + album_name = ( + album_ctx.get("name") + or track_info.get("album") + or original_search.get("album") + or "" + ) + track_name = ( + track_info.get("name") + or original_search.get("title") + or "" + ) + artist_name = extract_artist_name(artist_context) or get_import_clean_artist(context, default="") + + if album_name and track_name and album_name.strip().lower() not in { + track_name.strip().lower(), + artist_name.strip().lower(), + }: + return build_import_album_info( + context, + album_info={ + "album_name": album_name, + "track_number": track_info.get("track_number", 1), + "disc_number": track_info.get("disc_number", 1), + "album_image_url": album_ctx.get("image_url", ""), + "confidence": 0.5, + }, + force_album=True, + ) + + return None diff --git a/core/imports/pipeline.py b/core/imports/pipeline.py index 99eee1a5..68429c85 100644 --- a/core/imports/pipeline.py +++ b/core/imports/pipeline.py @@ -18,6 +18,7 @@ from core.imports.file_ops import ( ) from core.imports.context import ( build_import_album_info, + detect_album_info_web, extract_artist_name, get_import_clean_artist, get_import_clean_title, @@ -38,17 +39,16 @@ from core.imports.side_effects import ( record_retag_download, record_soulsync_library_entry, ) -from core.imports.runtime_state import ( +from core.runtime_state import ( add_activity_item, - detect_album_info_web, download_batches, download_tasks, matched_context_lock, matched_downloads_context, mark_task_completed as _mark_task_completed, - _post_process_locks, - _post_process_locks_lock, - _processed_download_ids, + post_process_locks, + post_process_locks_lock, + processed_download_ids, tasks_lock, ) from core.metadata_artwork import download_cover_art @@ -79,10 +79,10 @@ def post_process_matched_download(context_key, context, file_path, runtime): if on_download_completed: on_download_completed(batch_id, task_id, success=success) - with _post_process_locks_lock: - if context_key not in _post_process_locks: - _post_process_locks[context_key] = threading.Lock() - file_lock = _post_process_locks[context_key] + with post_process_locks_lock: + if context_key not in post_process_locks: + post_process_locks[context_key] = threading.Lock() + file_lock = post_process_locks[context_key] file_lock.acquire() try: @@ -747,8 +747,8 @@ def post_process_matched_download(context_key, context, file_path, runtime): source_exists = os.path.exists(file_path) if file_path else False if source_exists: - if context_key in _processed_download_ids: - _processed_download_ids.remove(context_key) + if context_key in processed_download_ids: + processed_download_ids.remove(context_key) logger.warning(f"Removed {context_key} from processed set - will retry on next check") with matched_context_lock: if context_key not in matched_downloads_context: @@ -758,8 +758,8 @@ def post_process_matched_download(context_key, context, file_path, runtime): logger.warning(f"Source file gone, not retrying: {context_key}") finally: file_lock.release() - with _post_process_locks_lock: - _post_process_locks.pop(context_key, None) + with post_process_locks_lock: + post_process_locks.pop(context_key, None) def post_process_matched_download_with_verification(context_key, context, file_path, task_id, batch_id, runtime): diff --git a/core/imports/runtime_state.py b/core/imports/runtime_state.py deleted file mode 100644 index 4e24d06f..00000000 --- a/core/imports/runtime_state.py +++ /dev/null @@ -1,121 +0,0 @@ -"""Shared runtime state and tiny helpers for import/post-processing code.""" - -from __future__ import annotations - -import threading -import time -from typing import Any, Dict, Optional - -from core.imports.context import ( - build_import_album_info, - extract_artist_name, - get_import_clean_artist, - get_import_context_album, - get_import_original_search, - get_import_track_info, - normalize_import_context, -) - -matched_context_lock = threading.Lock() -matched_downloads_context: Dict[str, Dict[str, Any]] = {} -tasks_lock = threading.Lock() -download_tasks: Dict[str, Dict[str, Any]] = {} -download_batches: Dict[str, Dict[str, Any]] = {} -_processed_download_ids = set() -_post_process_locks: Dict[str, threading.Lock] = {} -_post_process_locks_lock = threading.Lock() - -activity_feed = [] -activity_feed_lock = threading.Lock() -_activity_toast_emitter = None - - -def set_activity_toast_emitter(emitter) -> None: - """Set the WebSocket-style emitter used by add_activity_item.""" - global _activity_toast_emitter - _activity_toast_emitter = emitter - - -def add_activity_item(icon, title, subtitle, time_ago="Now", show_toast=True): - """Append an activity item and emit a toast if an emitter is configured.""" - activity_item = { - "icon": icon, - "title": title, - "subtitle": subtitle, - "time": time_ago, - "timestamp": time.time(), - "show_toast": show_toast, - } - with activity_feed_lock: - activity_feed.append(activity_item) - if len(activity_feed) > 20: - activity_feed.pop(0) - - if show_toast and _activity_toast_emitter is not None: - try: - _activity_toast_emitter("dashboard:toast", activity_item) - except Exception: - pass - - return activity_item - - -def mark_task_completed(task_id: str, track_info: Optional[Dict[str, Any]] = None) -> bool: - """Mark a download task as completed in the shared task registry.""" - with tasks_lock: - task = download_tasks.get(task_id) - if not task: - return False - - task["status"] = "completed" - task["stream_processed"] = True - task["status_change_time"] = time.time() - if track_info is not None: - task["track_info"] = track_info - return True - - -def detect_album_info_web(context, artist_context=None): - """Best-effort album detection for single-track downloads.""" - context = normalize_import_context(context) - if artist_context is None: - artist_context = context.get("artist") or {} - - album_info = build_import_album_info(context) - if album_info.get("is_album"): - return album_info - - album_ctx = get_import_context_album(context) - track_info = get_import_track_info(context) - original_search = get_import_original_search(context) - - album_name = ( - album_ctx.get("name") - or track_info.get("album") - or original_search.get("album") - or "" - ) - track_name = ( - track_info.get("name") - or original_search.get("title") - or "" - ) - artist_name = extract_artist_name(artist_context) or get_import_clean_artist(context, default="") - - if album_name and track_name and album_name.strip().lower() not in { - track_name.strip().lower(), - artist_name.strip().lower(), - }: - return build_import_album_info( - context, - album_info={ - "album_name": album_name, - "track_number": track_info.get("track_number", 1), - "disc_number": track_info.get("disc_number", 1), - "album_image_url": album_ctx.get("image_url", ""), - "confidence": 0.5, - }, - force_album=True, - ) - - return None diff --git a/core/runtime_state.py b/core/runtime_state.py new file mode 100644 index 00000000..8e4f4b64 --- /dev/null +++ b/core/runtime_state.py @@ -0,0 +1,65 @@ +"""Shared runtime state and tiny helpers for the app.""" + +from __future__ import annotations + +import threading +import time +from typing import Any, Dict, Optional + +matched_context_lock = threading.Lock() +matched_downloads_context: Dict[str, Dict[str, Any]] = {} +tasks_lock = threading.Lock() +download_tasks: Dict[str, Dict[str, Any]] = {} +download_batches: Dict[str, Dict[str, Any]] = {} +processed_download_ids = set() +post_process_locks: Dict[str, threading.Lock] = {} +post_process_locks_lock = threading.Lock() + +activity_feed = [] +activity_feed_lock = threading.Lock() +_activity_toast_emitter = None + + +def set_activity_toast_emitter(emitter) -> None: + """Set the WebSocket-style emitter used by add_activity_item.""" + global _activity_toast_emitter + _activity_toast_emitter = emitter + + +def add_activity_item(icon, title, subtitle, time_ago="Now", show_toast=True): + """Append an activity item and emit a toast if an emitter is configured.""" + activity_item = { + "icon": icon, + "title": title, + "subtitle": subtitle, + "time": time_ago, + "timestamp": time.time(), + "show_toast": show_toast, + } + with activity_feed_lock: + activity_feed.append(activity_item) + if len(activity_feed) > 20: + activity_feed.pop(0) + + if show_toast and _activity_toast_emitter is not None: + try: + _activity_toast_emitter("dashboard:toast", activity_item) + except Exception: + pass + + return activity_item + + +def mark_task_completed(task_id: str, track_info: Optional[Dict[str, Any]] = None) -> bool: + """Mark a download task as completed in the shared task registry.""" + with tasks_lock: + task = download_tasks.get(task_id) + if not task: + return False + + task["status"] = "completed" + task["stream_processed"] = True + task["status_change_time"] = time.time() + if track_info is not None: + task["track_info"] = track_info + return True diff --git a/tests/imports/test_import_pipeline.py b/tests/imports/test_import_pipeline.py index 7c50c942..20b71692 100644 --- a/tests/imports/test_import_pipeline.py +++ b/tests/imports/test_import_pipeline.py @@ -4,7 +4,7 @@ import types import core.imports.pipeline as import_pipeline import core.imports.paths as import_paths -import core.imports.runtime_state as runtime_state +import core.runtime_state as runtime_state class _Config: @@ -63,14 +63,14 @@ def test_verification_wrapper_handles_simple_download(tmp_path, monkeypatch): original_matched_context = dict(runtime_state.matched_downloads_context) original_download_tasks = dict(runtime_state.download_tasks) original_download_batches = dict(runtime_state.download_batches) - original_processed_ids = set(runtime_state._processed_download_ids) - original_post_process_locks = dict(runtime_state._post_process_locks) + original_processed_ids = set(runtime_state.processed_download_ids) + original_post_locks = dict(runtime_state.post_process_locks) runtime_state.matched_downloads_context.clear() runtime_state.download_tasks.clear() runtime_state.download_batches.clear() - runtime_state._processed_download_ids.clear() - runtime_state._post_process_locks.clear() + runtime_state.processed_download_ids.clear() + runtime_state.post_process_locks.clear() runtime = types.SimpleNamespace( automation_engine=None, @@ -122,7 +122,7 @@ def test_verification_wrapper_handles_simple_download(tmp_path, monkeypatch): runtime_state.download_tasks.update(original_download_tasks) runtime_state.download_batches.clear() runtime_state.download_batches.update(original_download_batches) - runtime_state._processed_download_ids.clear() - runtime_state._processed_download_ids.update(original_processed_ids) - runtime_state._post_process_locks.clear() - runtime_state._post_process_locks.update(original_post_process_locks) + runtime_state.processed_download_ids.clear() + runtime_state.processed_download_ids.update(original_processed_ids) + runtime_state.post_process_locks.clear() + runtime_state.post_process_locks.update(original_post_locks) diff --git a/tests/test_downloads_pagination.py b/tests/test_downloads_pagination.py index 24340b66..2a9c9f05 100644 --- a/tests/test_downloads_pagination.py +++ b/tests/test_downloads_pagination.py @@ -44,7 +44,7 @@ _install_flask_limiter_stub() from flask import Flask, Blueprint # noqa: E402 from api import downloads as downloads_mod # noqa: E402 -import core.imports.runtime_state as runtime_state # noqa: E402 +import core.runtime_state as runtime_state # noqa: E402 def _make_task(status="downloading", when=None): diff --git a/web_server.py b/web_server.py index 6d7da994..daa7c687 100644 --- a/web_server.py +++ b/web_server.py @@ -133,7 +133,7 @@ from core.imports.staging import ( ) from core.imports.paths import build_final_path_for_track as _build_final_path_for_track from core.metadata_common import get_file_lock -from core.imports.runtime_state import ( +from core.runtime_state import ( activity_feed, activity_feed_lock, add_activity_item, @@ -142,6 +142,7 @@ from core.imports.runtime_state import ( matched_context_lock, matched_downloads_context, mark_task_completed as _core_mark_task_completed, + processed_download_ids, set_activity_toast_emitter, tasks_lock, ) @@ -747,7 +748,7 @@ retag_executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="RetagWork # Download Missing Tracks Modal State Management # Thread-safe state tracking for modal download functionality. -# Shared task/batch state now lives in core.imports.runtime_state. +# Shared task/batch state now lives in core.runtime_state. missing_download_executor = ThreadPoolExecutor(max_workers=3, thread_name_prefix="MissingTrackWorker") # Automatic Wishlist / Watchlist Processing Flags @@ -1007,7 +1008,6 @@ session_completed_downloads = 0 session_stats_lock = threading.Lock() batch_locks = {} -_processed_download_ids = set() _orphaned_download_keys = set() _mb_release_cache = {} @@ -1940,7 +1940,7 @@ def get_download_status(): return jsonify({"transfers": []}) try: - global _processed_download_ids + global processed_download_ids transfers_data = run_async(soulseek_client._make_request('GET', 'transfers/downloads')) all_transfers = [] completed_matched_downloads = [] @@ -1994,7 +1994,7 @@ def get_download_status(): _orphaned_download_keys.discard(context_key) continue - if context_key in _processed_download_ids: + if context_key in processed_download_ids: continue with matched_context_lock: @@ -2025,7 +2025,7 @@ def get_download_status(): _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) + processed_download_ids.add(context_key) del _download_retry_attempts[context_key] if completed_matched_downloads: def process_completed_downloads(): @@ -2042,7 +2042,7 @@ def get_download_status(): thread = threading.Thread(target=_pp_target, args=_pp_args) thread.daemon = True thread.start() - _processed_download_ids.add(context_key) + 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) @@ -3967,7 +3967,7 @@ def _update_automation_progress(automation_id, **kwargs): pass # --- Global Matched Downloads Context Management --- -# Shared with core.imports.runtime_state so the refactored pipeline and web +# 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 @@ -11755,7 +11755,7 @@ def get_download_status(): return jsonify({"transfers": []}) try: - global _processed_download_ids + global processed_download_ids # Skip slskd API call if Soulseek is already known to be disconnected soulseek_known_down = not _status_cache.get('soulseek', {}).get('connected', True) transfers_data = None @@ -11835,7 +11835,7 @@ def get_download_status(): continue # Skip normal post-processing either way # Skip downloads we've already processed (prevents log spam) - if context_key in _processed_download_ids: + if context_key in processed_download_ids: continue with matched_context_lock: @@ -11865,7 +11865,7 @@ def get_download_status(): _files_claimed_this_cycle.add(_norm_path) logger.info(f"Found completed matched file on disk: {found_path}") completed_matched_downloads.append((context_key, context, found_path)) - # Don't add to _processed_download_ids yet - wait until thread starts successfully + # Don't add to processed_download_ids yet - wait until thread starts successfully # Clean up retry tracking if file was found after retries with _download_retry_lock: @@ -11894,7 +11894,7 @@ def get_download_status(): if retry_count >= _download_retry_max: # Max retries reached, give up logger.error(f"CRITICAL: Could not find '{os.path.basename(filename_from_api)}' after {retry_count} attempts over {elapsed:.1f}s. Giving up.") - _processed_download_ids.add(context_key) + processed_download_ids.add(context_key) # Clean up retry tracking del _download_retry_attempts[context_key] else: @@ -11921,7 +11921,7 @@ def get_download_status(): thread.start() # Only mark as processed AFTER thread starts successfully - _processed_download_ids.add(context_key) + processed_download_ids.add(context_key) logger.info(f"Marked as processed: {context_key}") # DON'T remove context immediately - verification worker needs it @@ -11968,7 +11968,7 @@ def get_download_status(): with matched_context_lock: context = matched_downloads_context.get(context_key) - if context and context_key not in _processed_download_ids: + if context and context_key not in processed_download_ids: download_dir = docker_resolve_path(config_manager.get('soulseek.download_path', './downloads')) found_result = _find_completed_file_robust(download_dir, download.filename) found_path = found_result[0] if found_result and found_result[0] else None @@ -11997,7 +11997,7 @@ def get_download_status(): thread = threading.Thread(target=_st_target, args=_st_args) thread.daemon = True thread.start() - _processed_download_ids.add(_ctx_key) + processed_download_ids.add(_ctx_key) logger.info(f"[{_label}] Marked as processed: {_ctx_key}") except Exception as e: logger.error(f"[{_label}] Error starting post-processing thread for {_ctx_key}: {e}") @@ -12008,7 +12008,7 @@ def get_download_status(): else: # File not found - likely already processed and moved to library # Mark as processed to prevent infinite checking - _processed_download_ids.add(context_key) + processed_download_ids.add(context_key) except Exception as streaming_error: import traceback logger.error(f"Could not fetch YouTube/Tidal downloads for status: {streaming_error}") @@ -18630,7 +18630,7 @@ def _search_track_in_album_context_web(context: dict, spotify_artist: dict) -> d if similarity > threshold: logger.info(f"FOUND: '{track_name}' (track #{track_number}) matches '{clean_track}' (similarity: {similarity:.2f})") - # Classify as album vs single using same logic as _detect_album_info_web + # Classify as album vs single using the shared detect_album_info_web helper ctx_album_type = getattr(album, 'album_type', 'album') or 'album' ctx_total_tracks = getattr(album, 'total_tracks', 1) or 1 ctx_is_album = (