import threading import queue import logging import sys import os import shutil import pathlib from typing import Dict, List from tidal_dl_ng.download import Download from tidal_dl_ng.config import Settings, Tidal class MockTask: def __init__(self, total): self.finished = False self.percentage = 0 self.total = total self.completed = 0 class MockProgressRich: def __init__(self): self.tasks = [] def add_task(self, *args, **kwargs): total = kwargs.get("total", 100) self.tasks.append(MockTask(total)) return len(self.tasks) - 1 def update(self, task_id, *args, **kwargs): task = self.tasks[task_id] if "completed" in kwargs: task.completed = kwargs["completed"] if "advance" in kwargs: task.completed += kwargs["advance"] if task.total and task.completed >= task.total: task.finished = True def advance(self, task_id, advance=1): self.update(task_id, advance=advance) def remove_task(self, *args, **kwargs): pass def start(self): pass def stop(self): pass from tidal_dl_ng.model.gui_data import ProgressBars from .tidal_wrapper import TidalWrapper logger = logging.getLogger(__name__) class DownloadManager: _instance = None def __new__(cls): if cls._instance is None: cls._instance = super(DownloadManager, cls).__new__(cls) cls._instance._initialized = False return cls._instance def __init__(self): if self._initialized: return self.queue = queue.Queue() self.active_downloads: Dict[str, Dict] = {} self.history: List[Dict] = [] self.worker_thread = threading.Thread(target=self._worker, daemon=True) self.worker_thread.start() self.tidal_wrapper = TidalWrapper() self._initialized = True def add_to_queue(self, item_type: str, item_id: str): task_id = f"{item_type}_{item_id}" task = { "id": task_id, "type": item_type, "item_id": item_id, "status": "queued", "progress": 0, "name": "Fetching metadata..." } self.active_downloads[task_id] = task self.queue.put(task) return task def get_queue(self): return list(self.active_downloads.values()) + self.history def _worker(self): while True: task = self.queue.get() if task is None: break task_id = task["id"] self.active_downloads[task_id]["status"] = "downloading" try: self._process_download(task) self.active_downloads[task_id]["status"] = "completed" self.active_downloads[task_id]["progress"] = 100 except Exception as e: logger.error(f"Download failed: {e}") self.active_downloads[task_id]["status"] = "failed" self.active_downloads[task_id]["error"] = str(e) finally: # Move to history self.history.append(self.active_downloads.pop(task_id)) self.queue.task_done() def cancel_task(self, task_id: str): if task_id in self.active_downloads: self.active_downloads[task_id]["control"] = "cancel" return True return False def pause_task(self, task_id: str): if task_id in self.active_downloads: self.active_downloads[task_id]["control"] = "pause" return True return False def resume_task(self, task_id: str): if task_id in self.active_downloads: self.active_downloads[task_id]["control"] = "resume" return True return False def _check_control(self, task): # Check for control flags while task.get("control") == "pause": time.sleep(1) # If cancelled while paused if task.get("control") == "cancel": break if task.get("control") == "cancel": task["status"] = "cancelled" raise Exception("Download cancelled by user") def _process_download(self, task): # Mock Signal class to capture emit calls class MockSignal: def __init__(self, callback): self.callback = callback def emit(self, value): self.callback(value) # Mock ProgressBars class class MockProgressBars: def __init__(self, task_updater): self.item = MockSignal(lambda p: task_updater("progress", p)) self.item_name = MockSignal(lambda n: task_updater("name", n)) self.list_item = MockSignal(lambda p: task_updater("list_progress", p)) self.list_name = MockSignal(lambda n: task_updater("list_name", n)) def update_task(key, value): if key == "progress": # For single track, this is the main progress if task["type"] == "track": self.active_downloads[task["id"]]["progress"] = value elif key == "name": # Update status with current track name self.active_downloads[task["id"]]["current_item"] = value elif key == "list_progress": # For albums/playlists, this is the main progress if task["type"] != "track": self.active_downloads[task["id"]]["progress"] = value mock_progress = MockProgressBars(update_task) settings = Settings() tidal = Tidal(settings) # Attempt to load token from storage tidal.login_token() # Ensure we are logged in if not tidal.session.check_login(): raise Exception("Not logged in") # Move any existing albums to destination before starting new download if task["type"] in ["album", "artist"]: self._move_albums_to_destination() # Use environment variable for download path, default to /app/downloads import os settings.data.download_base_path = os.getenv("DOWNLOAD_PATH", "/app/downloads") settings.data.path_binary_ffmpeg = "ffmpeg" # Configure logger logger_tidal = logging.getLogger("tidal_dl_ng") logger_tidal.setLevel(logging.DEBUG) handler = logging.StreamHandler(sys.stdout) handler.setLevel(logging.DEBUG) formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') handler.setFormatter(formatter) if not logger_tidal.handlers: logger_tidal.addHandler(handler) mock_progress_rich = MockProgressRich() event_abort = threading.Event() event_run = threading.Event() event_run.set() # Allow running downloader = Download( tidal_obj=tidal, path_base=settings.data.download_base_path, fn_logger=logger_tidal, skip_existing=False, progress_gui=mock_progress, progress=mock_progress_rich, event_abort=event_abort, event_run=event_run ) # Fetch media object media = None if task["type"] == "track": media = tidal.session.track(task["item_id"]) elif task["type"] == "album": media = tidal.session.album(task["item_id"]) elif task["type"] == "video": media = tidal.session.video(task["item_id"]) elif task["type"] == "playlist": media = tidal.session.playlist(task["item_id"]) elif task["type"] == "artist": media = tidal.session.artist(task["item_id"]) if not media: raise Exception("Media not found") task["name"] = f"{media.name}" if hasattr(media, "artist"): task["name"] += f" - {media.artist.name}" self._check_control(task) if task["type"] == "track": result, path = downloader.item( file_template=settings.data.format_track, media=media, quality_audio=settings.data.quality_audio ) if not result: raise Exception("Download failed (downloader returned False)") logger.info(f"Download successful: {path}") elif task["type"] == "album": tracks = media.tracks() total = len(tracks) task["total_items"] = total for i, track in enumerate(tracks): self._check_control(task) task["current_index"] = i + 1 task["progress"] = int((i / total) * 100) task["current_item"] = track.name downloader.item( file_template=settings.data.format_album, media=track, quality_audio=settings.data.quality_audio, is_parent_album=True, list_position=i+1, list_total=total ) elif task["type"] == "playlist": tracks = media.tracks() total = len(tracks) task["total_items"] = total for i, track in enumerate(tracks): self._check_control(task) task["current_index"] = i + 1 task["progress"] = int((i / total) * 100) task["current_item"] = track.name downloader.item( file_template=settings.data.format_track, media=track, quality_audio=settings.data.quality_audio, is_parent_album=False, # Playlist tracks are treated as individual tracks usually, or we can use playlist format list_position=i+1, list_total=total ) elif task["type"] == "artist": # For artist, download all albums albums = media.get_albums() total_albums = len(albums) # We can't easily know total tracks upfront without fetching all albums first. # Let's track progress by Album for the top level, and maybe tracks within? # Or just flatten everything. Flattening is better for "Track X of Y". all_tracks = [] for album in albums: all_tracks.extend(album.tracks()) total = len(all_tracks) task["total_items"] = total for i, track in enumerate(all_tracks): self._check_control(task) task["current_index"] = i + 1 task["progress"] = int((i / total) * 100) task["current_item"] = track.name downloader.item( file_template=settings.data.format_album, media=track, quality_audio=settings.data.quality_audio, is_parent_album=True, # Treat as album tracks to keep folder structure list_position=track.track_num, # Use original track num list_total=track.album.num_tracks ) # Move albums to destination after download completes if task["type"] in ["album", "artist"]: self._move_albums_to_destination() def _move_albums_to_destination(self): """Move all albums from downloads/Albums/ to destination path.""" dest_base = os.getenv("ALBUM_DESTINATION_PATH") if not dest_base: return download_base = os.getenv("DOWNLOAD_PATH", "/app/downloads") albums_path = pathlib.Path(download_base) / "Albums" if not albums_path.exists(): return try: moved_count = 0 # Move all album folders to destination for album_dir in albums_path.iterdir(): if not album_dir.is_dir(): continue # Destination path: [ALBUM_DESTINATION_PATH]/[Album folder name] dest_album_path = pathlib.Path(dest_base) / album_dir.name logger.info(f"Moving {album_dir.name} to {dest_album_path}") # Remove destination if it exists if dest_album_path.exists(): shutil.rmtree(dest_album_path) # Move the album shutil.move(str(album_dir), str(dest_album_path)) moved_count += 1 if moved_count > 0: logger.info(f"Successfully moved {moved_count} album(s) to {dest_base}") except Exception as e: logger.error(f"Failed to move albums: {e}")