312 lines
14 KiB
Python
312 lines
14 KiB
Python
# tidal_dl_ng/gui_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
|