diff options
Diffstat (limited to 'tidal_dl_ng/gui/queue.py')
| -rw-r--r-- | tidal_dl_ng/gui/queue.py | 311 |
1 files changed, 311 insertions, 0 deletions
diff --git a/tidal_dl_ng/gui/queue.py b/tidal_dl_ng/gui/queue.py new file mode 100644 index 0000000..ec425bc --- /dev/null +++ b/tidal_dl_ng/gui/queue.py @@ -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 |