danvics/soulsync
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Lift shared runtime state into core

Browse source 
- Move app-wide task and activity registries out of core/imports
- Share one runtime-state module across the web server, API, and import pipeline
- Keep import-specific helpers focused on context and post-processing
This commit is contained in:
Antti Kettunen 2026-04-25 13:11:55 +03:00
parent e10df4caf2
commit bdef127dd6
No known key found for this signature in database
GPG key ID: C6B2A3D250359BD7
9 changed files with 154 additions and 164 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
api/downloads.py
View file

@ -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):

4
api/system.py
View file

@ -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()

46
core/imports/context.py
View file

@ -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

26
core/imports/pipeline.py
View file

@ -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):

121
core/imports/runtime_state.py
View file

@ -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

65
core/runtime_state.py Normal file
View file

@ -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

18
tests/imports/test_import_pipeline.py
View file

@ -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)

2
tests/test_downloads_pagination.py
View file

@ -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):

34
web_server.py
View file

@ -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 = (