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