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