mirror of https://github.com/Nezreka/SoulSync.git
First lift in the new PR5 discovery-workers series. Pulls the 448-line playlist sync background worker out of `web_server.py` into its own focused module under `core/discovery/`. Pure 1:1 lift — wrappers keep the original entry-point name so the four callers (`sync_executor.submit(_run_sync_task, ...)`) continue to work without changes. What the sync worker does: 1. Convert frontend JSON tracks → SpotifyTrack/SpotifyPlaylist objects. 2. Normalize artist/album shapes for downstream wishlist parity. 3. Wire a progress_callback that updates `sync_states` + automation card. 4. Patch sync_service for database-only fallback when no media server is connected. 5. `run_async(sync_service.sync_playlist(...))` and capture the result. 6. Update sync_states to 'finished', push playlist poster image to Plex / Jellyfin / Emby, record sync history (with re-sync vs new-sync branching), emit `playlist_synced` event for automation engine, and persist sync status with a tracks_hash for smart-skip on the next scheduled sync. 7. On exception → mark error in sync_states + automation; finally clear progress callback + drop `_original_tracks_map` from sync_service. Dependencies injected via `SyncDeps` (11 fields) — config_manager, sync_service, plex_client, jellyfin_client, automation_engine, run_async, record_sync_history_start, update_automation_progress, update_and_save_sync_status, sync_states dict, sync_lock. The only structural drift from a pure paste is the top-of-function variable binding: original used `global sync_states, sync_service`, lifted version rebinds them as locals from deps (`sync_states = deps.sync_states` etc.) since the names aren't module-level in the new file. Same behaviour otherwise — diff against the original after `deps.X` → global X normalization is **zero differences**. Tests: 18 new under tests/discovery/test_discovery_sync.py covering sync history recording (new + resync), setup error path (with and without automation_id), missing sync_service handling, sync_playlist exception handling, successful sync state transition, unmatched-tracks summary, playlist image upload (plex + jellyfin + zero-synced gate), automation engine emit, automation progress finished call, sync history DB persistence (completion + match_details), tracks_hash persistence, and finally-block cleanup (callback clear + map drop). Full suite: 1068 passing (was 1050). Ruff clean. Kicks off the PR5 series — 9 discovery workers totaling ~2,400 lines across `_run_sync_task`, `_run_*_discovery_worker` family, `_run_quality_scanner`, and `_process_watchlist_scan_automatically`. Wishlist-related extractions deliberately skipped to avoid overlap with kettui's planned `core/wishlist/` package.pull/404/head
Broque Thomas
1 month ago
parent
ed26e0f726
commit
bdb7a3139d
@ -0,0 +1,502 @@
|
||||
"""Background worker for the playlist sync task.
|
||||
|
||||
`run_sync_task(playlist_id, playlist_name, tracks_json, automation_id, profile_id,
|
||||
playlist_image_url, deps)` is the function `sync_executor.submit(...)` invokes
|
||||
to drive the entire playlist-sync workflow:
|
||||
|
||||
1. Convert frontend JSON tracks → SpotifyTrack/SpotifyPlaylist objects.
|
||||
2. Normalize artist/album shapes for downstream wishlist parity.
|
||||
3. Wire a progress_callback that updates `sync_states` + automation card.
|
||||
4. Patch sync_service for database-only fallback when no media server is connected.
|
||||
5. `run_async(sync_service.sync_playlist(...))` and capture the result.
|
||||
6. Update sync_states to 'finished', push playlist poster image to Plex/Jellyfin/Emby,
|
||||
record sync history (with re-sync vs new-sync branching), emit
|
||||
`playlist_synced` event for automation engine, and persist sync status with a
|
||||
tracks_hash for smart-skip on the next scheduled sync.
|
||||
7. On exception → mark error in sync_states + automation; finally clear progress
|
||||
callback + drop `_original_tracks_map` from sync_service.
|
||||
|
||||
Lifted verbatim from web_server.py. Wide dependency surface (sync_service,
|
||||
sync_states, plex/jellyfin clients, automation engine, multiple helper funcs)
|
||||
all injected via `SyncDeps`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Callable
|
||||
|
||||
from core.spotify_client import Playlist as SpotifyPlaylist, Track as SpotifyTrack
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class SyncDeps:
|
||||
"""Bundle of cross-cutting deps the sync worker needs."""
|
||||
config_manager: Any
|
||||
sync_service: Any
|
||||
plex_client: Any
|
||||
jellyfin_client: Any
|
||||
automation_engine: Any
|
||||
run_async: Callable[..., Any]
|
||||
record_sync_history_start: Callable
|
||||
update_automation_progress: Callable
|
||||
update_and_save_sync_status: Callable
|
||||
sync_states: dict
|
||||
sync_lock: Any # threading.Lock
|
||||
|
||||
|
||||
def run_sync_task(playlist_id, playlist_name, tracks_json, automation_id=None, profile_id=1, playlist_image_url='', deps: SyncDeps = None):
|
||||
"""The actual sync function that runs in the background thread."""
|
||||
sync_states = deps.sync_states
|
||||
sync_lock = deps.sync_lock
|
||||
sync_service = deps.sync_service
|
||||
|
||||
task_start_time = time.time()
|
||||
logger.info(f"[TIMING] _run_sync_task STARTED for playlist '{playlist_name}' at {time.strftime('%H:%M:%S')}")
|
||||
logger.info(f"Received {len(tracks_json)} tracks from frontend")
|
||||
|
||||
# Record sync history start (skip for re-syncs triggered from history)
|
||||
_is_resync = playlist_id.startswith('resync_')
|
||||
_resync_entry_id = None
|
||||
sync_batch_id = f"sync_{playlist_id}_{int(time.time())}"
|
||||
if _is_resync:
|
||||
# Extract the original entry ID from resync_{entryId}_{timestamp}
|
||||
try:
|
||||
_resync_entry_id = int(playlist_id.split('_')[1])
|
||||
except (IndexError, ValueError):
|
||||
pass
|
||||
else:
|
||||
deps.record_sync_history_start(
|
||||
batch_id=sync_batch_id,
|
||||
playlist_id=playlist_id,
|
||||
playlist_name=playlist_name,
|
||||
tracks=tracks_json,
|
||||
is_album_download=False,
|
||||
album_context=None,
|
||||
artist_context=None,
|
||||
playlist_folder_mode=False,
|
||||
source_page='sync'
|
||||
)
|
||||
|
||||
try:
|
||||
# Recreate a Playlist object from the JSON data sent by the frontend
|
||||
# This avoids needing to re-fetch it from Spotify
|
||||
logger.info("Converting JSON tracks to SpotifyTrack objects...")
|
||||
|
||||
# Store original track data with full album objects (for wishlist with cover art)
|
||||
# Normalize formats for wishlist: album must be dict {'name': ...}, artists must be [{'name': ...}]
|
||||
# Important: copy data — don't mutate tracks_json since SpotifyTrack expects List[str] artists
|
||||
original_tracks_map = {}
|
||||
for t in tracks_json:
|
||||
track_id = t.get('id', '')
|
||||
if track_id:
|
||||
normalized = dict(t)
|
||||
# Normalize album to dict format, preserving images and metadata
|
||||
raw_album = normalized.get('album', '')
|
||||
if isinstance(raw_album, str):
|
||||
normalized['album'] = {
|
||||
'name': raw_album or normalized.get('name', 'Unknown Album'),
|
||||
'images': [], 'album_type': 'single', 'total_tracks': 1, 'release_date': ''
|
||||
}
|
||||
elif not isinstance(raw_album, dict):
|
||||
normalized['album'] = {
|
||||
'name': str(raw_album) if raw_album else normalized.get('name', 'Unknown Album'),
|
||||
'images': [], 'album_type': 'single', 'total_tracks': 1, 'release_date': ''
|
||||
}
|
||||
else:
|
||||
# Dict — ensure required keys exist
|
||||
raw_album.setdefault('name', 'Unknown Album')
|
||||
raw_album.setdefault('images', [])
|
||||
# Normalize artists to list of dicts
|
||||
raw_artists = normalized.get('artists', [])
|
||||
if raw_artists and isinstance(raw_artists[0], str):
|
||||
normalized['artists'] = [{'name': a} for a in raw_artists]
|
||||
original_tracks_map[track_id] = normalized
|
||||
|
||||
tracks = []
|
||||
for i, t in enumerate(tracks_json):
|
||||
# Handle album field - extract name if it's a dictionary
|
||||
raw_album = t.get('album', '')
|
||||
if isinstance(raw_album, dict) and 'name' in raw_album:
|
||||
album_name = raw_album['name']
|
||||
elif isinstance(raw_album, str):
|
||||
album_name = raw_album
|
||||
else:
|
||||
album_name = str(raw_album)
|
||||
|
||||
# Extract image URL from album data if available
|
||||
_track_image = ''
|
||||
if isinstance(raw_album, dict):
|
||||
_imgs = raw_album.get('images', [])
|
||||
if _imgs and isinstance(_imgs, list) and len(_imgs) > 0:
|
||||
_track_image = _imgs[0].get('url', '') if isinstance(_imgs[0], dict) else ''
|
||||
if not _track_image:
|
||||
_track_image = t.get('image_url', '')
|
||||
|
||||
# Create SpotifyTrack objects with proper default values for missing fields
|
||||
track = SpotifyTrack(
|
||||
id=t.get('id', ''),
|
||||
name=t.get('name', ''),
|
||||
artists=t.get('artists', []),
|
||||
album=album_name,
|
||||
duration_ms=t.get('duration_ms', 0),
|
||||
popularity=t.get('popularity', 0),
|
||||
preview_url=t.get('preview_url'),
|
||||
external_urls=t.get('external_urls'),
|
||||
image_url=_track_image or None
|
||||
)
|
||||
tracks.append(track)
|
||||
if i < 3: # Log first 3 tracks for debugging
|
||||
logger.info(f" Track {i+1}: '{track.name}' by {track.artists}")
|
||||
|
||||
logger.info(f"Created {len(tracks)} SpotifyTrack objects")
|
||||
|
||||
playlist = SpotifyPlaylist(
|
||||
id=playlist_id,
|
||||
name=playlist_name,
|
||||
description=None, # Not needed for sync
|
||||
owner="web_user", # Placeholder
|
||||
public=False, # Default
|
||||
collaborative=False, # Default
|
||||
tracks=tracks,
|
||||
total_tracks=len(tracks)
|
||||
)
|
||||
logger.info(f"Created SpotifyPlaylist object: '{playlist.name}' with {playlist.total_tracks} tracks")
|
||||
|
||||
first_callback_time = [None] # Use list to allow modification in nested function
|
||||
|
||||
def progress_callback(progress):
|
||||
"""Callback to update the shared state."""
|
||||
if first_callback_time[0] is None:
|
||||
first_callback_time[0] = time.time()
|
||||
first_callback_duration = (first_callback_time[0] - task_start_time) * 1000
|
||||
logger.info(f"⏱️ [TIMING] FIRST progress callback at {time.strftime('%H:%M:%S')} (took {first_callback_duration:.1f}ms from start)")
|
||||
|
||||
logger.info(f"PROGRESS CALLBACK: {progress.current_step} - {progress.current_track}")
|
||||
logger.error(f" Progress: {progress.progress}% ({progress.matched_tracks}/{progress.total_tracks} matched, {progress.failed_tracks} failed)")
|
||||
|
||||
with sync_lock:
|
||||
sync_states[playlist_id] = {
|
||||
"status": "syncing",
|
||||
"progress": progress.__dict__ # Convert dataclass to dict
|
||||
}
|
||||
logger.info(f" Updated sync_states for {playlist_id}")
|
||||
|
||||
# Update automation progress card
|
||||
if automation_id:
|
||||
step = getattr(progress, 'current_step', '')
|
||||
track = getattr(progress, 'current_track', '')
|
||||
pct = getattr(progress, 'progress', 0)
|
||||
matched = getattr(progress, 'matched_tracks', 0)
|
||||
failed = getattr(progress, 'failed_tracks', 0)
|
||||
total = getattr(progress, 'total_tracks', 0)
|
||||
log_type = 'success' if 'matched' in step.lower() or 'found' in step.lower() else 'info'
|
||||
if 'not found' in step.lower() or 'failed' in step.lower():
|
||||
log_type = 'error'
|
||||
deps.update_automation_progress(automation_id, progress=pct,
|
||||
phase=f'Syncing: {step}',
|
||||
processed=matched + failed, total=total,
|
||||
current_item=track,
|
||||
log_line=f'{track} — {step}' if track else step, log_type=log_type)
|
||||
|
||||
except Exception as setup_error:
|
||||
logger.error(f"SETUP ERROR in _run_sync_task: {setup_error}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
with sync_lock:
|
||||
sync_states[playlist_id] = {
|
||||
"status": "error",
|
||||
"error": f"Setup error: {str(setup_error)}"
|
||||
}
|
||||
if automation_id:
|
||||
deps.update_automation_progress(automation_id, status='error', progress=100,
|
||||
phase='Error', log_line=f'Setup error: {str(setup_error)}', log_type='error')
|
||||
return
|
||||
|
||||
try:
|
||||
logger.info("Setting up sync service...")
|
||||
logger.info(f" sync_service available: {sync_service is not None}")
|
||||
|
||||
if sync_service is None:
|
||||
raise Exception("sync_service is None - not initialized properly")
|
||||
|
||||
# Check sync service components
|
||||
logger.info(f" spotify_client: {sync_service.spotify_client is not None}")
|
||||
logger.info(f" deps.plex_client: {sync_service.plex_client is not None}")
|
||||
logger.info(f" deps.jellyfin_client: {sync_service.jellyfin_client is not None}")
|
||||
|
||||
# Check media server connection before starting
|
||||
from config.settings import config_manager
|
||||
active_server = config_manager.get_active_media_server()
|
||||
logger.info(f" Active media server: {active_server}")
|
||||
|
||||
media_client, server_type = sync_service._get_active_media_client()
|
||||
logger.info(f" Media client available: {media_client is not None}")
|
||||
|
||||
if media_client:
|
||||
is_connected = media_client.is_connected()
|
||||
logger.info(f" Media client connected: {is_connected}")
|
||||
|
||||
# Check database access
|
||||
try:
|
||||
from database.music_database import MusicDatabase
|
||||
db = MusicDatabase()
|
||||
logger.debug(f" Database initialized: {db is not None}")
|
||||
except Exception as db_error:
|
||||
logger.error(f" Database initialization failed: {db_error}")
|
||||
|
||||
logger.info("Attaching progress callback...")
|
||||
# Attach the progress callback
|
||||
sync_service.set_progress_callback(progress_callback, playlist.name)
|
||||
logger.info(f"Progress callback attached for playlist: {playlist.name}")
|
||||
|
||||
# CRITICAL FIX: Add database-only fallback for web context
|
||||
# If media client is not connected, patch the sync service to use database-only matching
|
||||
if media_client is None or not media_client.is_connected():
|
||||
logger.info("Media client not connected - patching sync service for database-only matching")
|
||||
|
||||
# Store original method
|
||||
original_find_track = sync_service._find_track_in_media_server
|
||||
|
||||
# Create database-only replacement method
|
||||
async def database_only_find_track(spotify_track):
|
||||
logger.info(f"Database-only search for: '{spotify_track.name}' by {spotify_track.artists}")
|
||||
try:
|
||||
from database.music_database import MusicDatabase
|
||||
from config.settings import config_manager
|
||||
|
||||
db = MusicDatabase()
|
||||
active_server = config_manager.get_active_media_server()
|
||||
original_title = spotify_track.name
|
||||
spotify_id = getattr(spotify_track, 'id', '') or ''
|
||||
|
||||
# --- Sync match cache fast-path ---
|
||||
if spotify_id:
|
||||
try:
|
||||
cached = db.read_sync_match_cache(spotify_id, active_server)
|
||||
if cached:
|
||||
db_track_check = db.get_track_by_id(cached['server_track_id'])
|
||||
if db_track_check:
|
||||
class DatabaseTrackCached:
|
||||
def __init__(self, db_t):
|
||||
self.ratingKey = db_t.id
|
||||
self.title = db_t.title
|
||||
self.id = db_t.id
|
||||
logger.debug(f"Sync cache hit: '{original_title}' → server track {cached['server_track_id']}")
|
||||
return DatabaseTrackCached(db_track_check), cached['confidence']
|
||||
logger.warning(f"Sync cache stale for '{original_title}' — track gone")
|
||||
except Exception:
|
||||
pass
|
||||
# --- End cache fast-path ---
|
||||
|
||||
# Try each artist (same logic as original)
|
||||
for artist in spotify_track.artists:
|
||||
# Extract artist name from both string and dict formats
|
||||
if isinstance(artist, str):
|
||||
artist_name = artist
|
||||
elif isinstance(artist, dict) and 'name' in artist:
|
||||
artist_name = artist['name']
|
||||
else:
|
||||
artist_name = str(artist)
|
||||
|
||||
db_track, confidence = db.check_track_exists(
|
||||
original_title, artist_name,
|
||||
confidence_threshold=0.80,
|
||||
server_source=active_server
|
||||
)
|
||||
|
||||
if db_track and confidence >= 0.80:
|
||||
logger.info(f"Database match: '{db_track.title}' (confidence: {confidence:.2f})")
|
||||
|
||||
# Save to sync match cache
|
||||
if spotify_id:
|
||||
try:
|
||||
from core.matching_engine import MusicMatchingEngine
|
||||
me = MusicMatchingEngine()
|
||||
db.save_sync_match_cache(
|
||||
spotify_id, me.clean_title(original_title), me.clean_artist(artist_name),
|
||||
active_server, db_track.id, db_track.title, confidence
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Create mock track object for playlist creation
|
||||
class DatabaseTrackMock:
|
||||
def __init__(self, db_track):
|
||||
self.ratingKey = db_track.id
|
||||
self.title = db_track.title
|
||||
self.id = db_track.id
|
||||
|
||||
return DatabaseTrackMock(db_track), confidence
|
||||
|
||||
logger.warning(f"No database match found for: '{original_title}'")
|
||||
return None, 0.0
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Database search error: {e}")
|
||||
return None, 0.0
|
||||
|
||||
# Patch the method
|
||||
sync_service._find_track_in_media_server = database_only_find_track
|
||||
logger.info("Patched sync service to use database-only matching")
|
||||
|
||||
sync_start_time = time.time()
|
||||
setup_duration = (sync_start_time - task_start_time) * 1000
|
||||
logger.info(f"⏱️ [TIMING] Setup completed at {time.strftime('%H:%M:%S')} (took {setup_duration:.1f}ms)")
|
||||
logger.info("Starting actual sync process with run_async()...")
|
||||
|
||||
# Attach original tracks map to sync_service for wishlist with album images
|
||||
sync_service._original_tracks_map = original_tracks_map
|
||||
|
||||
# Wing It mode — skip wishlist for unmatched tracks
|
||||
with sync_lock:
|
||||
is_wing_it = sync_states.get(playlist_id, {}).get('wing_it', False)
|
||||
sync_service._skip_wishlist = is_wing_it
|
||||
|
||||
# Run the sync (this is a blocking call within this thread)
|
||||
result = deps.run_async(sync_service.sync_playlist(playlist, download_missing=False, profile_id=profile_id))
|
||||
|
||||
# Clear progress callback immediately to prevent race condition where a
|
||||
# late-firing progress callback overwrites the "finished" state below
|
||||
if sync_service:
|
||||
sync_service.clear_progress_callback(playlist.name)
|
||||
|
||||
sync_duration = (time.time() - sync_start_time) * 1000
|
||||
total_duration = (time.time() - task_start_time) * 1000
|
||||
logger.info(f"⏱️ [TIMING] Sync completed at {time.strftime('%H:%M:%S')} (sync: {sync_duration:.1f}ms, total: {total_duration:.1f}ms)")
|
||||
logger.info(f"Sync process completed! Result type: {type(result)}")
|
||||
logger.info(f" Result details: matched={getattr(result, 'matched_tracks', 'N/A')}, total={getattr(result, 'total_tracks', 'N/A')}")
|
||||
|
||||
# Update final state on completion
|
||||
# Convert result to JSON-serializable dict (datetime/errors can't be emitted via SocketIO)
|
||||
# Exclude match_details (large) but include a summary of unmatched tracks
|
||||
result_dict = {
|
||||
k: (v.isoformat() if hasattr(v, 'isoformat') else v)
|
||||
for k, v in result.__dict__.items()
|
||||
if k != 'match_details'
|
||||
}
|
||||
# Include unmatched track names so the frontend can show which tracks failed
|
||||
match_details = getattr(result, 'match_details', None)
|
||||
if match_details:
|
||||
unmatched_summary = [
|
||||
{'name': d.get('name', ''), 'artist': d.get('artist', ''), 'image_url': d.get('image_url', '')}
|
||||
for d in match_details if d.get('status') == 'not_found'
|
||||
]
|
||||
if unmatched_summary:
|
||||
result_dict['unmatched_tracks'] = unmatched_summary
|
||||
with sync_lock:
|
||||
sync_states[playlist_id] = {
|
||||
"status": "finished",
|
||||
"progress": result_dict,
|
||||
"result": result_dict
|
||||
}
|
||||
logger.info(f"Sync finished for {playlist_id} - state updated")
|
||||
|
||||
# Set playlist poster image if available (Plex, Jellyfin, Emby)
|
||||
_synced = getattr(result, 'synced_tracks', 0)
|
||||
logger.info(f"[PLAYLIST IMAGE] image_url={playlist_image_url!r}, synced_tracks={_synced}")
|
||||
if playlist_image_url and _synced > 0:
|
||||
try:
|
||||
active_server = deps.config_manager.get_active_media_server()
|
||||
logger.info(f"[PLAYLIST IMAGE] active_server={active_server}")
|
||||
if active_server == 'plex' and deps.plex_client:
|
||||
ok = deps.plex_client.set_playlist_image(playlist_name, playlist_image_url)
|
||||
logger.info(f"[PLAYLIST IMAGE] Plex upload result: {ok}")
|
||||
elif active_server in ('jellyfin', 'emby') and deps.jellyfin_client:
|
||||
ok = deps.jellyfin_client.set_playlist_image(playlist_name, playlist_image_url)
|
||||
logger.info(f"[PLAYLIST IMAGE] Jellyfin upload result: {ok}")
|
||||
# Navidrome doesn't support custom playlist images
|
||||
except Exception as img_err:
|
||||
logger.error(f"[PLAYLIST IMAGE] Exception: {img_err}")
|
||||
|
||||
# Record sync history completion with per-track data
|
||||
try:
|
||||
matched = getattr(result, 'matched_tracks', 0)
|
||||
failed = getattr(result, 'failed_tracks', 0)
|
||||
synced = getattr(result, 'synced_tracks', 0)
|
||||
db = MusicDatabase()
|
||||
target_batch_id = sync_batch_id
|
||||
if _is_resync and _resync_entry_id:
|
||||
db.refresh_sync_history_entry(_resync_entry_id, matched, synced, failed)
|
||||
# For re-sync, get the batch_id from the original entry
|
||||
try:
|
||||
entry = db.get_sync_history_entry(_resync_entry_id)
|
||||
if entry:
|
||||
target_batch_id = entry.get('batch_id', sync_batch_id)
|
||||
except Exception:
|
||||
pass
|
||||
else:
|
||||
db.update_sync_history_completion(sync_batch_id, matched, synced, failed)
|
||||
|
||||
# Save per-track match details from sync service
|
||||
match_details = getattr(result, 'match_details', None)
|
||||
if match_details:
|
||||
try:
|
||||
track_results_json = json.dumps(match_details, default=str)
|
||||
saved = db.update_sync_history_track_results(target_batch_id, track_results_json)
|
||||
logger.info(f"[Sync History] Saved {len(match_details)} track results for batch {target_batch_id} (saved={saved})")
|
||||
except Exception as json_err:
|
||||
logger.error(f"[Sync History] Failed to serialize track results: {json_err}")
|
||||
else:
|
||||
logger.warning(f"[Sync History] No match_details on SyncResult for batch {target_batch_id}")
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to record sync history completion: {e}")
|
||||
|
||||
if automation_id:
|
||||
matched = getattr(result, 'matched_tracks', 0)
|
||||
total = getattr(result, 'total_tracks', 0)
|
||||
failed = getattr(result, 'failed_tracks', 0)
|
||||
deps.update_automation_progress(automation_id, status='finished', progress=100,
|
||||
phase='Sync complete',
|
||||
log_line=f'Done: {matched}/{total} matched, {failed} failed', log_type='success')
|
||||
|
||||
# Emit playlist_synced event for automation engine
|
||||
try:
|
||||
if deps.automation_engine:
|
||||
deps.automation_engine.emit('playlist_synced', {
|
||||
'playlist_name': playlist_name,
|
||||
'total_tracks': str(getattr(result, 'total_tracks', 0)),
|
||||
'matched_tracks': str(getattr(result, 'matched_tracks', 0)),
|
||||
'synced_tracks': str(getattr(result, 'synced_tracks', 0)),
|
||||
'failed_tracks': str(getattr(result, 'failed_tracks', 0)),
|
||||
})
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Save sync status with match counts and track hash for smart-skip on next scheduled sync
|
||||
import hashlib as _hl
|
||||
_track_ids_str = ','.join(sorted(t.get('id', '') for t in tracks_json))
|
||||
_tracks_hash = _hl.md5(_track_ids_str.encode()).hexdigest()
|
||||
snapshot_id = getattr(playlist, 'snapshot_id', None)
|
||||
deps.update_and_save_sync_status(playlist_id, playlist_name, playlist.owner, snapshot_id,
|
||||
matched_tracks=getattr(result, 'matched_tracks', 0),
|
||||
total_tracks=getattr(result, 'total_tracks', 0),
|
||||
discovered_tracks=len(tracks_json),
|
||||
tracks_hash=_tracks_hash)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"SYNC FAILED for {playlist_id}: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
with sync_lock:
|
||||
sync_states[playlist_id] = {
|
||||
"status": "error",
|
||||
"error": str(e)
|
||||
}
|
||||
if automation_id:
|
||||
deps.update_automation_progress(automation_id, status='error', progress=100,
|
||||
phase='Error', log_line=f'Sync failed: {str(e)}', log_type='error')
|
||||
finally:
|
||||
logger.info(f"Cleaning up progress callback for {playlist.name}")
|
||||
# Clean up the callback
|
||||
if sync_service:
|
||||
sync_service.clear_progress_callback(playlist.name)
|
||||
# Clean up original tracks map
|
||||
if hasattr(sync_service, '_original_tracks_map'):
|
||||
del sync_service._original_tracks_map
|
||||
logger.info(f"Cleanup completed for {playlist_id}")
|
||||
@ -0,0 +1,455 @@
|
||||
"""Tests for core/discovery/sync.py — playlist sync background worker."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
from dataclasses import dataclass
|
||||
|
||||
import pytest
|
||||
|
||||
from core.discovery import sync as ds
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fakes
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@dataclass
|
||||
class _FakeSyncResult:
|
||||
matched_tracks: int = 5
|
||||
failed_tracks: int = 1
|
||||
synced_tracks: int = 4
|
||||
total_tracks: int = 6
|
||||
match_details: list = None
|
||||
|
||||
def __post_init__(self):
|
||||
if self.match_details is None:
|
||||
self.match_details = []
|
||||
|
||||
|
||||
class _FakeMediaClient:
|
||||
def __init__(self, connected=True):
|
||||
self._connected = connected
|
||||
|
||||
def is_connected(self):
|
||||
return self._connected
|
||||
|
||||
|
||||
class _FakeSyncService:
|
||||
def __init__(self, *, media_client=None, server_type='plex',
|
||||
sync_result=None, raise_on_sync=None,
|
||||
spotify_client=True, plex_client=True, jellyfin_client=True):
|
||||
self._media_client = media_client
|
||||
self._server_type = server_type
|
||||
self._sync_result = sync_result or _FakeSyncResult()
|
||||
self._raise_on_sync = raise_on_sync
|
||||
self.spotify_client = object() if spotify_client else None
|
||||
self.plex_client = object() if plex_client else None
|
||||
self.jellyfin_client = object() if jellyfin_client else None
|
||||
self.progress_callback = None
|
||||
self.progress_playlist_name = None
|
||||
self.cleared_callbacks = []
|
||||
|
||||
def _get_active_media_client(self):
|
||||
return (self._media_client, self._server_type)
|
||||
|
||||
def set_progress_callback(self, cb, playlist_name):
|
||||
self.progress_callback = cb
|
||||
self.progress_playlist_name = playlist_name
|
||||
|
||||
def clear_progress_callback(self, playlist_name):
|
||||
self.cleared_callbacks.append(playlist_name)
|
||||
|
||||
async def sync_playlist(self, playlist, download_missing=False, profile_id=1):
|
||||
if self._raise_on_sync:
|
||||
raise self._raise_on_sync
|
||||
return self._sync_result
|
||||
|
||||
async def _find_track_in_media_server(self, spotify_track):
|
||||
return None, 0.0
|
||||
|
||||
|
||||
class _FakeConfig:
|
||||
def __init__(self, server='plex'):
|
||||
self._server = server
|
||||
|
||||
def get_active_media_server(self):
|
||||
return self._server
|
||||
|
||||
|
||||
class _FakePlex:
|
||||
def __init__(self):
|
||||
self.image_calls = []
|
||||
|
||||
def set_playlist_image(self, name, url):
|
||||
self.image_calls.append((name, url))
|
||||
return True
|
||||
|
||||
|
||||
class _FakeJellyfin:
|
||||
def __init__(self):
|
||||
self.image_calls = []
|
||||
|
||||
def set_playlist_image(self, name, url):
|
||||
self.image_calls.append((name, url))
|
||||
return True
|
||||
|
||||
|
||||
class _FakeAutomationEngine:
|
||||
def __init__(self):
|
||||
self.events = []
|
||||
|
||||
def emit(self, event_type, data):
|
||||
self.events.append((event_type, data))
|
||||
|
||||
|
||||
def _run_async_sync(coro):
|
||||
"""Run a coroutine to completion using a new event loop."""
|
||||
import asyncio
|
||||
loop = asyncio.new_event_loop()
|
||||
try:
|
||||
return loop.run_until_complete(coro)
|
||||
finally:
|
||||
loop.close()
|
||||
|
||||
|
||||
def _build_deps(
|
||||
*,
|
||||
sync_service=None,
|
||||
config=None,
|
||||
plex=None,
|
||||
jellyfin=None,
|
||||
automation=None,
|
||||
sync_states=None,
|
||||
sync_lock=None,
|
||||
record_sync_history_start=None,
|
||||
update_automation_progress=None,
|
||||
update_and_save_sync_status=None,
|
||||
run_async=None,
|
||||
):
|
||||
return ds.SyncDeps(
|
||||
config_manager=config or _FakeConfig(),
|
||||
sync_service=sync_service or _FakeSyncService(media_client=_FakeMediaClient()),
|
||||
plex_client=plex or _FakePlex(),
|
||||
jellyfin_client=jellyfin or _FakeJellyfin(),
|
||||
automation_engine=automation or _FakeAutomationEngine(),
|
||||
run_async=run_async or _run_async_sync,
|
||||
record_sync_history_start=record_sync_history_start or (lambda **kw: None),
|
||||
update_automation_progress=update_automation_progress or (lambda *a, **kw: None),
|
||||
update_and_save_sync_status=update_and_save_sync_status or (lambda *a, **kw: None),
|
||||
sync_states=sync_states if sync_states is not None else {},
|
||||
sync_lock=sync_lock or threading.Lock(),
|
||||
)
|
||||
|
||||
|
||||
def _track(name='Song', artists=None, album='Album', track_id='id1'):
|
||||
return {
|
||||
'id': track_id,
|
||||
'name': name,
|
||||
'artists': artists or ['Artist'],
|
||||
'album': album,
|
||||
'duration_ms': 1000,
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def patched_db(monkeypatch):
|
||||
"""Stubs database access — never hits a real DB."""
|
||||
class _StubDB:
|
||||
def __init__(self):
|
||||
self.completion_calls = []
|
||||
self.track_results_calls = []
|
||||
|
||||
def update_sync_history_completion(self, batch_id, matched, synced, failed):
|
||||
self.completion_calls.append((batch_id, matched, synced, failed))
|
||||
|
||||
def update_sync_history_track_results(self, batch_id, results_json):
|
||||
self.track_results_calls.append((batch_id, results_json))
|
||||
return True
|
||||
|
||||
def refresh_sync_history_entry(self, *args):
|
||||
pass
|
||||
|
||||
def get_sync_history_entry(self, entry_id):
|
||||
return None
|
||||
|
||||
def read_sync_match_cache(self, sp_id, server):
|
||||
return None
|
||||
|
||||
stub = _StubDB()
|
||||
monkeypatch.setattr('database.music_database.MusicDatabase', lambda: stub)
|
||||
return stub
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# History recording
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_records_sync_history_for_new_sync(patched_db):
|
||||
"""Non-resync playlist_id triggers record_sync_history_start callback."""
|
||||
history_calls = []
|
||||
deps = _build_deps(record_sync_history_start=lambda **kw: history_calls.append(kw))
|
||||
|
||||
ds.run_sync_task('p1', 'My Playlist', [_track()], deps=deps)
|
||||
|
||||
assert len(history_calls) == 1
|
||||
assert history_calls[0]['playlist_id'] == 'p1'
|
||||
assert history_calls[0]['playlist_name'] == 'My Playlist'
|
||||
assert history_calls[0]['source_page'] == 'sync'
|
||||
|
||||
|
||||
def test_resync_skips_history_record(patched_db):
|
||||
"""Re-sync playlist_id (resync_<id>_<ts>) skips record_sync_history_start."""
|
||||
history_calls = []
|
||||
deps = _build_deps(record_sync_history_start=lambda **kw: history_calls.append(kw))
|
||||
|
||||
ds.run_sync_task('resync_42_1234', 'Replayed', [_track()], deps=deps)
|
||||
|
||||
assert history_calls == []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Setup error path
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_setup_error_marks_state_error(patched_db, monkeypatch):
|
||||
"""Exception during track conversion → sync_states[id] = 'error'."""
|
||||
states = {}
|
||||
|
||||
# Force SpotifyTrack constructor to raise to trigger setup error path
|
||||
class BoomSpotifyTrack:
|
||||
def __init__(self, **kw):
|
||||
raise ValueError("boom!")
|
||||
|
||||
monkeypatch.setattr(ds, 'SpotifyTrack', BoomSpotifyTrack)
|
||||
deps = _build_deps(sync_states=states)
|
||||
|
||||
ds.run_sync_task('pX', 'Playlist X', [_track()], deps=deps)
|
||||
|
||||
assert states['pX']['status'] == 'error'
|
||||
assert 'boom!' in states['pX']['error']
|
||||
|
||||
|
||||
def test_setup_error_with_automation_id_updates_progress(patched_db, monkeypatch):
|
||||
"""Setup error with automation_id calls update_automation_progress with status=error."""
|
||||
auto_calls = []
|
||||
|
||||
class BoomSpotifyTrack:
|
||||
def __init__(self, **kw):
|
||||
raise ValueError("setup boom")
|
||||
|
||||
monkeypatch.setattr(ds, 'SpotifyTrack', BoomSpotifyTrack)
|
||||
deps = _build_deps(update_automation_progress=lambda *a, **kw: auto_calls.append((a, kw)))
|
||||
|
||||
ds.run_sync_task('pY', 'PY', [_track()], automation_id='auto-1', deps=deps)
|
||||
|
||||
assert any(kw.get('status') == 'error' for _, kw in auto_calls)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Sync service errors
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_no_sync_service_marks_error(patched_db):
|
||||
"""sync_service None → caught by outer except, sync_states marked error."""
|
||||
states = {}
|
||||
deps = _build_deps(sync_states=states)
|
||||
deps.sync_service = None # explicit override past the default fallback
|
||||
|
||||
ds.run_sync_task('pZ', 'PZ', [_track()], deps=deps)
|
||||
|
||||
assert states['pZ']['status'] == 'error'
|
||||
|
||||
|
||||
def test_sync_playlist_exception_marks_error(patched_db):
|
||||
"""sync_playlist raising propagates → sync_states marked error."""
|
||||
states = {}
|
||||
svc = _FakeSyncService(media_client=_FakeMediaClient(),
|
||||
raise_on_sync=RuntimeError("network down"))
|
||||
deps = _build_deps(sync_service=svc, sync_states=states)
|
||||
|
||||
ds.run_sync_task('pErr', 'PErr', [_track()], deps=deps)
|
||||
|
||||
assert states['pErr']['status'] == 'error'
|
||||
assert 'network down' in states['pErr']['error']
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Successful sync
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_successful_sync_marks_state_finished(patched_db):
|
||||
"""Successful sync transitions sync_states to 'finished' with result_dict."""
|
||||
states = {}
|
||||
result = _FakeSyncResult(matched_tracks=10, total_tracks=12, synced_tracks=10, failed_tracks=2)
|
||||
svc = _FakeSyncService(media_client=_FakeMediaClient(), sync_result=result)
|
||||
deps = _build_deps(sync_service=svc, sync_states=states)
|
||||
|
||||
ds.run_sync_task('pOK', 'POK', [_track()], deps=deps)
|
||||
|
||||
assert states['pOK']['status'] == 'finished'
|
||||
assert states['pOK']['progress']['matched_tracks'] == 10
|
||||
|
||||
|
||||
def test_unmatched_tracks_summary_added_to_state(patched_db):
|
||||
"""match_details with not_found entries → unmatched_tracks summary on result_dict."""
|
||||
states = {}
|
||||
md = [
|
||||
{'name': 'Lost1', 'artist': 'A', 'image_url': '', 'status': 'not_found'},
|
||||
{'name': 'Found1', 'artist': 'B', 'status': 'matched'},
|
||||
{'name': 'Lost2', 'artist': 'C', 'image_url': '', 'status': 'not_found'},
|
||||
]
|
||||
result = _FakeSyncResult(match_details=md)
|
||||
svc = _FakeSyncService(media_client=_FakeMediaClient(), sync_result=result)
|
||||
deps = _build_deps(sync_service=svc, sync_states=states)
|
||||
|
||||
ds.run_sync_task('pU', 'PU', [_track()], deps=deps)
|
||||
|
||||
unmatched = states['pU']['progress'].get('unmatched_tracks', [])
|
||||
assert len(unmatched) == 2
|
||||
assert unmatched[0]['name'] == 'Lost1'
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Playlist image upload
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_playlist_image_uploaded_to_plex(patched_db):
|
||||
"""Plex active server + image_url + synced > 0 → plex_client.set_playlist_image called."""
|
||||
plex = _FakePlex()
|
||||
cfg = _FakeConfig(server='plex')
|
||||
result = _FakeSyncResult(synced_tracks=5)
|
||||
svc = _FakeSyncService(media_client=_FakeMediaClient(), sync_result=result)
|
||||
deps = _build_deps(sync_service=svc, plex=plex, config=cfg)
|
||||
|
||||
ds.run_sync_task('pImg', 'PImg', [_track()],
|
||||
playlist_image_url='https://img/x.png', deps=deps)
|
||||
|
||||
assert plex.image_calls == [('PImg', 'https://img/x.png')]
|
||||
|
||||
|
||||
def test_playlist_image_uploaded_to_jellyfin(patched_db):
|
||||
"""Jellyfin/Emby active → jellyfin_client.set_playlist_image."""
|
||||
jf = _FakeJellyfin()
|
||||
cfg = _FakeConfig(server='jellyfin')
|
||||
result = _FakeSyncResult(synced_tracks=3)
|
||||
svc = _FakeSyncService(media_client=_FakeMediaClient(), sync_result=result)
|
||||
deps = _build_deps(sync_service=svc, jellyfin=jf, config=cfg)
|
||||
|
||||
ds.run_sync_task('pJF', 'PJF', [_track()],
|
||||
playlist_image_url='https://img/y.png', deps=deps)
|
||||
|
||||
assert jf.image_calls == [('PJF', 'https://img/y.png')]
|
||||
|
||||
|
||||
def test_no_image_upload_when_zero_synced(patched_db):
|
||||
"""synced_tracks == 0 → no playlist image upload."""
|
||||
plex = _FakePlex()
|
||||
result = _FakeSyncResult(synced_tracks=0)
|
||||
svc = _FakeSyncService(media_client=_FakeMediaClient(), sync_result=result)
|
||||
deps = _build_deps(sync_service=svc, plex=plex)
|
||||
|
||||
ds.run_sync_task('pNoImg', 'PNoImg', [_track()],
|
||||
playlist_image_url='https://img/z.png', deps=deps)
|
||||
|
||||
assert plex.image_calls == []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Automation engine
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_automation_engine_emits_playlist_synced(patched_db):
|
||||
"""Successful sync emits 'playlist_synced' event on automation_engine."""
|
||||
ae = _FakeAutomationEngine()
|
||||
result = _FakeSyncResult(matched_tracks=7, total_tracks=8, synced_tracks=7, failed_tracks=1)
|
||||
svc = _FakeSyncService(media_client=_FakeMediaClient(), sync_result=result)
|
||||
deps = _build_deps(sync_service=svc, automation=ae)
|
||||
|
||||
ds.run_sync_task('pE', 'PE', [_track()], deps=deps)
|
||||
|
||||
assert any(evt == 'playlist_synced' for evt, _ in ae.events)
|
||||
|
||||
|
||||
def test_automation_progress_finished_called(patched_db):
|
||||
"""automation_id provided + sync OK → update_automation_progress called with status=finished."""
|
||||
auto_calls = []
|
||||
svc = _FakeSyncService(media_client=_FakeMediaClient())
|
||||
deps = _build_deps(sync_service=svc,
|
||||
update_automation_progress=lambda *a, **kw: auto_calls.append(kw))
|
||||
|
||||
ds.run_sync_task('pA', 'PA', [_track()], automation_id='auto-99', deps=deps)
|
||||
|
||||
assert any(kw.get('status') == 'finished' for kw in auto_calls)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Sync history persistence
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_sync_history_completion_saved(patched_db):
|
||||
"""Successful sync calls update_sync_history_completion on the DB."""
|
||||
result = _FakeSyncResult(matched_tracks=4, synced_tracks=4, failed_tracks=0)
|
||||
svc = _FakeSyncService(media_client=_FakeMediaClient(), sync_result=result)
|
||||
deps = _build_deps(sync_service=svc)
|
||||
|
||||
ds.run_sync_task('pHist', 'PHist', [_track()], deps=deps)
|
||||
|
||||
assert len(patched_db.completion_calls) == 1
|
||||
bid, matched, synced, failed = patched_db.completion_calls[0]
|
||||
assert matched == 4 and synced == 4 and failed == 0
|
||||
|
||||
|
||||
def test_match_details_persisted_to_track_results(patched_db):
|
||||
"""match_details on result → update_sync_history_track_results called with JSON."""
|
||||
md = [{'name': 'T1', 'status': 'matched'}]
|
||||
result = _FakeSyncResult(match_details=md)
|
||||
svc = _FakeSyncService(media_client=_FakeMediaClient(), sync_result=result)
|
||||
deps = _build_deps(sync_service=svc)
|
||||
|
||||
ds.run_sync_task('pMD', 'PMD', [_track()], deps=deps)
|
||||
|
||||
assert len(patched_db.track_results_calls) == 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Sync status save (smart-skip hash)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_update_and_save_sync_status_called(patched_db):
|
||||
"""update_and_save_sync_status called with a tracks_hash for smart-skip."""
|
||||
save_calls = []
|
||||
svc = _FakeSyncService(media_client=_FakeMediaClient())
|
||||
deps = _build_deps(sync_service=svc,
|
||||
update_and_save_sync_status=lambda *a, **kw: save_calls.append((a, kw)))
|
||||
|
||||
ds.run_sync_task('pSS', 'PSS', [_track(track_id='abc'), _track(track_id='def')], deps=deps)
|
||||
|
||||
assert len(save_calls) == 1
|
||||
args, kwargs = save_calls[0]
|
||||
assert kwargs.get('tracks_hash') # md5 hash present
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Cleanup (finally)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_finally_clears_progress_callback(patched_db):
|
||||
"""finally block clears sync_service progress callback."""
|
||||
svc = _FakeSyncService(media_client=_FakeMediaClient())
|
||||
deps = _build_deps(sync_service=svc)
|
||||
|
||||
ds.run_sync_task('pCB', 'PCB', [_track()], deps=deps)
|
||||
|
||||
# Both the explicit clear (after run_async) and the finally block run
|
||||
assert 'PCB' in svc.cleared_callbacks
|
||||
|
||||
|
||||
def test_finally_drops_original_tracks_map(patched_db):
|
||||
"""finally block deletes _original_tracks_map attribute when present."""
|
||||
svc = _FakeSyncService(media_client=_FakeMediaClient())
|
||||
deps = _build_deps(sync_service=svc)
|
||||
|
||||
ds.run_sync_task('pTM', 'PTM', [_track()], deps=deps)
|
||||
|
||||
assert not hasattr(svc, '_original_tracks_map')
|
||||
Loading…
Reference in new issue