PR4g: lift batch lifecycle to core/downloads/lifecycle.py
Seventh sub-PR in the download orchestrator series. Strict 1:1 lift —
zero behavior change. ~570 lines moved out of web_server.py.
What moved (lifted as 3 tightly-coupled functions in one module):
- _start_next_batch_of_downloads → start_next_batch_of_downloads
- _on_download_completed → on_download_completed
- _check_batch_completion_v2 → check_batch_completion_v2
Dependencies bundled in `LifecycleDeps` (15+ refs):
- config_manager, automation_engine, download_monitor, repair_worker,
mb_worker (live globals)
- is_shutting_down (lambda over IS_SHUTTING_DOWN flag)
- get_batch_lock (web_server helper for batch_locks dict)
- submit_download_track_worker (lambda wrapping
missing_download_executor.submit + _download_track_worker)
- submit_failed_to_wishlist + submit_failed_to_wishlist_with_auto_completion
(async, used by on_download_completed) AND process_failed_to_wishlist
+ process_failed_to_wishlist_with_auto_completion (sync, used by
check_batch_completion_v2 — direct call matches original v2 behavior;
the non-v2 path always submitted to executor)
- ensure_spotify_track_format, get_track_artist_name,
check_and_remove_from_wishlist, regenerate_batch_m3u (web_server
helpers — large, will lift in follow-up PRs)
- youtube_playlist_states, tidal_discovery_states,
deezer_discovery_states, spotify_public_discovery_states
(per-source playlist state dicts — phase transitions on batch
completion)
Direct imports for already-lifted helpers:
- core.runtime_state.{download_tasks, download_batches, tasks_lock,
add_activity_item}
- core.downloads.history.record_sync_history_completion (PR4a)
- core.album_consistency.run_album_consistency
- core.metadata.common.get_file_lock
Behavior parity verified line-by-line:
- start_next_batch: same batch lock acquisition, same shutdown gate,
same V2-cancelled-task skip, same searching-status-set-before-submit,
same submit-fails-no-ghost-worker semantics
- on_download_completed: same duplicate-call detection (skip decrement
but still check completion), same failed-track tracking with
spotify_track formatting + activity items + automation event emission,
same wishlist removal on success, same active_count decrement, same
stuck-detection (searching > 10min → not_found, post_processing >
5min → completed), same M3U regeneration + repair worker hand-off
+ album consistency pass + wishlist failed-tracks submission
- check_batch_completion_v2: same finished-count tally, same stuck
detection, same already-complete short-circuit returning True, same
per-source playlist phase updates, same album consistency pass,
same DIRECT (sync) wishlist processing call (NOT submit-to-executor
— matches original v2 which called process_* functions directly)
CRITICAL drift caught + fixed during review:
- Initial lift had v2 routing wishlist calls through submit_* deps
(async). Original v2 called process_* directly (sync). Added separate
process_* deps to LifecycleDeps and routed v2 to them. Tests updated.
Two minor defensive additions documented:
- `is_auto_batch = False` initialized before conditional in v2 (Python
scope rules made this unnecessary in original, but explicit is safer)
- Variable rename inside the queue-completion-check loop in
on_download_completed: `task_id` → `queue_task_id` to avoid shadowing
the outer parameter. Log output preserves the same task ID.
Tests: 28 new under tests/downloads/test_downloads_lifecycle.py
covering start-next (early-returns, shutdown gate, max_concurrent,
cancelled-task skip, searching-status-set, submit-failure-no-ghost,
orphan task), on-complete (decrement, duplicate skip, failed/cancelled
tracking, automation emit, wishlist removal, batch completion + emit
+ source phase update, stuck detection, auto vs manual routing),
check-v2 (missing batch, not-complete, complete-marking, already-
complete, auto routing, exception handling).
Full suite: 1029 passing (was 1001). Ruff clean.
This commit is contained in:
Broque Thomas
parent
678fc890a8
commit
0a6d1759b7
3 changed files with 1206 additions and 558 deletions
653
core/downloads/lifecycle.py
Normal file
653
core/downloads/lifecycle.py
Normal file
|
|
@ -0,0 +1,653 @@
|
|||
"""Batch lifecycle: start workers, on-completion accounting, completion check.
|
||||
|
||||
Three deeply-coupled functions:
|
||||
|
||||
- `start_next_batch_of_downloads(batch_id, deps)` — launches workers up to
|
||||
the batch's max_concurrent. Skips cancelled tasks, sets searching status,
|
||||
submits to the executor, decrement-safe on submit failures (no ghost
|
||||
workers).
|
||||
|
||||
- `on_download_completed(batch_id, task_id, success, deps)` — called when
|
||||
a single track download finishes (good or bad). Tracks failed/cancelled
|
||||
tracks for wishlist replay, decrements active count, then runs the full
|
||||
batch-completion check — which is its own beast: stuck-task detection
|
||||
(searching > 10min → not_found, post_processing > 5min → completed),
|
||||
M3U regeneration, repair worker hand-off, album consistency pass,
|
||||
wishlist failed-tracks processing.
|
||||
|
||||
- `check_batch_completion_v2(batch_id, deps)` — same completion check
|
||||
but called from the V2 atomic cancel path (which bypasses
|
||||
on_download_completed). Duplicate logic preserved verbatim.
|
||||
|
||||
Lifted verbatim from web_server.py. Dependencies injected via
|
||||
`LifecycleDeps` since the surface is wide (15+ callbacks/refs).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import time
|
||||
import traceback
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Callable, Optional
|
||||
|
||||
from core.downloads.history import record_sync_history_completion
|
||||
from core.runtime_state import (
|
||||
add_activity_item,
|
||||
download_batches,
|
||||
download_tasks,
|
||||
tasks_lock,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class LifecycleDeps:
|
||||
"""Bundle of cross-cutting deps the batch lifecycle needs."""
|
||||
config_manager: Any
|
||||
automation_engine: Any
|
||||
download_monitor: Any
|
||||
repair_worker: Any
|
||||
mb_worker: Any
|
||||
is_shutting_down: Callable[[], bool]
|
||||
get_batch_lock: Callable[[str], Any] # (batch_id) -> threading.Lock
|
||||
submit_download_track_worker: Callable # (task_id, batch_id) -> None (submits to executor)
|
||||
submit_failed_to_wishlist: Callable[[str], None] # async — submits to executor
|
||||
submit_failed_to_wishlist_with_auto_completion: Callable[[str], None] # async — submits to executor
|
||||
process_failed_to_wishlist: Callable[[str], None] # sync — direct call (used by v2 path)
|
||||
process_failed_to_wishlist_with_auto_completion: Callable[[str], None] # sync — direct call (used by v2 path)
|
||||
ensure_spotify_track_format: Callable
|
||||
get_track_artist_name: Callable
|
||||
check_and_remove_from_wishlist: Callable
|
||||
regenerate_batch_m3u: Callable
|
||||
youtube_playlist_states: dict
|
||||
tidal_discovery_states: dict
|
||||
deezer_discovery_states: dict
|
||||
spotify_public_discovery_states: dict
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# start_next_batch_of_downloads
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def start_next_batch_of_downloads(batch_id: str, deps: LifecycleDeps) -> None:
|
||||
"""Start the next batch of downloads up to the concurrent limit (like GUI)."""
|
||||
# ENHANCED: Use batch-specific lock to prevent race conditions when multiple threads
|
||||
# try to start workers for the same batch concurrently
|
||||
batch_lock = deps.get_batch_lock(batch_id)
|
||||
|
||||
with batch_lock:
|
||||
# Prevent starting new tasks if shutting down
|
||||
if deps.is_shutting_down():
|
||||
logger.info(f"[Batch Manager] Server shutting down - skipping new tasks for batch {batch_id}")
|
||||
return
|
||||
|
||||
with tasks_lock:
|
||||
if batch_id not in download_batches:
|
||||
return
|
||||
|
||||
batch = download_batches[batch_id]
|
||||
max_concurrent = batch['max_concurrent']
|
||||
queue = batch['queue']
|
||||
queue_index = batch['queue_index']
|
||||
active_count = batch['active_count']
|
||||
|
||||
logger.info(f"[Batch Lock] Starting workers for {batch_id}: active={active_count}, max={max_concurrent}, queue_pos={queue_index}/{len(queue)}")
|
||||
|
||||
# Start downloads up to the concurrent limit
|
||||
while active_count < max_concurrent and queue_index < len(queue):
|
||||
task_id = queue[queue_index]
|
||||
|
||||
# CRITICAL V2 FIX: Skip cancelled tasks instead of trying to restart them
|
||||
if task_id in download_tasks:
|
||||
current_status = download_tasks[task_id]['status']
|
||||
if current_status == 'cancelled':
|
||||
logger.warning(f"[Batch Lock] Skipping cancelled task {task_id} (queue position {queue_index + 1})")
|
||||
download_batches[batch_id]['queue_index'] += 1
|
||||
queue_index += 1
|
||||
continue # Skip to next task without consuming worker slot
|
||||
|
||||
# IMPORTANT: Set status to 'searching' BEFORE starting worker (like GUI)
|
||||
# Must be done INSIDE the lock to prevent race conditions with status polling
|
||||
download_tasks[task_id]['status'] = 'searching'
|
||||
download_tasks[task_id]['status_change_time'] = time.time()
|
||||
logger.info(f"[Batch Manager] Set task {task_id} status to 'searching'")
|
||||
else:
|
||||
logger.warning(f"[Batch Lock] Task {task_id} not found in download_tasks - skipping")
|
||||
download_batches[batch_id]['queue_index'] += 1
|
||||
queue_index += 1
|
||||
continue
|
||||
|
||||
# CRITICAL FIX: Submit to executor BEFORE incrementing counters to prevent ghost workers
|
||||
try:
|
||||
# Submit to executor first - this can fail
|
||||
deps.submit_download_track_worker(task_id, batch_id)
|
||||
|
||||
# Only increment counters AFTER successful submit
|
||||
download_batches[batch_id]['active_count'] += 1
|
||||
download_batches[batch_id]['queue_index'] += 1
|
||||
|
||||
logger.info(f"[Batch Lock] Started download {queue_index + 1}/{len(queue)} - Active: {active_count + 1}/{max_concurrent}")
|
||||
|
||||
# Update local counters for next iteration
|
||||
active_count += 1
|
||||
queue_index += 1
|
||||
|
||||
except Exception as submit_error:
|
||||
logger.error(f"[Batch Lock] CRITICAL: Failed to submit task {task_id} to executor: {submit_error}")
|
||||
logger.info("[Batch Lock] Worker slot NOT consumed - preventing ghost worker")
|
||||
|
||||
# Reset task status since worker never started
|
||||
if task_id in download_tasks:
|
||||
download_tasks[task_id]['status'] = 'failed'
|
||||
logger.error(f"[Batch Lock] Set task {task_id} status to 'failed' due to submit failure")
|
||||
|
||||
# Don't increment counters - no worker was actually started
|
||||
# This prevents the "ghost worker" issue where active_count is incremented but no actual worker runs
|
||||
break # Stop trying to start more workers if executor is failing
|
||||
|
||||
logger.info(f"[Batch Lock] Finished starting workers for {batch_id}: final_active={download_batches[batch_id]['active_count']}, max={max_concurrent}")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# on_download_completed
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def on_download_completed(batch_id: str, task_id: str, success: bool, deps: LifecycleDeps) -> None:
|
||||
"""Called when a download completes to start the next one in queue."""
|
||||
with tasks_lock:
|
||||
if batch_id not in download_batches:
|
||||
logger.warning(f"[Batch Manager] Batch {batch_id} not found for completed task {task_id}")
|
||||
return
|
||||
|
||||
# Guard against double-calling: track which tasks have already been completed
|
||||
# This prevents active_count from being decremented multiple times for the same task
|
||||
# (e.g. monitor detects completion AND post-processing calls this again)
|
||||
# NOTE: On duplicate calls, we skip decrement/tracking but STILL check batch completion.
|
||||
# This is critical because the first call may see the task in 'post_processing' (not finished),
|
||||
# and the second call (from post-processing worker) arrives after the task is truly 'completed'.
|
||||
# Without the fallthrough, batch_complete would never be emitted.
|
||||
completed_tasks = download_batches[batch_id].setdefault('_completed_task_ids', set())
|
||||
_is_duplicate_completion = task_id in completed_tasks
|
||||
if _is_duplicate_completion:
|
||||
logger.info(f"[Batch Manager] Task {task_id} already completed — skipping decrement, still checking batch completion")
|
||||
# Set terminal status so the monitor loop stops re-processing this task
|
||||
if task_id in download_tasks and download_tasks[task_id].get('status') in ('downloading', 'queued'):
|
||||
download_tasks[task_id]['status'] = 'completed'
|
||||
# Fall through to batch completion check below (don't return)
|
||||
else:
|
||||
completed_tasks.add(task_id)
|
||||
|
||||
if not _is_duplicate_completion:
|
||||
# Track failed/cancelled tasks in batch state (replicating sync.py)
|
||||
if not success and task_id in download_tasks:
|
||||
task = download_tasks[task_id]
|
||||
task_status = task.get('status', 'unknown')
|
||||
|
||||
# Build track_info structure matching sync.py's permanently_failed_tracks format
|
||||
original_track_info = task.get('track_info', {})
|
||||
|
||||
# Ensure spotify_track has proper structure for wishlist service
|
||||
spotify_track_data = deps.ensure_spotify_track_format(original_track_info)
|
||||
|
||||
track_info = {
|
||||
'download_index': task.get('track_index', 0),
|
||||
'table_index': task.get('track_index', 0),
|
||||
'track_name': original_track_info.get('name', 'Unknown Track'),
|
||||
'artist_name': deps.get_track_artist_name(original_track_info),
|
||||
'retry_count': task.get('retry_count', 0),
|
||||
'spotify_track': spotify_track_data, # Properly formatted spotify track for wishlist
|
||||
'failure_reason': 'Download cancelled' if task_status == 'cancelled' else ('No matching track found' if task_status == 'not_found' else 'Download failed'),
|
||||
'candidates': task.get('cached_candidates', []), # Include search results if available
|
||||
}
|
||||
|
||||
if task_status == 'cancelled':
|
||||
download_batches[batch_id]['cancelled_tracks'].add(task.get('track_index', 0))
|
||||
logger.warning(f"[Batch Manager] Added cancelled track to batch tracking: {track_info['track_name']}")
|
||||
add_activity_item("", "Download Cancelled", f"'{track_info['track_name']}'", "Now")
|
||||
elif task_status in ('failed', 'not_found'):
|
||||
download_batches[batch_id]['permanently_failed_tracks'].append(track_info)
|
||||
if task_status == 'not_found':
|
||||
logger.info(f"[Batch Manager] Added not-found track to batch tracking: {track_info['track_name']}")
|
||||
add_activity_item("", "Not Found", f"'{track_info['track_name']}'", "Now")
|
||||
else:
|
||||
logger.error(f"[Batch Manager] Added failed track to batch tracking: {track_info['track_name']}")
|
||||
add_activity_item("", "Download Failed", f"'{track_info['track_name']}'", "Now")
|
||||
|
||||
try:
|
||||
if deps.automation_engine:
|
||||
deps.automation_engine.emit('download_failed', {
|
||||
'artist': track_info.get('artist_name', ''),
|
||||
'title': track_info.get('track_name', ''),
|
||||
'reason': track_info.get('failure_reason', 'Unknown'),
|
||||
})
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# WISHLIST REMOVAL: Handle successful downloads for wishlist removal
|
||||
if success and task_id in download_tasks:
|
||||
try:
|
||||
task = download_tasks[task_id]
|
||||
track_info = task.get('track_info', {})
|
||||
logger.info(f"[Batch Manager] Successful download - checking wishlist removal for task {task_id}")
|
||||
|
||||
# Add activity for successful download
|
||||
track_name = track_info.get('name', 'Unknown Track')
|
||||
|
||||
# Safely extract artist name (handle both list and string formats)
|
||||
artists = track_info.get('artists', [])
|
||||
if isinstance(artists, list) and len(artists) > 0:
|
||||
first_artist = artists[0]
|
||||
artist_name = first_artist.get('name', 'Unknown Artist') if isinstance(first_artist, dict) else str(first_artist)
|
||||
elif isinstance(artists, str):
|
||||
artist_name = artists
|
||||
else:
|
||||
artist_name = 'Unknown Artist'
|
||||
|
||||
add_activity_item("", "Download Complete", f"'{track_name}' by {artist_name}", "Now")
|
||||
|
||||
# Try to remove from wishlist using track info
|
||||
if track_info:
|
||||
# Create a context-like structure for the wishlist removal function
|
||||
context = {
|
||||
'track_info': track_info,
|
||||
'original_search_result': track_info, # fallback
|
||||
}
|
||||
deps.check_and_remove_from_wishlist(context)
|
||||
except Exception as wishlist_error:
|
||||
logger.error(f"[Batch Manager] Error checking wishlist removal for successful download: {wishlist_error}")
|
||||
|
||||
# Decrement active count
|
||||
old_active = download_batches[batch_id]['active_count']
|
||||
download_batches[batch_id]['active_count'] -= 1
|
||||
new_active = download_batches[batch_id]['active_count']
|
||||
|
||||
logger.error(f"[Batch Manager] Task {task_id} completed ({'success' if success else 'failed/cancelled'}). Active workers: {old_active} → {new_active}/{download_batches[batch_id]['max_concurrent']}")
|
||||
|
||||
# ENHANCED: Always check batch completion after any task completes (including duplicate calls)
|
||||
# This ensures completion is detected even when mixing normal downloads with cancelled tasks
|
||||
logger.info(f"[Batch Manager] Checking batch completion after task {task_id} completed")
|
||||
|
||||
# FIXED: Check if batch is truly complete (all tasks finished, not just workers freed)
|
||||
batch = download_batches[batch_id]
|
||||
all_tasks_started = batch['queue_index'] >= len(batch['queue'])
|
||||
no_active_workers = batch['active_count'] == 0
|
||||
|
||||
# Count actually finished tasks (completed, failed, or cancelled)
|
||||
# CRITICAL: Don't include 'post_processing' as finished - it's still in progress (unless stuck)!
|
||||
# CRITICAL: Don't include 'searching' as finished - task is being retried (unless stuck)!
|
||||
finished_count = 0
|
||||
retrying_count = 0
|
||||
queue = batch.get('queue', [])
|
||||
current_time = time.time()
|
||||
for queue_task_id in queue:
|
||||
if queue_task_id in download_tasks:
|
||||
task = download_tasks[queue_task_id]
|
||||
task_status = task['status']
|
||||
|
||||
# STUCK DETECTION: Force fail tasks that have been in transitional states too long
|
||||
if task_status == 'searching':
|
||||
task_age = current_time - task.get('status_change_time', current_time)
|
||||
if task_age > 600: # 10 minutes
|
||||
logger.info(f"⏰ [Stuck Detection] Task {queue_task_id} stuck in searching for {task_age:.0f}s - forcing not_found")
|
||||
task['status'] = 'not_found'
|
||||
task['error_message'] = f'Search stuck for {int(task_age // 60)} minutes with no results — timed out'
|
||||
finished_count += 1
|
||||
else:
|
||||
retrying_count += 1
|
||||
elif task_status == 'post_processing':
|
||||
task_age = current_time - task.get('status_change_time', current_time)
|
||||
if task_age > 300: # 5 minutes (post-processing should be fast)
|
||||
logger.info(f"⏰ [Stuck Detection] Task {queue_task_id} stuck in post_processing for {task_age:.0f}s - forcing completion")
|
||||
task['status'] = 'completed' # Assume it worked if file verification is taking too long
|
||||
finished_count += 1
|
||||
else:
|
||||
retrying_count += 1
|
||||
elif task_status in ['completed', 'failed', 'cancelled', 'not_found']:
|
||||
finished_count += 1
|
||||
else:
|
||||
# Task ID in queue but not in download_tasks - treat as completed to prevent blocking
|
||||
logger.warning(f"[Orphaned Task] Task {queue_task_id} in queue but not in download_tasks - counting as finished")
|
||||
finished_count += 1
|
||||
|
||||
all_tasks_truly_finished = finished_count >= len(queue)
|
||||
has_retrying_tasks = retrying_count > 0
|
||||
|
||||
if all_tasks_started and no_active_workers and all_tasks_truly_finished and not has_retrying_tasks:
|
||||
logger.error(f"[Batch Manager] Batch {batch_id} truly complete - all {finished_count}/{len(queue)} tasks finished - processing failed tracks to wishlist")
|
||||
elif all_tasks_started and no_active_workers and has_retrying_tasks:
|
||||
logger.warning(f"[Batch Manager] Batch {batch_id}: all workers free but {retrying_count} tasks retrying - continuing monitoring")
|
||||
elif all_tasks_started and no_active_workers:
|
||||
# This used to incorrectly mark batch as complete!
|
||||
logger.info(f"[Batch Manager] Batch {batch_id}: all workers free but only {finished_count}/{len(queue)} tasks finished - continuing monitoring")
|
||||
|
||||
if all_tasks_started and no_active_workers and all_tasks_truly_finished and not has_retrying_tasks:
|
||||
|
||||
# Check if this is an auto-initiated batch
|
||||
is_auto_batch = batch.get('auto_initiated', False)
|
||||
|
||||
# FIXED: Ensure batch is not already marked as complete to prevent duplicate processing
|
||||
if batch.get('phase') != 'complete':
|
||||
# Mark batch as complete and set completion timestamp for auto-cleanup
|
||||
batch['phase'] = 'complete'
|
||||
batch['completion_time'] = time.time() # Track when batch completed
|
||||
|
||||
# Record sync history completion
|
||||
from database.music_database import MusicDatabase
|
||||
record_sync_history_completion(MusicDatabase(), batch_id, batch)
|
||||
|
||||
# Add activity for batch completion
|
||||
playlist_name = batch.get('playlist_name', 'Unknown Playlist')
|
||||
failed_count = len(batch.get('permanently_failed_tracks', []))
|
||||
successful_downloads = finished_count - failed_count
|
||||
add_activity_item("", "Download Batch Complete", f"'{playlist_name}' - {successful_downloads} tracks downloaded", "Now")
|
||||
|
||||
# Emit batch_complete event for automation engine (only if something downloaded)
|
||||
if successful_downloads > 0:
|
||||
try:
|
||||
if deps.automation_engine:
|
||||
deps.automation_engine.emit('batch_complete', {
|
||||
'playlist_name': playlist_name,
|
||||
'total_tracks': str(len(queue)),
|
||||
'completed_tracks': str(successful_downloads),
|
||||
'failed_tracks': str(failed_count),
|
||||
})
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Update YouTube playlist phase to 'download_complete' if this is a YouTube playlist
|
||||
playlist_id = batch.get('playlist_id')
|
||||
if playlist_id and playlist_id.startswith('youtube_'):
|
||||
url_hash = playlist_id.replace('youtube_', '')
|
||||
if url_hash in deps.youtube_playlist_states:
|
||||
deps.youtube_playlist_states[url_hash]['phase'] = 'download_complete'
|
||||
logger.info(f"Updated YouTube playlist {url_hash} to download_complete phase")
|
||||
|
||||
# Update Tidal playlist phase to 'download_complete' if this is a Tidal playlist
|
||||
if playlist_id and playlist_id.startswith('tidal_'):
|
||||
tidal_playlist_id = playlist_id.replace('tidal_', '')
|
||||
if tidal_playlist_id in deps.tidal_discovery_states:
|
||||
deps.tidal_discovery_states[tidal_playlist_id]['phase'] = 'download_complete'
|
||||
logger.info(f"Updated Tidal playlist {tidal_playlist_id} to download_complete phase")
|
||||
|
||||
# Update Deezer playlist phase to 'download_complete' if this is a Deezer playlist
|
||||
if playlist_id and playlist_id.startswith('deezer_'):
|
||||
deezer_playlist_id = playlist_id.replace('deezer_', '')
|
||||
if deezer_playlist_id in deps.deezer_discovery_states:
|
||||
deps.deezer_discovery_states[deezer_playlist_id]['phase'] = 'download_complete'
|
||||
logger.info(f"Updated Deezer playlist {deezer_playlist_id} to download_complete phase")
|
||||
|
||||
# Update Spotify Public playlist phase to 'download_complete' if this is a Spotify Public playlist
|
||||
if playlist_id and playlist_id.startswith('spotify_public_'):
|
||||
spotify_public_url_hash = playlist_id.replace('spotify_public_', '')
|
||||
if spotify_public_url_hash in deps.spotify_public_discovery_states:
|
||||
deps.spotify_public_discovery_states[spotify_public_url_hash]['phase'] = 'download_complete'
|
||||
logger.info(f"Updated Spotify Public playlist {spotify_public_url_hash} to download_complete phase")
|
||||
|
||||
logger.info(f"[Batch Manager] Batch {batch_id} complete - stopping monitor")
|
||||
deps.download_monitor.stop_monitoring(batch_id)
|
||||
|
||||
# M3U REGENERATION: Regenerate M3U with real library paths now that
|
||||
# all post-processing (tagging, moving, DB writes) is complete.
|
||||
# The frontend M3U save may fire too early — this ensures paths resolve.
|
||||
if deps.config_manager.get('m3u_export.enabled', False):
|
||||
try:
|
||||
m3u_tracks = []
|
||||
for tid in queue:
|
||||
if tid in download_tasks and download_tasks[tid].get('status') == 'completed':
|
||||
ti = download_tasks[tid].get('track_info', {})
|
||||
artists = ti.get('artists', [])
|
||||
artist_str = artists[0] if isinstance(artists, list) and artists else ''
|
||||
if isinstance(artist_str, dict):
|
||||
artist_str = artist_str.get('name', '')
|
||||
m3u_tracks.append({
|
||||
'name': ti.get('name', ''),
|
||||
'artist': artist_str,
|
||||
'duration_ms': ti.get('duration_ms', 0),
|
||||
})
|
||||
if m3u_tracks:
|
||||
deps.regenerate_batch_m3u(batch, m3u_tracks)
|
||||
except Exception as m3u_err:
|
||||
logger.error(f"[M3U] Error regenerating M3U on batch complete: {m3u_err}")
|
||||
|
||||
# REPAIR: Scan all album folders from this batch for track number issues
|
||||
if deps.repair_worker:
|
||||
deps.repair_worker.process_batch(batch_id)
|
||||
|
||||
# ALBUM CONSISTENCY: Picard-style post-batch pass — pick ONE MusicBrainz
|
||||
# release and overwrite album-level tags on all files to guarantee consistency.
|
||||
# This is the safety net: even if per-track MB lookups drifted (different cache
|
||||
# keys, API hiccups), this pass forces every file to share the same release MBID,
|
||||
# album artist ID, release group ID, etc. — preventing Navidrome album splits.
|
||||
_cons_files = batch.get('_consistency_files', [])
|
||||
if batch.get('is_album_download') and _cons_files and len(_cons_files) >= 2:
|
||||
_cons_album = batch.get('album_context', {})
|
||||
_cons_artist = batch.get('artist_context', {})
|
||||
_cons_album_name = _cons_album.get('name', '') if isinstance(_cons_album, dict) else ''
|
||||
_cons_artist_name = _cons_artist.get('name', '') if isinstance(_cons_artist, dict) else ''
|
||||
if _cons_album_name and _cons_artist_name:
|
||||
try:
|
||||
_cons_mb_svc = deps.mb_worker.mb_service if deps.mb_worker else None
|
||||
if _cons_mb_svc and deps.config_manager.get('musicbrainz.embed_tags', True):
|
||||
from core.album_consistency import run_album_consistency
|
||||
from core.metadata.common import get_file_lock
|
||||
_cons_result = run_album_consistency(
|
||||
file_infos=_cons_files,
|
||||
album_name=_cons_album_name,
|
||||
artist_name=_cons_artist_name,
|
||||
mb_service=_cons_mb_svc,
|
||||
total_discs=_cons_album.get('total_discs', 1),
|
||||
file_lock_fn=get_file_lock,
|
||||
)
|
||||
if _cons_result.get('success'):
|
||||
logger.info(f"[Album Consistency] {_cons_result['tags_written']}/{_cons_result['total_files']} files "
|
||||
f"harmonized to release {_cons_result.get('release_mbid', '')[:8]}...")
|
||||
elif _cons_result.get('error'):
|
||||
logger.error(f"[Album Consistency] Skipped: {_cons_result['error']}")
|
||||
except Exception as cons_err:
|
||||
logger.error(f"[Album Consistency] Failed (non-fatal): {cons_err}")
|
||||
|
||||
# Mark that wishlist processing is starting (prevents premature cleanup)
|
||||
batch['wishlist_processing_started'] = True
|
||||
|
||||
# Process wishlist outside of the lock to prevent threading issues
|
||||
if is_auto_batch:
|
||||
# For auto-initiated batches, handle completion and schedule next cycle
|
||||
deps.submit_failed_to_wishlist_with_auto_completion(batch_id)
|
||||
else:
|
||||
# For manual batches, use standard wishlist processing
|
||||
deps.submit_failed_to_wishlist(batch_id)
|
||||
else:
|
||||
logger.warning(f"[Batch Manager] Batch {batch_id} already marked complete - skipping duplicate processing")
|
||||
|
||||
return # Don't start next batch if we're done
|
||||
|
||||
# Start next downloads in queue
|
||||
logger.info(f"[Batch Manager] Starting next batch for {batch_id}")
|
||||
start_next_batch_of_downloads(batch_id, deps)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# check_batch_completion_v2
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def check_batch_completion_v2(batch_id: str, deps: LifecycleDeps) -> Optional[bool]:
|
||||
"""V2 SYSTEM: Check if batch is complete after worker slot changes.
|
||||
|
||||
This is needed because V2 atomic cancel bypasses on_download_completed,
|
||||
so we need to manually check for batch completion.
|
||||
"""
|
||||
try:
|
||||
with tasks_lock:
|
||||
if batch_id not in download_batches:
|
||||
logger.warning(f"[Completion Check V2] Batch {batch_id} not found")
|
||||
return
|
||||
|
||||
batch = download_batches[batch_id]
|
||||
all_tasks_started = batch['queue_index'] >= len(batch['queue'])
|
||||
no_active_workers = batch['active_count'] == 0
|
||||
|
||||
# Count actually finished tasks (completed, failed, or cancelled)
|
||||
finished_count = 0
|
||||
retrying_count = 0
|
||||
queue = batch.get('queue', [])
|
||||
current_time = time.time()
|
||||
|
||||
for task_id in queue:
|
||||
if task_id in download_tasks:
|
||||
task = download_tasks[task_id]
|
||||
task_status = task['status']
|
||||
|
||||
# STUCK DETECTION: Force fail tasks that have been in transitional states too long
|
||||
if task_status == 'searching':
|
||||
task_age = current_time - task.get('status_change_time', current_time)
|
||||
if task_age > 600: # 10 minutes
|
||||
logger.info(f"⏰ [Stuck Detection V2] Task {task_id} stuck in searching for {task_age:.0f}s - forcing not_found")
|
||||
task['status'] = 'not_found'
|
||||
task['error_message'] = f'Search stuck for {int(task_age // 60)} minutes with no results — timed out'
|
||||
finished_count += 1
|
||||
else:
|
||||
retrying_count += 1
|
||||
elif task_status == 'post_processing':
|
||||
task_age = current_time - task.get('status_change_time', current_time)
|
||||
if task_age > 300: # 5 minutes (post-processing should be fast)
|
||||
logger.info(f"⏰ [Stuck Detection V2] Task {task_id} stuck in post_processing for {task_age:.0f}s - forcing completion")
|
||||
task['status'] = 'completed' # Assume it worked if file verification is taking too long
|
||||
finished_count += 1
|
||||
else:
|
||||
retrying_count += 1
|
||||
elif task_status in ['completed', 'failed', 'cancelled', 'not_found']:
|
||||
finished_count += 1
|
||||
else:
|
||||
# Task ID in queue but not in download_tasks - treat as completed to prevent blocking
|
||||
logger.warning(f"[Orphaned Task V2] Task {task_id} in queue but not in download_tasks - counting as finished")
|
||||
finished_count += 1
|
||||
|
||||
all_tasks_truly_finished = finished_count >= len(queue)
|
||||
has_retrying_tasks = retrying_count > 0
|
||||
|
||||
logger.warning(f"[Completion Check V2] Batch {batch_id}: tasks_started={all_tasks_started}, workers={no_active_workers}, finished={finished_count}/{len(queue)}, retrying={retrying_count}")
|
||||
|
||||
is_auto_batch = False
|
||||
if all_tasks_started and no_active_workers and all_tasks_truly_finished and not has_retrying_tasks:
|
||||
# FIXED: Ensure batch is not already marked as complete to prevent duplicate processing
|
||||
if batch.get('phase') != 'complete':
|
||||
logger.info(f"[Completion Check V2] Batch {batch_id} is complete - marking as finished")
|
||||
|
||||
# Check if this is an auto-initiated batch
|
||||
is_auto_batch = batch.get('auto_initiated', False)
|
||||
|
||||
# Mark batch as complete and set completion timestamp for auto-cleanup
|
||||
batch['phase'] = 'complete'
|
||||
batch['completion_time'] = time.time() # Track when batch completed
|
||||
|
||||
# Add activity for batch completion
|
||||
playlist_name = batch.get('playlist_name', 'Unknown Playlist')
|
||||
failed_count = len(batch.get('permanently_failed_tracks', []))
|
||||
successful_downloads = finished_count - failed_count
|
||||
add_activity_item("", "Download Batch Complete", f"'{playlist_name}' - {successful_downloads} tracks downloaded", "Now")
|
||||
|
||||
# Emit batch_complete event for automation engine (only if something downloaded)
|
||||
if successful_downloads > 0:
|
||||
try:
|
||||
if deps.automation_engine:
|
||||
deps.automation_engine.emit('batch_complete', {
|
||||
'playlist_name': playlist_name,
|
||||
'total_tracks': str(len(queue)),
|
||||
'completed_tracks': str(successful_downloads),
|
||||
'failed_tracks': str(failed_count),
|
||||
})
|
||||
except Exception:
|
||||
pass
|
||||
else:
|
||||
logger.warning(f"[Completion Check V2] Batch {batch_id} already marked complete - skipping duplicate processing")
|
||||
return True # Already complete
|
||||
|
||||
# Update YouTube playlist phase to 'download_complete' if this is a YouTube playlist
|
||||
playlist_id = batch.get('playlist_id')
|
||||
if playlist_id and playlist_id.startswith('youtube_'):
|
||||
url_hash = playlist_id.replace('youtube_', '')
|
||||
if url_hash in deps.youtube_playlist_states:
|
||||
deps.youtube_playlist_states[url_hash]['phase'] = 'download_complete'
|
||||
logger.info(f"[Completion Check V2] Updated YouTube playlist {url_hash} to download_complete phase")
|
||||
|
||||
# Update Tidal playlist phase to 'download_complete' if this is a Tidal playlist
|
||||
if playlist_id and playlist_id.startswith('tidal_'):
|
||||
tidal_playlist_id = playlist_id.replace('tidal_', '')
|
||||
if tidal_playlist_id in deps.tidal_discovery_states:
|
||||
deps.tidal_discovery_states[tidal_playlist_id]['phase'] = 'download_complete'
|
||||
logger.info(f"[Completion Check V2] Updated Tidal playlist {tidal_playlist_id} to download_complete phase")
|
||||
|
||||
# Update Deezer playlist phase to 'download_complete' if this is a Deezer playlist
|
||||
if playlist_id and playlist_id.startswith('deezer_'):
|
||||
deezer_playlist_id = playlist_id.replace('deezer_', '')
|
||||
if deezer_playlist_id in deps.deezer_discovery_states:
|
||||
deps.deezer_discovery_states[deezer_playlist_id]['phase'] = 'download_complete'
|
||||
logger.info(f"[Completion Check V2] Updated Deezer playlist {deezer_playlist_id} to download_complete phase")
|
||||
|
||||
# Update Spotify Public playlist phase to 'download_complete' if this is a Spotify Public playlist
|
||||
if playlist_id and playlist_id.startswith('spotify_public_'):
|
||||
spotify_public_url_hash = playlist_id.replace('spotify_public_', '')
|
||||
if spotify_public_url_hash in deps.spotify_public_discovery_states:
|
||||
deps.spotify_public_discovery_states[spotify_public_url_hash]['phase'] = 'download_complete'
|
||||
logger.info(f"[Completion Check V2] Updated Spotify Public playlist {spotify_public_url_hash} to download_complete phase")
|
||||
|
||||
logger.info(f"[Completion Check V2] Batch {batch_id} complete - stopping monitor")
|
||||
deps.download_monitor.stop_monitoring(batch_id)
|
||||
|
||||
# REPAIR: Scan all album folders from this batch for track number issues
|
||||
if deps.repair_worker:
|
||||
deps.repair_worker.process_batch(batch_id)
|
||||
|
||||
# ALBUM CONSISTENCY: Same Picard-style pass as the primary completion path
|
||||
_cons_files = batch.get('_consistency_files', [])
|
||||
if batch.get('is_album_download') and _cons_files and len(_cons_files) >= 2:
|
||||
_cons_album = batch.get('album_context', {})
|
||||
_cons_artist = batch.get('artist_context', {})
|
||||
_cons_album_name = _cons_album.get('name', '') if isinstance(_cons_album, dict) else ''
|
||||
_cons_artist_name = _cons_artist.get('name', '') if isinstance(_cons_artist, dict) else ''
|
||||
if _cons_album_name and _cons_artist_name:
|
||||
try:
|
||||
_cons_mb_svc = deps.mb_worker.mb_service if deps.mb_worker else None
|
||||
if _cons_mb_svc and deps.config_manager.get('musicbrainz.embed_tags', True):
|
||||
from core.album_consistency import run_album_consistency
|
||||
from core.metadata.common import get_file_lock
|
||||
_cons_result = run_album_consistency(
|
||||
file_infos=_cons_files,
|
||||
album_name=_cons_album_name,
|
||||
artist_name=_cons_artist_name,
|
||||
mb_service=_cons_mb_svc,
|
||||
total_discs=_cons_album.get('total_discs', 1),
|
||||
file_lock_fn=get_file_lock,
|
||||
)
|
||||
if _cons_result.get('success'):
|
||||
logger.info(f"[Album Consistency V2] {_cons_result['tags_written']}/{_cons_result['total_files']} files "
|
||||
f"harmonized to release {_cons_result.get('release_mbid', '')[:8]}...")
|
||||
elif _cons_result.get('error'):
|
||||
logger.error(f"[Album Consistency V2] Skipped: {_cons_result['error']}")
|
||||
except Exception as cons_err:
|
||||
logger.error(f"[Album Consistency V2] Failed (non-fatal): {cons_err}")
|
||||
|
||||
# Process wishlist outside of the lock to prevent threading issues
|
||||
if all_tasks_started and no_active_workers and all_tasks_truly_finished and not has_retrying_tasks:
|
||||
# Call wishlist processing outside the lock — DIRECT (synchronous) call
|
||||
# to match original v2 behavior. The non-v2 path (on_download_completed)
|
||||
# uses the async submit_* deps; v2 calls directly because v2 itself runs
|
||||
# from a context where blocking is acceptable.
|
||||
if is_auto_batch:
|
||||
logger.info("[Completion Check V2] Processing auto-initiated batch completion")
|
||||
deps.process_failed_to_wishlist_with_auto_completion(batch_id)
|
||||
else:
|
||||
logger.info("[Completion Check V2] Processing regular batch completion")
|
||||
deps.process_failed_to_wishlist(batch_id)
|
||||
|
||||
return True # Batch was completed
|
||||
else:
|
||||
logger.warning(f"[Completion Check V2] Batch {batch_id} not yet complete: finished={finished_count}/{len(queue)}, retrying={retrying_count}, workers={batch['active_count']}")
|
||||
return False # Batch still in progress
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"[Completion Check V2] Error checking batch completion: {e}")
|
||||
traceback.print_exc()
|
||||
return False
|
||||
509
tests/downloads/test_downloads_lifecycle.py
Normal file
509
tests/downloads/test_downloads_lifecycle.py
Normal file
|
|
@ -0,0 +1,509 @@
|
|||
"""Tests for core/downloads/lifecycle.py — batch lifecycle (start, complete, check)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
|
||||
import pytest
|
||||
|
||||
from core.downloads import lifecycle as lc
|
||||
from core.runtime_state import (
|
||||
download_batches,
|
||||
download_tasks,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def reset_state():
|
||||
download_tasks.clear()
|
||||
download_batches.clear()
|
||||
yield
|
||||
download_tasks.clear()
|
||||
download_batches.clear()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fakes
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class _Recorder:
|
||||
def __init__(self):
|
||||
self.calls = []
|
||||
|
||||
def __call__(self, name):
|
||||
def _inner(*args, **kwargs):
|
||||
self.calls.append((name, args, kwargs))
|
||||
return None
|
||||
return _inner
|
||||
|
||||
|
||||
class _FakeAutoEngine:
|
||||
def __init__(self):
|
||||
self.events = []
|
||||
|
||||
def emit(self, event_type, data):
|
||||
self.events.append((event_type, data))
|
||||
|
||||
|
||||
class _FakeMonitor:
|
||||
def __init__(self):
|
||||
self.stopped = []
|
||||
|
||||
def stop_monitoring(self, batch_id):
|
||||
self.stopped.append(batch_id)
|
||||
|
||||
|
||||
class _FakeRepair:
|
||||
def __init__(self):
|
||||
self.batches = []
|
||||
|
||||
def process_batch(self, batch_id):
|
||||
self.batches.append(batch_id)
|
||||
|
||||
|
||||
class _FakeConfig:
|
||||
def __init__(self, values=None):
|
||||
self._v = values or {}
|
||||
|
||||
def get(self, key, default=None):
|
||||
return self._v.get(key, default)
|
||||
|
||||
|
||||
def _build_deps(
|
||||
*,
|
||||
automation=None,
|
||||
monitor=None,
|
||||
repair=None,
|
||||
config=None,
|
||||
is_shutting_down=lambda: False,
|
||||
submit_dl_worker=None,
|
||||
submit_failed=None,
|
||||
submit_failed_auto=None,
|
||||
process_failed=None,
|
||||
process_failed_auto=None,
|
||||
yt_states=None,
|
||||
tidal_states=None,
|
||||
deezer_states=None,
|
||||
spotify_states=None,
|
||||
):
|
||||
rec = _Recorder()
|
||||
return lc.LifecycleDeps(
|
||||
config_manager=config or _FakeConfig(),
|
||||
automation_engine=automation,
|
||||
download_monitor=monitor or _FakeMonitor(),
|
||||
repair_worker=repair,
|
||||
mb_worker=None,
|
||||
is_shutting_down=is_shutting_down,
|
||||
get_batch_lock=lambda bid: threading.Lock(),
|
||||
submit_download_track_worker=submit_dl_worker or rec('submit_dl'),
|
||||
submit_failed_to_wishlist=submit_failed or rec('submit_failed'),
|
||||
submit_failed_to_wishlist_with_auto_completion=submit_failed_auto or rec('submit_failed_auto'),
|
||||
process_failed_to_wishlist=process_failed or rec('process_failed'),
|
||||
process_failed_to_wishlist_with_auto_completion=process_failed_auto or rec('process_failed_auto'),
|
||||
ensure_spotify_track_format=lambda track: track,
|
||||
get_track_artist_name=lambda track: 'Artist',
|
||||
check_and_remove_from_wishlist=rec('check_wishlist'),
|
||||
regenerate_batch_m3u=rec('regen_m3u'),
|
||||
youtube_playlist_states=yt_states or {},
|
||||
tidal_discovery_states=tidal_states or {},
|
||||
deezer_discovery_states=deezer_states or {},
|
||||
spotify_public_discovery_states=spotify_states or {},
|
||||
), rec
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# start_next_batch_of_downloads
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_start_next_returns_silently_for_missing_batch():
|
||||
deps, rec = _build_deps()
|
||||
lc.start_next_batch_of_downloads('absent', deps)
|
||||
assert rec.calls == []
|
||||
|
||||
|
||||
def test_start_next_skipped_when_shutting_down():
|
||||
download_batches['b1'] = {'queue': ['t1'], 'queue_index': 0, 'active_count': 0, 'max_concurrent': 1}
|
||||
deps, rec = _build_deps(is_shutting_down=lambda: True)
|
||||
lc.start_next_batch_of_downloads('b1', deps)
|
||||
assert rec.calls == [] # no submit
|
||||
|
||||
|
||||
def test_start_next_submits_up_to_max_concurrent():
|
||||
download_tasks['t1'] = {'status': 'queued'}
|
||||
download_tasks['t2'] = {'status': 'queued'}
|
||||
download_tasks['t3'] = {'status': 'queued'}
|
||||
download_batches['b1'] = {
|
||||
'queue': ['t1', 't2', 't3'], 'queue_index': 0,
|
||||
'active_count': 0, 'max_concurrent': 2,
|
||||
}
|
||||
deps, rec = _build_deps()
|
||||
lc.start_next_batch_of_downloads('b1', deps)
|
||||
submits = [c for c in rec.calls if c[0] == 'submit_dl']
|
||||
assert len(submits) == 2
|
||||
assert download_batches['b1']['active_count'] == 2
|
||||
assert download_batches['b1']['queue_index'] == 2
|
||||
|
||||
|
||||
def test_start_next_skips_cancelled_tasks_without_consuming_slots():
|
||||
download_tasks['t1'] = {'status': 'cancelled'}
|
||||
download_tasks['t2'] = {'status': 'queued'}
|
||||
download_batches['b1'] = {
|
||||
'queue': ['t1', 't2'], 'queue_index': 0,
|
||||
'active_count': 0, 'max_concurrent': 1,
|
||||
}
|
||||
deps, rec = _build_deps()
|
||||
lc.start_next_batch_of_downloads('b1', deps)
|
||||
submits = [c for c in rec.calls if c[0] == 'submit_dl']
|
||||
assert len(submits) == 1
|
||||
# t2 should be the one submitted (t1 skipped)
|
||||
assert submits[0][1] == ('t2', 'b1')
|
||||
|
||||
|
||||
def test_start_next_sets_searching_status_before_submit():
|
||||
download_tasks['t1'] = {'status': 'queued'}
|
||||
download_batches['b1'] = {
|
||||
'queue': ['t1'], 'queue_index': 0,
|
||||
'active_count': 0, 'max_concurrent': 1,
|
||||
}
|
||||
deps, _ = _build_deps()
|
||||
lc.start_next_batch_of_downloads('b1', deps)
|
||||
assert download_tasks['t1']['status'] == 'searching'
|
||||
assert download_tasks['t1']['status_change_time'] is not None
|
||||
|
||||
|
||||
def test_start_next_submit_failure_marks_task_failed_no_ghost_worker():
|
||||
download_tasks['t1'] = {'status': 'queued'}
|
||||
download_batches['b1'] = {
|
||||
'queue': ['t1'], 'queue_index': 0,
|
||||
'active_count': 0, 'max_concurrent': 1,
|
||||
}
|
||||
|
||||
def _broken_submit(task_id, batch_id):
|
||||
raise RuntimeError("executor dead")
|
||||
|
||||
deps, _ = _build_deps(submit_dl_worker=_broken_submit)
|
||||
lc.start_next_batch_of_downloads('b1', deps)
|
||||
# No counters incremented
|
||||
assert download_batches['b1']['active_count'] == 0
|
||||
assert download_batches['b1']['queue_index'] == 0
|
||||
# Task marked failed
|
||||
assert download_tasks['t1']['status'] == 'failed'
|
||||
|
||||
|
||||
def test_start_next_orphan_task_in_queue_skipped():
|
||||
download_batches['b1'] = {
|
||||
'queue': ['absent', 't2'], 'queue_index': 0,
|
||||
'active_count': 0, 'max_concurrent': 2,
|
||||
}
|
||||
download_tasks['t2'] = {'status': 'queued'}
|
||||
deps, rec = _build_deps()
|
||||
lc.start_next_batch_of_downloads('b1', deps)
|
||||
submits = [c for c in rec.calls if c[0] == 'submit_dl']
|
||||
# Only t2 submitted
|
||||
assert len(submits) == 1
|
||||
assert submits[0][1] == ('t2', 'b1')
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# on_download_completed
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_on_complete_missing_batch_returns_silently():
|
||||
deps, rec = _build_deps()
|
||||
lc.on_download_completed('absent', 't1', True, deps)
|
||||
assert rec.calls == []
|
||||
|
||||
|
||||
def test_on_complete_decrements_active_count():
|
||||
download_tasks['t1'] = {'status': 'completed', 'track_info': {'name': 'X'}}
|
||||
download_batches['b1'] = {
|
||||
'queue': ['t1'], 'queue_index': 1, 'active_count': 1,
|
||||
'max_concurrent': 1, 'permanently_failed_tracks': [],
|
||||
'cancelled_tracks': set(),
|
||||
}
|
||||
deps, _ = _build_deps()
|
||||
lc.on_download_completed('b1', 't1', True, deps)
|
||||
assert download_batches['b1']['active_count'] == 0
|
||||
|
||||
|
||||
def test_on_complete_duplicate_call_skips_decrement_but_checks_completion():
|
||||
download_tasks['t1'] = {'status': 'completed', 'track_info': {'name': 'X'}}
|
||||
download_batches['b1'] = {
|
||||
'queue': ['t1'], 'queue_index': 1, 'active_count': 1,
|
||||
'max_concurrent': 1, 'permanently_failed_tracks': [],
|
||||
'cancelled_tracks': set(),
|
||||
}
|
||||
deps, _ = _build_deps()
|
||||
# First call: decrements
|
||||
lc.on_download_completed('b1', 't1', True, deps)
|
||||
assert download_batches['b1']['active_count'] == 0
|
||||
# Mark as complete to prevent batch completion path
|
||||
download_batches['b1']['phase'] = 'complete'
|
||||
# Second call: should NOT decrement again (would go negative)
|
||||
lc.on_download_completed('b1', 't1', True, deps)
|
||||
assert download_batches['b1']['active_count'] == 0 # still 0, not -1
|
||||
|
||||
|
||||
def test_on_complete_failed_task_appended_to_permanently_failed_tracks():
|
||||
download_tasks['t1'] = {
|
||||
'status': 'failed', 'track_info': {'name': 'Money'},
|
||||
'track_index': 0, 'retry_count': 0,
|
||||
}
|
||||
download_batches['b1'] = {
|
||||
'queue': ['t1'], 'queue_index': 1, 'active_count': 1,
|
||||
'max_concurrent': 1, 'permanently_failed_tracks': [],
|
||||
'cancelled_tracks': set(),
|
||||
}
|
||||
deps, _ = _build_deps()
|
||||
lc.on_download_completed('b1', 't1', False, deps)
|
||||
assert len(download_batches['b1']['permanently_failed_tracks']) == 1
|
||||
assert download_batches['b1']['permanently_failed_tracks'][0]['track_name'] == 'Money'
|
||||
|
||||
|
||||
def test_on_complete_cancelled_task_added_to_cancelled_tracks():
|
||||
download_tasks['t1'] = {
|
||||
'status': 'cancelled', 'track_info': {'name': 'X'},
|
||||
'track_index': 5,
|
||||
}
|
||||
download_batches['b1'] = {
|
||||
'queue': ['t1'], 'queue_index': 1, 'active_count': 1,
|
||||
'max_concurrent': 1, 'permanently_failed_tracks': [],
|
||||
'cancelled_tracks': set(),
|
||||
}
|
||||
deps, _ = _build_deps()
|
||||
lc.on_download_completed('b1', 't1', False, deps)
|
||||
assert 5 in download_batches['b1']['cancelled_tracks']
|
||||
|
||||
|
||||
def test_on_complete_emits_download_failed_for_not_found():
|
||||
download_tasks['t1'] = {
|
||||
'status': 'not_found', 'track_info': {'name': 'X'},
|
||||
'track_index': 0, 'retry_count': 0,
|
||||
}
|
||||
download_batches['b1'] = {
|
||||
'queue': ['t1'], 'queue_index': 1, 'active_count': 1,
|
||||
'max_concurrent': 1, 'permanently_failed_tracks': [],
|
||||
'cancelled_tracks': set(),
|
||||
}
|
||||
auto = _FakeAutoEngine()
|
||||
deps, _ = _build_deps(automation=auto)
|
||||
lc.on_download_completed('b1', 't1', False, deps)
|
||||
events = [e for e in auto.events if e[0] == 'download_failed']
|
||||
assert len(events) == 1
|
||||
|
||||
|
||||
def test_on_complete_success_calls_check_and_remove_wishlist():
|
||||
download_tasks['t1'] = {'status': 'completed', 'track_info': {'name': 'X', 'artists': ['A']}}
|
||||
download_batches['b1'] = {
|
||||
'queue': ['t1'], 'queue_index': 1, 'active_count': 1,
|
||||
'max_concurrent': 1, 'permanently_failed_tracks': [],
|
||||
'cancelled_tracks': set(),
|
||||
}
|
||||
deps, rec = _build_deps()
|
||||
lc.on_download_completed('b1', 't1', True, deps)
|
||||
assert any(c[0] == 'check_wishlist' for c in rec.calls)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Batch completion (via on_download_completed)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_batch_completion_emits_batch_complete_when_all_done():
|
||||
download_tasks['t1'] = {'status': 'completed', 'track_info': {'name': 'X'}}
|
||||
download_tasks['t2'] = {'status': 'completed', 'track_info': {'name': 'Y'}}
|
||||
download_batches['b1'] = {
|
||||
'queue': ['t1', 't2'], 'queue_index': 2, 'active_count': 1,
|
||||
'max_concurrent': 2, 'permanently_failed_tracks': [],
|
||||
'cancelled_tracks': set(),
|
||||
'playlist_name': 'PL',
|
||||
}
|
||||
auto = _FakeAutoEngine()
|
||||
monitor = _FakeMonitor()
|
||||
deps, _ = _build_deps(automation=auto, monitor=monitor)
|
||||
# Final task completes
|
||||
lc.on_download_completed('b1', 't2', True, deps)
|
||||
assert download_batches['b1']['phase'] == 'complete'
|
||||
assert ('batch_complete', {'playlist_name': 'PL', 'total_tracks': '2', 'completed_tracks': '2', 'failed_tracks': '0'}) in auto.events
|
||||
assert 'b1' in monitor.stopped
|
||||
|
||||
|
||||
def test_batch_completion_skips_emit_when_zero_successful():
|
||||
"""Don't emit batch_complete if nothing actually downloaded."""
|
||||
download_tasks['t1'] = {'status': 'failed', 'track_info': {'name': 'X'}, 'track_index': 0}
|
||||
download_batches['b1'] = {
|
||||
'queue': ['t1'], 'queue_index': 1, 'active_count': 1,
|
||||
'max_concurrent': 1,
|
||||
'permanently_failed_tracks': [{'track_name': 'X'}], # 1 failed
|
||||
'cancelled_tracks': set(),
|
||||
'playlist_name': 'PL',
|
||||
}
|
||||
auto = _FakeAutoEngine()
|
||||
deps, _ = _build_deps(automation=auto)
|
||||
lc.on_download_completed('b1', 't1', False, deps)
|
||||
# Pre-existing failed already counted, so successful = 1 (count) - 1 (already failed) = 0
|
||||
# but on_download_completed appends another failure for t1, so failed count = 2 > finished = 1
|
||||
# the emit is only triggered if successful_downloads > 0
|
||||
events = [e for e in auto.events if e[0] == 'batch_complete']
|
||||
# successful = finished_count(1) - failed_count(2) = -1, which is not > 0 → no emit
|
||||
assert events == []
|
||||
|
||||
|
||||
def test_batch_completion_routes_auto_batch_to_auto_completion_handler():
|
||||
download_tasks['t1'] = {'status': 'completed', 'track_info': {'name': 'X'}}
|
||||
download_batches['b1'] = {
|
||||
'queue': ['t1'], 'queue_index': 1, 'active_count': 1,
|
||||
'max_concurrent': 1,
|
||||
'permanently_failed_tracks': [], 'cancelled_tracks': set(),
|
||||
'auto_initiated': True,
|
||||
'playlist_name': 'PL',
|
||||
}
|
||||
deps, rec = _build_deps()
|
||||
lc.on_download_completed('b1', 't1', True, deps)
|
||||
assert any(c[0] == 'submit_failed_auto' and c[1] == ('b1',) for c in rec.calls)
|
||||
|
||||
|
||||
def test_batch_completion_routes_manual_batch_to_regular_handler():
|
||||
download_tasks['t1'] = {'status': 'completed', 'track_info': {'name': 'X'}}
|
||||
download_batches['b1'] = {
|
||||
'queue': ['t1'], 'queue_index': 1, 'active_count': 1,
|
||||
'max_concurrent': 1,
|
||||
'permanently_failed_tracks': [], 'cancelled_tracks': set(),
|
||||
# no auto_initiated → manual
|
||||
}
|
||||
deps, rec = _build_deps()
|
||||
lc.on_download_completed('b1', 't1', True, deps)
|
||||
assert any(c[0] == 'submit_failed' and c[1] == ('b1',) for c in rec.calls)
|
||||
|
||||
|
||||
def test_batch_completion_does_not_complete_when_tasks_still_searching():
|
||||
download_tasks['t1'] = {'status': 'searching', 'track_info': {'name': 'X'},
|
||||
'status_change_time': lc.time.time()} # fresh
|
||||
download_tasks['t2'] = {'status': 'completed', 'track_info': {'name': 'Y'}}
|
||||
download_batches['b1'] = {
|
||||
'queue': ['t1', 't2'], 'queue_index': 2, 'active_count': 1,
|
||||
'max_concurrent': 2, 'permanently_failed_tracks': [],
|
||||
'cancelled_tracks': set(),
|
||||
}
|
||||
auto = _FakeAutoEngine()
|
||||
deps, _ = _build_deps(automation=auto)
|
||||
lc.on_download_completed('b1', 't2', True, deps)
|
||||
# Batch NOT marked complete (t1 still searching)
|
||||
assert download_batches['b1'].get('phase') != 'complete'
|
||||
|
||||
|
||||
def test_stuck_searching_task_forced_to_not_found():
|
||||
"""Task searching > 10min gets forced to not_found."""
|
||||
download_tasks['t1'] = {
|
||||
'status': 'searching', 'track_info': {'name': 'X'},
|
||||
'status_change_time': 0, # very ancient
|
||||
}
|
||||
download_batches['b1'] = {
|
||||
'queue': ['t1'], 'queue_index': 1, 'active_count': 1,
|
||||
'max_concurrent': 1, 'permanently_failed_tracks': [],
|
||||
'cancelled_tracks': set(),
|
||||
}
|
||||
deps, _ = _build_deps()
|
||||
# Trigger completion check
|
||||
lc.on_download_completed('b1', 't1', False, deps)
|
||||
# t1 forced to not_found
|
||||
assert download_tasks['t1']['status'] == 'not_found'
|
||||
|
||||
|
||||
def test_stuck_post_processing_task_forced_to_completed():
|
||||
"""Task post_processing > 5min gets forced to completed."""
|
||||
download_tasks['t1'] = {
|
||||
'status': 'post_processing', 'track_info': {'name': 'X'},
|
||||
'status_change_time': 0,
|
||||
}
|
||||
download_batches['b1'] = {
|
||||
'queue': ['t1'], 'queue_index': 1, 'active_count': 1,
|
||||
'max_concurrent': 1, 'permanently_failed_tracks': [],
|
||||
'cancelled_tracks': set(),
|
||||
}
|
||||
deps, _ = _build_deps()
|
||||
lc.on_download_completed('b1', 't1', True, deps)
|
||||
assert download_tasks['t1']['status'] == 'completed'
|
||||
|
||||
|
||||
def test_youtube_playlist_phase_updated_on_completion():
|
||||
download_tasks['t1'] = {'status': 'completed', 'track_info': {'name': 'X'}}
|
||||
download_batches['b1'] = {
|
||||
'queue': ['t1'], 'queue_index': 1, 'active_count': 1,
|
||||
'max_concurrent': 1, 'permanently_failed_tracks': [],
|
||||
'cancelled_tracks': set(),
|
||||
'playlist_id': 'youtube_abc123',
|
||||
}
|
||||
yt = {'abc123': {'phase': 'downloading'}}
|
||||
deps, _ = _build_deps(yt_states=yt)
|
||||
lc.on_download_completed('b1', 't1', True, deps)
|
||||
assert yt['abc123']['phase'] == 'download_complete'
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# check_batch_completion_v2
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_check_v2_returns_none_for_missing_batch():
|
||||
deps, _ = _build_deps()
|
||||
result = lc.check_batch_completion_v2('absent', deps)
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_check_v2_returns_false_when_not_complete():
|
||||
download_tasks['t1'] = {'status': 'downloading', 'track_info': {}}
|
||||
download_batches['b1'] = {
|
||||
'queue': ['t1'], 'queue_index': 1, 'active_count': 1,
|
||||
'max_concurrent': 1, 'permanently_failed_tracks': [],
|
||||
}
|
||||
deps, _ = _build_deps()
|
||||
result = lc.check_batch_completion_v2('b1', deps)
|
||||
assert result is False
|
||||
|
||||
|
||||
def test_check_v2_returns_true_when_complete():
|
||||
download_tasks['t1'] = {'status': 'completed', 'track_info': {'name': 'X'}}
|
||||
download_batches['b1'] = {
|
||||
'queue': ['t1'], 'queue_index': 1, 'active_count': 0,
|
||||
'max_concurrent': 1, 'permanently_failed_tracks': [],
|
||||
}
|
||||
deps, _ = _build_deps()
|
||||
result = lc.check_batch_completion_v2('b1', deps)
|
||||
assert result is True
|
||||
assert download_batches['b1']['phase'] == 'complete'
|
||||
|
||||
|
||||
def test_check_v2_already_complete_returns_true_without_reprocessing():
|
||||
download_tasks['t1'] = {'status': 'completed', 'track_info': {'name': 'X'}}
|
||||
download_batches['b1'] = {
|
||||
'queue': ['t1'], 'queue_index': 1, 'active_count': 0,
|
||||
'max_concurrent': 1, 'permanently_failed_tracks': [],
|
||||
'phase': 'complete', # already marked
|
||||
}
|
||||
deps, rec = _build_deps()
|
||||
result = lc.check_batch_completion_v2('b1', deps)
|
||||
assert result is True
|
||||
# Wishlist NOT submitted again
|
||||
assert not any(c[0] in ('process_failed', 'process_failed_auto') for c in rec.calls)
|
||||
|
||||
|
||||
def test_check_v2_routes_auto_batch_to_auto_handler():
|
||||
"""v2 calls wishlist processing DIRECTLY (sync), not via executor submit.
|
||||
Different from on_download_completed which uses async submit."""
|
||||
download_tasks['t1'] = {'status': 'completed', 'track_info': {'name': 'X'}}
|
||||
download_batches['b1'] = {
|
||||
'queue': ['t1'], 'queue_index': 1, 'active_count': 0,
|
||||
'max_concurrent': 1, 'permanently_failed_tracks': [],
|
||||
'auto_initiated': True,
|
||||
}
|
||||
deps, rec = _build_deps()
|
||||
lc.check_batch_completion_v2('b1', deps)
|
||||
assert any(c[0] == 'process_failed_auto' for c in rec.calls)
|
||||
|
||||
|
||||
def test_check_v2_exception_returns_false():
|
||||
download_batches['b1'] = {'queue': ['t1'], 'queue_index': 1, 'active_count': 0}
|
||||
# Force an exception by putting invalid type in batch
|
||||
download_batches['b1']['queue'] = None # will raise TypeError on len()
|
||||
deps, _ = _build_deps()
|
||||
result = lc.check_batch_completion_v2('b1', deps)
|
||||
assert result is False
|
||||
602
web_server.py
602
web_server.py
|
|
@ -21129,83 +21129,47 @@ def _get_batch_lock(batch_id):
|
|||
batch_locks[batch_id] = threading.Lock()
|
||||
return batch_locks[batch_id]
|
||||
|
||||
def _start_next_batch_of_downloads(batch_id):
|
||||
"""Start the next batch of downloads up to the concurrent limit (like GUI)"""
|
||||
# ENHANCED: Use batch-specific lock to prevent race conditions when multiple threads
|
||||
# try to start workers for the same batch concurrently
|
||||
batch_lock = _get_batch_lock(batch_id)
|
||||
|
||||
with batch_lock:
|
||||
# Prevent starting new tasks if shutting down
|
||||
if IS_SHUTTING_DOWN:
|
||||
logger.info(f"[Batch Manager] Server shutting down - skipping new tasks for batch {batch_id}")
|
||||
return
|
||||
# Batch lifecycle logic lives in core/downloads/lifecycle.py.
|
||||
from core.downloads import lifecycle as _downloads_lifecycle
|
||||
|
||||
|
||||
def _build_lifecycle_deps():
|
||||
"""Build LifecycleDeps bundle from web_server.py globals on each call."""
|
||||
return _downloads_lifecycle.LifecycleDeps(
|
||||
config_manager=config_manager,
|
||||
automation_engine=automation_engine,
|
||||
download_monitor=download_monitor,
|
||||
repair_worker=repair_worker,
|
||||
mb_worker=mb_worker,
|
||||
is_shutting_down=lambda: IS_SHUTTING_DOWN,
|
||||
get_batch_lock=_get_batch_lock,
|
||||
submit_download_track_worker=lambda task_id, batch_id: missing_download_executor.submit(
|
||||
_download_track_worker, task_id, batch_id,
|
||||
),
|
||||
submit_failed_to_wishlist=lambda batch_id: missing_download_executor.submit(
|
||||
_process_failed_tracks_to_wishlist_exact, batch_id,
|
||||
),
|
||||
submit_failed_to_wishlist_with_auto_completion=lambda batch_id: missing_download_executor.submit(
|
||||
_process_failed_tracks_to_wishlist_exact_with_auto_completion, batch_id,
|
||||
),
|
||||
process_failed_to_wishlist=_process_failed_tracks_to_wishlist_exact,
|
||||
process_failed_to_wishlist_with_auto_completion=_process_failed_tracks_to_wishlist_exact_with_auto_completion,
|
||||
ensure_spotify_track_format=_ensure_spotify_track_format,
|
||||
get_track_artist_name=_get_track_artist_name,
|
||||
check_and_remove_from_wishlist=_check_and_remove_from_wishlist,
|
||||
regenerate_batch_m3u=_regenerate_batch_m3u,
|
||||
youtube_playlist_states=youtube_playlist_states,
|
||||
tidal_discovery_states=tidal_discovery_states,
|
||||
deezer_discovery_states=deezer_discovery_states,
|
||||
spotify_public_discovery_states=spotify_public_discovery_states,
|
||||
)
|
||||
|
||||
|
||||
def _start_next_batch_of_downloads(batch_id):
|
||||
"""Start the next batch of downloads up to the concurrent limit."""
|
||||
_downloads_lifecycle.start_next_batch_of_downloads(batch_id, _build_lifecycle_deps())
|
||||
|
||||
|
||||
with tasks_lock:
|
||||
if batch_id not in download_batches:
|
||||
return
|
||||
|
||||
batch = download_batches[batch_id]
|
||||
max_concurrent = batch['max_concurrent']
|
||||
queue = batch['queue']
|
||||
queue_index = batch['queue_index']
|
||||
active_count = batch['active_count']
|
||||
|
||||
logger.info(f"[Batch Lock] Starting workers for {batch_id}: active={active_count}, max={max_concurrent}, queue_pos={queue_index}/{len(queue)}")
|
||||
|
||||
# Start downloads up to the concurrent limit
|
||||
while active_count < max_concurrent and queue_index < len(queue):
|
||||
task_id = queue[queue_index]
|
||||
|
||||
# CRITICAL V2 FIX: Skip cancelled tasks instead of trying to restart them
|
||||
if task_id in download_tasks:
|
||||
current_status = download_tasks[task_id]['status']
|
||||
if current_status == 'cancelled':
|
||||
logger.warning(f"[Batch Lock] Skipping cancelled task {task_id} (queue position {queue_index + 1})")
|
||||
download_batches[batch_id]['queue_index'] += 1
|
||||
queue_index += 1
|
||||
continue # Skip to next task without consuming worker slot
|
||||
|
||||
# IMPORTANT: Set status to 'searching' BEFORE starting worker (like GUI)
|
||||
# Must be done INSIDE the lock to prevent race conditions with status polling
|
||||
download_tasks[task_id]['status'] = 'searching'
|
||||
download_tasks[task_id]['status_change_time'] = time.time()
|
||||
logger.info(f"[Batch Manager] Set task {task_id} status to 'searching'")
|
||||
else:
|
||||
logger.warning(f"[Batch Lock] Task {task_id} not found in download_tasks - skipping")
|
||||
download_batches[batch_id]['queue_index'] += 1
|
||||
queue_index += 1
|
||||
continue
|
||||
|
||||
# CRITICAL FIX: Submit to executor BEFORE incrementing counters to prevent ghost workers
|
||||
try:
|
||||
# Submit to executor first - this can fail
|
||||
future = missing_download_executor.submit(_download_track_worker, task_id, batch_id)
|
||||
|
||||
# Only increment counters AFTER successful submit
|
||||
download_batches[batch_id]['active_count'] += 1
|
||||
download_batches[batch_id]['queue_index'] += 1
|
||||
|
||||
logger.info(f"[Batch Lock] Started download {queue_index + 1}/{len(queue)} - Active: {active_count + 1}/{max_concurrent}")
|
||||
|
||||
# Update local counters for next iteration
|
||||
active_count += 1
|
||||
queue_index += 1
|
||||
|
||||
except Exception as submit_error:
|
||||
logger.error(f"[Batch Lock] CRITICAL: Failed to submit task {task_id} to executor: {submit_error}")
|
||||
logger.info("[Batch Lock] Worker slot NOT consumed - preventing ghost worker")
|
||||
|
||||
# Reset task status since worker never started
|
||||
if task_id in download_tasks:
|
||||
download_tasks[task_id]['status'] = 'failed'
|
||||
logger.error(f"[Batch Lock] Set task {task_id} status to 'failed' due to submit failure")
|
||||
|
||||
# Don't increment counters - no worker was actually started
|
||||
# This prevents the "ghost worker" issue where active_count is incremented but no actual worker runs
|
||||
break # Stop trying to start more workers if executor is failing
|
||||
|
||||
logger.info(f"[Batch Lock] Finished starting workers for {batch_id}: final_active={download_batches[batch_id]['active_count']}, max={max_concurrent}")
|
||||
|
||||
def _get_track_artist_name(track_info):
|
||||
"""Extract artist name from track info, handling different data formats (replicating sync.py)"""
|
||||
|
|
@ -21613,315 +21577,10 @@ def _process_failed_tracks_to_wishlist_exact_with_auto_completion(batch_id):
|
|||
return {'tracks_added': 0, 'errors': 1, 'total_failed': 0}
|
||||
|
||||
def _on_download_completed(batch_id, task_id, success=True):
|
||||
"""Called when a download completes to start the next one in queue"""
|
||||
with tasks_lock:
|
||||
if batch_id not in download_batches:
|
||||
logger.warning(f"[Batch Manager] Batch {batch_id} not found for completed task {task_id}")
|
||||
return
|
||||
"""Called when a download completes to start the next one in queue."""
|
||||
_downloads_lifecycle.on_download_completed(batch_id, task_id, success, _build_lifecycle_deps())
|
||||
|
||||
# Guard against double-calling: track which tasks have already been completed
|
||||
# This prevents active_count from being decremented multiple times for the same task
|
||||
# (e.g. monitor detects completion AND post-processing calls this again)
|
||||
# NOTE: On duplicate calls, we skip decrement/tracking but STILL check batch completion.
|
||||
# This is critical because the first call may see the task in 'post_processing' (not finished),
|
||||
# and the second call (from post-processing worker) arrives after the task is truly 'completed'.
|
||||
# Without the fallthrough, batch_complete would never be emitted.
|
||||
completed_tasks = download_batches[batch_id].setdefault('_completed_task_ids', set())
|
||||
_is_duplicate_completion = task_id in completed_tasks
|
||||
if _is_duplicate_completion:
|
||||
logger.info(f"[Batch Manager] Task {task_id} already completed — skipping decrement, still checking batch completion")
|
||||
# Set terminal status so the monitor loop stops re-processing this task
|
||||
if task_id in download_tasks and download_tasks[task_id].get('status') in ('downloading', 'queued'):
|
||||
download_tasks[task_id]['status'] = 'completed'
|
||||
# Fall through to batch completion check below (don't return)
|
||||
else:
|
||||
completed_tasks.add(task_id)
|
||||
|
||||
if not _is_duplicate_completion:
|
||||
# Track failed/cancelled tasks in batch state (replicating sync.py)
|
||||
if not success and task_id in download_tasks:
|
||||
task = download_tasks[task_id]
|
||||
task_status = task.get('status', 'unknown')
|
||||
|
||||
# Build track_info structure matching sync.py's permanently_failed_tracks format
|
||||
original_track_info = task.get('track_info', {})
|
||||
|
||||
# Ensure spotify_track has proper structure for wishlist service
|
||||
spotify_track_data = _ensure_spotify_track_format(original_track_info)
|
||||
|
||||
track_info = {
|
||||
'download_index': task.get('track_index', 0),
|
||||
'table_index': task.get('track_index', 0),
|
||||
'track_name': original_track_info.get('name', 'Unknown Track'),
|
||||
'artist_name': _get_track_artist_name(original_track_info),
|
||||
'retry_count': task.get('retry_count', 0),
|
||||
'spotify_track': spotify_track_data, # Properly formatted spotify track for wishlist
|
||||
'failure_reason': 'Download cancelled' if task_status == 'cancelled' else ('No matching track found' if task_status == 'not_found' else 'Download failed'),
|
||||
'candidates': task.get('cached_candidates', []) # Include search results if available
|
||||
}
|
||||
|
||||
if task_status == 'cancelled':
|
||||
download_batches[batch_id]['cancelled_tracks'].add(task.get('track_index', 0))
|
||||
logger.warning(f"[Batch Manager] Added cancelled track to batch tracking: {track_info['track_name']}")
|
||||
add_activity_item("", "Download Cancelled", f"'{track_info['track_name']}'", "Now")
|
||||
elif task_status in ('failed', 'not_found'):
|
||||
download_batches[batch_id]['permanently_failed_tracks'].append(track_info)
|
||||
if task_status == 'not_found':
|
||||
logger.info(f"[Batch Manager] Added not-found track to batch tracking: {track_info['track_name']}")
|
||||
add_activity_item("", "Not Found", f"'{track_info['track_name']}'", "Now")
|
||||
else:
|
||||
logger.error(f"[Batch Manager] Added failed track to batch tracking: {track_info['track_name']}")
|
||||
add_activity_item("", "Download Failed", f"'{track_info['track_name']}'", "Now")
|
||||
|
||||
try:
|
||||
if automation_engine:
|
||||
automation_engine.emit('download_failed', {
|
||||
'artist': track_info.get('artist_name', ''),
|
||||
'title': track_info.get('track_name', ''),
|
||||
'reason': track_info.get('failure_reason', 'Unknown'),
|
||||
})
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# WISHLIST REMOVAL: Handle successful downloads for wishlist removal
|
||||
if success and task_id in download_tasks:
|
||||
try:
|
||||
task = download_tasks[task_id]
|
||||
track_info = task.get('track_info', {})
|
||||
logger.info(f"[Batch Manager] Successful download - checking wishlist removal for task {task_id}")
|
||||
|
||||
# Add activity for successful download
|
||||
track_name = track_info.get('name', 'Unknown Track')
|
||||
|
||||
# Safely extract artist name (handle both list and string formats)
|
||||
artists = track_info.get('artists', [])
|
||||
if isinstance(artists, list) and len(artists) > 0:
|
||||
first_artist = artists[0]
|
||||
artist_name = first_artist.get('name', 'Unknown Artist') if isinstance(first_artist, dict) else str(first_artist)
|
||||
elif isinstance(artists, str):
|
||||
artist_name = artists
|
||||
else:
|
||||
artist_name = 'Unknown Artist'
|
||||
|
||||
add_activity_item("", "Download Complete", f"'{track_name}' by {artist_name}", "Now")
|
||||
|
||||
# Try to remove from wishlist using track info
|
||||
if track_info:
|
||||
# Create a context-like structure for the wishlist removal function
|
||||
context = {
|
||||
'track_info': track_info,
|
||||
'original_search_result': track_info # fallback
|
||||
}
|
||||
_check_and_remove_from_wishlist(context)
|
||||
except Exception as wishlist_error:
|
||||
logger.error(f"[Batch Manager] Error checking wishlist removal for successful download: {wishlist_error}")
|
||||
|
||||
# Decrement active count
|
||||
old_active = download_batches[batch_id]['active_count']
|
||||
download_batches[batch_id]['active_count'] -= 1
|
||||
new_active = download_batches[batch_id]['active_count']
|
||||
|
||||
logger.error(f"[Batch Manager] Task {task_id} completed ({'success' if success else 'failed/cancelled'}). Active workers: {old_active} → {new_active}/{download_batches[batch_id]['max_concurrent']}")
|
||||
|
||||
# ENHANCED: Always check batch completion after any task completes (including duplicate calls)
|
||||
# This ensures completion is detected even when mixing normal downloads with cancelled tasks
|
||||
logger.info(f"[Batch Manager] Checking batch completion after task {task_id} completed")
|
||||
|
||||
# FIXED: Check if batch is truly complete (all tasks finished, not just workers freed)
|
||||
batch = download_batches[batch_id]
|
||||
all_tasks_started = batch['queue_index'] >= len(batch['queue'])
|
||||
no_active_workers = batch['active_count'] == 0
|
||||
|
||||
# Count actually finished tasks (completed, failed, or cancelled)
|
||||
# CRITICAL: Don't include 'post_processing' as finished - it's still in progress (unless stuck)!
|
||||
# CRITICAL: Don't include 'searching' as finished - task is being retried (unless stuck)!
|
||||
finished_count = 0
|
||||
retrying_count = 0
|
||||
queue = batch.get('queue', [])
|
||||
current_time = time.time()
|
||||
for task_id in queue:
|
||||
if task_id in download_tasks:
|
||||
task = download_tasks[task_id]
|
||||
task_status = task['status']
|
||||
|
||||
# STUCK DETECTION: Force fail tasks that have been in transitional states too long
|
||||
if task_status == 'searching':
|
||||
task_age = current_time - task.get('status_change_time', current_time)
|
||||
if task_age > 600: # 10 minutes
|
||||
logger.info(f"⏰ [Stuck Detection] Task {task_id} stuck in searching for {task_age:.0f}s - forcing not_found")
|
||||
task['status'] = 'not_found'
|
||||
task['error_message'] = f'Search stuck for {int(task_age // 60)} minutes with no results — timed out'
|
||||
finished_count += 1
|
||||
else:
|
||||
retrying_count += 1
|
||||
elif task_status == 'post_processing':
|
||||
task_age = current_time - task.get('status_change_time', current_time)
|
||||
if task_age > 300: # 5 minutes (post-processing should be fast)
|
||||
logger.info(f"⏰ [Stuck Detection] Task {task_id} stuck in post_processing for {task_age:.0f}s - forcing completion")
|
||||
task['status'] = 'completed' # Assume it worked if file verification is taking too long
|
||||
finished_count += 1
|
||||
else:
|
||||
retrying_count += 1
|
||||
elif task_status in ['completed', 'failed', 'cancelled', 'not_found']:
|
||||
finished_count += 1
|
||||
else:
|
||||
# Task ID in queue but not in download_tasks - treat as completed to prevent blocking
|
||||
logger.warning(f"[Orphaned Task] Task {task_id} in queue but not in download_tasks - counting as finished")
|
||||
finished_count += 1
|
||||
|
||||
all_tasks_truly_finished = finished_count >= len(queue)
|
||||
has_retrying_tasks = retrying_count > 0
|
||||
|
||||
if all_tasks_started and no_active_workers and all_tasks_truly_finished and not has_retrying_tasks:
|
||||
logger.error(f"[Batch Manager] Batch {batch_id} truly complete - all {finished_count}/{len(queue)} tasks finished - processing failed tracks to wishlist")
|
||||
elif all_tasks_started and no_active_workers and has_retrying_tasks:
|
||||
logger.warning(f"[Batch Manager] Batch {batch_id}: all workers free but {retrying_count} tasks retrying - continuing monitoring")
|
||||
elif all_tasks_started and no_active_workers:
|
||||
# This used to incorrectly mark batch as complete!
|
||||
logger.info(f"[Batch Manager] Batch {batch_id}: all workers free but only {finished_count}/{len(queue)} tasks finished - continuing monitoring")
|
||||
|
||||
if all_tasks_started and no_active_workers and all_tasks_truly_finished and not has_retrying_tasks:
|
||||
|
||||
# Check if this is an auto-initiated batch
|
||||
is_auto_batch = batch.get('auto_initiated', False)
|
||||
|
||||
# FIXED: Ensure batch is not already marked as complete to prevent duplicate processing
|
||||
if batch.get('phase') != 'complete':
|
||||
# Mark batch as complete and set completion timestamp for auto-cleanup
|
||||
batch['phase'] = 'complete'
|
||||
batch['completion_time'] = time.time() # Track when batch completed
|
||||
|
||||
# Record sync history completion
|
||||
_record_sync_history_completion(batch_id, batch)
|
||||
|
||||
# Add activity for batch completion
|
||||
playlist_name = batch.get('playlist_name', 'Unknown Playlist')
|
||||
failed_count = len(batch.get('permanently_failed_tracks', []))
|
||||
successful_downloads = finished_count - failed_count
|
||||
add_activity_item("", "Download Batch Complete", f"'{playlist_name}' - {successful_downloads} tracks downloaded", "Now")
|
||||
|
||||
# Emit batch_complete event for automation engine (only if something downloaded)
|
||||
if successful_downloads > 0:
|
||||
try:
|
||||
if automation_engine:
|
||||
automation_engine.emit('batch_complete', {
|
||||
'playlist_name': playlist_name,
|
||||
'total_tracks': str(len(queue)),
|
||||
'completed_tracks': str(successful_downloads),
|
||||
'failed_tracks': str(failed_count),
|
||||
})
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Update YouTube playlist phase to 'download_complete' if this is a YouTube playlist
|
||||
playlist_id = batch.get('playlist_id')
|
||||
if playlist_id and playlist_id.startswith('youtube_'):
|
||||
url_hash = playlist_id.replace('youtube_', '')
|
||||
if url_hash in youtube_playlist_states:
|
||||
youtube_playlist_states[url_hash]['phase'] = 'download_complete'
|
||||
logger.info(f"Updated YouTube playlist {url_hash} to download_complete phase")
|
||||
|
||||
# Update Tidal playlist phase to 'download_complete' if this is a Tidal playlist
|
||||
if playlist_id and playlist_id.startswith('tidal_'):
|
||||
tidal_playlist_id = playlist_id.replace('tidal_', '')
|
||||
if tidal_playlist_id in tidal_discovery_states:
|
||||
tidal_discovery_states[tidal_playlist_id]['phase'] = 'download_complete'
|
||||
logger.info(f"Updated Tidal playlist {tidal_playlist_id} to download_complete phase")
|
||||
|
||||
# Update Deezer playlist phase to 'download_complete' if this is a Deezer playlist
|
||||
if playlist_id and playlist_id.startswith('deezer_'):
|
||||
deezer_playlist_id = playlist_id.replace('deezer_', '')
|
||||
if deezer_playlist_id in deezer_discovery_states:
|
||||
deezer_discovery_states[deezer_playlist_id]['phase'] = 'download_complete'
|
||||
logger.info(f"Updated Deezer playlist {deezer_playlist_id} to download_complete phase")
|
||||
|
||||
# Update Spotify Public playlist phase to 'download_complete' if this is a Spotify Public playlist
|
||||
if playlist_id and playlist_id.startswith('spotify_public_'):
|
||||
spotify_public_url_hash = playlist_id.replace('spotify_public_', '')
|
||||
if spotify_public_url_hash in spotify_public_discovery_states:
|
||||
spotify_public_discovery_states[spotify_public_url_hash]['phase'] = 'download_complete'
|
||||
logger.info(f"Updated Spotify Public playlist {spotify_public_url_hash} to download_complete phase")
|
||||
|
||||
logger.info(f"[Batch Manager] Batch {batch_id} complete - stopping monitor")
|
||||
download_monitor.stop_monitoring(batch_id)
|
||||
|
||||
# M3U REGENERATION: Regenerate M3U with real library paths now that
|
||||
# all post-processing (tagging, moving, DB writes) is complete.
|
||||
# The frontend M3U save may fire too early — this ensures paths resolve.
|
||||
if config_manager.get('m3u_export.enabled', False):
|
||||
try:
|
||||
m3u_tracks = []
|
||||
for tid in queue:
|
||||
if tid in download_tasks and download_tasks[tid].get('status') == 'completed':
|
||||
ti = download_tasks[tid].get('track_info', {})
|
||||
artists = ti.get('artists', [])
|
||||
artist_str = artists[0] if isinstance(artists, list) and artists else ''
|
||||
if isinstance(artist_str, dict):
|
||||
artist_str = artist_str.get('name', '')
|
||||
m3u_tracks.append({
|
||||
'name': ti.get('name', ''),
|
||||
'artist': artist_str,
|
||||
'duration_ms': ti.get('duration_ms', 0),
|
||||
})
|
||||
if m3u_tracks:
|
||||
_regenerate_batch_m3u(batch, m3u_tracks)
|
||||
except Exception as m3u_err:
|
||||
logger.error(f"[M3U] Error regenerating M3U on batch complete: {m3u_err}")
|
||||
|
||||
# REPAIR: Scan all album folders from this batch for track number issues
|
||||
if repair_worker:
|
||||
repair_worker.process_batch(batch_id)
|
||||
|
||||
# ALBUM CONSISTENCY: Picard-style post-batch pass — pick ONE MusicBrainz
|
||||
# release and overwrite album-level tags on all files to guarantee consistency.
|
||||
# This is the safety net: even if per-track MB lookups drifted (different cache
|
||||
# keys, API hiccups), this pass forces every file to share the same release MBID,
|
||||
# album artist ID, release group ID, etc. — preventing Navidrome album splits.
|
||||
_cons_files = batch.get('_consistency_files', [])
|
||||
if batch.get('is_album_download') and _cons_files and len(_cons_files) >= 2:
|
||||
_cons_album = batch.get('album_context', {})
|
||||
_cons_artist = batch.get('artist_context', {})
|
||||
_cons_album_name = _cons_album.get('name', '') if isinstance(_cons_album, dict) else ''
|
||||
_cons_artist_name = _cons_artist.get('name', '') if isinstance(_cons_artist, dict) else ''
|
||||
if _cons_album_name and _cons_artist_name:
|
||||
try:
|
||||
_cons_mb_svc = mb_worker.mb_service if mb_worker else None
|
||||
if _cons_mb_svc and config_manager.get('musicbrainz.embed_tags', True):
|
||||
from core.album_consistency import run_album_consistency
|
||||
_cons_result = run_album_consistency(
|
||||
file_infos=_cons_files,
|
||||
album_name=_cons_album_name,
|
||||
artist_name=_cons_artist_name,
|
||||
mb_service=_cons_mb_svc,
|
||||
total_discs=_cons_album.get('total_discs', 1),
|
||||
file_lock_fn=get_file_lock,
|
||||
)
|
||||
if _cons_result.get('success'):
|
||||
logger.info(f"[Album Consistency] {_cons_result['tags_written']}/{_cons_result['total_files']} files "
|
||||
f"harmonized to release {_cons_result.get('release_mbid', '')[:8]}...")
|
||||
elif _cons_result.get('error'):
|
||||
logger.error(f"[Album Consistency] Skipped: {_cons_result['error']}")
|
||||
except Exception as cons_err:
|
||||
logger.error(f"[Album Consistency] Failed (non-fatal): {cons_err}")
|
||||
|
||||
# Mark that wishlist processing is starting (prevents premature cleanup)
|
||||
batch['wishlist_processing_started'] = True
|
||||
|
||||
# Process wishlist outside of the lock to prevent threading issues
|
||||
if is_auto_batch:
|
||||
# For auto-initiated batches, handle completion and schedule next cycle
|
||||
missing_download_executor.submit(_process_failed_tracks_to_wishlist_exact_with_auto_completion, batch_id)
|
||||
else:
|
||||
# For manual batches, use standard wishlist processing
|
||||
missing_download_executor.submit(_process_failed_tracks_to_wishlist_exact, batch_id)
|
||||
else:
|
||||
logger.warning(f"[Batch Manager] Batch {batch_id} already marked complete - skipping duplicate processing")
|
||||
|
||||
return # Don't start next batch if we're done
|
||||
|
||||
# Start next downloads in queue
|
||||
logger.info(f"[Batch Manager] Starting next batch for {batch_id}")
|
||||
_start_next_batch_of_downloads(batch_id)
|
||||
|
||||
def _run_full_missing_tracks_process(batch_id, playlist_id, tracks_json):
|
||||
"""
|
||||
|
|
@ -23909,183 +23568,10 @@ def cancel_task_v2():
|
|||
return jsonify({"success": False, "error": str(e)}), 500
|
||||
|
||||
def _check_batch_completion_v2(batch_id):
|
||||
"""
|
||||
V2 SYSTEM: Check if batch is complete after worker slot changes.
|
||||
|
||||
This is needed because V2 atomic cancel bypasses _on_download_completed,
|
||||
so we need to manually check for batch completion.
|
||||
"""
|
||||
try:
|
||||
with tasks_lock:
|
||||
if batch_id not in download_batches:
|
||||
logger.warning(f"[Completion Check V2] Batch {batch_id} not found")
|
||||
return
|
||||
|
||||
batch = download_batches[batch_id]
|
||||
all_tasks_started = batch['queue_index'] >= len(batch['queue'])
|
||||
no_active_workers = batch['active_count'] == 0
|
||||
|
||||
# Count actually finished tasks (completed, failed, or cancelled)
|
||||
finished_count = 0
|
||||
retrying_count = 0
|
||||
queue = batch.get('queue', [])
|
||||
current_time = time.time()
|
||||
"""V2 SYSTEM: Check if batch is complete after worker slot changes."""
|
||||
return _downloads_lifecycle.check_batch_completion_v2(batch_id, _build_lifecycle_deps())
|
||||
|
||||
for task_id in queue:
|
||||
if task_id in download_tasks:
|
||||
task = download_tasks[task_id]
|
||||
task_status = task['status']
|
||||
|
||||
# STUCK DETECTION: Force fail tasks that have been in transitional states too long
|
||||
if task_status == 'searching':
|
||||
task_age = current_time - task.get('status_change_time', current_time)
|
||||
if task_age > 600: # 10 minutes
|
||||
logger.info(f"⏰ [Stuck Detection V2] Task {task_id} stuck in searching for {task_age:.0f}s - forcing not_found")
|
||||
task['status'] = 'not_found'
|
||||
task['error_message'] = f'Search stuck for {int(task_age // 60)} minutes with no results — timed out'
|
||||
finished_count += 1
|
||||
else:
|
||||
retrying_count += 1
|
||||
elif task_status == 'post_processing':
|
||||
task_age = current_time - task.get('status_change_time', current_time)
|
||||
if task_age > 300: # 5 minutes (post-processing should be fast)
|
||||
logger.info(f"⏰ [Stuck Detection V2] Task {task_id} stuck in post_processing for {task_age:.0f}s - forcing completion")
|
||||
task['status'] = 'completed' # Assume it worked if file verification is taking too long
|
||||
finished_count += 1
|
||||
else:
|
||||
retrying_count += 1
|
||||
elif task_status in ['completed', 'failed', 'cancelled', 'not_found']:
|
||||
finished_count += 1
|
||||
else:
|
||||
# Task ID in queue but not in download_tasks - treat as completed to prevent blocking
|
||||
logger.warning(f"[Orphaned Task V2] Task {task_id} in queue but not in download_tasks - counting as finished")
|
||||
finished_count += 1
|
||||
|
||||
all_tasks_truly_finished = finished_count >= len(queue)
|
||||
has_retrying_tasks = retrying_count > 0
|
||||
|
||||
logger.warning(f"[Completion Check V2] Batch {batch_id}: tasks_started={all_tasks_started}, workers={no_active_workers}, finished={finished_count}/{len(queue)}, retrying={retrying_count}")
|
||||
|
||||
if all_tasks_started and no_active_workers and all_tasks_truly_finished and not has_retrying_tasks:
|
||||
# FIXED: Ensure batch is not already marked as complete to prevent duplicate processing
|
||||
if batch.get('phase') != 'complete':
|
||||
logger.info(f"[Completion Check V2] Batch {batch_id} is complete - marking as finished")
|
||||
|
||||
# Check if this is an auto-initiated batch
|
||||
is_auto_batch = batch.get('auto_initiated', False)
|
||||
|
||||
# Mark batch as complete and set completion timestamp for auto-cleanup
|
||||
batch['phase'] = 'complete'
|
||||
batch['completion_time'] = time.time() # Track when batch completed
|
||||
|
||||
# Add activity for batch completion
|
||||
playlist_name = batch.get('playlist_name', 'Unknown Playlist')
|
||||
failed_count = len(batch.get('permanently_failed_tracks', []))
|
||||
successful_downloads = finished_count - failed_count
|
||||
add_activity_item("", "Download Batch Complete", f"'{playlist_name}' - {successful_downloads} tracks downloaded", "Now")
|
||||
|
||||
# Emit batch_complete event for automation engine (only if something downloaded)
|
||||
if successful_downloads > 0:
|
||||
try:
|
||||
if automation_engine:
|
||||
automation_engine.emit('batch_complete', {
|
||||
'playlist_name': playlist_name,
|
||||
'total_tracks': str(len(queue)),
|
||||
'completed_tracks': str(successful_downloads),
|
||||
'failed_tracks': str(failed_count),
|
||||
})
|
||||
except Exception:
|
||||
pass
|
||||
else:
|
||||
logger.warning(f"[Completion Check V2] Batch {batch_id} already marked complete - skipping duplicate processing")
|
||||
return True # Already complete
|
||||
|
||||
# Update YouTube playlist phase to 'download_complete' if this is a YouTube playlist
|
||||
playlist_id = batch.get('playlist_id')
|
||||
if playlist_id and playlist_id.startswith('youtube_'):
|
||||
url_hash = playlist_id.replace('youtube_', '')
|
||||
if url_hash in youtube_playlist_states:
|
||||
youtube_playlist_states[url_hash]['phase'] = 'download_complete'
|
||||
logger.info(f"[Completion Check V2] Updated YouTube playlist {url_hash} to download_complete phase")
|
||||
|
||||
# Update Tidal playlist phase to 'download_complete' if this is a Tidal playlist
|
||||
if playlist_id and playlist_id.startswith('tidal_'):
|
||||
tidal_playlist_id = playlist_id.replace('tidal_', '')
|
||||
if tidal_playlist_id in tidal_discovery_states:
|
||||
tidal_discovery_states[tidal_playlist_id]['phase'] = 'download_complete'
|
||||
logger.info(f"[Completion Check V2] Updated Tidal playlist {tidal_playlist_id} to download_complete phase")
|
||||
|
||||
# Update Deezer playlist phase to 'download_complete' if this is a Deezer playlist
|
||||
if playlist_id and playlist_id.startswith('deezer_'):
|
||||
deezer_playlist_id = playlist_id.replace('deezer_', '')
|
||||
if deezer_playlist_id in deezer_discovery_states:
|
||||
deezer_discovery_states[deezer_playlist_id]['phase'] = 'download_complete'
|
||||
logger.info(f"[Completion Check V2] Updated Deezer playlist {deezer_playlist_id} to download_complete phase")
|
||||
|
||||
# Update Spotify Public playlist phase to 'download_complete' if this is a Spotify Public playlist
|
||||
if playlist_id and playlist_id.startswith('spotify_public_'):
|
||||
spotify_public_url_hash = playlist_id.replace('spotify_public_', '')
|
||||
if spotify_public_url_hash in spotify_public_discovery_states:
|
||||
spotify_public_discovery_states[spotify_public_url_hash]['phase'] = 'download_complete'
|
||||
logger.info(f"[Completion Check V2] Updated Spotify Public playlist {spotify_public_url_hash} to download_complete phase")
|
||||
|
||||
logger.info(f"[Completion Check V2] Batch {batch_id} complete - stopping monitor")
|
||||
download_monitor.stop_monitoring(batch_id)
|
||||
|
||||
# REPAIR: Scan all album folders from this batch for track number issues
|
||||
if repair_worker:
|
||||
repair_worker.process_batch(batch_id)
|
||||
|
||||
# ALBUM CONSISTENCY: Same Picard-style pass as the primary completion path
|
||||
_cons_files = batch.get('_consistency_files', [])
|
||||
if batch.get('is_album_download') and _cons_files and len(_cons_files) >= 2:
|
||||
_cons_album = batch.get('album_context', {})
|
||||
_cons_artist = batch.get('artist_context', {})
|
||||
_cons_album_name = _cons_album.get('name', '') if isinstance(_cons_album, dict) else ''
|
||||
_cons_artist_name = _cons_artist.get('name', '') if isinstance(_cons_artist, dict) else ''
|
||||
if _cons_album_name and _cons_artist_name:
|
||||
try:
|
||||
_cons_mb_svc = mb_worker.mb_service if mb_worker else None
|
||||
if _cons_mb_svc and config_manager.get('musicbrainz.embed_tags', True):
|
||||
from core.album_consistency import run_album_consistency
|
||||
_cons_result = run_album_consistency(
|
||||
file_infos=_cons_files,
|
||||
album_name=_cons_album_name,
|
||||
artist_name=_cons_artist_name,
|
||||
mb_service=_cons_mb_svc,
|
||||
total_discs=_cons_album.get('total_discs', 1),
|
||||
file_lock_fn=get_file_lock,
|
||||
)
|
||||
if _cons_result.get('success'):
|
||||
logger.info(f"[Album Consistency V2] {_cons_result['tags_written']}/{_cons_result['total_files']} files "
|
||||
f"harmonized to release {_cons_result.get('release_mbid', '')[:8]}...")
|
||||
elif _cons_result.get('error'):
|
||||
logger.error(f"[Album Consistency V2] Skipped: {_cons_result['error']}")
|
||||
except Exception as cons_err:
|
||||
logger.error(f"[Album Consistency V2] Failed (non-fatal): {cons_err}")
|
||||
|
||||
# Process wishlist outside of the lock to prevent threading issues
|
||||
if all_tasks_started and no_active_workers and all_tasks_truly_finished and not has_retrying_tasks:
|
||||
# Call wishlist processing outside the lock
|
||||
if is_auto_batch:
|
||||
logger.info("[Completion Check V2] Processing auto-initiated batch completion")
|
||||
# Use the existing auto-completion function
|
||||
_process_failed_tracks_to_wishlist_exact_with_auto_completion(batch_id)
|
||||
else:
|
||||
logger.info("[Completion Check V2] Processing regular batch completion")
|
||||
# Use the regular completion function
|
||||
_process_failed_tracks_to_wishlist_exact(batch_id)
|
||||
|
||||
return True # Batch was completed
|
||||
else:
|
||||
logger.warning(f"[Completion Check V2] Batch {batch_id} not yet complete: finished={finished_count}/{len(queue)}, retrying={retrying_count}, workers={batch['active_count']}")
|
||||
return False # Batch still in progress
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"[Completion Check V2] Error checking batch completion: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
def _add_cancelled_task_to_wishlist(task):
|
||||
"""
|
||||
|
|
|
|||
Loading…
Reference in a new issue