refactor: Restructure GUI components into separate modules for improved maintainability

This commit is contained in:
Warry
2025-12-04 12:03:31 +01:00
parent 428d34056b
commit 75868b32d0
23 changed files with 1901 additions and 1791 deletions
+19
View File
@@ -0,0 +1,19 @@
"""GUI module for TIDAL Downloader Next Generation.
This module provides the main window and all GUI-related functionality
organized into manageable components.
"""
try:
from PySide6 import QtCore, QtGui, QtWidgets
except ImportError: # pragma: no cover - GUI deps optional in some environments
QtCore = QtGui = QtWidgets = None
from tidal_dl_ng.gui.main_window import MainWindow
__all__ = [
"MainWindow",
"QtCore",
"QtGui",
"QtWidgets",
]
+64
View File
@@ -0,0 +1,64 @@
"""GUI activation and entry point for TIDAL Downloader Next Generation."""
import sys
from tidal_dl_ng import __version__
from tidal_dl_ng.config import Tidal
try:
import qdarktheme
from PySide6 import QtCore, QtGui, QtWidgets
except ImportError as e:
print(e)
print("Qt dependencies missing. Cannot start GUI. Please read the 'README.md' carefully.")
sys.exit(1)
def gui_activate(tidal: Tidal | None = None):
"""Activate the GUI application.
Args:
tidal (Tidal | None): Optional pre-existing Tidal session.
"""
from tidal_dl_ng.gui.main_window import MainWindow
# Set dark theme and create QT app
qdarktheme.enable_hi_dpi()
app = QtWidgets.QApplication(sys.argv)
# Fix for Windows: Tooltips have bright font color
# https://github.com/5yutan5/PyQtDarkTheme/issues/239
qdarktheme.setup_theme(additional_qss="QToolTip { border: 0px; }")
# Create icon object and apply it to app window
icon: QtGui.QIcon = QtGui.QIcon()
icon.addFile("tidal_dl_ng/ui/icon16.png", QtCore.QSize(16, 16))
icon.addFile("tidal_dl_ng/ui/icon32.png", QtCore.QSize(32, 32))
icon.addFile("tidal_dl_ng/ui/icon48.png", QtCore.QSize(48, 48))
icon.addFile("tidal_dl_ng/ui/icon64.png", QtCore.QSize(64, 64))
icon.addFile("tidal_dl_ng/ui/icon256.png", QtCore.QSize(256, 256))
icon.addFile("tidal_dl_ng/ui/icon512.png", QtCore.QSize(512, 512))
app.setWindowIcon(icon)
# This bit gets the taskbar icon working properly in Windows
if sys.platform.startswith("win"):
import ctypes
# Make sure Pyinstaller icons are still grouped
if not sys.argv[0].endswith(".exe"):
# Arbitrary string
my_app_id: str = "exislow.tidal.dl-ng." + __version__
ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID(my_app_id)
window = MainWindow(tidal=tidal)
window.show()
# Check for updates
window.s_update_check.emit(True)
sys.exit(app.exec())
if __name__ == "__main__":
gui_activate()
+278
View File
@@ -0,0 +1,278 @@
"""Context menus mixin for MainWindow.
Handles context menu creation and actions.
"""
import time
import urllib.parse
from PySide6 import QtCore, QtGui, QtWidgets
from tidalapi import Album, Artist, Mix, Playlist, Track, UserPlaylist, Video
from tidal_dl_ng.constants import QueueDownloadStatus
from tidal_dl_ng.helper.gui import get_results_media_item, get_user_list_media_item
from tidal_dl_ng.helper.tidal import items_results_all, name_builder_artist
from tidal_dl_ng.logger import logger_gui
from tidal_dl_ng.model.gui_data import StatusbarMessage
class ContextMenusMixin:
"""Mixin containing context menu methods."""
def menu_context_tree_results(self, point: QtCore.QPoint) -> None:
"""Show context menu for results tree."""
index = self.tr_results.indexAt(point)
if not index.isValid():
return
media = get_results_media_item(index, self.proxy_tr_results, self.model_tr_results)
menu = QtWidgets.QMenu()
if isinstance(media, Track | Video) and hasattr(media, "album") and media.album:
menu.addAction("Download Full Album", lambda: self.thread_download_album_from_track(point))
if isinstance(media, Track):
track_id = str(media.id)
is_downloaded = self.history_service.is_downloaded(track_id)
if is_downloaded:
menu.addAction(
"✖️ Mark as Not Downloaded", lambda: self.on_mark_track_as_not_downloaded(track_id, index)
)
else:
menu.addAction("✅ Mark as Downloaded", lambda: self.on_mark_track_as_downloaded(media, index))
menu.addAction("Copy Share URL", lambda: self.on_copy_url_share(self.tr_results, point))
menu.exec(self.tr_results.mapToGlobal(point))
def menu_context_queue_download(self, point: QtCore.QPoint) -> None:
"""Show context menu for download queue."""
item = self.tr_queue_download.itemAt(point)
if not item:
return
menu = QtWidgets.QMenu()
status = item.text(0)
if status == QueueDownloadStatus.Waiting:
menu.addAction("🗑️ Remove from Queue", lambda: self.on_queue_download_remove_item(item))
if menu.isEmpty():
return
menu.exec(self.tr_queue_download.mapToGlobal(point))
def on_queue_download_remove_item(self, item: QtWidgets.QTreeWidgetItem) -> None:
"""Remove a specific item from the download queue."""
index = self.tr_queue_download.indexOfTopLevelItem(item)
if index >= 0:
self.tr_queue_download.takeTopLevelItem(index)
logger_gui.info("Removed item from download queue")
def on_copy_url_share(
self, tree_target: QtWidgets.QTreeWidget | QtWidgets.QTreeView, point: QtCore.QPoint = None
) -> None:
"""Copy the share URL of a media item to the clipboard."""
if isinstance(tree_target, QtWidgets.QTreeWidget):
item: QtWidgets.QTreeWidgetItem = tree_target.itemAt(point)
media = get_user_list_media_item(item)
else:
index: QtCore.QModelIndex = tree_target.indexAt(point)
media = get_results_media_item(index, self.proxy_tr_results, self.model_tr_results)
clipboard = QtWidgets.QApplication.clipboard()
url_share = media.share_url if hasattr(media, "share_url") else "No share URL available."
clipboard.clear()
clipboard.setText(url_share)
def thread_download_list_media(self, point: QtCore.QPoint) -> None:
"""Start download of a list media item in a thread."""
self.thread_it(self.playlist_manager.on_download_list_media, point)
def thread_download_album_from_track(self, point: QtCore.QPoint) -> None:
"""Starts the download of the full album from a selected track in a new thread."""
self.thread_it(self.on_download_album_from_track, point)
def on_download_album_from_track(self, point: QtCore.QPoint) -> None:
"""Adds the album associated with a selected track to the download queue."""
index: QtCore.QModelIndex = self.tr_results.indexAt(point)
media_track: Track = get_results_media_item(index, self.proxy_tr_results, self.model_tr_results)
if isinstance(media_track, Track) and media_track.album and media_track.album.id:
try:
full_album_object = self.tidal.session.album(media_track.album.id)
queue_dl_item = self.search_manager.media_to_queue_download_model(full_album_object)
if queue_dl_item:
self.queue_download_media(queue_dl_item)
else:
logger_gui.warning(f"Failed to create a queue item for album ID: {full_album_object.id}")
except Exception as e:
logger_gui.error(f"Could not fetch the full album from TIDAL. Error: {e}")
else:
logger_gui.warning("Could not retrieve album information from the selected track.")
def on_download_all_albums_from_playlist(self, point: QtCore.QPoint) -> None:
"""Download all unique albums from tracks in a playlist."""
try:
item = self.tr_lists_user.itemAt(point)
media_list = get_user_list_media_item(item)
if not isinstance(media_list, Playlist | UserPlaylist | Mix):
logger_gui.error("Please select a playlist or mix.")
return
logger_gui.info(f"Fetching all tracks from: {media_list.name}")
media_items = items_results_all(self.tidal.session, media_list)
album_ids = self._extract_album_ids_from_tracks(media_items)
if not album_ids:
logger_gui.warning("No albums found in this playlist.")
return
logger_gui.info(f"Found {len(album_ids)} unique albums. Loading with rate limiting...")
albums_dict = self._load_albums_with_rate_limiting(album_ids)
if not albums_dict:
logger_gui.error("Failed to load any albums from playlist.")
return
self._queue_loaded_albums(albums_dict)
message = f"Added {len(albums_dict)} albums to download queue"
self.s_statusbar_message.emit(StatusbarMessage(message=message, timeout=3000))
logger_gui.info(message)
except Exception as e:
error_msg = f"Error downloading albums from playlist: {e!s}"
logger_gui.error(error_msg)
self.s_statusbar_message.emit(StatusbarMessage(message=error_msg, timeout=3000))
def _extract_album_ids_from_tracks(self, media_items: list) -> dict[int, Album]:
"""Extract unique album IDs from a list of media items."""
album_ids = {}
for media_item in media_items:
if not isinstance(media_item, Track | Video):
continue
if not hasattr(media_item, "album") or not media_item.album:
continue
try:
album_id = media_item.album.id
if album_id:
album_ids[album_id] = media_item.album
except Exception as e:
logger_gui.debug(f"Skipping track with unavailable album: {e!s}")
continue
return album_ids
def _load_albums_with_rate_limiting(self, album_ids: dict[int, Album]) -> dict[int, Album]:
"""Load full album objects with rate limiting to prevent API throttling."""
albums_dict = {}
batch_size = self.settings.data.api_rate_limit_batch_size
delay_sec = self.settings.data.api_rate_limit_delay_sec
for idx, album_id in enumerate(album_ids.keys(), start=1):
try:
if idx > 1 and (idx - 1) % batch_size == 0:
logger_gui.info(f"⏰ RATE LIMITING: Processed {idx - 1} albums, pausing for {delay_sec} seconds...")
time.sleep(delay_sec)
if not self.tidal.session.check_login():
logger_gui.error("Session expired. Please restart the application and login again.")
return albums_dict
album = self.tidal.session.album(album_id)
albums_dict[album.id] = album
logger_gui.debug(f"Loaded album {idx}/{len(album_ids)}: {name_builder_artist(album)} - {album.name}")
except Exception as e:
if not self._handle_album_load_error(e, album_id):
return albums_dict
continue
logger_gui.info(f"Successfully loaded {len(albums_dict)} albums.")
return albums_dict
def _handle_album_load_error(self, error: Exception, album_id: int) -> bool:
"""Handle errors that occur when loading an album."""
if self.tidal.is_authentication_error(error):
error_msg = str(error)
logger_gui.error(f"Authentication error: {error_msg}")
logger_gui.error("Your session has expired. Please restart the application and login again.")
self.s_statusbar_message.emit(
StatusbarMessage(message="Session expired - please restart and login", timeout=5000)
)
return False
logger_gui.warning(f"Failed to load album {album_id}: {error!s}")
logger_gui.info(
"Note: Some albums may be unavailable due to region restrictions or removal from TIDAL. This is normal."
)
return True
def _queue_loaded_albums(self, albums_dict: dict[int, Album]) -> None:
"""Prepare and add loaded albums to the download queue."""
logger_gui.info(f"Preparing queue items for {len(albums_dict)} albums...")
queue_items = []
for album in albums_dict.values():
queue_dl_item = self.queue_manager.media_to_queue_download_model(album)
if queue_dl_item:
queue_items.append((queue_dl_item, album))
logger_gui.debug(f"Prepared: {name_builder_artist(album)} - {album.name}")
logger_gui.info(f"Adding {len(queue_items)} albums to queue...")
for queue_dl_item, album in queue_items:
self.queue_manager.queue_download_media(queue_dl_item)
logger_gui.info(f"Added: {name_builder_artist(album)} - {album.name}")
def on_search_in_app(self, search_term: str, search_type: str):
"""Perform a search within the application, selecting the correct category."""
self.l_search.setText(search_term)
search_type_map = {
"artist": Artist,
"album": Album,
"track": Track,
"video": Video,
"playlist": Playlist,
}
search_category = search_type_map.get(search_type.lower(), self.cb_search_type.currentData())
for i in range(self.cb_search_type.count()):
if self.cb_search_type.itemData(i) == search_category:
self.cb_search_type.setCurrentIndex(i)
break
self.search_manager.search_populate_results(search_term, search_category)
def on_search_in_browser(self, search_term: str, search_type: str):
"""Open a search in the default web browser."""
safe_term = urllib.parse.quote(search_term)
search_path_map = {
"artist": "artists",
"album": "albums",
"track": "tracks",
"video": "videos",
"playlist": "playlists",
}
search_path = search_path_map.get(search_type.lower())
if search_path:
url = QtCore.QUrl(f"https://listen.tidal.com/search/{search_path}?q={safe_term}")
else:
url = QtCore.QUrl(f"https://listen.tidal.com/search?q={safe_term}")
QtGui.QDesktopServices.openUrl(url)
+161
View File
@@ -0,0 +1,161 @@
"""Cover management for GUI - Handles cover loading, caching and display."""
from contextlib import suppress
from PySide6 import QtGui
from tidal_dl_ng.cache import CoverPixmapCache
from tidal_dl_ng.download import Download
from tidal_dl_ng.helper.path import resource_path
from tidal_dl_ng.logger import logger_gui
from tidal_dl_ng.worker import Worker
class CoverManager:
"""Manages cover art loading, caching and display operations."""
def __init__(self, parent_window, threadpool, info_tab_widget):
"""Initialize the cover manager.
Args:
parent_window: Main window instance
threadpool: QThreadPool for async operations
info_tab_widget: InfoTabWidget instance for display
"""
self.parent = parent_window
self.threadpool = threadpool
self.info_tab = info_tab_widget
self.cache = CoverPixmapCache()
self.cover_url_current = ""
def load_cover(self, media, use_cache_check: bool = True):
"""Load and display cover for media item.
Args:
media: Media object (Track, Album, etc.)
use_cache_check: If True, check cache before loading
"""
if use_cache_check:
# Try cache first for instant display
cover_url = self._get_cover_url(media)
if cover_url:
if cover_url == self.cover_url_current:
return # Already displayed
cached_pixmap = self.cache.get(cover_url)
if cached_pixmap:
self._display_cover(cached_pixmap, cover_url)
return
# Load asynchronously
worker = Worker(self._load_cover_async, media)
self.threadpool.start(worker)
def _load_cover_async(self, media):
"""Load cover in background thread."""
# Emit spinner on tab widget instead of InfoTabWidget (which is QObject, not QWidget)
tab_widget = self.info_tab.tab_widget if hasattr(self.info_tab, "tab_widget") else None
if tab_widget:
self.parent.s_spinner_start.emit(tab_widget)
try:
cover_url = self._get_cover_url(media)
if cover_url and self.cover_url_current != cover_url:
# Check cache again (thread-safe)
cached_pixmap = self.cache.get(cover_url)
if cached_pixmap:
self._display_cover(cached_pixmap, cover_url)
else:
# Download and cache
data_cover = Download.cover_data(cover_url)
pixmap = QtGui.QPixmap()
pixmap.loadFromData(data_cover)
self.cache.set(cover_url, pixmap)
self._display_cover(pixmap, cover_url)
elif not cover_url:
self._display_default_cover()
except Exception as e:
logger_gui.warning(f"Failed to load cover: {e}")
self._display_default_cover()
finally:
self.parent.s_spinner_stop.emit()
def _get_cover_url(self, media) -> str | None:
"""Extract cover URL from media object."""
with suppress(Exception):
if hasattr(media, "album") and media.album:
return media.album.image()
if hasattr(media, "image") and callable(getattr(media, "image", None)):
return media.image()
return None
def _display_cover(self, pixmap: QtGui.QPixmap, url: str):
"""Display a pixmap on the info tab."""
self.info_tab.set_cover_pixmap(pixmap)
self.info_tab.cover_url_current = url
self.cover_url_current = url
def _display_default_cover(self):
"""Display default cover image."""
path_image = resource_path("tidal_dl_ng/ui/default_album_image.png")
pixmap = QtGui.QPixmap(path_image)
self.info_tab.set_cover_pixmap(pixmap)
self.info_tab.cover_url_current = ""
self.cover_url_current = ""
def preload_covers_for_playlist(self, items: list) -> None:
"""Preload cover pixmaps for a list of tracks in background.
Args:
items: List of Track/Video objects to preload covers for.
"""
def worker() -> None:
# Extract unique cover URLs
cover_urls = set()
for item in items[:50]: # Limit to first 50
with suppress(Exception):
url = self._get_cover_url(item)
if url:
cover_urls.add(url)
# Preload each unique cover
for cover_url in cover_urls:
if self.cache.get(cover_url):
continue # Already cached
with suppress(Exception):
data_cover = Download.cover_data(cover_url)
pixmap = QtGui.QPixmap()
pixmap.loadFromData(data_cover)
self.cache.set(cover_url, pixmap)
logger_gui.debug(f"Preloaded cover: {cover_url[:50]}...")
worker_obj = Worker(worker)
self.threadpool.start(worker_obj)
def _queue_cover_fetch(self, media):
"""Queue a cover fetch operation for a media item."""
with suppress(Exception):
# previously try/except/pass
self.threadpool.start(Worker(self.load_cover, media))
def _fetch_cover_pixmap(self, media, use_cache_check: bool):
"""Fetch cover pixmap for media, with optional cache check."""
with suppress(Exception):
# previously try/except/continue inside loop
pixmap = self.cache.get(self._get_cover_url(media))
if pixmap:
return pixmap
# If not in cache, download cover
cover_url = self._get_cover_url(media)
if cover_url:
data_cover = Download.cover_data(cover_url)
pixmap = QtGui.QPixmap()
pixmap.loadFromData(data_cover)
self.cache.set(cover_url, pixmap)
return pixmap
return None
+209
View File
@@ -0,0 +1,209 @@
"""Downloads mixin for MainWindow.
Handles download queue and download operations.
"""
import time
from PySide6 import QtCore, QtWidgets
from tidalapi import Album, Artist, Mix, Playlist, Quality, Track, UserPlaylist, Video
from tidal_dl_ng.config import HandlingApp
from tidal_dl_ng.constants import QualityVideo, QueueDownloadStatus
from tidal_dl_ng.download import Download
from tidal_dl_ng.helper.gui import (
get_queue_download_media,
get_queue_download_quality_audio,
get_queue_download_quality_video,
get_results_media_item,
set_queue_download_media,
)
from tidal_dl_ng.helper.path import get_format_template
from tidal_dl_ng.helper.tidal import items_results_all
from tidal_dl_ng.logger import logger_gui
from tidal_dl_ng.model.gui_data import QueueDownloadItem, StatusbarMessage
class DownloadsMixin:
"""Mixin containing download-related methods."""
def on_download_results(self) -> None:
"""Download the selected results in the results tree."""
items = self.tr_results.selectionModel().selectedRows()
if len(items) == 0:
logger_gui.error("Please select a row first.")
else:
for item in items:
media = get_results_media_item(item, self.proxy_tr_results, self.model_tr_results)
queue_dl_item: QueueDownloadItem = self.search_manager.media_to_queue_download_model(media)
if queue_dl_item:
self.queue_download_media(queue_dl_item)
def queue_download_media(self, queue_dl_item: QueueDownloadItem) -> None:
"""Add a media item to the download queue."""
child: QtWidgets.QTreeWidgetItem = QtWidgets.QTreeWidgetItem()
child.setText(0, queue_dl_item.status)
set_queue_download_media(child, queue_dl_item.obj)
child.setText(2, queue_dl_item.name)
child.setText(3, queue_dl_item.type_media)
child.setText(4, queue_dl_item.quality_audio)
child.setText(5, queue_dl_item.quality_video)
self.tr_queue_download.addTopLevelItem(child)
def watcher_queue_download(self) -> None:
"""Monitor the download queue and process items as they become available."""
handling_app: HandlingApp = HandlingApp()
while not handling_app.event_abort.is_set():
items = self.tr_queue_download.findItems(
QueueDownloadStatus.Waiting, QtCore.Qt.MatchFlag.MatchExactly, column=0
)
if len(items) > 0:
item: QtWidgets.QTreeWidgetItem = items[0]
media = get_queue_download_media(item)
quality_audio: Quality = get_queue_download_quality_audio(item)
quality_video: QualityVideo = get_queue_download_quality_video(item)
try:
self.s_queue_download_item_downloading.emit(item)
result = self.on_queue_download(media, quality_audio=quality_audio, quality_video=quality_video)
if result == QueueDownloadStatus.Finished:
self.s_queue_download_item_finished.emit(item)
elif result == QueueDownloadStatus.Skipped:
self.s_queue_download_item_skipped.emit(item)
except Exception as e:
logger_gui.error(e)
self.s_queue_download_item_failed.emit(item)
else:
time.sleep(2)
def on_queue_download_item_downloading(self, item: QtWidgets.QTreeWidgetItem) -> None:
"""Update the status of a queue download item to 'Downloading'."""
self.queue_download_item_status(item, QueueDownloadStatus.Downloading)
def on_queue_download_item_finished(self, item: QtWidgets.QTreeWidgetItem) -> None:
"""Update the status of a queue download item to 'Finished'."""
self.queue_download_item_status(item, QueueDownloadStatus.Finished)
def on_queue_download_item_failed(self, item: QtWidgets.QTreeWidgetItem) -> None:
"""Update the status of a queue download item to 'Failed'."""
self.queue_download_item_status(item, QueueDownloadStatus.Failed)
def on_queue_download_item_skipped(self, item: QtWidgets.QTreeWidgetItem) -> None:
"""Update the status of a queue download item to 'Skipped'."""
self.queue_download_item_status(item, QueueDownloadStatus.Skipped)
def queue_download_item_status(self, item: QtWidgets.QTreeWidgetItem, status: str) -> None:
"""Set the status text of a queue download item."""
item.setText(0, status)
def on_queue_download(
self,
media: Track | Album | Playlist | Video | Mix | Artist,
quality_audio: Quality | None = None,
quality_video: QualityVideo | None = None,
) -> QueueDownloadStatus:
"""Download the specified media item(s) and return the result status."""
result: QueueDownloadStatus
items_media: list = items_results_all(self.tidal.session, media) if isinstance(media, Artist) else [media]
download_delay: bool = bool(isinstance(media, Track | Video) and self.settings.data.download_delay)
for item_media in items_media:
result = self.download(
item_media,
self.dl,
delay_track=download_delay,
quality_audio=quality_audio,
quality_video=quality_video,
)
return result
def download(
self,
media: Track | Album | Playlist | Video | Mix | Artist,
dl: Download,
delay_track: bool = False,
quality_audio: Quality | None = None,
quality_video: QualityVideo | None = None,
) -> QueueDownloadStatus:
"""Download a media item and return the result status."""
result_dl: bool
path_file: str
result: QueueDownloadStatus
self.s_pb_reset.emit()
self.s_statusbar_message.emit(StatusbarMessage(message="Download started..."))
file_template = get_format_template(media, self.settings)
# Determine source information
source_type = "manual"
source_id = None
source_name = None
if isinstance(media, Album):
source_type = "album"
source_id = str(media.id)
source_name = media.name
elif isinstance(media, Playlist | UserPlaylist):
source_type = "playlist"
source_id = str(media.id) if hasattr(media, "id") else None
source_name = media.name if hasattr(media, "name") else None
elif isinstance(media, Mix):
source_type = "mix"
source_id = str(media.id)
source_name = media.title
elif isinstance(media, Track):
if hasattr(media, "album") and media.album:
source_type = "album"
source_id = str(media.album.id)
source_name = media.album.name
else:
source_type = "track"
source_id = str(media.id)
source_name = media.name
if isinstance(media, Track | Video):
result_dl, path_file = dl.item(
media=media,
file_template=file_template,
download_delay=delay_track,
quality_audio=quality_audio,
quality_video=quality_video,
source_type=source_type,
source_id=source_id,
source_name=source_name,
)
elif isinstance(media, Album | Playlist | Mix):
dl.items(
media=media,
file_template=file_template,
video_download=self.settings.data.video_download,
download_delay=self.settings.data.download_delay,
quality_audio=quality_audio,
quality_video=quality_video,
source_type=source_type,
source_id=source_id,
source_name=source_name,
)
result_dl = True
path_file = "dummy"
self.s_statusbar_message.emit(StatusbarMessage(message="Download finished.", timeout=2000))
if result_dl and path_file:
result = QueueDownloadStatus.Finished
elif not result_dl and path_file:
result = QueueDownloadStatus.Skipped
else:
result = QueueDownloadStatus.Failed
return result
+85
View File
@@ -0,0 +1,85 @@
"""History management mixin for MainWindow.
Handles download history and duplicate prevention.
"""
from PySide6 import QtCore
from tidalapi import Track
from tidal_dl_ng.dialog import DialogPreferences
from tidal_dl_ng.dialog_history import DialogHistory
from tidal_dl_ng.logger import logger_gui
from tidal_dl_ng.model.gui_data import StatusbarMessage
class HistoryMixin:
"""Mixin containing download history management methods."""
def on_view_history(self) -> None:
"""Open the download history dialog."""
DialogHistory(history_service=self.history_service, parent=self)
def on_toggle_duplicate_prevention(self, enabled: bool) -> None:
"""Toggle duplicate download prevention on or off."""
self.history_service.update_settings(preventDuplicates=enabled)
status_msg = "enabled" if enabled else "disabled"
logger_gui.info(f"Duplicate download prevention {status_msg}")
self.s_statusbar_message.emit(StatusbarMessage(message=f"Duplicate prevention {status_msg}.", timeout=2500))
def on_mark_track_as_downloaded(self, track: Track, index: QtCore.QModelIndex) -> None:
"""Mark a track as downloaded in history."""
track_id = str(track.id)
source_type = "manual"
source_id = None
source_name = None
if hasattr(track, "album") and track.album:
source_type = "album"
source_id = str(track.album.id)
source_name = track.album.name
self.history_service.add_track_to_history(
track_id=track_id, source_type=source_type, source_id=source_id, source_name=source_name
)
self._update_downloaded_column(index, True)
logger_gui.info(f"Marked track as downloaded: {track.name}")
def on_mark_track_as_not_downloaded(self, track_id: str, index: QtCore.QModelIndex) -> None:
"""Remove a track from download history."""
success = self.history_service.remove_track_from_history(track_id)
if success:
self._update_downloaded_column(index, False)
logger_gui.info(f"Unmarked track (ID: {track_id})")
def _update_downloaded_column(self, index: QtCore.QModelIndex, is_downloaded: bool) -> None:
"""Update the Downloaded? column for a specific index."""
source_index = self.proxy_tr_results.mapToSource(index)
item = self.model_tr_results.itemFromIndex(source_index)
if not item:
return
row = item.row()
parent = item.parent()
downloaded_item = self.model_tr_results.item(row, 8) if parent is None else parent.child(row, 8)
if downloaded_item:
if is_downloaded:
downloaded_item.setText("✅")
downloaded_item.setTextAlignment(QtCore.Qt.AlignmentFlag.AlignCenter)
else:
downloaded_item.setText("")
def on_preferences(self) -> None:
"""Open the preferences dialog."""
DialogPreferences(settings=self.settings, settings_save=self.s_settings_save, parent=self)
def on_settings_save(self) -> None:
"""Save settings and re-apply them to the GUI."""
self.settings.save()
self.apply_settings(self.settings)
self._init_dl()
+192
View File
@@ -0,0 +1,192 @@
"""Initialization mixin for MainWindow.
Handles all initialization-related methods for the GUI components.
"""
from collections.abc import Iterable
from typing import Any
from ansi2html import Ansi2HTMLConverter
from PySide6 import QtCore, QtGui, QtWidgets
from rich.progress import Progress
from tidal_dl_ng.config import HandlingApp
from tidal_dl_ng.download import Download
from tidal_dl_ng.helper.gui import FilterHeader, HumanProxyModel
from tidal_dl_ng.helper.path import resource_path
from tidal_dl_ng.logger import logger_gui
from tidal_dl_ng.model.gui_data import ProgressBars
from tidal_dl_ng.ui.spinner import QtWaitingSpinner
class InitializationMixin:
"""Mixin containing initialization methods for MainWindow."""
def _init_gui(self) -> None:
"""Initialize GUI-specific variables and state."""
self.setGeometry(
self.settings.data.window_x,
self.settings.data.window_y,
self.settings.data.window_w,
self.settings.data.window_h,
)
self.spinners: dict[QtWidgets.QWidget, QtWaitingSpinner] = {}
self.converter_ansi_html: Ansi2HTMLConverter = Ansi2HTMLConverter()
def _init_threads(self):
"""Initialize thread pool and start background workers."""
self.threadpool = QtCore.QThreadPool()
self.thread_it(self.queue_manager.watcher_queue_download)
def _init_dl(self):
"""Initialize Download object and related progress bars."""
data_pb: ProgressBars = ProgressBars(
item=self.s_item_advance,
list_item=self.s_list_advance,
item_name=self.s_item_name,
list_name=self.s_list_name,
)
progress: Progress = Progress()
handling_app: HandlingApp = HandlingApp()
self.dl = Download(
tidal_obj=self.tidal,
skip_existing=self.tidal.settings.data.skip_existing,
path_base=self.settings.data.download_base_path,
fn_logger=logger_gui,
progress_gui=data_pb,
progress=progress,
event_abort=handling_app.event_abort,
event_run=handling_app.event_run,
)
def _init_progressbar(self):
"""Initialize and add progress bars to the status bar."""
self.pb_list = QtWidgets.QProgressBar()
self.pb_item = QtWidgets.QProgressBar()
pbs = [self.pb_list, self.pb_item]
for pb in pbs:
pb.setRange(0, 100)
self.statusbar.addPermanentWidget(pb)
def _init_info(self):
"""Set default album cover image in the GUI."""
path_image: str = resource_path("tidal_dl_ng/ui/default_album_image.png")
self.l_pm_cover.setPixmap(QtGui.QPixmap(path_image))
def _init_tree_results(self, tree: QtWidgets.QTreeView, model: QtGui.QStandardItemModel) -> None:
"""Initialize the results tree view and its model."""
header: FilterHeader = FilterHeader(tree)
self.proxy_tr_results: HumanProxyModel = HumanProxyModel(self)
tree.setHeader(header)
tree.setModel(model)
self.proxy_tr_results.setSourceModel(model)
tree.setModel(self.proxy_tr_results)
header.set_filter_boxes(model.columnCount())
header.filter_activated.connect(self.handle_filter_activated)
tree.sortByColumn(0, QtCore.Qt.SortOrder.AscendingOrder)
tree.setColumnHidden(1, True)
normal_width = max(150, (self.width() * 0.13))
narrow_width = max(90, (self.width() * 0.06))
skinny_width = max(60, (self.width() * 0.03))
tree.setColumnWidth(2, normal_width)
tree.setColumnWidth(3, normal_width)
tree.setColumnWidth(4, normal_width)
tree.setColumnWidth(5, skinny_width)
tree.setColumnWidth(6, narrow_width)
tree.setColumnWidth(7, narrow_width)
tree.setColumnWidth(8, skinny_width)
header.setSectionResizeMode(0, QtWidgets.QHeaderView.ResizeToContents)
tree.setContextMenuPolicy(QtCore.Qt.CustomContextMenu)
tree.customContextMenuRequested.connect(self.menu_context_tree_results)
from tidal_dl_ng.helper.hover_manager import HoverManager
self.hover_manager = HoverManager(
tree_view=tree,
proxy_model=self.proxy_tr_results,
source_model=model,
debounce_delay_ms=50,
parent=self,
)
self.hover_manager.s_hover_confirmed.connect(self.on_track_hover_confirmed)
self.hover_manager.s_hover_left.connect(self.on_track_hover_left)
def _init_tree_results_model(self, model: QtGui.QStandardItemModel) -> None:
"""Initialize the model for the results tree view."""
labels_column: list[str] = [
"#",
"obj",
"Artist",
"Title",
"Album",
"Duration",
"Quality",
"Date",
"Downloaded?",
]
model.setColumnCount(len(labels_column))
model.setRowCount(0)
model.setHorizontalHeaderLabels(labels_column)
def _init_tree_queue(self, tree: QtWidgets.QTableWidget) -> None:
"""Initialize the download queue table widget."""
tree.setColumnHidden(1, True)
tree.setColumnWidth(2, 200)
header = tree.header()
if hasattr(header, "setSectionResizeMode"):
header.setSectionResizeMode(0, QtWidgets.QHeaderView.ResizeToContents)
tree.setContextMenuPolicy(QtCore.Qt.CustomContextMenu)
tree.customContextMenuRequested.connect(self.menu_context_queue_download)
def _init_tree_lists(self, tree: QtWidgets.QTreeWidget) -> None:
"""Initialize the user lists tree widget."""
tree.setColumnWidth(0, 200)
tree.setColumnHidden(1, True)
tree.setColumnWidth(2, 300)
tree.expandAll()
tree.setContextMenuPolicy(QtCore.Qt.CustomContextMenu)
tree.customContextMenuRequested.connect(self.playlist_manager.menu_context_tree_lists)
def _init_menu_actions(self) -> None:
"""Initialize custom menu actions."""
menubar = self.menuBar()
tools_menu = None
for action in menubar.actions():
if action.text() == "Tools":
tools_menu = action.menu()
break
if not tools_menu:
tools_menu = menubar.addMenu("Tools")
self.a_view_history = QtGui.QAction("View Download History...", self)
self.a_view_history.triggered.connect(self.on_view_history)
tools_menu.addAction(self.a_view_history)
tools_menu.addSeparator()
self.a_toggle_duplicate_prevention = QtGui.QAction("Prevent Duplicate Downloads", self)
self.a_toggle_duplicate_prevention.setCheckable(True)
is_preventing = self.history_service.get_settings().get("preventDuplicates", True)
self.a_toggle_duplicate_prevention.setChecked(is_preventing)
self.a_toggle_duplicate_prevention.triggered.connect(self.on_toggle_duplicate_prevention)
tools_menu.addAction(self.a_toggle_duplicate_prevention)
def _populate_quality(self, ui_target: QtWidgets.QComboBox, options: Iterable[Any]) -> None:
"""Populate a combo box with quality options."""
for item in options:
ui_target.addItem(item.name, item)
def _populate_search_types(self, ui_target: QtWidgets.QComboBox, options: Iterable[Any]) -> None:
"""Populate a combo box with search type options."""
for item in options:
if item:
ui_target.addItem(item.__name__, item)
self.cb_search_type.setCurrentIndex(2)
+224
View File
@@ -0,0 +1,224 @@
"""Main window for TIDAL Downloader Next Generation.
This module combines all GUI functionality through mixins to create
the main application window.
"""
from collections.abc import Callable
from PySide6 import QtCore, QtGui, QtWidgets
from tidalapi import Quality
from tidal_dl_ng.cache import TrackExtrasCache
from tidal_dl_ng.config import HandlingApp, Settings, Tidal
from tidal_dl_ng.constants import QualityVideo
from tidal_dl_ng.download import Download
from tidal_dl_ng.gui.context_menus import ContextMenusMixin
from tidal_dl_ng.gui.covers import CoverManager
from tidal_dl_ng.gui.downloads import DownloadsMixin
from tidal_dl_ng.gui.history import HistoryMixin
from tidal_dl_ng.gui.initialization import InitializationMixin
from tidal_dl_ng.gui.playlist import GuiPlaylistManager
from tidal_dl_ng.gui.progress import ProgressMixin
from tidal_dl_ng.gui.queue import GuiQueueManager
from tidal_dl_ng.gui.search import GuiSearchManager
from tidal_dl_ng.gui.signals import SignalsMixin
from tidal_dl_ng.gui.tidal_session import TidalSessionMixin
from tidal_dl_ng.gui.track_extras import TrackExtrasMixin
from tidal_dl_ng.gui.trees_results import TreesResultsMixin
from tidal_dl_ng.gui.ui_helpers import UIHelpersMixin
from tidal_dl_ng.gui.updates import UpdatesMixin
from tidal_dl_ng.helper.gui import HumanProxyModel
from tidal_dl_ng.helper.hover_manager import HoverManager
from tidal_dl_ng.history import HistoryService
from tidal_dl_ng.logger import XStream, logger_gui
from tidal_dl_ng.ui.info_tab_widget import InfoTabWidget
from tidal_dl_ng.ui.main import Ui_MainWindow
from tidal_dl_ng.worker import Worker
# TODO: Make more use of Exceptions
class MainWindow(
QtWidgets.QMainWindow,
Ui_MainWindow,
InitializationMixin,
TidalSessionMixin,
SignalsMixin,
ProgressMixin,
UIHelpersMixin,
TrackExtrasMixin,
UpdatesMixin,
DownloadsMixin,
TreesResultsMixin,
ContextMenusMixin,
HistoryMixin,
):
"""Main application window for TIDAL Downloader Next Generation.
Handles GUI setup, user interactions, and download logic through
a combination of mixins for better code organization.
"""
# Type hints for class attributes
settings: Settings
tidal: Tidal
dl: Download
history_service: HistoryService
threadpool: QtCore.QThreadPool
tray: QtWidgets.QSystemTrayIcon
spinners: dict
cover_url_current: str = ""
shutdown: bool = False
model_tr_results: QtGui.QStandardItemModel = QtGui.QStandardItemModel()
proxy_tr_results: HumanProxyModel
info_tab_widget: InfoTabWidget
hover_manager: HoverManager
queue_manager: GuiQueueManager
playlist_manager: GuiPlaylistManager
search_manager: GuiSearchManager
cover_manager: CoverManager
track_extras_cache: TrackExtrasCache
_pending_extras_workers: dict[str, Worker]
_track_extras_callbacks: dict[str, Callable]
# Qt Signals
s_spinner_start: QtCore.Signal = QtCore.Signal(QtWidgets.QWidget)
s_spinner_stop: QtCore.Signal = QtCore.Signal()
s_track_extras_ready: QtCore.Signal = QtCore.Signal(str, object)
s_invoke_callback: QtCore.Signal = QtCore.Signal(str, object)
pb_item: QtWidgets.QProgressBar
s_item_advance: QtCore.Signal = QtCore.Signal(float)
s_item_name: QtCore.Signal = QtCore.Signal(str)
s_list_name: QtCore.Signal = QtCore.Signal(str)
pb_list: QtWidgets.QProgressBar
s_list_advance: QtCore.Signal = QtCore.Signal(float)
s_pb_reset: QtCore.Signal = QtCore.Signal()
s_populate_tree_lists: QtCore.Signal = QtCore.Signal(dict)
s_populate_folder_children: QtCore.Signal = QtCore.Signal(object, list, list)
s_statusbar_message: QtCore.Signal = QtCore.Signal(object)
s_tr_results_add_top_level_item: QtCore.Signal = QtCore.Signal(object)
s_settings_save: QtCore.Signal = QtCore.Signal()
s_pb_reload_status: QtCore.Signal = QtCore.Signal(bool)
s_update_check: QtCore.Signal = QtCore.Signal(bool)
s_update_show: QtCore.Signal = QtCore.Signal(bool, bool, object)
s_queue_download_item_downloading: QtCore.Signal = QtCore.Signal(object)
s_queue_download_item_finished: QtCore.Signal = QtCore.Signal(object)
s_queue_download_item_failed: QtCore.Signal = QtCore.Signal(object)
s_queue_download_item_skipped: QtCore.Signal = QtCore.Signal(object)
def __init__(self, tidal: Tidal | None = None) -> None:
"""Initialize the main window and all components.
Args:
tidal (Tidal | None): Optional Tidal session object.
"""
super().__init__()
self.setupUi(self)
self.setWindowTitle("TIDAL Downloader Next Generation!")
# Initialize settings first
self.settings = Settings()
# Initialize managers that depend on settings
self.queue_manager = GuiQueueManager(self)
self.playlist_manager = GuiPlaylistManager(self)
self.search_manager = GuiSearchManager(self)
self.info_tab_widget = InfoTabWidget(self, self.tabWidget)
# Logging redirect
XStream.stdout().messageWritten.connect(self._log_output)
self.history_service = HistoryService()
# Core components
self._init_threads()
self._init_gui()
self.track_extras_cache = TrackExtrasCache()
self._pending_extras_workers: dict[str, Worker] = {}
self._track_extras_callbacks: dict[str, Callable] = {}
# Managers that have dependencies
self.cover_manager = CoverManager(self, self.threadpool, self.info_tab_widget)
# Initialize the rest of the UI
self.info_tab_widget.set_track_extras_provider(self.get_track_extras)
self._init_tree_results_model(self.model_tr_results)
self._init_tree_results(self.tr_results, self.model_tr_results)
self.playlist_manager.init_ui()
self.queue_manager.init_ui()
self._init_tree_lists(self.tr_lists_user)
self._init_tree_queue(self.tr_queue_download)
self._init_info()
self._init_progressbar()
self._populate_quality(self.cb_quality_audio, Quality)
self._populate_quality(self.cb_quality_video, QualityVideo)
from tidalapi.session import SearchTypes
self._populate_search_types(self.cb_search_type, SearchTypes)
self.apply_settings(self.settings)
self._init_menu_actions()
self._init_signals()
# Connect signal for invoking track extras callbacks
self.s_invoke_callback.connect(self._on_invoke_callback)
# Connect playlist manager signals
self.s_populate_tree_lists.connect(self.on_populate_tree_lists)
self.init_tidal(tidal)
logger_gui.debug("All setup.")
def closeEvent(self, event: QtGui.QCloseEvent) -> None:
"""Handle the close event of the main window.
Args:
event (QtGui.QCloseEvent): The close event.
"""
# Save the main window size and position
self.settings.data.window_x = self.x()
self.settings.data.window_y = self.y()
self.settings.data.window_w = self.width()
self.settings.data.window_h = self.height()
self.settings.save()
self.shutdown = True
handling_app: HandlingApp = HandlingApp()
handling_app.event_abort.set()
event.accept()
def apply_settings(self, settings: Settings) -> None:
"""Apply user settings to the GUI.
Args:
settings (Settings): The settings object.
"""
quality_audio = getattr(getattr(settings, "data", None), "quality_audio", 1)
quality_video = getattr(getattr(settings, "data", None), "quality_video", 0)
elements = [
{"element": self.cb_quality_audio, "setting": quality_audio, "default_id": 1},
{"element": self.cb_quality_video, "setting": quality_video, "default_id": 0},
]
for item in elements:
idx = item["element"].findData(item["setting"])
if idx > -1:
item["element"].setCurrentIndex(idx)
else:
item["element"].setCurrentIndex(item["default_id"])
def thread_it(self, fn: Callable, *args, **kwargs) -> None:
"""Run a function in a separate thread.
Args:
fn (Callable): The function to run.
*args: Positional arguments for the function.
**kwargs: Keyword arguments for the function.
"""
worker = Worker(fn, *args, **kwargs)
self.threadpool.start(worker)
+697
View File
@@ -0,0 +1,697 @@
# tidal_dl_ng/playlist.py
from typing import TYPE_CHECKING, Any
from PySide6 import QtCore, QtWidgets
from tidalapi import Album, Artist, Mix, Playlist, Track, UserPlaylist
from tidalapi.playlist import Folder
from tidal_dl_ng.constants import FAVORITES, TidalLists
from tidal_dl_ng.helper.gui import get_user_list_media_item, set_user_list_media
from tidal_dl_ng.helper.tidal import (
favorite_function_factory,
items_results_all,
user_media_lists,
)
from tidal_dl_ng.logger import logger_gui
from tidal_dl_ng.model.gui_data import QueueDownloadItem, ResultItem, StatusbarMessage
if TYPE_CHECKING:
from tidal_dl_ng.gui import MainWindow
class GuiPlaylistManager:
"""Manages the playlist, mixes, and favorites GUI and logic."""
def __init__(self, main_window: "MainWindow"):
"""Initialize the playlist manager."""
self.main_window: "MainWindow" = main_window
self.settings = main_window.settings
def init_ui(self):
"""Initialize UI elements related to playlists."""
self._init_tree_lists(self.main_window.tr_lists_user)
def connect_signals(self):
"""Connect signals for playlist-related widgets."""
self.main_window.pb_reload_user_lists.clicked.connect(lambda: self.main_window.thread_it(self.tidal_user_lists))
self.main_window.tr_lists_user.itemClicked.connect(self.on_list_items_show)
self.main_window.tr_lists_user.itemExpanded.connect(self.on_tr_lists_user_expanded)
self.main_window.tr_lists_user.customContextMenuRequested.connect(self.menu_context_tree_lists)
self.main_window.s_populate_tree_lists.connect(self.on_populate_tree_lists)
self.main_window.s_populate_folder_children.connect(self.on_populate_folder_children)
def _init_tree_lists(self, tree: QtWidgets.QTreeWidget) -> None:
"""Initialize the user lists tree widget."""
tree.setColumnWidth(0, 200)
tree.setColumnHidden(1, True)
tree.setColumnWidth(2, 300)
tree.expandAll()
tree.setContextMenuPolicy(QtCore.Qt.CustomContextMenu)
def tidal_user_lists(self) -> None:
"""Fetch and emit user playlists, mixes, and favorites from Tidal."""
self.main_window.s_spinner_start.emit(self.main_window.tr_lists_user)
self.main_window.s_pb_reload_status.emit(False)
user_all = user_media_lists(self.main_window.tidal.session)
self.main_window.s_populate_tree_lists.emit(user_all)
def on_populate_tree_lists(self, user_lists: dict[str, list]) -> None:
"""Populate the user lists tree with playlists, mixes, and favorites."""
twi_playlists = self.main_window.tr_lists_user.findItems(TidalLists.Playlists, QtCore.Qt.MatchExactly, 0)[0]
twi_mixes = self.main_window.tr_lists_user.findItems(TidalLists.Mixes, QtCore.Qt.MatchExactly, 0)[0]
twi_favorites = self.main_window.tr_lists_user.findItems(TidalLists.Favorites, QtCore.Qt.MatchExactly, 0)[0]
for twi in [twi_playlists, twi_mixes, twi_favorites]:
for i in reversed(range(twi.childCount())):
twi.removeChild(twi.child(i))
for item in user_lists.get("playlists", []):
if isinstance(item, Folder):
twi_child = QtWidgets.QTreeWidgetItem(twi_playlists)
twi_child.setText(0, f"📁 {item.name}")
set_user_list_media(twi_child, item)
info = f"({item.total_number_of_items} items)" if item.total_number_of_items else ""
twi_child.setText(2, info)
dummy_child = QtWidgets.QTreeWidgetItem(twi_child)
dummy_child.setDisabled(True)
elif isinstance(item, UserPlaylist | Playlist):
twi_child = QtWidgets.QTreeWidgetItem(twi_playlists)
name = item.name or ""
description = f" {item.description}" if item.description else ""
info = f"({item.num_tracks + item.num_videos} Tracks){description}"
twi_child.setText(0, name)
set_user_list_media(twi_child, item)
twi_child.setText(2, info)
for item in user_lists.get("mixes", []):
if isinstance(item, Mix):
twi_child = QtWidgets.QTreeWidgetItem(twi_mixes)
twi_child.setText(0, item.title)
set_user_list_media(twi_child, item)
twi_child.setText(2, item.sub_title)
for key, favorite in FAVORITES.items():
twi_child = QtWidgets.QTreeWidgetItem(twi_favorites)
twi_child.setText(0, favorite["name"])
set_user_list_media(twi_child, key)
self.main_window.s_spinner_stop.emit()
self.main_window.s_pb_reload_status.emit(True)
def menu_context_tree_lists(self, point: QtCore.QPoint) -> None:
"""Show context menu for user lists tree.
Args:
point (QPoint): The point where the menu is requested.
"""
# Infos about the node selected.
index = self.main_window.tr_lists_user.indexAt(point)
# Do not open menu if something went wrong or a parent node is clicked.
if not index.isValid() or not index.parent().data():
return
# Get the media item to determine type
item = self.main_window.tr_lists_user.itemAt(point)
media = get_user_list_media_item(item)
# We build the menu.
menu = QtWidgets.QMenu()
if isinstance(media, Folder):
# Folder-specific menu items
menu.addAction(
"Download All Playlists in Folder",
lambda: self.main_window.thread_it(self.on_download_folder_playlists, point),
)
menu.addAction(
"Download All Albums from Folder",
lambda: self.main_window.thread_it(self.on_download_folder_albums, point),
)
elif isinstance(media, str):
# Favorites items (stored as string keys like "fav_tracks", "fav_albums")
menu.addAction("Download All Items", lambda: self.main_window.thread_it(self.on_download_favorites, point))
menu.addAction(
"Download All Albums from Items",
lambda: self.main_window.thread_it(self.on_download_albums_from_favorites, point),
)
else:
# Playlist/Mix menu items (existing)
menu.addAction("Download Playlist", lambda: self.main_window.thread_download_list_media(point))
menu.addAction(
"Download All Albums in Playlist",
lambda: self.main_window.thread_it(self.on_download_all_albums_from_playlist, point),
)
menu.addAction(
"Copy Share URL", lambda: self.main_window.on_copy_url_share(self.main_window.tr_lists_user, point)
)
menu.exec(self.main_window.tr_lists_user.mapToGlobal(point))
def on_download_list_media(self, point: QtCore.QPoint | None = None) -> None:
"""Download all media items in a selected list.
Args:
point (QPoint | None, optional): The point in the tree. Defaults to None.
"""
items: list[QtWidgets.QTreeWidgetItem] = []
if point:
items = [self.main_window.tr_lists_user.itemAt(point)]
else:
items = self.main_window.tr_lists_user.selectedItems()
if len(items) == 0:
logger_gui.error("Please select a mix or playlist first.")
for item in items:
media = get_user_list_media_item(item)
queue_dl_item: QueueDownloadItem | None = self.main_window.search_manager.media_to_queue_download_model(
media
)
if queue_dl_item:
self.main_window.queue_download_media(queue_dl_item)
def on_download_folder_playlists(self, point: QtCore.QPoint) -> None:
"""Download all playlists in a folder.
Args:
point (QPoint): The point in the tree where the folder was right-clicked.
"""
try:
# Get and validate the folder
item = self.main_window.tr_lists_user.itemAt(point)
media = get_user_list_media_item(item)
if not isinstance(media, Folder):
logger_gui.error("Please select a folder.")
return
# Fetch all playlists in the folder
logger_gui.info(f"Fetching playlists from folder: {media.name}")
playlists = self._get_folder_playlists(media)
if not playlists:
logger_gui.info(f"No playlists found in folder: {media.name}")
return
# Queue each playlist for download
logger_gui.info(f"Queueing {len(playlists)} playlists from folder: {media.name}")
for playlist in playlists:
queue_dl_item: QueueDownloadItem | None = self.main_window.search_manager.media_to_queue_download_model(
playlist
)
if queue_dl_item:
self.main_window.queue_download_media(queue_dl_item)
logger_gui.info(f"✅ Successfully queued {len(playlists)} playlists from folder: {media.name}")
except Exception as e:
logger_gui.exception(f"Error downloading playlists from folder: {e}")
logger_gui.error("Failed to download playlists from folder. See log for details.")
def on_download_folder_albums(self, point: QtCore.QPoint) -> None:
"""Download all unique albums from all playlists in a folder.
Args:
point (QPoint): The point in the tree where the folder was right-clicked.
"""
try:
# Get and validate the folder
item = self.main_window.tr_lists_user.itemAt(point)
media = get_user_list_media_item(item)
if not isinstance(media, Folder):
logger_gui.error("Please select a folder.")
return
# Fetch all playlists in the folder
logger_gui.info(f"Fetching playlists from folder: {media.name}")
playlists = self._get_folder_playlists(media)
if not playlists:
logger_gui.info(f"No playlists found in folder: {media.name}")
return
logger_gui.info(f"Found {len(playlists)} playlists in folder: {media.name}")
# Collect all tracks from all playlists
all_tracks: list[Track] = []
for playlist in playlists:
try:
tracks = self._get_playlist_tracks(playlist)
all_tracks.extend(tracks)
logger_gui.debug(f"Collected {len(tracks)} tracks from playlist: {playlist.name}")
except Exception as e:
logger_gui.error(f"Error getting tracks from playlist '{playlist.name}': {e}")
continue
if not all_tracks:
logger_gui.info(f"No tracks found in folder playlists: {media.name}")
return
logger_gui.info(f"Collected {len(all_tracks)} total tracks from all playlists")
# Extract unique album IDs
album_ids = self.main_window._extract_album_ids_from_tracks(all_tracks)
logger_gui.info(f"Found {len(album_ids)} unique albums across all playlists in folder: {media.name}")
if not album_ids:
logger_gui.info("No albums found to download.")
return
# Load full album objects with rate limiting
albums_dict = self.main_window._load_albums_with_rate_limiting(album_ids)
if not albums_dict:
logger_gui.error("Failed to load any albums.")
return
# Queue the albums for download
self.main_window._queue_loaded_albums(albums_dict)
logger_gui.info(f"✅ Successfully queued {len(albums_dict)} unique albums from folder: {media.name}")
except Exception as e:
logger_gui.exception(f"Error downloading albums from folder: {e}")
logger_gui.error("Failed to download albums from folder. See log for details.")
def on_download_favorites(self, point: QtCore.QPoint) -> None:
"""Download all items from a Favorites category.
Args:
point (QPoint): The point in the tree where the favorites item was right-clicked.
"""
try:
# Get and validate the favorites item
item = self.main_window.tr_lists_user.itemAt(point)
media = get_user_list_media_item(item)
if not isinstance(media, str):
logger_gui.error("Please select a favorites category.")
return
# Get the favorites category name for logging
favorite_name = FAVORITES.get(media, {}).get("name", media)
logger_gui.info(f"Fetching all items from favorites: {favorite_name}")
# Use the factory to get the appropriate favorites function
favorite_function = favorite_function_factory(self.main_window.tidal, media)
# Fetch all items from this favorites category
media_items = favorite_function()
if not media_items:
logger_gui.info(f"No items found in favorites: {favorite_name}")
return
logger_gui.info(f"Found {len(media_items)} items in favorites: {favorite_name}")
# Queue each item for download
queued_count = 0
for media_item in media_items:
queue_dl_item: QueueDownloadItem | None = self.main_window.search_manager.media_to_queue_download_model(
media_item
)
if queue_dl_item:
self.main_window.queue_download_media(queue_dl_item)
queued_count += 1
logger_gui.info(f"✅ Successfully queued {queued_count} items from favorites: {favorite_name}")
except Exception as e:
logger_gui.exception(f"Error downloading favorites: {e}")
logger_gui.error("Failed to download favorites. See log for details.")
def _download_albums_from_fav_artists(self, media_items: list) -> None:
"""Download all albums from favorite artists."""
all_albums = {}
for artist in media_items:
try:
artist_albums = items_results_all(self.main_window.tidal.session, artist)
for album in artist_albums:
if isinstance(album, Album) and album.id:
all_albums[album.id] = album
except Exception as e:
logger_gui.error(f"Error getting albums from artist '{artist.name}': {e}")
if all_albums:
self.main_window._queue_loaded_albums(all_albums)
def _download_albums_from_fav_tracks(self, media_items: list) -> None:
"""Download all albums from favorite tracks."""
album_ids = self.main_window._extract_album_ids_from_tracks(media_items)
if album_ids:
albums_dict = self.main_window._load_albums_with_rate_limiting(album_ids)
if albums_dict:
self.main_window._queue_loaded_albums(albums_dict)
def on_download_albums_from_favorites(self, point: QtCore.QPoint) -> None:
"""Download all unique albums from items in a Favorites category.
Args:
point (QPoint): The point in the tree where the favorites item was right-clicked.
"""
try:
# Get and validate the favorites item
item = self.main_window.tr_lists_user.itemAt(point)
media = get_user_list_media_item(item)
if not isinstance(media, str):
logger_gui.error("Please select a favorites category.")
return
# Get the favorites category name for logging
favorite_name = FAVORITES.get(media, {}).get("name", media)
logger_gui.info(f"Fetching all items from favorites: {favorite_name}")
# Use the factory to get the appropriate favorites function
favorite_function = favorite_function_factory(self.main_window.tidal, media)
# Fetch all items from this favorites category
media_items = favorite_function()
if not media_items:
logger_gui.info(f"No items found in favorites: {favorite_name}")
return
logger_gui.info(f"Found {len(media_items)} items in favorites: {favorite_name}")
# Delegate to appropriate handler based on favorites type
if media == "fav_albums":
self._download_albums_from_favorites_albums(media_items, favorite_name)
elif media == "fav_artists":
self._download_albums_from_favorites_artists(media_items, favorite_name)
else:
self._download_albums_from_favorites_tracks(media_items, favorite_name)
except Exception as e:
logger_gui.exception(f"Error downloading albums from favorites: {e}")
logger_gui.error("Failed to download albums from favorites. See log for details.")
def on_download_all_albums_from_playlist(self, point: QtCore.QPoint) -> None:
"""Download all unique albums from tracks in a playlist."""
item = self.main_window.tr_lists_user.itemAt(point)
media_list = get_user_list_media_item(item)
if not isinstance(media_list, Playlist | UserPlaylist | Mix):
logger_gui.error("Please select a playlist or mix.")
return
media_items = items_results_all(self.main_window.tidal.session, media_list)
album_ids = self.main_window._extract_album_ids_from_tracks(media_items)
if not album_ids:
logger_gui.warning("No albums found in this playlist.")
return
albums_dict = self.main_window._load_albums_with_rate_limiting(album_ids)
if not albums_dict:
logger_gui.error("Failed to load any albums from playlist.")
return
self.main_window._queue_loaded_albums(albums_dict)
message = f"Added {len(albums_dict)} albums to download queue"
self.main_window.s_statusbar_message.emit(StatusbarMessage(message=message, timeout=3000))
def on_tr_lists_user_expanded(self, item: QtWidgets.QTreeWidgetItem) -> None:
"""Handle expansion of folders in the user lists tree.
Args:
item (QTreeWidgetItem): The expanded tree item.
"""
# Check if it's a first-time expansion (has disabled dummy child)
if item.childCount() > 0 and item.child(0).isDisabled():
# Run in thread to avoid blocking UI
self.main_window.thread_it(self.tr_lists_user_load_folder_children, item)
def tr_lists_user_load_folder_children(self, parent_item: QtWidgets.QTreeWidgetItem) -> None:
"""Load and display children of a folder in the user lists tree.
Args:
parent_item (QTreeWidgetItem): The parent folder item.
"""
folder: Folder | None = get_user_list_media_item(parent_item)
if not isinstance(folder, Folder):
return
# Show spinner while loading
self.main_window.s_spinner_start.emit(self.main_window.tr_lists_user)
try:
# Fetch folder contents
folders, playlists = self._fetch_folder_contents(folder)
# Emit signal to populate in main thread
self.main_window.s_populate_folder_children.emit(parent_item, folders, playlists)
finally:
self.main_window.s_spinner_stop.emit()
def on_populate_folder_children(
self, parent_item: QtWidgets.QTreeWidgetItem, folders: list[Folder], playlists: list[Playlist]
) -> None:
"""Populate folder children in the main thread (signal handler).
Args:
parent_item (QTreeWidgetItem): The parent folder item.
folders (list[Folder]): List of sub-folders.
playlists (list[Playlist]): List of playlists.
"""
# Remove dummy child
parent_item.takeChild(0)
# Add sub-folders as children
for sub_folder in folders:
twi_child = QtWidgets.QTreeWidgetItem(parent_item)
twi_child.setText(0, f"📁 {sub_folder.name}")
set_user_list_media(twi_child, sub_folder)
info = f"({sub_folder.total_number_of_items} items)" if sub_folder.total_number_of_items else ""
twi_child.setText(2, info)
# Add dummy child for potential sub-folders
dummy = QtWidgets.QTreeWidgetItem(twi_child)
dummy.setDisabled(True)
# Add playlists as children
for playlist in playlists:
twi_child = QtWidgets.QTreeWidgetItem(parent_item)
name = playlist.name if playlist.name else ""
twi_child.setText(0, name)
set_user_list_media(twi_child, playlist)
info = f"({playlist.num_tracks + playlist.num_videos} Tracks)"
if playlist.description:
info += f" {playlist.description}"
twi_child.setText(2, info)
def _fetch_folder_contents(self, folder: Folder) -> tuple[list[Folder], list[Playlist]]:
"""Fetch contents (sub-folders and playlists) of a folder.
Args:
folder (Folder): The folder to fetch contents for.
Returns:
tuple[list[Folder], list[Playlist]]: Sub-folders and playlists within the folder.
"""
folder_id = folder.id if folder.id else "root"
# Fetch sub-folders with manual pagination
offset = 0
limit = 50
folders = []
while True:
batch = self.main_window.tidal.session.user.favorites.playlist_folders(
limit=limit, offset=offset, parent_folder_id=folder_id
)
if not batch:
break
folders.extend(batch)
if len(batch) < limit:
break
offset += limit
# Fetch playlists in this folder using folder.items() method
offset = 0
playlists = []
while True:
batch = folder.items(offset=offset, limit=limit)
if not batch:
break
playlists.extend(batch)
if len(batch) < limit:
break
offset += limit
return folders, playlists
def _get_folder_playlists(self, folder: Folder) -> list[Playlist]:
"""Fetch all playlists from a folder.
Args:
folder (Folder): The folder to fetch playlists from.
Returns:
list[Playlist]: List of playlists in the folder.
"""
# Use existing method to fetch folder contents
# Since folders can't contain folders, we ignore the folders return value
_, playlists = self._fetch_folder_contents(folder)
logger_gui.debug(f"Found {len(playlists)} playlists in folder: {folder.name}")
return playlists
def _get_playlist_tracks(self, playlist: Playlist | UserPlaylist | Mix) -> list[Track]:
"""Fetch all tracks from a playlist.
Args:
playlist (Playlist | UserPlaylist | Mix): The playlist to fetch tracks from.
Returns:
list[Track]: List of tracks in the playlist.
"""
playlist_name = getattr(playlist, "name", "unknown")
logger_gui.debug(f"Fetching tracks from playlist: {playlist_name}")
media_items = items_results_all(self.main_window.tidal.session, playlist)
# Filter for Track objects only (items_results_all may return Videos too)
tracks = [item for item in media_items if isinstance(item, Track)]
logger_gui.debug(f"Found {len(tracks)} tracks in playlist: {playlist_name}")
return tracks
def on_list_items_show(self, item: QtWidgets.QTreeWidgetItem) -> None:
"""Show the items in the selected playlist or mix.
Args:
item (QtWidgets.QTreeWidgetItem): The selected tree widget item.
"""
self.main_window.thread_it(self.list_items_show, item)
def list_items_show(self, item: QtWidgets.QTreeWidgetItem) -> None:
"""Fetch and display the items in a playlist, mix, or folder.
Args:
item (QtWidgets.QTreeWidgetItem): The tree widget item representing a playlist, mix, or folder.
"""
media_list: Album | Playlist | Folder | str = get_user_list_media_item(item)
# Only if clicked item is not a top level item.
if media_list:
# Show spinner while loading list
self.main_window.s_spinner_start.emit(self.main_window.tr_results)
try:
if isinstance(media_list, Folder):
# Show folder contents
self._show_folder_contents(media_list)
elif isinstance(media_list, str) and media_list.startswith("fav_"):
function_list = favorite_function_factory(self.main_window.tidal, media_list)
self.main_window.list_items_show_result(favorite_function=function_list)
else:
self.main_window.list_items_show_result(media_list)
# Load cover asynchronously to avoid blocking the GUI
if hasattr(self.main_window, "cover_manager"):
self.main_window.cover_manager.load_cover(media_list)
finally:
self.main_window.s_spinner_stop.emit()
def _show_folder_contents(self, folder: Folder) -> None:
"""Display folder contents (nested playlists/folders) in results pane.
Args:
folder (Folder): The folder to display contents for.
"""
# Fetch folder contents using the shared helper method
folders, playlists = self._fetch_folder_contents(folder)
# Combine folders and playlists
items = folders + playlists
# Convert to ResultItems and display
result = self.main_window.search_manager.search_result_to_model(items)
self.main_window.populate_tree_results(result)
def _download_albums_from_favorites_albums(self, media_items: list, favorite_name: str) -> None:
"""Download albums from favorite albums list.
Args:
media_items (list): List of favorite albums.
favorite_name (str): Name of the favorites category for logging.
"""
logger_gui.info(f"Queueing {len(media_items)} albums from favorites: {favorite_name}")
albums_dict = {album.id: album for album in media_items if isinstance(album, Album) and album.id}
self.main_window._queue_loaded_albums(albums_dict)
logger_gui.info(f"✅ Successfully queued {len(albums_dict)} albums from favorites: {favorite_name}")
def _download_albums_from_favorites_artists(self, media_items: list, favorite_name: str) -> None:
"""Download albums from favorite artists list.
Args:
media_items (list): List of favorite artists.
favorite_name (str): Name of the favorites category for logging.
"""
logger_gui.info(f"Fetching albums from {len(media_items)} artists...")
all_albums = {}
for artist in media_items:
if isinstance(artist, Artist):
try:
artist_albums = items_results_all(self.main_window.tidal.session, artist)
for album in artist_albums:
if isinstance(album, Album) and album.id:
all_albums[album.id] = album
logger_gui.debug(f"Found {len(artist_albums)} albums from artist: {artist.name}")
except Exception as e:
logger_gui.error(f"Error getting albums from artist '{artist.name}': {e}")
continue
if not all_albums:
logger_gui.info("No albums found from favorite artists.")
return
logger_gui.info(f"Found {len(all_albums)} unique albums from favorite artists")
self.main_window._queue_loaded_albums(all_albums)
logger_gui.info(f"✅ Successfully queued {len(all_albums)} albums from favorites: {favorite_name}")
def _download_albums_from_favorites_tracks(self, media_items: list, favorite_name: str) -> None:
"""Download albums from favorite tracks/videos/mixes list.
Args:
media_items (list): List of favorite tracks/videos/mixes.
favorite_name (str): Name of the favorites category for logging.
"""
logger_gui.info("Extracting albums from tracks...")
album_ids = self.main_window._extract_album_ids_from_tracks(media_items)
if not album_ids:
logger_gui.info(f"No albums found in favorites: {favorite_name}")
return
logger_gui.info(f"Found {len(album_ids)} unique albums. Loading with rate limiting...")
# Load full album objects with rate limiting
albums_dict = self.main_window._load_albums_with_rate_limiting(album_ids)
if not albums_dict:
logger_gui.error("Failed to load any albums from favorites.")
return
# Queue the albums for download
self.main_window._queue_loaded_albums(albums_dict)
logger_gui.info(f"✅ Successfully queued {len(albums_dict)} unique albums from favorites: {favorite_name}")
def search_populate_results(self, query: str, type_media: Any) -> None:
"""Populate the results tree with search results.
Args:
query (str): The search query.
type_media (SearchTypes): The type of media to search for.
"""
results: list[ResultItem] = self.main_window.search_manager.search(query, [type_media])
self.main_window.populate_tree_results(results)
+31
View File
@@ -0,0 +1,31 @@
"""Progress bars mixin for MainWindow.
Handles progress bar updates and formatting.
"""
import math
class ProgressMixin:
"""Mixin containing progress bar management methods."""
def on_progress_reset(self):
"""Reset progress bars to zero."""
self.pb_list.setValue(0)
self.pb_item.setValue(0)
def on_progress_list(self, value: float) -> None:
"""Update the progress of the list progress bar."""
self.pb_list.setValue(int(math.ceil(value)))
def on_progress_item(self, value: float) -> None:
"""Update the progress of the item progress bar."""
self.pb_item.setValue(int(math.ceil(value)))
def on_progress_item_name(self, value: str) -> None:
"""Set the format of the item progress bar."""
self.pb_item.setFormat(f"%p% {value}")
def on_progress_list_name(self, value: str) -> None:
"""Set the format of the list progress bar."""
self.pb_list.setFormat(f"%p% {value}")
+311
View File
@@ -0,0 +1,311 @@
# tidal_dl_ng/queue.py
import time
from typing import TYPE_CHECKING
from PySide6 import QtCore, QtGui, QtWidgets
from tidalapi import Album, Artist, Mix, Playlist, Quality, Track, Video
from tidal_dl_ng.config import HandlingApp, Settings
from tidal_dl_ng.constants import QualityVideo, QueueDownloadStatus
from tidal_dl_ng.download import Download
from tidal_dl_ng.helper.gui import (
get_queue_download_media,
get_queue_download_quality_audio,
get_queue_download_quality_video,
set_queue_download_media,
)
from tidal_dl_ng.helper.path import get_format_template
from tidal_dl_ng.helper.tidal import items_results_all, name_builder_artist, name_builder_title, quality_audio_highest
from tidal_dl_ng.logger import logger_gui
from tidal_dl_ng.model.gui_data import QueueDownloadItem, StatusbarMessage
if TYPE_CHECKING:
from tidal_dl_ng.gui import MainWindow
class GuiQueueManager:
"""Manages the download queue GUI and logic."""
def __init__(self, main_window: "MainWindow"):
"""Initialize the queue manager."""
self.main_window: "MainWindow" = main_window
self.settings: Settings = main_window.settings
def init_ui(self):
"""Initialize UI elements related to the queue."""
self._init_tree_queue(self.main_window.tr_queue_download)
self.pb_queue_download_run()
def connect_signals(self):
"""Connect signals for queue-related widgets."""
self.main_window.pb_queue_download_clear_all.clicked.connect(self.on_queue_download_clear_all)
self.main_window.pb_queue_download_clear_finished.clicked.connect(self.on_queue_download_clear_finished)
self.main_window.pb_queue_download_remove.clicked.connect(self.on_queue_download_remove)
self.main_window.pb_queue_download_toggle.clicked.connect(self.on_pb_queue_download_toggle)
self.main_window.tr_queue_download.itemClicked.connect(self.on_queue_download_item_clicked)
self.main_window.tr_queue_download.customContextMenuRequested.connect(self.menu_context_queue_download)
self.main_window.s_queue_download_item_downloading.connect(self.on_queue_download_item_downloading)
self.main_window.s_queue_download_item_finished.connect(self.on_queue_download_item_finished)
self.main_window.s_queue_download_item_failed.connect(self.on_queue_download_item_failed)
self.main_window.s_queue_download_item_skipped.connect(self.on_queue_download_item_skipped)
def _init_tree_queue(self, tree: QtWidgets.QTableWidget) -> None:
"""Initialize the download queue table widget."""
tree.setColumnHidden(1, True)
tree.setColumnWidth(2, 200)
header = tree.header()
if hasattr(header, "setSectionResizeMode"):
header.setSectionResizeMode(0, QtWidgets.QHeaderView.ResizeToContents)
tree.setContextMenuPolicy(QtCore.Qt.CustomContextMenu)
def menu_context_queue_download(self, point: QtCore.QPoint) -> None:
"""Show context menu for download queue."""
item = self.main_window.tr_queue_download.itemAt(point)
if not item:
return
menu = QtWidgets.QMenu()
status = item.text(0)
if status == QueueDownloadStatus.Waiting:
menu.addAction("🗑️ Remove from Queue", lambda: self.on_queue_download_remove_item(item))
if menu.isEmpty():
return
menu.exec(self.main_window.tr_queue_download.mapToGlobal(point))
def on_queue_download_remove_item(self, item: QtWidgets.QTreeWidgetItem) -> None:
"""Remove a specific item from the download queue."""
index = self.main_window.tr_queue_download.indexOfTopLevelItem(item)
if index >= 0:
self.main_window.tr_queue_download.takeTopLevelItem(index)
logger_gui.info("Removed item from download queue")
def on_queue_download_clear_all(self) -> None:
"""Clear all items from the download queue."""
self.on_clear_queue_download(
f"({QueueDownloadStatus.Waiting}|{QueueDownloadStatus.Finished}|{QueueDownloadStatus.Failed})"
)
def on_queue_download_clear_finished(self) -> None:
"""Clear finished items from the download queue."""
self.on_clear_queue_download(f"[{QueueDownloadStatus.Finished}]")
def on_clear_queue_download(self, regex: str) -> None:
"""Clear items from the download queue matching the given regex.
Args:
regex (str): Regular expression to match items.
"""
items: list[QtWidgets.QTreeWidgetItem | None] = self.main_window.tr_queue_download.findItems(
regex, QtCore.Qt.MatchFlag.MatchRegularExpression, column=0
)
for item in items:
self.main_window.tr_queue_download.takeTopLevelItem(
self.main_window.tr_queue_download.indexOfTopLevelItem(item)
)
def on_queue_download_remove(self) -> None:
"""Remove selected items from the download queue."""
items: list[QtWidgets.QTreeWidgetItem | None] = self.main_window.tr_queue_download.selectedItems()
if len(items) == 0:
logger_gui.error("Please select an item from the queue first.")
else:
for item in items:
status: str = item.text(0)
if status != QueueDownloadStatus.Downloading:
self.main_window.tr_queue_download.takeTopLevelItem(
self.main_window.tr_queue_download.indexOfTopLevelItem(item)
)
else:
logger_gui.info("Cannot remove a currently downloading item from queue.")
def on_pb_queue_download_toggle(self) -> None:
"""Toggle download status (pause / resume) accordingly.
:return: None
"""
handling_app: HandlingApp = HandlingApp()
if handling_app.event_run.is_set():
self.pb_queue_download_pause()
else:
self.pb_queue_download_run()
def pb_queue_download_run(self) -> None:
"""Start the download queue and update the button state."""
handling_app: HandlingApp = HandlingApp()
handling_app.event_run.set()
icon = QtGui.QIcon(QtGui.QIcon.fromTheme(QtGui.QIcon.ThemeIcon.MediaPlaybackPause))
self.main_window.pb_queue_download_toggle.setIcon(icon)
self.main_window.pb_queue_download_toggle.setStyleSheet("background-color: #e0a800; color: #212529")
def pb_queue_download_pause(self) -> None:
"""Pause the download queue and update the button state."""
handling_app: HandlingApp = HandlingApp()
handling_app.event_run.clear()
icon = QtGui.QIcon(QtGui.QIcon.fromTheme(QtGui.QIcon.ThemeIcon.MediaPlaybackStart))
self.main_window.pb_queue_download_toggle.setIcon(icon)
self.main_window.pb_queue_download_toggle.setStyleSheet("background-color: #218838; color: #fff")
def queue_download_media(self, queue_dl_item: QueueDownloadItem) -> None:
"""Add a media item to the download queue."""
child = QtWidgets.QTreeWidgetItem()
child.setText(0, queue_dl_item.status)
set_queue_download_media(child, queue_dl_item.obj)
child.setText(2, queue_dl_item.name)
child.setText(3, queue_dl_item.type_media)
child.setText(4, str(queue_dl_item.quality_audio))
child.setText(5, str(queue_dl_item.quality_video))
self.main_window.tr_queue_download.addTopLevelItem(child)
def watcher_queue_download(self) -> None:
"""Monitor the download queue and process items as they become available."""
handling_app: HandlingApp = HandlingApp()
while not handling_app.event_abort.is_set():
items = self.main_window.tr_queue_download.findItems(
QueueDownloadStatus.Waiting, QtCore.Qt.MatchFlag.MatchExactly, column=0
)
if items:
item = items[0]
media = get_queue_download_media(item)
quality_audio = get_queue_download_quality_audio(item)
quality_video = get_queue_download_quality_video(item)
try:
self.main_window.s_queue_download_item_downloading.emit(item)
result = self.on_queue_download(media, quality_audio=quality_audio, quality_video=quality_video)
if result == QueueDownloadStatus.Finished:
self.main_window.s_queue_download_item_finished.emit(item)
elif result == QueueDownloadStatus.Skipped:
self.main_window.s_queue_download_item_skipped.emit(item)
except Exception as e:
logger_gui.error(e)
self.main_window.s_queue_download_item_failed.emit(item)
else:
time.sleep(2)
def on_queue_download_item_downloading(self, item: QtWidgets.QTreeWidgetItem) -> None:
"""Update the status of a queue download item to 'Downloading'."""
self.queue_download_item_status(item, QueueDownloadStatus.Downloading)
def on_queue_download_item_finished(self, item: QtWidgets.QTreeWidgetItem) -> None:
"""Update the status of a queue download item to 'Finished'."""
self.queue_download_item_status(item, QueueDownloadStatus.Finished)
def on_queue_download_item_failed(self, item: QtWidgets.QTreeWidgetItem) -> None:
"""Update the status of a queue download item to 'Failed'."""
self.queue_download_item_status(item, QueueDownloadStatus.Failed)
def on_queue_download_item_skipped(self, item: QtWidgets.QTreeWidgetItem) -> None:
"""Update the status of a queue download item to 'Skipped'."""
self.queue_download_item_status(item, QueueDownloadStatus.Skipped)
def queue_download_item_status(self, item: QtWidgets.QTreeWidgetItem, status: str) -> None:
"""Set the status text of a queue download item."""
item.setText(0, status)
def on_queue_download(
self,
media: Track | Album | Playlist | Video | Mix | Artist,
quality_audio: Quality | None = None,
quality_video: QualityVideo | None = None,
) -> QueueDownloadStatus:
"""Download the specified media item(s) and return the result status."""
items_media = items_results_all(self.main_window.tidal.session, media) if isinstance(media, Artist) else [media]
download_delay = bool(isinstance(media, Track | Video) and self.settings.data.download_delay)
result = QueueDownloadStatus.Failed
for item_media in items_media:
result = self.download(
item_media,
self.main_window.dl,
delay_track=download_delay,
quality_audio=quality_audio,
quality_video=quality_video,
)
return result
def download(
self,
media: Track | Album | Playlist | Video | Mix,
dl: Download,
delay_track: bool = False,
quality_audio: Quality | None = None,
quality_video: QualityVideo | None = None,
) -> QueueDownloadStatus:
"""Download a media item and return the result status."""
self.main_window.s_pb_reset.emit()
self.main_window.s_statusbar_message.emit(StatusbarMessage(message="Download started..."))
file_template = get_format_template(media, self.settings)
result_dl, path_file = False, None
if isinstance(media, Track | Video):
result_dl, path_file = dl.item(
media=media,
file_template=file_template,
download_delay=delay_track,
quality_audio=quality_audio,
quality_video=quality_video,
)
elif isinstance(media, Album | Playlist | Mix):
dl.items(
media=media,
file_template=file_template,
video_download=self.settings.data.video_download,
download_delay=self.settings.data.download_delay,
quality_audio=quality_audio,
quality_video=quality_video,
)
result_dl, path_file = True, "dummy"
self.main_window.s_statusbar_message.emit(StatusbarMessage(message="Download finished.", timeout=2000))
if result_dl and path_file:
return QueueDownloadStatus.Finished
if not result_dl and path_file:
return QueueDownloadStatus.Skipped
return QueueDownloadStatus.Failed
def on_queue_download_item_clicked(self, item: QtWidgets.QTreeWidgetItem) -> None:
"""Handle the event when a queue download item is clicked."""
media = get_queue_download_media(item)
self.main_window.info_tab_widget.update_on_selection(media)
self.main_window.thread_it(self.main_window.cover_manager.load_cover, media)
def media_to_queue_download_model(
self, media: Artist | Track | Video | Album | Playlist | Mix
) -> QueueDownloadItem | None:
"""Convert a media object to a QueueDownloadItem for the download queue."""
if hasattr(media, "available") and media.available is False:
return None
explicit = " 🅴" if isinstance(media, Track | Video | Album) and media.explicit else ""
name = ""
if isinstance(media, Track | Video):
name = f"{name_builder_artist(media)} - {name_builder_title(media)}{explicit}"
elif isinstance(media, Playlist | Artist):
name = media.name
elif isinstance(media, Album):
name = f"{name_builder_artist(media)} - {media.name}{explicit}"
elif isinstance(media, Mix):
name = media.title
quality_audio = self.settings.data.quality_audio
if isinstance(media, Track | Album):
quality_highest = quality_audio_highest(media)
if (
self.settings.data.quality_audio == quality_highest
or self.settings.data.quality_audio == Quality.hi_res_lossless
):
quality_audio = quality_highest
if name:
return QueueDownloadItem(
name=name,
quality_audio=quality_audio,
quality_video=self.settings.data.quality_video,
type_media=type(media).__name__,
status=QueueDownloadStatus.Waiting,
obj=media,
)
return None
+401
View File
@@ -0,0 +1,401 @@
# tidal_dl_ng/search.py
from typing import TYPE_CHECKING, Any
from tidalapi import Album, Artist, Mix, Playlist, Track, Video
from tidalapi import VideoQuality as QualityVideo
from tidalapi.media import AudioMode, Quality
from tidalapi.playlist import Folder
from tidalapi.session import SearchTypes
from tidal_dl_ng.constants import QueueDownloadStatus
from tidal_dl_ng.helper.tidal import (
get_tidal_media_id,
get_tidal_media_type,
instantiate_media,
name_builder_artist,
name_builder_title,
quality_audio_highest,
search_results_all,
url_ending_clean,
)
from tidal_dl_ng.logger import logger_gui
from tidal_dl_ng.model.gui_data import QueueDownloadItem, ResultItem
if TYPE_CHECKING:
from tidal_dl_ng.gui import MainWindow
class GuiSearchManager:
"""Manages the search GUI and logic."""
def __init__(self, main_window: "MainWindow"):
"""Initialize the search manager."""
self.main_window: "MainWindow" = main_window
def search_populate_results(self, query: str, type_media: Any) -> None:
"""Populate the results tree with search results."""
results = self.search(query, [type_media])
self.main_window.populate_tree_results(results)
def search(self, query: str, types_media: list[Any]) -> list[ResultItem]:
"""Perform a search and return a list of ResultItems.
Args:
query (str): The search query.
types_media (list[Any]): The types of media to search for.
Returns:
list[ResultItem]: The search results.
"""
query_clean: str = query.strip()
# If a direct link was searched for, skip search and create the object from the link directly.
if "http" in query_clean:
query_clean: str = url_ending_clean(query_clean)
media_type = get_tidal_media_type(query_clean)
item_id = get_tidal_media_id(query_clean)
try:
media = instantiate_media(self.main_window.tidal.session, media_type, item_id)
except:
logger_gui.error(f"Media not found (ID: {item_id}). Maybe it is not available anymore.")
media = None
result_search = {"direct": [media]}
else:
result_search: dict[str, list[SearchTypes]] = search_results_all(
session=self.main_window.tidal.session, needle=query_clean, types_media=types_media
)
result: list[ResultItem] = []
for _media_type, l_media in result_search.items():
if isinstance(l_media, list):
result = result + self.search_result_to_model(l_media)
return result
def search_result_to_model(self, items: list[SearchTypes]) -> list[ResultItem]:
"""Convert search results to ResultItem models.
Args:
items (list[SearchTypes]): List of search result items.
Returns:
list[ResultItem]: List of ResultItem models.
"""
result: list[ResultItem] = []
for idx, item in enumerate(items):
result_item = self._to_result_item(idx, item)
if result_item is not None:
result.append(result_item)
return result
def _to_result_item(self, idx: int, item) -> ResultItem | None:
"""Helper to convert a single item to ResultItem, or None if not valid.
Args:
idx (int): Index of the item.
item: The item to convert.
Returns:
ResultItem | None: The converted ResultItem or None if not valid.
"""
if not item or (hasattr(item, "available") and not item.available):
return None
# Prepare common data
explicit = " 🅴" if isinstance(item, Track | Video | Album) and item.explicit else ""
date_user_added = (
item.user_date_added.strftime("%Y-%m-%d_%H:%M") if getattr(item, "user_date_added", None) else ""
)
date_release = self._get_date_release(item)
# Map item types to their conversion methods
type_handlers = {
Track: lambda: self._result_item_from_track(idx, item, explicit, date_user_added, date_release),
Video: lambda: self._result_item_from_video(idx, item, explicit, date_user_added, date_release),
Playlist: lambda: self._result_item_from_playlist(idx, item, date_user_added, date_release),
Album: lambda: self._result_item_from_album(idx, item, explicit, date_user_added, date_release),
Mix: lambda: self._result_item_from_mix(idx, item, date_user_added, date_release),
Artist: lambda: self._result_item_from_artist(idx, item, date_user_added, date_release),
Folder: lambda: self._result_item_from_folder(idx, item, date_user_added),
}
# Find and execute the appropriate handler
for item_type, handler in type_handlers.items():
if isinstance(item, item_type):
return handler()
return None
def _get_date_release(self, item) -> str:
"""Get the release date string for an item.
Args:
item: The item to extract the release date from.
Returns:
str: The formatted release date or empty string.
"""
if hasattr(item, "album") and item.album and getattr(item.album, "release_date", None):
return item.album.release_date.strftime("%Y-%m-%d_%H:%M")
if hasattr(item, "release_date") and item.release_date:
return item.release_date.strftime("%Y-%m-%d_%H:%M")
return ""
def _result_item_from_track(
self, idx: int, item, explicit: str, date_user_added: str, date_release: str
) -> ResultItem:
"""Create a ResultItem from a Track.
Args:
idx (int): Index of the item.
item: The Track item.
explicit (str): Explicit tag.
date_user_added (str): Date user added.
date_release (str): Release date.
Returns:
ResultItem: The constructed ResultItem.
"""
final_quality = quality_audio_highest(item)
if hasattr(item, "audio_modes") and AudioMode.dolby_atmos.value in item.audio_modes:
final_quality = f"{final_quality} / Dolby Atmos"
return ResultItem(
position=idx,
artist=name_builder_artist(item),
title=f"{name_builder_title(item)}{explicit}",
album=item.album.name,
duration_sec=item.duration,
obj=item,
quality=final_quality,
explicit=bool(item.explicit),
date_user_added=date_user_added,
date_release=date_release,
)
def _result_item_from_video(
self, idx: int, item, explicit: str, date_user_added: str, date_release: str
) -> ResultItem:
"""Create a ResultItem from a Video.
Args:
idx (int): Index of the item.
item: The Video item.
explicit (str): Explicit tag.
date_user_added (str): Date user added.
date_release (str): Release date.
Returns:
ResultItem: The constructed ResultItem.
"""
return ResultItem(
position=idx,
artist=name_builder_artist(item),
title=f"{name_builder_title(item)}{explicit}",
album=item.album.name if item.album else "",
duration_sec=item.duration,
obj=item,
quality=item.video_quality,
explicit=bool(item.explicit),
date_user_added=date_user_added,
date_release=date_release,
)
def _result_item_from_playlist(self, idx: int, item, date_user_added: str, date_release: str) -> ResultItem:
"""Create a ResultItem from a Playlist.
Args:
idx (int): Index of the item.
item: The Playlist item.
date_user_added (str): Date user added.
date_release (str): Release date.
Returns:
ResultItem: The constructed ResultItem.
"""
return ResultItem(
position=idx,
artist=", ".join(artist.name for artist in item.promoted_artists) if item.promoted_artists else "",
title=item.name,
album="",
duration_sec=item.duration,
obj=item,
quality="",
explicit=False,
date_user_added=date_user_added,
date_release=date_release,
)
def _result_item_from_album(
self, idx: int, item, explicit: str, date_user_added: str, date_release: str
) -> ResultItem:
"""Create a ResultItem from an Album.
Args:
idx (int): Index of the item.
item: The Album item.
explicit (str): Explicit tag.
date_user_added (str): Date user added.
date_release (str): Release date.
Returns:
ResultItem: The constructed ResultItem.
"""
return ResultItem(
position=idx,
artist=name_builder_artist(item),
title="",
album=f"{item.name}{explicit}",
duration_sec=item.duration,
obj=item,
quality=quality_audio_highest(item),
explicit=bool(item.explicit),
date_user_added=date_user_added,
date_release=date_release,
)
def _result_item_from_mix(self, idx: int, item, date_user_added: str, date_release: str) -> ResultItem:
"""Create a ResultItem from a Mix.
Args:
idx (int): Index of the item.
item: The Mix item.
date_user_added (str): Date user added.
date_release (str): Release date.
Returns:
ResultItem: The constructed ResultItem.
"""
return ResultItem(
position=idx,
artist=item.sub_title,
title=item.title,
album="",
duration_sec=-1, # TODO: Calculate total duration.
obj=item,
quality="",
explicit=False,
date_user_added=date_user_added,
date_release=date_release,
)
def _result_item_from_artist(self, idx: int, item, date_user_added: str, date_release: str) -> ResultItem:
"""Create a ResultItem from an Artist.
Args:
idx (int): Index of the item.
item: The Artist item.
date_user_added (str): Date user added.
date_release (str): Release date.
Returns:
ResultItem: The constructed ResultItem.
"""
return ResultItem(
position=idx,
artist=item.name,
title="",
album="",
duration_sec=-1,
obj=item,
quality="",
explicit=False,
date_user_added=date_user_added,
date_release=date_release,
)
def _result_item_from_folder(self, idx: int, item: Folder, date_user_added: str) -> ResultItem:
"""Create a ResultItem from a Folder.
Args:
idx (int): Index of the item.
item (Folder): The Folder item.
date_user_added (str): Date user added.
Returns:
ResultItem: The constructed ResultItem.
"""
total_items: int = item.total_number_of_items if hasattr(item, "total_number_of_items") else 0
return ResultItem(
position=idx,
artist="",
title=f"📁 {item.name} ({total_items} items)",
album="",
duration_sec=-1,
obj=item,
quality="",
explicit=False,
date_user_added=date_user_added,
date_release="",
)
def media_to_queue_download_model(
self, media: Artist | Track | Video | Album | Playlist | Mix
) -> QueueDownloadItem | bool:
"""Convert a media object to a QueueDownloadItem for the download queue.
Args:
media (Artist | Track | Video | Album | Playlist | Mix): The media object.
Returns:
QueueDownloadItem | bool: The queue item or False if not available.
"""
result: QueueDownloadItem | False
name: str = ""
quality_audio: Quality = self.main_window.settings.data.quality_audio
quality_video: QualityVideo = self.main_window.settings.data.quality_video
explicit: str = ""
# Check if item is available on TIDAL.
# Note: Some albums have available=None, which should be treated as available
if hasattr(media, "available") and media.available is False:
return False
# Set "Explicit" tag
if isinstance(media, Track | Video | Album):
explicit = " 🅴" if media.explicit else ""
# Build name and set quality
if isinstance(media, Track | Video):
name = f"{name_builder_artist(media)} - {name_builder_title(media)}{explicit}"
elif isinstance(media, Playlist | Artist):
name = media.name
elif isinstance(media, Album):
name = f"{name_builder_artist(media)} - {media.name}{explicit}"
elif isinstance(media, Mix):
name = media.title
# Determine actual quality.
if isinstance(media, Track | Album):
quality_highest: Quality = quality_audio_highest(media)
if (
self.main_window.settings.data.quality_audio == quality_highest
or self.main_window.settings.data.quality_audio == Quality.hi_res_lossless
):
quality_audio = quality_highest
if name:
result = QueueDownloadItem(
name=name,
quality_audio=quality_audio,
quality_video=quality_video,
type_media=type(media).__name__,
status=QueueDownloadStatus.Waiting,
obj=media,
)
else:
result = False
return result
+85
View File
@@ -0,0 +1,85 @@
"""Signals mixin for MainWindow.
Handles all Qt signal definitions and connections.
"""
from PySide6 import QtCore, QtWidgets
from tidalapi import Album, Artist, Track, Video
from tidal_dl_ng.helper.gui import get_results_media_item
class SignalsMixin:
"""Mixin containing Qt signal definitions and signal connection methods."""
def _init_signals(self) -> None:
"""Connect signals to their respective slots."""
self.pb_download.clicked.connect(lambda: self.thread_it(self.on_download_results))
self.pb_download_list.clicked.connect(lambda: self.thread_it(self.playlist_manager.on_download_list_media))
self.l_search.returnPressed.connect(
lambda: self.search_manager.search_populate_results(self.l_search.text(), self.cb_search_type.currentData())
)
self.pb_search.clicked.connect(
lambda: self.search_manager.search_populate_results(self.l_search.text(), self.cb_search_type.currentData())
)
self.cb_quality_audio.currentIndexChanged.connect(self.on_quality_set_audio)
self.cb_quality_video.currentIndexChanged.connect(self.on_quality_set_video)
self.s_spinner_start[QtWidgets.QWidget].connect(self.on_spinner_start)
self.s_spinner_stop.connect(self.on_spinner_stop)
self.s_item_advance.connect(self.on_progress_item)
self.s_item_name.connect(self.on_progress_item_name)
self.s_list_name.connect(self.on_progress_list_name)
self.s_list_advance.connect(self.on_progress_list)
self.s_pb_reset.connect(self.on_progress_reset)
self.s_statusbar_message.connect(self.on_statusbar_message)
self.s_tr_results_add_top_level_item.connect(self.on_tr_results_add_top_level_item)
self.s_settings_save.connect(self.on_settings_save)
self.s_pb_reload_status.connect(self.button_reload_status)
self.s_update_check.connect(lambda: self.thread_it(self.on_update_check))
self.s_update_show.connect(self.on_version)
# Menubar
self.a_exit.triggered.connect(self.close)
self.a_version.triggered.connect(self.on_version)
self.a_preferences.triggered.connect(self.on_preferences)
self.a_logout.triggered.connect(self.on_logout)
self.a_updates_check.triggered.connect(lambda: self.on_update_check(False))
# Results
self.tr_results.expanded.connect(self.on_tr_results_expanded)
self.tr_results.clicked.connect(self.on_result_item_clicked)
self.tr_results.doubleClicked.connect(lambda: self.thread_it(self.on_download_results))
# Managers
self.queue_manager.connect_signals()
self.playlist_manager.connect_signals()
self.info_tab_widget.s_search_in_app.connect(self.on_search_in_app)
self.info_tab_widget.s_search_in_browser.connect(self.on_search_in_browser)
def on_result_item_clicked(self, index: QtCore.QModelIndex) -> None:
"""Handle the event when a result item is clicked."""
media: Track | Video | Album | Artist = get_results_media_item(
index, self.proxy_tr_results, self.model_tr_results
)
self.info_tab_widget.update_on_selection(media)
self.thread_it(self.cover_manager.load_cover, media)
def on_quality_set_audio(self, index: int) -> None:
"""Set the audio quality for downloads."""
from tidalapi import Quality
quality_data = self.cb_quality_audio.itemData(index)
self.settings.data.quality_audio = Quality(quality_data)
self.settings.save()
if self.tidal:
self.tidal.settings_apply()
def on_quality_set_video(self, index: int) -> None:
"""Set the video quality for downloads."""
from tidal_dl_ng.constants import QualityVideo
self.settings.data.quality_video = QualityVideo(self.cb_quality_video.itemData(index))
self.settings.save()
if self.tidal:
self.tidal.settings_apply()
+64
View File
@@ -0,0 +1,64 @@
"""Tidal session mixin for MainWindow.
Handles Tidal authentication and session management.
"""
import sys
from requests.exceptions import HTTPError
from tidalapi.session import LinkLogin
from tidal_dl_ng.config import Tidal
from tidal_dl_ng.dialog import DialogLogin
from tidal_dl_ng.logger import logger_gui
class TidalSessionMixin:
"""Mixin containing Tidal session management methods."""
def init_tidal(self, tidal: Tidal | None = None):
"""Initialize Tidal session and handle login flow."""
result: bool = False
if tidal:
self.tidal = tidal
result = True
else:
self.tidal = Tidal(self.settings)
result = self.tidal.login_token()
if not result:
hint: str = "After you have finished the TIDAL login via web browser click the 'OK' button."
while not result:
link_login: LinkLogin = self.tidal.session.get_link_login()
expires_in = int(link_login.expires_in) if hasattr(link_login, "expires_in") else 0
d_login: DialogLogin = DialogLogin(
url_login=link_login.verification_uri_complete,
hint=hint,
expires_in=expires_in,
parent=self,
)
if d_login.return_code == 1:
try:
self.tidal.session.process_link_login(link_login, until_expiry=False)
self.tidal.login_finalize()
result = True
logger_gui.info("Login successful. Have fun!")
except (HTTPError, Exception):
hint = "Something was wrong with your redirect url. Please try again!"
logger_gui.warning("Login not successful. Try again...")
else:
sys.exit(1)
if result:
self._init_dl()
self.thread_it(self.playlist_manager.tidal_user_lists)
def on_logout(self) -> None:
"""Log out from TIDAL and close the application."""
result: bool = self.tidal.logout()
if result:
sys.exit(0)
+74
View File
@@ -0,0 +1,74 @@
"""Track extras mixin for MainWindow.
Handles track extras caching and retrieval.
"""
import contextlib
from collections.abc import Callable
from PySide6 import QtCore
from tidal_dl_ng.helper.tidal import extract_contributor_names, fetch_raw_track_and_album, parse_track_and_album_extras
from tidal_dl_ng.worker import Worker
class TrackExtrasMixin:
"""Mixin containing track extras management methods."""
def get_track_extras(self, track_id: str, callback: Callable[[str, dict | None], None]) -> dict | None:
"""Return cached extras for a track or start async fetch."""
cached = self.track_extras_cache.get(track_id)
if cached is not None:
return cached
if track_id in self._pending_extras_workers:
return None
if callback:
self._track_extras_callbacks[track_id] = callback
def worker() -> None:
extras = None
try:
track_json, album_json = fetch_raw_track_and_album(self.tidal.session, track_id)
extras = parse_track_and_album_extras(track_json, album_json)
extras = self._decorate_extras(extras)
self.track_extras_cache.set(track_id, extras)
except Exception:
extras = None
finally:
self._pending_extras_workers.pop(track_id, None)
self.s_track_extras_ready.emit(track_id, extras)
self.s_invoke_callback.emit(track_id, extras)
worker_obj = Worker(worker)
self._pending_extras_workers[track_id] = worker_obj
self.threadpool.start(worker_obj)
return None
@QtCore.Slot(str, object)
def _on_invoke_callback(self, track_id: str, extras: dict | None) -> None:
"""Invoke the stored callback for a track in the main thread."""
callback = self._track_extras_callbacks.pop(track_id, None)
if callback:
with contextlib.suppress(Exception):
callback(track_id, extras)
def _decorate_extras(self, extras: dict | None) -> dict:
"""Add formatted string fields to extras dict."""
if not extras:
return {}
result = dict(extras)
result["genres_text"] = ", ".join(result.get("genres", []))
for role, key in [
("producer", "producers_text"),
("composer", "composers_text"),
("lyricist", "lyricists_text"),
]:
result[key] = extract_contributor_names(result.get("contributors_by_role"), role)
return result
def preload_covers_for_playlist(self, items: list) -> None:
"""Preload cover pixmaps for a list of tracks in background."""
if self.cover_manager:
self.cover_manager.preload_covers_for_playlist(items)
+232
View File
@@ -0,0 +1,232 @@
"""Trees and results mixin for MainWindow.
Handles tree view population and results display.
"""
import contextlib
import math
from collections.abc import Callable, Sequence
from PySide6 import QtCore, QtGui, QtWidgets
from tidalapi import Album, Artist, Mix, Playlist, Track, UserPlaylist, Video
from tidalapi.playlist import Folder
from tidal_dl_ng.constants import FAVORITES, TidalLists
from tidal_dl_ng.helper.gui import (
HumanProxyModel,
get_results_media_item,
get_user_list_media_item,
set_user_list_media,
)
from tidal_dl_ng.helper.tidal import favorite_function_factory, items_results_all, user_media_lists
from tidal_dl_ng.model.gui_data import ResultItem
class TreesResultsMixin:
"""Mixin containing tree view and results management methods."""
def handle_filter_activated(self) -> None:
"""Handle activation of filter headers in the results tree."""
header = self.tr_results.header()
filters: list[tuple[int, str]] = []
for i in range(header.count()):
text: str = header.filter_text(i)
if text:
filters.append((i, text))
proxy_model: HumanProxyModel = self.tr_results.model()
proxy_model.filters = filters
def populate_tree_results(self, results: list[ResultItem], parent: QtGui.QStandardItem | None = None) -> None:
"""Populate the results tree with ResultItem objects."""
if not parent:
self.model_tr_results.removeRows(0, self.model_tr_results.rowCount())
count_digits: int = int(math.log10(len(results) if results else 1)) + 1
for item in results:
child: tuple = self.populate_tree_result_child(item=item, index_count_digits=count_digits)
if parent:
parent.appendRow(child)
else:
self.s_tr_results_add_top_level_item.emit(child)
def populate_tree_result_child(self, item: ResultItem, index_count_digits: int) -> Sequence[QtGui.QStandardItem]:
"""Create a row of QStandardItems for a ResultItem."""
duration: str = ""
if item.duration_sec > -1:
m, s = divmod(item.duration_sec, 60)
duration: str = f"{m:02d}:{s:02d}"
index: str = str(item.position + 1).zfill(index_count_digits)
child_index: QtGui.QStandardItem = QtGui.QStandardItem(index)
child_obj: QtGui.QStandardItem = QtGui.QStandardItem()
child_obj.setData(item.obj, QtCore.Qt.ItemDataRole.UserRole)
child_artist: QtGui.QStandardItem = QtGui.QStandardItem(item.artist)
child_title: QtGui.QStandardItem = QtGui.QStandardItem(item.title)
child_album: QtGui.QStandardItem = QtGui.QStandardItem(item.album)
child_duration: QtGui.QStandardItem = QtGui.QStandardItem(duration)
child_quality: QtGui.QStandardItem = QtGui.QStandardItem(item.quality)
child_date: QtGui.QStandardItem = QtGui.QStandardItem(
item.date_user_added if item.date_user_added != "" else item.date_release
)
child_downloaded: QtGui.QStandardItem = QtGui.QStandardItem()
if isinstance(item.obj, Track):
track_id = str(item.obj.id)
if self.history_service.is_downloaded(track_id):
child_downloaded.setText("✅")
child_downloaded.setTextAlignment(QtCore.Qt.AlignmentFlag.AlignCenter)
if isinstance(item.obj, Mix | Playlist | Album | Artist):
child_dummy: QtGui.QStandardItem = QtGui.QStandardItem()
child_dummy.setEnabled(False)
child_index.appendRow(child_dummy)
return (
child_index,
child_obj,
child_artist,
child_title,
child_album,
child_duration,
child_quality,
child_date,
child_downloaded,
)
def on_tr_results_add_top_level_item(self, item_child: Sequence[QtGui.QStandardItem]):
"""Add a top-level item to the results tree model."""
self.model_tr_results.appendRow(item_child)
def on_tr_results_expanded(self, index: QtCore.QModelIndex) -> None:
"""Handle the event when a result item group is expanded."""
self.thread_it(self.tr_results_expanded, index)
def tr_results_expanded(self, index: QtCore.QModelIndex) -> None:
"""Load and display the children of an expanded result item."""
item: QtGui.QStandardItem = self.model_tr_results.itemFromIndex(self.proxy_tr_results.mapToSource(index))
load_children: bool = not item.child(0, 0).isEnabled()
if load_children:
item.removeRow(0)
media_list = get_results_media_item(index, self.proxy_tr_results, self.model_tr_results)
self.s_spinner_start.emit(self.tr_results)
try:
self.list_items_show_result(media_list=media_list, parent=item)
finally:
self.s_spinner_stop.emit()
def list_items_show_result(
self,
media_list: Album | Playlist | Mix | Artist | None = None,
point: QtCore.QPoint | None = None,
parent: QtGui.QStandardItem | None = None,
favorite_function: Callable | None = None,
) -> None:
"""Populate the results tree with the items of a media list."""
if point:
item = self.tr_lists_user.itemAt(point)
media_list = get_user_list_media_item(item)
if favorite_function or isinstance(media_list, str):
if isinstance(media_list, str):
favorite_function = favorite_function_factory(self.tidal, media_list)
media_items: list[Track | Video | Album] = favorite_function()
else:
media_items: list[Track | Video | Album] = items_results_all(self.tidal.session, media_list)
result: list[ResultItem] = self.search_manager.search_result_to_model(media_items)
self.populate_tree_results(result, parent=parent)
def tidal_user_lists(self) -> None:
"""Fetch and emit user playlists, mixes, and favorites from Tidal."""
self.s_spinner_start.emit(self.tr_lists_user)
self.s_pb_reload_status.emit(False)
user_all: dict[str, list] = user_media_lists(self.tidal.session)
self.s_populate_tree_lists.emit(user_all)
def on_populate_tree_lists(self, user_lists: dict[str, list]) -> None:
"""Populate the user lists tree with playlists, mixes, and favorites."""
twi_playlists: QtWidgets.QTreeWidgetItem = self.tr_lists_user.findItems(
TidalLists.Playlists, QtCore.Qt.MatchExactly, 0
)[0]
twi_mixes: QtWidgets.QTreeWidgetItem = self.tr_lists_user.findItems(
TidalLists.Mixes, QtCore.Qt.MatchExactly, 0
)[0]
twi_favorites: QtWidgets.QTreeWidgetItem = self.tr_lists_user.findItems(
TidalLists.Favorites, QtCore.Qt.MatchExactly, 0
)[0]
for twi in [twi_playlists, twi_mixes]:
for i in reversed(range(twi.childCount())):
twi.removeChild(twi.child(i))
for item in user_lists.get("playlists", []):
if isinstance(item, Folder):
twi_child = QtWidgets.QTreeWidgetItem(twi_playlists)
name: str = f"📁 {item.name}"
info: str = f"({item.total_number_of_items} items)" if item.total_number_of_items else ""
twi_child.setText(0, name)
set_user_list_media(twi_child, item)
twi_child.setText(2, info)
dummy_child = QtWidgets.QTreeWidgetItem(twi_child)
dummy_child.setDisabled(True)
elif isinstance(item, UserPlaylist | Playlist):
twi_child = QtWidgets.QTreeWidgetItem(twi_playlists)
name: str = item.name if getattr(item, "name", None) is not None else ""
description: str = f" {item.description}" if item.description else ""
info: str = f"({item.num_tracks + item.num_videos} Tracks){description}"
twi_child.setText(0, name)
set_user_list_media(twi_child, item)
twi_child.setText(2, info)
for item in user_lists.get("mixes", []):
if isinstance(item, Mix):
twi_child = QtWidgets.QTreeWidgetItem(twi_mixes)
name: str = item.title
info: str = item.sub_title
twi_child.setText(0, name)
set_user_list_media(twi_child, item)
twi_child.setText(2, info)
for i in reversed(range(twi_favorites.childCount())):
twi_favorites.removeChild(twi_favorites.child(i))
for key, favorite in FAVORITES.items():
twi_child = QtWidgets.QTreeWidgetItem(twi_favorites)
name: str = favorite["name"]
info: str = ""
twi_child.setText(0, name)
set_user_list_media(twi_child, key)
twi_child.setText(2, info)
self.s_spinner_stop.emit()
self.s_pb_reload_status.emit(True)
def on_track_hover_confirmed(self, media: Track | Video | Album | Mix | Playlist | Artist) -> None:
"""Handle confirmed hover event over a track (after debounce delay)."""
if not media:
return
self.info_tab_widget.update_on_hover(media)
if self.cover_manager:
self.cover_manager.load_cover(media, use_cache_check=True)
def on_track_hover_left(self) -> None:
"""Handle hover leaving the track list."""
with contextlib.suppress(Exception):
self.info_tab_widget.revert_to_selection()
if self.info_tab_widget.current_media_selected and self.cover_manager:
self.cover_manager.load_cover(self.info_tab_widget.current_media_selected, use_cache_check=True)
+53
View File
@@ -0,0 +1,53 @@
"""UI helpers mixin for MainWindow.
Handles UI helper functions like spinners, logs, and status bar messages.
"""
from PySide6 import QtGui, QtWidgets
from tidal_dl_ng.model.gui_data import StatusbarMessage
from tidal_dl_ng.ui.spinner import QtWaitingSpinner
class UIHelpersMixin:
"""Mixin containing UI helper methods."""
def on_spinner_start(self, parent: QtWidgets.QWidget) -> None:
"""Start a loading spinner on the given parent widget."""
if parent in self.spinners:
spinner = self.spinners[parent]
spinner.stop()
spinner.deleteLater()
del self.spinners[parent]
spinner = QtWaitingSpinner(parent, True, True)
spinner.setColor(QtGui.QColor(255, 255, 255))
spinner.start()
self.spinners[parent] = spinner
def on_spinner_stop(self) -> None:
"""Stop all active loading spinners."""
for spinner in list(self.spinners.values()):
spinner.stop()
spinner.deleteLater()
self.spinners.clear()
def on_statusbar_message(self, data: StatusbarMessage):
"""Show a message in the status bar."""
self.statusbar.showMessage(data.message, data.timeout)
def _log_output(self, text: str) -> None:
"""Redirect log output to the debug text area."""
cursor: QtGui.QTextCursor = self.te_debug.textCursor()
html = self.converter_ansi_html.convert(text)
cursor.movePosition(QtGui.QTextCursor.End)
cursor.insertHtml(html)
self.te_debug.setTextCursor(cursor)
self.te_debug.ensureCursorVisible()
def button_reload_status(self, status: bool) -> None:
"""Update the reload button's state and text."""
button_text: str = "Reloading..." if not status else "Reload"
self.pb_reload_user_lists.setEnabled(status)
self.pb_reload_user_lists.setText(button_text)
+25
View File
@@ -0,0 +1,25 @@
"""Updates mixin for MainWindow.
Handles application update checking.
"""
from tidal_dl_ng import update_available
from tidal_dl_ng.dialog import DialogVersion
from tidal_dl_ng.model.meta import ReleaseLatest
class UpdatesMixin:
"""Mixin containing update checking methods."""
def on_update_check(self, on_startup: bool = True) -> None:
"""Check for application updates and emit update signals."""
is_available, info = update_available()
if (on_startup and is_available) or not on_startup:
self.s_update_show.emit(True, is_available, info)
def on_version(
self, update_check: bool = False, update_available: bool = False, update_info: ReleaseLatest | None = None
) -> None:
"""Show the version information dialog."""
DialogVersion(self, update_check, update_available, update_info)