diff options
Diffstat (limited to 'tidal_dl_ng/gui.py')
| -rw-r--r-- | tidal_dl_ng/gui.py | 481 |
1 files changed, 481 insertions, 0 deletions
diff --git a/tidal_dl_ng/gui.py b/tidal_dl_ng/gui.py new file mode 100644 index 0000000..a995c01 --- /dev/null +++ b/tidal_dl_ng/gui.py @@ -0,0 +1,481 @@ +import math +import sys +from collections.abc import Callable + +try: + import qdarktheme + from PySide6 import QtCore, QtGui, QtWidgets +except ImportError as e: + print(e) + print("Qt dependencies missing. Cannot start GUI. Please execute: 'pip install pyside6 pyqtdarktheme'") + sys.exit(1) + +import coloredlogs.converter +from rich.progress import Progress +from tidalapi import Album, Mix, Playlist, Quality, Track, UserPlaylist, Video +from tidalapi.session import SearchTypes +from tidal_dl_ng.config import Settings, Tidal +from tidal_dl_ng.constants import QualityVideo +from tidal_dl_ng.download import Download +from tidal_dl_ng.logger import XStream, logger_cli, logger_gui +from tidal_dl_ng.model.gui_data import ProgressBars, ResultSearch +from tidal_dl_ng.ui.main import Ui_MainWindow +from tidal_dl_ng.ui.spinner import QtWaitingSpinner +from tidal_dl_ng.worker import Worker + + +# TODO: Make more use of Exceptions +class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow): + settings: Settings = None + tidal: Tidal = None + download: Download = None + threadpool: QtCore.QThreadPool = None + spinner: QtWaitingSpinner = None + spinner_start: QtCore.Signal = QtCore.Signal(QtWidgets.QWidget) + spinner_stop: QtCore.Signal = QtCore.Signal() + pb_item: QtWidgets.QProgressBar = None + s_item_advance: QtCore.Signal = QtCore.Signal(float) + s_item_name: QtCore.Signal = QtCore.Signal(str) + pb_list: QtWidgets.QProgressBar = None + s_list_advance: QtCore.Signal = QtCore.Signal(float) + + def __init__(self, tidal: Tidal | None = None): + super().__init__() + self.setupUi(self) + # self.setGeometry(50, 50, 500, 300) + self.setWindowTitle("TIDAL Downloader Next Gen!") + # TODO: Fix icons (make them visible). + my_pixmap = QtGui.QPixmap("tidal_dl_ng/ui/icon.png") + my_icon = QtGui.QIcon("tidal_dl_ng/ui/icon.png") + self.setWindowIcon(my_icon) + tray = QtWidgets.QSystemTrayIcon() + tray.setIcon(my_icon) + tray.setVisible(True) + + # Logging redirect. + XStream.stdout().messageWritten.connect(self._log_output) + # XStream.stderr().messageWritten.connect(self._log_output) + + self.settings = Settings() + self.threadpool = QtCore.QThreadPool() + + # TODO: Show GUI, create a progress bar showing the TIDAL querying progress. + self._init_tree_results(self.tr_results) + self._init_tree_lists(self.tr_lists_user) + self._init_progressbar() + self._populate_quality(self.cb_quality_audio, Quality) + self._populate_quality(self.cb_quality_video, QualityVideo) + self._populate_search_types(self.cb_search_type, SearchTypes) + self.apply_settings(self.settings) + self._init_signals() + self.thread_it(self.init_tidal, tidal) + + logger_gui.debug("Multithreading with maximum %d threads" % self.threadpool.maxThreadCount()) + logger_gui.debug("All setup.") + + def init_tidal(self, tidal: Tidal = None): + result: bool = False + + if tidal: + self.tidal = tidal + result = True + else: + self.tidal = Tidal(self.settings) + + while True: + result = self.tidal.login(logger_gui.info) + + if result: + break + + if result: + self.download = Download(self.tidal.session, self.tidal.settings.data.skip_existing) + + self.thread_it(self.populate_tree_lists, self.tidal) + + def _init_progressbar(self): + 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.pb_progress.setVisible() + self.statusbar.addPermanentWidget(pb) + + def progress_reset(self): + self.pb_list.setValue(0) + self.pb_item.setValue(0) + + def _log_output(self, text): + display_msg = coloredlogs.converter.convert(text) + + cursor: QtGui.QTextCursor = self.te_debug.textCursor() + cursor.movePosition(QtGui.QTextCursor.End) + cursor.insertHtml(display_msg) + + self.te_debug.setTextCursor(cursor) + self.te_debug.ensureCursorVisible() + + def _populate_quality(self, ui_target: QtWidgets.QComboBox, options: type[Quality | QualityVideo]): + for item in options: + ui_target.addItem(item.name, item) + + def _populate_search_types(self, ui_target: QtWidgets.QComboBox, options: SearchTypes): + for item in options: + if item and not item.__name__ == "Artist": + ui_target.addItem(item.__name__, item) + + self.cb_search_type.setCurrentIndex(1) + + def _init_tree_results(self, tree: QtWidgets.QTableWidget): + tree.setColumnHidden(5, True) + + def populate_tree_lists(self, tidal: Tidal): + # Start loading spinner + self.spinner_start.emit(self.tr_lists_user) + self.tr_results.clear() + + # TODO: Refactor list item names to constants. + twi_playlists: QtWidgets.QTableWidgetItem = self.tr_lists_user.findItems( + "Playlists", QtCore.Qt.MatchExactly, 0 + )[0] + twi_mixes: QtWidgets.QTableWidgetItem = self.tr_lists_user.findItems("Favorites", QtCore.Qt.MatchExactly, 0)[0] + twi_favorites: QtWidgets.QTableWidgetItem = self.tr_lists_user.findItems("Mixes", QtCore.Qt.MatchExactly, 0)[0] + + user_playlists: [Playlist | UserPlaylist] = tidal.session.user.playlist_and_favorite_playlists() + user_mixes: [Mix] = tidal.session.mixes().categories[0].items + user_all: [Playlist | UserPlaylist | Mix] = user_playlists + user_mixes + + for item in user_all: + if isinstance(item, UserPlaylist): + twi_child = QtWidgets.QTreeWidgetItem(twi_playlists) + name: str = item.name + info: str = f"({item.num_tracks + item.num_videos} Tracks)" + elif isinstance(item, Playlist): + twi_child = QtWidgets.QTreeWidgetItem(twi_mixes) + name: str = item.name + info: str = f"({item.num_tracks + item.num_videos} Tracks) {item.description}" + elif isinstance(item, Mix): + twi_child = QtWidgets.QTreeWidgetItem(twi_favorites) + name: str = item.title + info: str = item.sub_title + + twi_child.setText(0, name) + twi_child.setText(1, info) + twi_child.setData(3, QtCore.Qt.ItemDataRole.UserRole, item) + + # Stop load spinner + self.spinner_stop.emit() + + def _init_tree_lists(self, tree: QtWidgets.QTreeWidget): + tree.setColumnWidth(0, 200) + tree.setColumnWidth(1, 300) + tree.setColumnHidden(2, True) + tree.expandAll() + + # Connect the contextmenu + tree.setContextMenuPolicy(QtCore.Qt.CustomContextMenu) + tree.customContextMenuRequested.connect(self.menu_context_tree_lists) + + def apply_settings(self, settings: Settings): + l_cb = [ + {"element": self.cb_quality_audio, "setting": settings.data.quality_audio.name, "default_id": 1}, + {"element": self.cb_quality_video, "setting": settings.data.quality_video.name, "default_id": 0}, + ] + + for item in l_cb: + idx = item["element"].findText(item["setting"]) + if idx > -1: + item["element"].setCurrentIndex(idx) + else: + item["element"].setCurrentIndex(item["default_id"]) + + @QtCore.Slot(QtWidgets.QWidget) + def on_spinner_start(self, parent: QtWidgets.QWidget): + self.spinner = QtWaitingSpinner(parent, True, True) + self.spinner.setColor(QtGui.QColor(255, 255, 255)) + self.spinner.start() + + @QtCore.Slot() + def on_spinner_stop(self): + self.spinner.stop() + self.spinner = None + + def menu_context_tree_lists(self, point): + # Infos about the node selected. + index = self.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 + + # We build the menu. + menu = QtWidgets.QMenu() + action = menu.addAction("Dowloading Playlist", lambda: self.thread_download_list_media(point)) + + menu.exec(self.tr_lists_user.mapToGlobal(point)) + + def thread_download_list_media(self, point): + self.thread_it(self.on_download_list_media, point) + self.thread_it(self.list_items_show, point=point) + + def on_download_list_media(self, point: QtCore.QPoint): + item = self.tr_lists_user.itemAt(point) + media = item.data(3, QtCore.Qt.ItemDataRole.UserRole) + + # TODO: Implement disable download button etc. + self.download_item(media, self.download) + + def search_populate_results(self, query: str, type_media: SearchTypes): + self.tr_results.clear() + + results: [ResultSearch] = self.search(query, [type_media]) + + self.populate_tree_results(results) + + def populate_tree_results(self, results: [ResultSearch]): + self.tr_results.clear() + + for item in results: + # Format seconds to mm:ss. + m, s = divmod(item.duration_sec, 60) + duration: str = f"{m:02d}:{s:02d}" + child = QtWidgets.QTreeWidgetItem() + + child.setText(0, str(item.position)) + child.setText(1, item.artist) + child.setText(2, item.title) + child.setText(3, item.album) + child.setText(4, duration) + child.setData(5, QtCore.Qt.ItemDataRole.UserRole, item.obj) + + self.tr_results.addTopLevelItem(child) + + def search(self, query: str, types_media: SearchTypes) -> [ResultSearch]: + result_search: [dict[str, *SearchTypes]] = self.tidal.session.search(query, models=types_media, limit=999) + result: [ResultSearch] = [] + + 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: [*SearchTypes]) -> [ResultSearch]: + result = [] + + for idx, item in enumerate(items): + if isinstance(item, Track): + result_item: ResultSearch = ResultSearch( + position=idx, + artist=item.artist.name, + title=item.name, + album=item.album.name, + duration_sec=item.duration, + obj=item, + ) + + result.append(result_item) + elif isinstance(item, Video): + result_item: ResultSearch = ResultSearch( + position=idx, + artist=item.artist.name, + title=item.name, + album=item.album.name if item.album else "", + duration_sec=item.duration, + obj=item, + ) + + result.append(result_item) + elif isinstance(item, Playlist): + result_item: ResultSearch = ResultSearch( + position=idx, + artist=f"{item.promoted_artists[0].name}, ..." if item.promoted_artists else "", + title=item.name, + album="", + duration_sec=item.duration, + obj=item, + ) + + result.append(result_item) + elif isinstance(item, Album): + result_item: ResultSearch = ResultSearch( + position=idx, + artist=item.artist.name, + title="", + album=item.name, + duration_sec=item.duration, + obj=item, + ) + + result.append(result_item) + + return result + + def _init_signals(self): + self.b_download.clicked.connect(lambda: self.thread_it(self.on_download_results)) + self.l_search.returnPressed.connect( + lambda: self.search_populate_results(self.l_search.text(), self.cb_search_type.currentData()) + ) + self.b_search.clicked.connect( + lambda: self.search_populate_results(self.l_search.text(), self.cb_search_type.currentData()) + ) + self.cb_quality_audio.currentIndexChanged.connect(self.quality_set_audio) + self.cb_quality_video.currentIndexChanged.connect(self.quality_set_video) + self.tr_lists_user.itemClicked.connect(self.on_list_items_show) + self.spinner_start[QtWidgets.QWidget].connect(self.on_spinner_start) + self.spinner_stop.connect(self.on_spinner_stop) + self.s_item_advance.connect(self.progress_item) + self.s_item_name.connect(self.progress_item_name) + self.s_list_advance.connect(self.progress_list) + + def progress_list(self, value: float): + self.pb_list.setValue(int(math.ceil(value))) + + def progress_item(self, value: float): + self.pb_item.setValue(int(math.ceil(value))) + + def progress_item_name(self, value: str): + self.pb_item.setFormat(f"%p% {value}") + + def progress_list_name(self, value: str): + self.pb_list.setFormat(f"%p% {value}") + + def quality_set_audio(self, index): + self.settings.data.quality_audio = Quality(self.cb_quality_audio.itemData(index).value) + self.settings.save() + + if self.tidal: + self.tidal.settings_apply() + + def quality_set_video(self, index): + self.settings.data.quality_video = QualityVideo(self.cb_quality_video.itemData(index).value) + self.settings.save() + + if self.tidal: + self.tidal.settings_apply() + + def on_list_items_show(self, item: QtWidgets.QTreeWidgetItem): + media_list: Album | Playlist = item.data(3, QtCore.Qt.ItemDataRole.UserRole) + + # Only if clicked item is not a top level item. + if media_list: + self.list_items_show(media_list) + + def list_items_show(self, media_list: Album | Playlist | None = None, point: QtCore.QPoint | None = None): + if point: + item = self.tr_lists_user.itemAt(point) + media_list = item.data(3, QtCore.Qt.ItemDataRole.UserRole) + + media_items = media_list.items() + result = self.search_result_to_model(media_items) + + self.populate_tree_results(result) + + def thread_it(self, fn: Callable, *args, **kwargs): + # Any other args, kwargs are passed to the run function + worker = Worker(fn, *args, **kwargs) + + # Execute + self.threadpool.start(worker) + + def on_download_results(self): + self.b_download.setEnabled(False) + self.b_download.setText(f"Downloading...") + + items: [QtWidgets.QTreeWidgetItem] = self.tr_results.selectedItems() + + if len(items) == 0: + logger_gui.error("Please select a row first.") + else: + for item in items: + media: Track | Album | Playlist | Video = item.data(5, QtCore.Qt.ItemDataRole.UserRole) + + result_dl: bool = self.download_item(media, self.download) + + self.b_download.setText(f"Download") + self.b_download.setEnabled(True) + + def download_item(self, media: Track | Album | Playlist | Video | Mix, dl: Download) -> bool: + # TODO: Refactor this. Move this logic to `Download` class and provide a generic interface. + self.progress_reset() + self.statusbar.showMessage("Download started...") + + data_pb: ProgressBars = ProgressBars( + item=self.s_item_advance, list_item=self.s_list_advance, item_name=self.s_item_name + ) + progress: Progress = Progress() + + if isinstance(media, Track) or isinstance(media, Video): + if isinstance(media, Track): + file_template: str = self.settings.data.format_track + elif isinstance(media, Video): + file_template: str = self.settings.data.format_video + + result_download, download_path_file = dl.item( + media=media, + path_base=self.settings.data.download_base_path, + file_template=file_template, + progress_gui=data_pb, + progress=progress, + fn_logger=logger_gui + ) + + if result_download: + logger_gui.info(f"Download successful: {download_path_file}") + else: + logger_gui.info(f"Download skipped (file exists): {download_path_file}") + elif isinstance(media, Album) or isinstance(media, Playlist) or isinstance(media, Mix): + file_template: str | bool = False + + if isinstance(media, Album): + file_template: str = self.settings.data.format_album + progress_name = media.name + elif isinstance(media, Playlist): + file_template: str = self.settings.data.format_playlist + progress_name = media.name + elif isinstance(media, Mix): + file_template: str = self.settings.data.format_mix + progress_name = media.title + else: + progress_name = "List N/A" + + self.progress_list_name(progress_name) + + dl.list( + path_base=self.settings.data.download_base_path, + file_template=file_template, + list_media=media, + video_download=self.settings.data.video_download, + progress_gui=data_pb, + progress=progress, + download_delay=self.settings.data.download_delay, + fn_logger=logger_gui, + ) + + self.statusbar.showMessage("Download finished.", 2000) + progress.stop() + + # TODO: Refactor to useful return value. + return True + + def progress_reset(self): + self.pb_list.setValue(0) + self.pb_item.setValue(0) + + +# TODO: Comment with Google Docstrings. +def gui_activate(tidal: Tidal | None = None): + qdarktheme.enable_hi_dpi() + app = QtWidgets.QApplication(sys.argv) + qdarktheme.setup_theme() + + window = MainWindow(tidal) + window.show() + + sys.exit(app.exec()) + + +if __name__ == "__main__": + gui_activate() |