210 lines
8.3 KiB
Python
210 lines
8.3 KiB
Python
|
"""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
|