summaryrefslogtreecommitdiff
path: root/tidal_dl_ng
diff options
context:
space:
mode:
authorexislow <exislow@users.noreply.github.com>2025-07-03 09:57:26 +0200
committerGitHub <noreply@github.com>2025-07-03 09:57:26 +0200
commit7da0d4c41e4abb6ec03ee1b0f64474ee970a230e (patch)
treea5e625565e2439ad92c590f7f9d2aeaf49824fe9 /tidal_dl_ng
parent29ea5c51825628dbe96ce4404488b87d5fadb80d (diff)
parent4662d18fdfc25d42d496cf6abb1c8d44be0c76dd (diff)
downloadtidal-dl-ng-7da0d4c41e4abb6ec03ee1b0f64474ee970a230e.tar.gz
Merge pull request #478 from exislow/ai-refactoring
Diffstat (limited to 'tidal_dl_ng')
-rw-r--r--tidal_dl_ng/cli.py363
-rw-r--r--tidal_dl_ng/download.py1098
-rw-r--r--tidal_dl_ng/gui.py810
-rw-r--r--tidal_dl_ng/helper/decryption.py12
-rw-r--r--tidal_dl_ng/helper/path.py581
5 files changed, 2154 insertions, 710 deletions
diff --git a/tidal_dl_ng/cli.py b/tidal_dl_ng/cli.py
index 75deba4..85e9f6e 100644
--- a/tidal_dl_ng/cli.py
+++ b/tidal_dl_ng/cli.py
@@ -22,12 +22,6 @@ from tidal_dl_ng.constants import CTX_TIDAL, MediaType
from tidal_dl_ng.download import Download
from tidal_dl_ng.helper.path import get_format_template, path_file_settings
from tidal_dl_ng.helper.tidal import (
- Album,
- Artist,
- Mix,
- Playlist,
- Track,
- Video,
all_artist_album_ids,
get_tidal_media_id,
get_tidal_media_type,
@@ -47,30 +41,167 @@ app.add_typer(dl_fav_group, name="dl_fav")
def version_callback(value: bool):
+ """Callback to print version and exit if version flag is set.
+
+ Args:
+ value (bool): If True, prints version and exits.
+ """
if value:
print(f"{__version__}")
+
raise typer.Exit()
+def _handle_track_or_video(
+ dl: Download, ctx: typer.Context, item: str, media: object, file_template: str, idx: int, urls_pos_last: int
+) -> None:
+ """Handle downloading a track or video item.
+
+ Args:
+ dl (Download): The Download instance.
+ ctx (typer.Context): Typer context object.
+ item (str): The URL or identifier of the item.
+ media: The media object to download.
+ file_template (str): The file template for saving the media.
+ idx (int): The index of the item in the list.
+ urls_pos_last (int): The last index in the URLs list.
+ """
+ settings = ctx.obj[CTX_TIDAL].settings
+ download_delay: bool = bool(settings.data.download_delay and idx < urls_pos_last)
+
+ dl.item(
+ media=media,
+ file_template=file_template,
+ download_delay=download_delay,
+ quality_audio=settings.data.quality_audio,
+ quality_video=settings.data.quality_video,
+ )
+
+
+def _handle_album_playlist_mix_artist(
+ ctx: typer.Context,
+ dl: Download,
+ handling_app: HandlingApp,
+ media_type: MediaType,
+ media: object,
+ item_id: str,
+ file_template: str,
+) -> bool:
+ """Handle downloading albums, playlists, mixes, or artist collections.
+
+ Args:
+ ctx (typer.Context): Typer context object.
+ dl (Download): The Download instance.
+ handling_app (HandlingApp): The HandlingApp instance.
+ media_type (MediaType): The type of media (album, playlist, mix, or artist).
+ media: The media object to download.
+ item_id (str): The ID of the media item.
+ file_template (str): The file template for saving the media.
+
+ Returns:
+ bool: False if aborted, True otherwise.
+ """
+ item_ids: list[str] = []
+ settings = ctx.obj[CTX_TIDAL].settings
+
+ if media_type == MediaType.ARTIST:
+ media_type = MediaType.ALBUM
+ item_ids += all_artist_album_ids(media)
+ else:
+ item_ids.append(item_id)
+
+ for _item_id in item_ids:
+ if handling_app.event_abort.is_set():
+ return False
+
+ dl.items(
+ media_id=_item_id,
+ media_type=media_type,
+ file_template=file_template,
+ video_download=settings.data.video_download,
+ download_delay=settings.data.download_delay,
+ quality_audio=settings.data.quality_audio,
+ quality_video=settings.data.quality_video,
+ )
+
+ return True
+
+
+def _process_url(
+ dl: Download,
+ ctx: typer.Context,
+ handling_app: HandlingApp,
+ item: str,
+ idx: int,
+ urls_pos_last: int,
+) -> bool:
+ """Process a single URL or ID for download.
+
+ Args:
+ dl (Download): The Download instance.
+ ctx (typer.Context): Typer context object.
+ handling_app (HandlingApp): The HandlingApp instance.
+ item (str): The URL or identifier to process.
+ idx (int): The index of the item in the list.
+ urls_pos_last (int): The last index in the URLs list.
+
+ Returns:
+ bool: False if aborted, True otherwise.
+ """
+ settings = ctx.obj[CTX_TIDAL].settings
+
+ if handling_app.event_abort.is_set():
+ return False
+
+ if "http" not in item:
+ print(f"It seems like that you have supplied an invalid URL: {item}")
+ return True
+
+ media_type = get_tidal_media_type(item)
+ if not isinstance(media_type, MediaType):
+ print(f"Could not determine media type for: {item}")
+ return True
+
+ item_id = get_tidal_media_id(item)
+ if not isinstance(item_id, str):
+ print(f"Could not determine media id for: {item}")
+ return True
+
+ file_template = get_format_template(media_type, settings)
+ if not isinstance(file_template, str):
+ print(f"Could not determine file template for: {item}")
+ return True
+
+ try:
+ media = instantiate_media(ctx.obj[CTX_TIDAL].session, media_type, item_id)
+ except Exception:
+ print(f"Media not found (ID: {item_id}). Maybe it is not available anymore.")
+ return True
+
+ if media_type in [MediaType.TRACK, MediaType.VIDEO]:
+ _handle_track_or_video(dl, ctx, item, media, file_template, idx, urls_pos_last)
+ elif media_type in [MediaType.ALBUM, MediaType.PLAYLIST, MediaType.MIX, MediaType.ARTIST]:
+ return _handle_album_playlist_mix_artist(ctx, dl, handling_app, media_type, media, item_id, file_template)
+ return True
+
+
def _download(ctx: typer.Context, urls: list[str], try_login: bool = True) -> bool:
"""Invokes download function and tracks progress.
- :param ctx: The typer context object.
- :type ctx: typer.Context
- :param urls: The list of URLs to download.
- :type urls: list[str]
- :param try_login: If true, attempts to login to TIDAL.
- :type try_login: bool
- :return: True if ran successfully.
- :rtype: bool
+ Args:
+ ctx (typer.Context): The typer context object.
+ urls (list[str]): The list of URLs to download.
+ try_login (bool, optional): If true, attempts to login to TIDAL. Defaults to True.
+
+ Returns:
+ bool: True if ran successfully.
"""
if try_login:
- # Call login method to validate the token.
ctx.invoke(login, ctx)
- # Create initial objects.
- settings: Settings = Settings()
+ settings: Settings = ctx.obj[CTX_TIDAL].settings
handling_app: HandlingApp = HandlingApp()
+
progress: Progress = Progress(
TextColumn("[progress.description]{task.description}"),
SpinnerColumn(),
@@ -79,8 +210,9 @@ def _download(ctx: typer.Context, urls: list[str], try_login: bool = True) -> bo
refresh_per_second=20,
auto_refresh=True,
expand=True,
- transient=False, # Prevent progress from disappearing
+ transient=False,
)
+
progress_overall = Progress(
TextColumn("[progress.description]{task.description}"),
SpinnerColumn(),
@@ -89,12 +221,14 @@ def _download(ctx: typer.Context, urls: list[str], try_login: bool = True) -> bo
refresh_per_second=20,
auto_refresh=True,
expand=True,
- transient=False, # Prevent progress from disappearing
+ transient=False,
)
+
fn_logger = LoggerWrapped(progress.print)
+
dl = Download(
session=ctx.obj[CTX_TIDAL].session,
- skip_existing=ctx.obj[CTX_TIDAL].settings.data.skip_existing,
+ skip_existing=settings.data.skip_existing,
path_base=settings.data.download_base_path,
fn_logger=fn_logger,
progress=progress,
@@ -102,80 +236,20 @@ def _download(ctx: typer.Context, urls: list[str], try_login: bool = True) -> bo
event_abort=handling_app.event_abort,
event_run=handling_app.event_run,
)
- progress_table = Table.grid()
- # Style Progress display.
+ progress_table = Table.grid()
progress_table.add_row(progress)
progress_table.add_row(progress_overall)
-
- progress_group = Group(
- progress_table,
- )
+ progress_group = Group(progress_table)
urls_pos_last = len(urls) - 1
- # Use a single Live display for both progress and table
with Live(progress_group, refresh_per_second=20, vertical_overflow="visible"):
try:
- for item in urls:
- media_type: MediaType | bool = False
-
- # Exit loop if abort signal is set.
- if handling_app.event_abort.is_set():
+ for idx, item in enumerate(urls):
+ if _process_url(dl, ctx, handling_app, item, idx, urls_pos_last) is False:
return False
-
- # Extract media name and id from link.
- if "http" in item:
- media_type = get_tidal_media_type(item)
- item_id = get_tidal_media_id(item)
- file_template = get_format_template(media_type, settings)
- else:
- print(f"It seems like that you have supplied an invalid URL: {item}")
-
- continue
-
- try:
- media: Track | Video | Album | Playlist | Mix | Artist = instantiate_media(
- ctx.obj[CTX_TIDAL].session, media_type, item_id
- )
- except Exception:
- print(f"Media not found (ID: {item_id}). Maybe it is not available anymore.")
- continue
-
- # Download media.
- if media_type in [MediaType.TRACK, MediaType.VIDEO]:
- download_delay: bool = bool(settings.data.download_delay and urls.index(item) < urls_pos_last)
-
- dl.item(
- media=media,
- file_template=file_template,
- download_delay=download_delay,
- quality_audio=settings.data.quality_audio,
- quality_video=settings.data.quality_video,
- )
- elif media_type in [MediaType.ALBUM, MediaType.PLAYLIST, MediaType.MIX, MediaType.ARTIST]:
- item_ids: [int] = []
-
- if media_type == MediaType.ARTIST:
- media_type = MediaType.ALBUM
- item_ids = item_ids + all_artist_album_ids(media)
- else:
- item_ids.append(item_id)
-
- for item_id in item_ids:
- # Exit loop if abort signal is set.
- if handling_app.event_abort.is_set():
- return False
-
- dl.items(
- media_id=item_id,
- media_type=media_type,
- file_template=file_template,
- video_download=ctx.obj[CTX_TIDAL].settings.data.video_download,
- download_delay=settings.data.download_delay,
- )
finally:
- # Clear and stop progress display
progress.refresh()
progress.stop()
@@ -187,6 +261,12 @@ def callback_app(
ctx: typer.Context,
version: Annotated[bool | None, typer.Option("--version", "-v", callback=version_callback, is_eager=True)] = None,
):
+ """App callback to initialize context and handle version option.
+
+ Args:
+ ctx (typer.Context): Typer context object.
+ version (bool | None, optional): Version flag. Defaults to None.
+ """
ctx.obj = {"tidal": None}
@@ -196,16 +276,12 @@ def settings_management(
editor: Annotated[
bool, typer.Option("--editor", "-e", help="Open the settings file in your default editor.")
] = False,
-):
- """
- Print or set an option.
- If no arguments are given, all options will be listed.
- If only one argument is given, the value will be printed for this option.
- To set a value for an option simply pass the value as the second argument
+) -> None:
+ """Print or set an option, or open the settings file in an editor.
- :param editor: If set, your favorite system editor will be opened.
- :param names: (Optional) None (list all options), one (list the value only for this option) or two arguments
- (set the value for the option).
+ Args:
+ names (list[str] | None, optional): None (list all options), one (list the value only for this option) or two arguments (set the value for the option). Defaults to None.
+ editor (bool, optional): If set, your favorite system editor will be opened. Defaults to False.
"""
if editor:
config_path: Path = Path(path_file_settings())
@@ -223,12 +299,11 @@ def settings_management(
if names:
if names[0] not in d_settings:
print(f'Option "{names[0]}" is not valid!')
- else:
- if len(names) == 1:
- print(f'{names[0]}: "{d_settings[names[0]]}"')
- elif len(names) > 1:
- settings.set_option(names[0], names[1])
- settings.save()
+ elif len(names) == 1:
+ print(f'{names[0]}: "{d_settings[names[0]]}"')
+ elif len(names) > 1:
+ settings.set_option(names[0], names[1])
+ settings.save()
else:
help_settings: dict = HelpSettings().to_dict()
table = Table(title=f"Config: {path_file_settings()}")
@@ -236,7 +311,6 @@ def settings_management(
table.add_column("Value", style="magenta")
table.add_column("Description", style="green")
- # Iterate over the attributes of the dataclass
for key, value in sorted(d_settings.items()):
table.add_row(key, str(value), help_settings[key])
@@ -246,6 +320,14 @@ def settings_management(
@app.command(name="login")
def login(ctx: typer.Context) -> bool:
+ """Login to TIDAL and update context object.
+
+ Args:
+ ctx (typer.Context): Typer context object.
+
+ Returns:
+ bool: True if login was successful, False otherwise.
+ """
print("Let us check, if you are already logged in... ", end="")
settings = Settings()
@@ -258,6 +340,11 @@ def login(ctx: typer.Context) -> bool:
@app.command(name="logout")
def logout() -> bool:
+ """Logout from TIDAL.
+
+ Returns:
+ bool: True if logout was successful, False otherwise.
+ """
settings = Settings()
tidal = Tidal(settings)
result = tidal.logout()
@@ -287,6 +374,16 @@ def download(
),
] = None,
) -> bool:
+ """Download media from provided URLs or a file containing URLs.
+
+ Args:
+ ctx (typer.Context): Typer context object.
+ urls (list[str] | None, optional): List of URLs to download. Defaults to None.
+ file_urls (Path | None, optional): Path to file containing URLs. Defaults to None.
+
+ Returns:
+ bool: True if download was successful, False otherwise.
+ """
if not urls:
# Read the text file provided.
if file_urls:
@@ -307,10 +404,11 @@ def download(
def download_fav_tracks(ctx: typer.Context) -> bool:
"""Download your favorite track collection.
- :param ctx: Typer context object.
- :type ctx: typer.Context
- :return: Download result.
- :rtype: bool
+ Args:
+ ctx (typer.Context): Typer context object.
+
+ Returns:
+ bool: Download result.
"""
# Method name
func_name_favorites: str = "tracks"
@@ -325,10 +423,11 @@ def download_fav_tracks(ctx: typer.Context) -> bool:
def download_fav_artists(ctx: typer.Context) -> bool:
"""Download your favorite artist collection.
- :param ctx: Typer context object.
- :type ctx: typer.Context
- :return: Download result.
- :rtype: bool
+ Args:
+ ctx (typer.Context): Typer context object.
+
+ Returns:
+ bool: Download result.
"""
# Method name
func_name_favorites: str = "artists"
@@ -343,10 +442,11 @@ def download_fav_artists(ctx: typer.Context) -> bool:
def download_fav_albums(ctx: typer.Context) -> bool:
"""Download your favorite album collection.
- :param ctx: Typer context object.
- :type ctx: typer.Context
- :return: Download result.
- :rtype: bool
+ Args:
+ ctx (typer.Context): Typer context object.
+
+ Returns:
+ bool: Download result.
"""
# Method name
func_name_favorites: str = "albums"
@@ -361,10 +461,11 @@ def download_fav_albums(ctx: typer.Context) -> bool:
def download_fav_videos(ctx: typer.Context) -> bool:
"""Download your favorite video collection.
- :param ctx: Typer context object.
- :type ctx: typer.Context
- :return: Download result.
- :rtype: bool
+ Args:
+ ctx (typer.Context): Typer context object.
+
+ Returns:
+ bool: Download result.
"""
# Method name
func_name_favorites: str = "videos"
@@ -375,26 +476,26 @@ def download_fav_videos(ctx: typer.Context) -> bool:
def _download_fav_factory(ctx: typer.Context, func_name_favorites: str) -> bool:
"""Factory which helps to download items from the favorites collections.
- :param ctx: Typer context object.
- :type ctx: typer.Context
- :param func_name_favorites: Method name to call from `tidalapi` favorites object.
- :type func_name_favorites: str
- :return: Download result.
- :rtype: bool
+ Args:
+ ctx (typer.Context): Typer context object.
+ func_name_favorites (str): Method name to call from `tidalapi` favorites object.
+
+ Returns:
+ bool: Download result.
"""
- # Call login method to validate the token.
ctx.invoke(login, ctx)
-
- # Get the method from the module
func_favorites: Callable = getattr(ctx.obj[CTX_TIDAL].session.user.favorites, func_name_favorites)
- # Get favorite videos
- media_urls: [str] = [media.share_url for media in func_favorites()]
-
+ media_urls: list[str] = [media.share_url for media in func_favorites()]
return _download(ctx, media_urls, try_login=False)
@app.command()
def gui(ctx: typer.Context):
+ """Launch the GUI for the application.
+
+ Args:
+ ctx (typer.Context): Typer context object.
+ """
from tidal_dl_ng.gui import gui_activate
ctx.invoke(login, ctx)
@@ -404,9 +505,9 @@ def gui(ctx: typer.Context):
def handle_sigint_term(signum, frame):
"""Set app abort event, so threads can check it and shutdown.
- :param signum:
- :param frame:
- :return:
+ Args:
+ signum: Signal number.
+ frame: Current stack frame.
"""
handling_app: HandlingApp = HandlingApp()
diff --git a/tidal_dl_ng/download.py b/tidal_dl_ng/download.py
index 9b13053..ba604b6 100644
--- a/tidal_dl_ng/download.py
+++ b/tidal_dl_ng/download.py
@@ -1,3 +1,13 @@
+"""
+download.py
+
+Implements the Download class and helpers for downloading media from TIDAL, including segment merging, file moving, metadata writing, and playlist creation.
+
+Classes:
+ RequestsClient: Simple HTTP client for downloading text content.
+ Download: Main class for managing downloads, segment merging, file operations, and metadata.
+"""
+
import os
import pathlib
import random
@@ -55,9 +65,22 @@ from tidal_dl_ng.model.gui_data import ProgressBars
# TODO: Set appropriate client string and use it for video download.
# https://github.com/globocom/m3u8#using-different-http-clients
class RequestsClient:
+ """HTTP client for downloading text content from a URI."""
+
def download(
self, uri: str, timeout: int = REQUESTS_TIMEOUT_SEC, headers: dict | None = None, verify_ssl: bool = True
- ):
+ ) -> tuple[str, str]:
+ """Download the content of a URI as text.
+
+ Args:
+ uri (str): The URI to download.
+ timeout (int, optional): Timeout in seconds. Defaults to REQUESTS_TIMEOUT_SEC.
+ headers (dict | None, optional): HTTP headers. Defaults to None.
+ verify_ssl (bool, optional): Whether to verify SSL. Defaults to True.
+
+ Returns:
+ tuple[str, str]: Tuple of (text content, final URL).
+ """
if not headers:
headers = {}
@@ -68,6 +91,8 @@ class RequestsClient:
# TODO: Use pathlib.Path everywhere
class Download:
+ """Main class for managing downloads, segment merging, file operations, and metadata for TIDAL media."""
+
settings: Settings
session: Session
skip_existing: bool = False
@@ -84,12 +109,25 @@ class Download:
path_base: str,
fn_logger: Callable,
skip_existing: bool = False,
- progress_gui: ProgressBars = None,
- progress: Progress = None,
- progress_overall: Progress = None,
- event_abort: Event = None,
- event_run: Event = None,
- ):
+ progress_gui: ProgressBars | None = None,
+ progress: Progress | None = None,
+ progress_overall: Progress | None = None,
+ event_abort: Event | None = None,
+ event_run: Event | None = None,
+ ) -> None:
+ """Initialize the Download object and its dependencies.
+
+ Args:
+ session (Session): TIDAL session object.
+ path_base (str): Base path for downloads.
+ fn_logger (Callable): Logger function or object.
+ skip_existing (bool, optional): Whether to skip existing files. Defaults to False.
+ progress_gui (ProgressBars | None, optional): GUI progress bars. Defaults to None.
+ progress (Progress | None, optional): Rich progress bar. Defaults to None.
+ progress_overall (Progress | None, optional): Overall progress bar. Defaults to None.
+ event_abort (Event | None, optional): Abort event. Defaults to None.
+ event_run (Event | None, optional): Run event. Defaults to None.
+ """
self.settings = Settings()
self.session = session
self.skip_existing = skip_existing
@@ -113,43 +151,54 @@ class Download:
"be set in (`path_binary_ffmpeg`)."
)
- def _download(
+ def _get_media_urls(
self,
media: Track | Video,
- path_file: pathlib.Path,
stream_manifest: StreamManifest | None = None,
- ) -> (bool, pathlib.Path):
- media_name: str = name_builder_item(media)
- urls: [str]
- path_base: pathlib.Path = path_file.parent
- result_segments: bool = True
- dl_segment_results: [DownloadSegmentResult] = []
- result_merge: bool = False
+ ) -> list[str]:
+ """Extract URLs for the given media item.
+ Args:
+ media (Track | Video): The media item to download.
+ stream_manifest (StreamManifest | None, optional): Stream manifest for tracks. Defaults to None.
+
+ Returns:
+ list[str]: List of URLs for the media segments.
+ """
# Get urls for media.
- try:
- if isinstance(media, Track):
- urls = stream_manifest.get_urls()
- elif isinstance(media, Video):
- m3u8_variant: m3u8.M3U8 = m3u8.load(media.get_url())
- # Find the desired video resolution or the next best one.
- m3u8_playlist, codecs = self._extract_video_stream(m3u8_variant, int(self.settings.data.quality_video))
- # Populate urls.
- urls = m3u8_playlist.files
- except Exception:
- return False, path_file
+ if isinstance(media, Track):
+ return stream_manifest.get_urls()
+ elif isinstance(media, Video):
+ quality_video = self.settings.data.quality_video
+ m3u8_variant: m3u8.M3U8 = m3u8.load(media.get_url())
+ # Find the desired video resolution or the next best one.
+ m3u8_playlist, _ = self._extract_video_stream(m3u8_variant, int(quality_video))
- # Set the correct progress output channel.
- if self.progress_gui is None:
- progress_to_stdout: bool = True
+ return m3u8_playlist.files
else:
- progress_to_stdout: bool = False
- # Send signal to GUI with media name
- self.progress_gui.item_name.emit(media_name[:30])
+ return []
- # Compute total iterations for progress
+ def _setup_progress(
+ self,
+ media_name: str,
+ urls: list[str],
+ progress_to_stdout: bool,
+ ) -> tuple[TaskID, int | float | None, int | None]:
+ """Set up the progress bar/task and compute progress total and block size.
+
+ Args:
+ media_name (str): Name of the media item.
+ urls (list[str]): List of segment URLs.
+ progress_to_stdout (bool): Whether to show progress in stdout.
+
+ Returns:
+ tuple[TaskID, int | float | None, int | None]: (TaskID, progress_total, block_size)
+ """
urls_count: int = len(urls)
+ progress_total: int | float | None = None
+ block_size: int | None = None
+ # Compute total iterations for progress
if urls_count > 1:
progress_total: int = urls_count
block_size: int | None = None
@@ -157,9 +206,10 @@ class Download:
try:
# Get file size and compute progress steps
r = requests.head(urls[0], timeout=REQUESTS_TIMEOUT_SEC)
+
total_size_in_bytes: int = int(r.headers.get("content-length", 0))
- block_size: int | None = 1048576
- progress_total: float = total_size_in_bytes / block_size
+ block_size = 1048576
+ progress_total = total_size_in_bytes / block_size
finally:
r.close()
else:
@@ -171,6 +221,30 @@ class Download:
total=progress_total,
visible=progress_to_stdout,
)
+ return p_task, progress_total, block_size
+
+ def _download_segments(
+ self,
+ urls: list[str],
+ path_base: pathlib.Path,
+ block_size: int | None,
+ p_task: TaskID,
+ progress_to_stdout: bool,
+ ) -> tuple[bool, list[DownloadSegmentResult]]:
+ """Download all segments with progress reporting and abort handling.
+
+ Args:
+ urls (list[str]): List of segment URLs.
+ path_base (pathlib.Path): Base path for segment files.
+ block_size (int | None): Block size for streaming.
+ p_task (TaskID): Progress bar task ID.
+ progress_to_stdout (bool): Whether to show progress in stdout.
+
+ Returns:
+ tuple[bool, list[DownloadSegmentResult]]: (result_segments, list of segment results)
+ """
+ result_segments: bool = True
+ dl_segment_results: list[DownloadSegmentResult] = []
# Download segments until progress is finished.
# TODO: Compute download speed (https://github.com/Textualize/rich/blob/master/examples/downloader.py)
@@ -179,7 +253,7 @@ class Download:
max_workers=self.settings.data.downloads_simultaneous_per_track_max
) as executor:
# Dispatch all download tasks to worker threads
- l_futures: [futures.Future] = [
+ l_futures: list[futures.Future] = [
executor.submit(self._download_segment, url, path_base, block_size, p_task, progress_to_stdout)
for url in urls
]
@@ -191,14 +265,15 @@ class Download:
dl_segment_results.append(result_dl_segment)
- # check for a link that was skipped
+ # Check for a link that was skipped
if not result_dl_segment.result and (result_dl_segment.url is not urls[-1]):
# Sometimes it happens, if a track is very short (< 8 seconds or so), that the last URL in `urls` is
# invalid (HTTP Error 500) and not necessary. File won't be corrupt.
# If this is NOT the case, but any other URL has resulted in an error,
# mark the whole thing as corrupt.
result_segments = False
- self.fn_logger.error(f"Something went wrong while downloading {media_name}. File is corrupt!")
+
+ self.fn_logger.error("Something went wrong while downloading. File is corrupt!")
# If app is terminated (CTRL+C)
if self.event_abort.is_set():
@@ -206,26 +281,106 @@ class Download:
for f in l_futures:
f.cancel()
- return False, path_file
+ return False, dl_segment_results
+
+ return result_segments, dl_segment_results
+
+ def _download_postprocess(
+ self,
+ result_segments: bool,
+ path_file: pathlib.Path,
+ dl_segment_results: list[DownloadSegmentResult],
+ media: Track | Video,
+ stream_manifest: StreamManifest | None = None,
+ ) -> tuple[bool, pathlib.Path]:
+ """Merge segments, decrypt if needed, and return the final file path.
+
+ Args:
+ result_segments (bool): Whether all segments downloaded successfully.
+ path_file (pathlib.Path): Path to the output file.
+ dl_segment_results (list[DownloadSegmentResult]): List of segment download results.
+ media (Track | Video): The media item.
+ stream_manifest (StreamManifest | None, optional): Stream manifest for tracks. Defaults to None.
+ Returns:
+ tuple[bool, pathlib.Path]: (Success, path to downloaded or decrypted file)
+ """
tmp_path_file_decrypted: pathlib.Path = path_file
+ result_merge: bool = False
# Only if no error happened while downloading.
if result_segments:
# Bring list into right order, so segments can be easily merged.
dl_segment_results.sort(key=lambda x: x.id_segment)
- result_merge: bool = self._segments_merge(path_file, dl_segment_results)
+
+ result_merge = self._segments_merge(path_file, dl_segment_results)
if not result_merge:
- self.fn_logger.error(f"Something went wrong while writing to {media_name}. File is corrupt!")
- elif result_merge and isinstance(media, Track) and stream_manifest.is_encrypted:
+ self.fn_logger.error(f"Something went wrong while writing to {media.name}. File is corrupt!")
+ elif isinstance(media, Track) and stream_manifest.is_encrypted:
key, nonce = decrypt_security_token(stream_manifest.encryption_key)
tmp_path_file_decrypted = path_file.with_suffix(".decrypted")
+
decrypt_file(path_file, tmp_path_file_decrypted, key, nonce)
return result_merge, tmp_path_file_decrypted
- def _segments_merge(self, path_file, dl_segment_results) -> bool:
+ def _download(
+ self,
+ media: Track | Video,
+ path_file: pathlib.Path,
+ stream_manifest: StreamManifest | None = None,
+ ) -> tuple[bool, pathlib.Path]:
+ """Download a media item (track or video), handling segments and merging.
+
+ Args:
+ media (Track | Video): The media item to download.
+ path_file (pathlib.Path): Path to the output file.
+ stream_manifest (StreamManifest | None, optional): Stream manifest for tracks. Defaults to None.
+
+ Returns:
+ tuple[bool, pathlib.Path]: (Success, path to downloaded or decrypted file)
+ """
+ media_name: str = name_builder_item(media)
+
+ try:
+ urls: list[str] = self._get_media_urls(media, stream_manifest)
+ except Exception:
+ return False, path_file
+
+ # Set the correct progress output channel.
+ if self.progress_gui is None:
+ progress_to_stdout: bool = True
+ else:
+ progress_to_stdout: bool = False
+ # Send signal to GUI with media name
+ self.progress_gui.item_name.emit(media_name[:30])
+
+ try:
+ p_task, progress_total, block_size = self._setup_progress(media_name, urls, progress_to_stdout)
+ except Exception:
+ return False, path_file
+
+ result_segments, dl_segment_results = self._download_segments(
+ urls, path_file.parent, block_size, p_task, progress_to_stdout
+ )
+
+ result_merge, tmp_path_file_decrypted = self._download_postprocess(
+ result_segments, path_file, dl_segment_results, media, stream_manifest
+ )
+
+ return result_merge, tmp_path_file_decrypted
+
+ def _segments_merge(self, path_file: pathlib.Path, dl_segment_results: list[DownloadSegmentResult]) -> bool:
+ """Merge downloaded segments into a single file and clean up segment files.
+
+ Args:
+ path_file (pathlib.Path): Path to the output file.
+ dl_segment_results (list[DownloadSegmentResult]): List of segment download results.
+
+ Returns:
+ bool: True if merge succeeded, False otherwise.
+ """
result: bool = True
# Copy the content of all segments into one file.
@@ -249,6 +404,18 @@ class Download:
def _download_segment(
self, url: str, path_base: pathlib.Path, block_size: int | None, p_task: TaskID, progress_to_stdout: bool
) -> DownloadSegmentResult:
+ """Download a single segment of a media file.
+
+ Args:
+ url (str): URL of the segment.
+ path_base (pathlib.Path): Base path for segment file.
+ block_size (int | None): Block size for streaming.
+ p_task (TaskID): Progress bar task ID.
+ progress_to_stdout (bool): Whether to show progress in stdout.
+
+ Returns:
+ DownloadSegmentResult: Result of the segment download.
+ """
result: bool = False
path_segment: pathlib.Path = path_base / url_to_filename(url)
# Calculate the segment ID based on the file name within the URL.
@@ -298,8 +465,18 @@ class Download:
)
def extension_guess(
- self, quality_audio: Quality, metadata_tags: [str], is_video: bool
+ self, quality_audio: Quality, metadata_tags: list[str], is_video: bool
) -> AudioExtensions | VideoExtensions:
+ """Guess the file extension for a media item based on quality and type.
+
+ Args:
+ quality_audio (Quality): Audio quality.
+ metadata_tags (list[str]): Metadata tags for the media.
+ is_video (bool): Whether the media is a video.
+
+ Returns:
+ AudioExtensions | VideoExtensions: Guessed file extension.
+ """
result: AudioExtensions | VideoExtensions
if is_video:
@@ -321,9 +498,9 @@ class Download:
def item(
self,
file_template: str,
- media: Track | Video = None,
- media_id: str = None,
- media_type: MediaType = None,
+ media_id: str | None = None,
+ media_type: MediaType | None = None,
+ media: Track | Video | None = None,
video_download: bool = True,
download_delay: bool = False,
quality_audio: Quality | None = None,
@@ -331,43 +508,152 @@ class Download:
is_parent_album: bool = False,
list_position: int = 0,
list_total: int = 0,
- ) -> (bool, pathlib.Path):
+ ) -> tuple[bool, pathlib.Path | str]:
+ """Download a single media item, handling file naming, skipping, and post-processing.
+
+ Args:
+ file_template (str): Template for file naming.
+ media_id (str | None, optional): Media ID. Defaults to None.
+ media_type (MediaType | None, optional): Media type. Defaults to None.
+ media (Track | Video | None, optional): Media item. Defaults to None.
+ video_download (bool, optional): Whether to allow video downloads. Defaults to True.
+ download_delay (bool, optional): Whether to delay between downloads. Defaults to False.
+ quality_audio (Quality | None, optional): Audio quality. Defaults to None.
+ quality_video (QualityVideo | None, optional): Video quality. Defaults to None.
+ is_parent_album (bool, optional): Whether this is a parent album. Defaults to False.
+ list_position (int, optional): Position in list. Defaults to 0.
+ list_total (int, optional): Total items in list. Defaults to 0.
+
+ Returns:
+ tuple[bool, pathlib.Path | str]: (Downloaded, path to file)
+ """
+ # Step 1: Validate and prepare media
+ validated_media = self._validate_and_prepare_media(media, media_id, media_type, video_download)
+ if validated_media is None or not isinstance(validated_media, Track | Video):
+ return False, ""
+
+ media = validated_media
+
+ # Step 2: Create file paths and determine skip logic
+ path_media_dst, file_extension_dummy, skip_file, skip_download = self._prepare_file_paths_and_skip_logic(
+ media, file_template, quality_audio, list_position, list_total
+ )
+
+ if skip_file:
+ self.fn_logger.debug(f"Download skipped, since file exists: '{path_media_dst}'")
+
+ return True, path_media_dst
+
+ # Step 3: Handle quality settings
+ quality_audio_old, quality_video_old = self._adjust_quality_settings(quality_audio, quality_video)
+
+ # Step 4: Download and process media
+ download_success = self._download_and_process_media(
+ media, path_media_dst, skip_download, is_parent_album, file_extension_dummy
+ )
+
+ # Step 5: Post-processing
+ self._perform_post_processing(
+ media,
+ path_media_dst,
+ quality_audio,
+ quality_video,
+ quality_audio_old,
+ quality_video_old,
+ download_delay,
+ skip_file,
+ )
+
+ return download_success, path_media_dst
+
+ def _validate_and_prepare_media(
+ self,
+ media: Track | Video | Album | Playlist | UserPlaylist | Mix | None,
+ media_id: str | None,
+ media_type: MediaType | None,
+ video_download: bool = True,
+ ) -> Track | Video | Album | Playlist | UserPlaylist | Mix | None:
+ """Validate and prepare media instance for download.
+
+ Args:
+ media (Track | Video | Album | Playlist | UserPlaylist | Mix | None): Media instance.
+ media_id (str | None): Media ID if creating new instance.
+ media_type (MediaType | None): Media type if creating new instance.
+ video_download (bool, optional): Whether video downloads are allowed. Defaults to True.
+
+ Returns:
+ Track | Video | Album | Playlist | UserPlaylist | Mix | None: Prepared media instance or None if invalid.
+ """
try:
if media_id and media_type:
# If no media instance is provided, we need to create the media instance.
+ # Throws `tidalapi.exceptions.ObjectNotFound` if item is not available anymore.
media = instantiate_media(self.session, media_type, media_id)
- elif isinstance(media, Track | Video): # Check if media is available not deactivated / removed from TIDAL.
+ elif isinstance(media, Track | Video):
+ # Check if media is available not deactivated / removed from TIDAL.
if not media.available:
self.fn_logger.info(
f"This item is not available for listening anymore on TIDAL. Skipping: {name_builder_item(media)}"
)
-
- return False, ""
+ return None
elif isinstance(media, Track):
# Re-create media instance with full album information
- media = self.session.track(media.id, with_album=True)
+ media = self.session.track(str(media.id), with_album=True)
+ elif isinstance(media, Album):
+ # Check if media is available not deactivated / removed from TIDAL.
+ if not media.available:
+ self.fn_logger.info(
+ f"This item is not available for listening anymore on TIDAL. Skipping: {name_builder_title(media)}"
+ )
+ return None
elif not media:
raise MediaMissing
except:
- return False, ""
+ return None
- # If video download is not allowed end here
+ # If video download is not allowed and this is a video, return None
if not video_download and isinstance(media, Video):
self.fn_logger.info(
f"Video downloads are deactivated (see settings). Skipping video: {name_builder_item(media)}"
)
+ return None
- return False, ""
+ return media
+
+ def _prepare_file_paths_and_skip_logic(
+ self,
+ media: Track | Video,
+ file_template: str,
+ quality_audio: Quality | None,
+ list_position: int,
+ list_total: int,
+ ) -> tuple[pathlib.Path, str, bool, bool]:
+ """Prepare file paths and determine skip logic.
+
+ Args:
+ media (Track | Video): Media item.
+ file_template (str): Template for file naming.
+ quality_audio (Quality | None): Audio quality setting.
+ list_position (int): Position in list.
+ list_total (int): Total items in list.
+ Returns:
+ tuple[pathlib.Path, str, bool, bool]: (path_media_dst, file_extension_dummy, skip_file, skip_download)
+ """
# Create file name and path
+ metadata_tags = [] if isinstance(media, Video) else (media.media_metadata_tags or [])
+ quality_for_extension = quality_audio if quality_audio is not None else Quality.high_lossless
+
file_extension_dummy: str = self.extension_guess(
- quality_audio,
- metadata_tags=[] if isinstance(media, Video) else media.media_metadata_tags,
+ quality_for_extension,
+ metadata_tags=metadata_tags,
is_video=isinstance(media, Video),
)
+
file_name_relative: str = format_path_media(
file_template, media, self.settings.data.album_track_num_pad_min, list_position, list_total
)
+
path_media_dst: pathlib.Path = (
pathlib.Path(self.path_base).expanduser() / (file_name_relative + file_extension_dummy)
).absolute()
@@ -394,122 +680,246 @@ class Download:
)
skip_download = file_exists_playlist_dir or file_exists_track_dir
- # If
+ # If file exists in playlist dir but not in track dir, we don't skip the file itself
if skip_file and file_exists_playlist_dir:
skip_file = False
else:
skip_file: bool = False
- if not skip_file:
- # If a quality is explicitly set, change it and remember the previously set quality.
- quality_audio_old: Quality = self.adjust_quality_audio(quality_audio) if quality_audio else quality_audio
- quality_video_old: QualityVideo = (
- self.adjust_quality_video(quality_video) if quality_video else quality_video
- )
- do_flac_extract = False
- # Get extension.
- file_extension: str
- stream_manifest: StreamManifest | None = None
+ return path_media_dst, file_extension_dummy, skip_file, skip_download
- if isinstance(media, Track):
- try:
- media_stream: Stream = media.get_stream()
- stream_manifest = media_stream.get_stream_manifest()
- except TooManyRequests:
- self.fn_logger.exception(
- f"Too many requests against TIDAL backend. Skipping '{name_builder_item(media)}'. "
- f"Consider to activate delay between downloads."
- )
+ def _adjust_quality_settings(
+ self, quality_audio: Quality | None, quality_video: QualityVideo | None
+ ) -> tuple[Quality | None, QualityVideo | None]:
+ """Adjust quality settings and return previous values.
- return False, ""
- except Exception:
- self.fn_logger.exception(f"Something went wrong. Skipping '{name_builder_item(media)}'.")
+ Args:
+ quality_audio (Quality | None): Audio quality setting.
+ quality_video (QualityVideo | None): Video quality setting.
- return False, ""
+ Returns:
+ tuple[Quality | None, QualityVideo | None]: Previous quality settings.
+ """
+ quality_audio_old: Quality | None = None
+ quality_video_old: QualityVideo | None = None
- file_extension = stream_manifest.file_extension
+ if quality_audio:
+ quality_audio_old = self.adjust_quality_audio(quality_audio)
- if self.settings.data.extract_flac and (
- stream_manifest.codecs.upper() == Codec.FLAC and file_extension != AudioExtensions.FLAC
- ):
- file_extension = AudioExtensions.FLAC
- do_flac_extract = True
- elif isinstance(media, Video):
- file_extension = AudioExtensions.MP4 if self.settings.data.video_convert_mp4 else VideoExtensions.TS
+ if quality_video:
+ quality_video_old = self.adjust_quality_video(quality_video)
- # If file extension was guessed wrong in the beginning
- if path_media_dst.suffix != file_extension:
- # Compute file name, sanitize once again, because file extension could have been replaced after guessing it first and create destination directory
- path_media_dst = path_media_dst.with_suffix(file_extension)
- path_media_dst = pathlib.Path(path_file_sanitize(path_media_dst, adapt=True))
+ return quality_audio_old, quality_video_old
- os.makedirs(path_media_dst.parent, exist_ok=True)
+ def _download_and_process_media(
+ self,
+ media: Track | Video,
+ path_media_dst: pathlib.Path,
+ skip_download: bool,
+ is_parent_album: bool,
+ file_extension_dummy: str,
+ ) -> bool:
+ """Download and process media file.
- if not skip_download:
- # Create a temp directory and file.
- with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmp_path_dir:
- tmp_path_file: pathlib.Path = pathlib.Path(tmp_path_dir) / str(uuid4())
+ Args:
+ media (Track | Video): Media item.
+ path_media_dst (pathlib.Path): Destination file path.
+ skip_download (bool): Whether to skip download.
+ is_parent_album (bool): Whether this is a parent album.
+ file_extension_dummy (str): Dummy file extension.
- # Create empty file
- tmp_path_file.touch()
+ Returns:
+ bool: Whether download was successful.
+ """
+ if skip_download:
+ return True
- # Download media.
- result_download, tmp_path_file = self._download(
- media=media, stream_manifest=stream_manifest, path_file=tmp_path_file
- )
+ # Get stream information and final file extension
+ stream_manifest, file_extension, do_flac_extract, media_stream = self._get_stream_info(media)
- if result_download:
- # Convert video from TS to MP4
- if isinstance(media, Video) and self.settings.data.video_convert_mp4:
- # Convert `*.ts` file to `*.mp4` using ffmpeg
- tmp_path_file = self._video_convert(tmp_path_file)
+ if stream_manifest is None and isinstance(media, Track):
+ return False
- # Extract FLAC from MP4 container using ffmpeg
- if isinstance(media, Track) and self.settings.data.extract_flac and do_flac_extract:
- tmp_path_file = self._extract_flac(tmp_path_file)
+ # Update path if extension changed
+ if path_media_dst.suffix != file_extension:
+ path_media_dst = path_media_dst.with_suffix(file_extension)
+ path_media_dst = pathlib.Path(path_file_sanitize(path_media_dst, adapt=True))
- tmp_path_lyrics: pathlib.Path | None = None
- tmp_path_cover: pathlib.Path | None = None
+ os.makedirs(path_media_dst.parent, exist_ok=True)
- # Write metadata to file.
- if not isinstance(media, Video):
- result_metadata, tmp_path_lyrics, tmp_path_cover = self.metadata_write(
- media, tmp_path_file, is_parent_album, media_stream
- )
+ # Perform actual download
+ return self._perform_actual_download(
+ media, path_media_dst, stream_manifest, do_flac_extract, is_parent_album, media_stream
+ )
- # Move lyrics file
- if self.settings.data.lyrics_file and not isinstance(media, Video) and tmp_path_lyrics:
- self._move_lyrics(tmp_path_lyrics, path_media_dst)
+ def _get_stream_info(self, media: Track | Video) -> tuple[StreamManifest | None, str, bool, Stream | None]:
+ """Get stream information for media.
- # Move cover file
- # TODO: Cover is downloaded with every track of the album. Needs refactoring, so cover is only
- # downloaded for an album once.
- if self.settings.data.cover_album_file and tmp_path_cover:
- self._move_cover(tmp_path_cover, path_media_dst)
+ Args:
+ media (Track | Video): Media item.
- self.fn_logger.info(f"Downloaded item '{name_builder_item(media)}'.")
+ Returns:
+ tuple[StreamManifest | None, str, bool, Stream | None]: Stream info.
+ """
+ stream_manifest: StreamManifest | None = None
+ media_stream: Stream | None = None
+ do_flac_extract: bool = False
- # Move final file to the configured destination directory.
- shutil.move(tmp_path_file, path_media_dst)
+ if isinstance(media, Track):
+ try:
+ media_stream = media.get_stream()
+ stream_manifest = media_stream.get_stream_manifest()
+ except TooManyRequests:
+ self.fn_logger.exception(
+ f"Too many requests against TIDAL backend. Skipping '{name_builder_item(media)}'. "
+ f"Consider to activate delay between downloads."
+ )
- # If files needs to be symlinked, do postprocessing here.
- if self.settings.data.symlink_to_track and not isinstance(media, Video):
- path_media_track_dir: pathlib.Path = self.media_move_and_symlink(media, path_media_dst, file_extension)
+ return None, "", False, None
+ except Exception:
+ self.fn_logger.exception(f"Something went wrong. Skipping '{name_builder_item(media)}'.")
- if quality_audio:
- # Set quality back to the global user value
- self.adjust_quality_audio(quality_audio_old)
+ return None, "", False, None
- if quality_video:
- # Set quality back to the global user value
- self.adjust_quality_video(quality_video_old)
- else:
- self.fn_logger.debug(f"Download skipped, since file exists: '{path_media_dst}'")
+ file_extension = stream_manifest.file_extension
+
+ if self.settings.data.extract_flac and (
+ stream_manifest.codecs.upper() == Codec.FLAC and file_extension != AudioExtensions.FLAC
+ ):
+ file_extension = AudioExtensions.FLAC
+ do_flac_extract = True
+ elif isinstance(media, Video):
+ file_extension = AudioExtensions.MP4 if self.settings.data.video_convert_mp4 else VideoExtensions.TS
+
+ return stream_manifest, file_extension, do_flac_extract, media_stream
+
+ def _perform_actual_download(
+ self,
+ media: Track | Video,
+ path_media_dst: pathlib.Path,
+ stream_manifest: StreamManifest | None,
+ do_flac_extract: bool,
+ is_parent_album: bool,
+ media_stream: Stream | None,
+ ) -> bool:
+ """Perform the actual download and processing.
+
+ Args:
+ media (Track | Video): Media item.
+ path_media_dst (pathlib.Path): Destination file path.
+ stream_manifest (StreamManifest | None): Stream manifest.
+ do_flac_extract (bool): Whether to extract FLAC.
+ is_parent_album (bool): Whether this is a parent album.
+ media_stream (Stream | None): Media stream.
+
+ Returns:
+ bool: Whether download was successful.
+ """
+ # Create a temp directory and file.
+ with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmp_path_dir:
+ tmp_path_file: pathlib.Path = pathlib.Path(tmp_path_dir) / str(uuid4())
+ tmp_path_file.touch()
+
+ # Download media.
+ result_download, tmp_path_file = self._download(
+ media=media, stream_manifest=stream_manifest, path_file=tmp_path_file
+ )
+
+ if not result_download:
+ return False
+
+ # Convert video from TS to MP4
+ if isinstance(media, Video) and self.settings.data.video_convert_mp4:
+ tmp_path_file = self._video_convert(tmp_path_file)
+
+ # Extract FLAC from MP4 container using ffmpeg
+ if isinstance(media, Track) and self.settings.data.extract_flac and do_flac_extract:
+ tmp_path_file = self._extract_flac(tmp_path_file)
+
+ # Handle metadata, lyrics, and cover
+ self._handle_metadata_and_extras(media, tmp_path_file, path_media_dst, is_parent_album, media_stream)
+
+ self.fn_logger.info(f"Downloaded item '{name_builder_item(media)}'.")
+
+ # Move final file to the configured destination directory.
+ shutil.move(tmp_path_file, path_media_dst)
+
+ return True
+
+ def _handle_metadata_and_extras(
+ self,
+ media: Track | Video,
+ tmp_path_file: pathlib.Path,
+ path_media_dst: pathlib.Path,
+ is_parent_album: bool,
+ media_stream: Stream | None,
+ ) -> None:
+ """Handle metadata, lyrics, and cover processing.
+
+ Args:
+ media (Track | Video): Media item.
+ tmp_path_file (pathlib.Path): Temporary file path.
+ path_media_dst (pathlib.Path): Destination file path.
+ is_parent_album (bool): Whether this is a parent album.
+ media_stream (Stream | None): Media stream.
+ """
+ if isinstance(media, Video):
+ return
+
+ tmp_path_lyrics: pathlib.Path | None = None
+ tmp_path_cover: pathlib.Path | None = None
+
+ # Write metadata to file.
+ if media_stream:
+ result_metadata, tmp_path_lyrics, tmp_path_cover = self.metadata_write(
+ media, tmp_path_file, is_parent_album, media_stream
+ )
+
+ # Move lyrics file
+ if self.settings.data.lyrics_file and tmp_path_lyrics:
+ self._move_lyrics(tmp_path_lyrics, path_media_dst)
+
+ # Move cover file
+ if self.settings.data.cover_album_file and tmp_path_cover:
+ self._move_cover(tmp_path_cover, path_media_dst)
+
+ def _perform_post_processing(
+ self,
+ media: Track | Video,
+ path_media_dst: pathlib.Path,
+ quality_audio: Quality | None,
+ quality_video: QualityVideo | None,
+ quality_audio_old: Quality | None,
+ quality_video_old: QualityVideo | None,
+ download_delay: bool,
+ skip_file: bool,
+ ) -> None:
+ """Perform post-processing tasks.
+
+ Args:
+ media (Track | Video): Media item.
+ path_media_dst (pathlib.Path): Destination file path.
+ quality_audio (Quality | None): Audio quality setting.
+ quality_video (QualityVideo | None): Video quality setting.
+ quality_audio_old (Quality | None): Previous audio quality.
+ quality_video_old (QualityVideo | None): Previous video quality.
+ download_delay (bool): Whether to apply download delay.
+ skip_file (bool): Whether file was skipped.
+ """
+ # If files needs to be symlinked, do postprocessing here.
+ if self.settings.data.symlink_to_track and not isinstance(media, Video):
+ # Determine file extension for symlink
+ file_extension = path_media_dst.suffix
+ self.media_move_and_symlink(media, path_media_dst, file_extension)
- status_download: bool = not skip_file
+ # Reset quality settings
+ if quality_audio_old is not None:
+ self.adjust_quality_audio(quality_audio_old)
- # Whether a file was downloaded or skipped and the download delay is enabled, wait until the next download.
- # Only use this, if you have a list of several Track items.
+ if quality_video_old is not None:
+ self.adjust_quality_video(quality_video_old)
+
+ # Apply download delay if needed
if (download_delay and not skip_file) and not self.event_abort.is_set():
time_sleep: float = round(
random.SystemRandom().uniform(
@@ -521,11 +931,19 @@ class Download:
self.fn_logger.debug(f"Next download will start in {time_sleep} seconds.")
time.sleep(time_sleep)
- return status_download, path_media_dst
-
def media_move_and_symlink(
self, media: Track | Video, path_media_src: pathlib.Path, file_extension: str
) -> pathlib.Path:
+ """Move a media file and create a symlink if required.
+
+ Args:
+ media (Track | Video): Media item.
+ path_media_src (pathlib.Path): Source file path.
+ file_extension (str): File extension.
+
+ Returns:
+ pathlib.Path: Destination path.
+ """
# Compute tracks path, sanitize and ensure path exists
file_name_relative: str = format_path_media(self.settings.data.format_track, media)
path_media_dst: pathlib.Path = (
@@ -557,14 +975,30 @@ class Download:
return path_media_dst
- def adjust_quality_audio(self, quality) -> Quality:
+ def adjust_quality_audio(self, quality: Quality) -> Quality:
+ """Temporarily set audio quality and return the previous value.
+
+ Args:
+ quality (Quality): New audio quality.
+
+ Returns:
+ Quality: Previous audio quality.
+ """
# Save original quality settings
quality_old: Quality = self.session.audio_quality
self.session.audio_quality = quality
return quality_old
- def adjust_quality_video(self, quality) -> QualityVideo:
+ def adjust_quality_video(self, quality: QualityVideo) -> QualityVideo:
+ """Temporarily set video quality and return the previous value.
+
+ Args:
+ quality (QualityVideo): New video quality.
+
+ Returns:
+ QualityVideo: Previous video quality.
+ """
quality_old: QualityVideo = self.settings.data.quality_video
self.settings.data.quality_video = quality
@@ -572,6 +1006,15 @@ class Download:
return quality_old
def _move_file(self, path_file_source: pathlib.Path, path_file_destination: str | pathlib.Path) -> bool:
+ """Move a file from source to destination.
+
+ Args:
+ path_file_source (pathlib.Path): Source file path.
+ path_file_destination (str | pathlib.Path): Destination file path.
+
+ Returns:
+ bool: True if moved, False otherwise.
+ """
result: bool
# Check if the file was downloaded
@@ -586,6 +1029,15 @@ class Download:
return result
def _move_lyrics(self, path_lyrics: pathlib.Path, file_media_dst: pathlib.Path) -> bool:
+ """Move a lyrics file to the destination.
+
+ Args:
+ path_lyrics (pathlib.Path): Source lyrics file.
+ file_media_dst (pathlib.Path): Destination media file path.
+
+ Returns:
+ bool: True if moved, False otherwise.
+ """
# Build tmp lyrics filename
path_file_lyrics: pathlib.Path = file_media_dst.with_suffix(EXTENSION_LYRICS)
result: bool = self._move_file(path_lyrics, path_file_lyrics)
@@ -593,6 +1045,15 @@ class Download:
return result
def _move_cover(self, path_cover: pathlib.Path, file_media_dst: pathlib.Path) -> bool:
+ """Move a cover file to the destination.
+
+ Args:
+ path_cover (pathlib.Path): Source cover file.
+ file_media_dst (pathlib.Path): Destination media file path.
+
+ Returns:
+ bool: True if moved, False otherwise.
+ """
# Build tmp lyrics filename
path_file_cover: pathlib.Path = file_media_dst.parent / COVER_NAME
result: bool = self._move_file(path_cover, path_file_cover)
@@ -600,12 +1061,40 @@ class Download:
return result
def lyrics_to_file(self, dir_destination: pathlib.Path, lyrics: str) -> str:
+ """Write lyrics to a temporary file.
+
+ Args:
+ dir_destination (pathlib.Path): Directory for the temp file.
+ lyrics (str): Lyrics content.
+
+ Returns:
+ str: Path to the temp file.
+ """
return self.write_to_tmp_file(dir_destination, mode="x", content=lyrics)
def cover_to_file(self, dir_destination: pathlib.Path, image: bytes) -> str:
+ """Write cover image to a temporary file.
+
+ Args:
+ dir_destination (pathlib.Path): Directory for the temp file.
+ image (bytes): Image data.
+
+ Returns:
+ str: Path to the temp file.
+ """
return self.write_to_tmp_file(dir_destination, mode="xb", content=image)
def write_to_tmp_file(self, dir_destination: pathlib.Path, mode: str, content: str | bytes) -> str:
+ """Write content to a temporary file.
+
+ Args:
+ dir_destination (pathlib.Path): Directory for the temp file.
+ mode (str): File open mode.
+ content (str | bytes): Content to write.
+
+ Returns:
+ str: Path to the temp file.
+ """
result: pathlib.Path = dir_destination / str(uuid4())
encoding: str | None = "utf-8" if isinstance(content, str) else None
@@ -618,7 +1107,16 @@ class Download:
return result
@staticmethod
- def cover_data(url: str = None, path_file: str = None) -> str | bytes:
+ def cover_data(url: str | None = None, path_file: str | None = None) -> str | bytes:
+ """Retrieve cover image data from a URL or file.
+
+ Args:
+ url (str | None, optional): URL to download image from. Defaults to None.
+ path_file (str | None, optional): Path to image file. Defaults to None.
+
+ Returns:
+ str | bytes: Image data or empty string on failure.
+ """
result: str | bytes = ""
if url:
@@ -642,7 +1140,18 @@ class Download:
def metadata_write(
self, track: Track, path_media: pathlib.Path, is_parent_album: bool, media_stream: Stream
- ) -> (bool, pathlib.Path | None, pathlib.Path | None):
+ ) -> tuple[bool, pathlib.Path | None, pathlib.Path | None]:
+ """Write metadata, lyrics, and cover to a media file.
+
+ Args:
+ track (Track): Track object.
+ path_media (pathlib.Path): Path to media file.
+ is_parent_album (bool): Whether this is a parent album.
+ media_stream (Stream): Stream object.
+
+ Returns:
+ tuple[bool, pathlib.Path | None, pathlib.Path | None]: (Success, path to lyrics, path to cover)
+ """
result: bool = False
path_lyrics: pathlib.Path | None = None
path_cover: pathlib.Path | None = None
@@ -713,31 +1222,84 @@ class Download:
def items(
self,
file_template: str,
- media: Album | Playlist | UserPlaylist | Mix = None,
- media_id: str = None,
- media_type: MediaType = None,
+ media: Album | Playlist | UserPlaylist | Mix | None = None,
+ media_id: str | None = None,
+ media_type: MediaType | None = None,
video_download: bool = False,
download_delay: bool = True,
quality_audio: Quality | None = None,
quality_video: QualityVideo | None = None,
- ):
- try:
- if media_id and media_type:
- # If no media instance is provided, we need to create the media instance.
- # Throws `tidalapi.exceptions.ObjectNotFound` if item is not available anymore.
- media = instantiate_media(self.session, media_type, media_id)
- elif isinstance(media, Album): # Check if media is available not deactivated / removed from TIDAL.
- if not media.available:
- self.fn_logger.info(
- f"This item is not available for listening anymore on TIDAL. Skipping: {name_builder_title(media)}"
- )
+ ) -> None:
+ """Download all items in an album, playlist, or mix.
- return
- elif not media:
- raise MediaMissing
- except:
+ Args:
+ file_template (str): Template for file naming.
+ media (Album | Playlist | UserPlaylist | Mix | None, optional): Media item. Defaults to None.
+ media_id (str | None, optional): Media ID. Defaults to None.
+ media_type (MediaType | None, optional): Media type. Defaults to None.
+ video_download (bool, optional): Whether to allow video downloads. Defaults to False.
+ download_delay (bool, optional): Whether to delay between downloads. Defaults to True.
+ quality_audio (Quality | None, optional): Audio quality. Defaults to None.
+ quality_video (QualityVideo | None, optional): Video quality. Defaults to None.
+ """
+ # Validate and prepare media collection
+ validated_media = self._validate_and_prepare_media(media, media_id, media_type, video_download)
+ if validated_media is None or not isinstance(validated_media, Album | Playlist | UserPlaylist | Mix):
return
+ media = validated_media
+
+ # Set up download context
+ download_context = self._setup_collection_download_context(media, file_template, video_download)
+ file_name_relative, list_media_name, list_media_name_short, items, progress_stdout = download_context
+
+ # Set up progress tracking
+ progress: Progress = self.progress_overall if self.progress_overall else self.progress
+ progress_task: TaskID = progress.add_task(
+ f"[green]List '{list_media_name_short}'", total=len(items), visible=progress_stdout
+ )
+
+ # Download configuration
+ is_album: bool = isinstance(media, Album)
+ sort_by_track_num: bool = bool("album_track_num" in file_name_relative or "list_pos" in file_name_relative)
+ list_total: int = len(items)
+
+ # Execute downloads
+ result_dirs: list[pathlib.Path] = self._execute_collection_downloads(
+ items,
+ file_name_relative,
+ quality_audio,
+ quality_video,
+ download_delay,
+ is_album,
+ list_total,
+ progress,
+ progress_task,
+ progress_stdout,
+ )
+
+ # Create playlist file if requested
+ if self.settings.data.playlist_create:
+ self.playlist_populate(set(result_dirs), list_media_name, is_album, sort_by_track_num)
+
+ self.fn_logger.info(f"Finished list '{list_media_name}'.")
+
+ def _setup_collection_download_context(
+ self,
+ media: Album | Playlist | UserPlaylist | Mix,
+ file_template: str,
+ video_download: bool,
+ ) -> tuple[str, str, str, list, bool]:
+ """Set up download context for media collection.
+
+ Args:
+ media (Album | Playlist | UserPlaylist | Mix): Media collection.
+ file_template (str): Template for file naming.
+ video_download (bool): Whether to allow video downloads.
+
+ Returns:
+ tuple[str, str, str, list, bool]: (file_name_relative, list_media_name, list_media_name_short, items, progress_stdout)
+ """
# Create file name and path
file_name_relative: str = format_path_media(file_template, media)
@@ -753,26 +1315,48 @@ class Download:
progress_stdout: bool = True
else:
progress_stdout: bool = False
+
self.progress_gui.list_name.emit(list_media_name_short)
- progress: Progress = self.progress_overall if self.progress_overall else self.progress
+ return file_name_relative, list_media_name, list_media_name_short, items, progress_stdout
- # Create the list progress task.
- p_task1: TaskID = progress.add_task(
- f"[green]List '{list_media_name_short}'", total=len(items), visible=progress_stdout
- )
+ def _execute_collection_downloads(
+ self,
+ items: list,
+ file_name_relative: str,
+ quality_audio: Quality | None,
+ quality_video: QualityVideo | None,
+ download_delay: bool,
+ is_album: bool,
+ list_total: int,
+ progress: Progress,
+ progress_task: TaskID,
+ progress_stdout: bool,
+ ) -> list[pathlib.Path]:
+ """Execute downloads for all items in the collection.
- is_album: bool = isinstance(media, Album)
- # TODO: Refactor strings to constants (also in cfg.py)
- sort_by_track_num: bool = bool("album_track_num" in file_name_relative or "list_pos" in file_name_relative)
- result_dirs: [pathlib.Path] = []
- list_total: int = len(items)
+ Args:
+ items (list): List of media items to download.
+ file_name_relative (str): Relative file name template.
+ quality_audio (Quality | None): Audio quality setting.
+ quality_video (QualityVideo | None): Video quality setting.
+ download_delay (bool): Whether to apply download delay.
+ is_album (bool): Whether this is an album.
+ list_total (int): Total number of items.
+ progress (Progress): Progress bar instance.
+ progress_task (TaskID): Progress task ID.
+ progress_stdout (bool): Whether to show progress in stdout.
+
+ Returns:
+ list[pathlib.Path]: List of result directories.
+ """
+ result_dirs: list[pathlib.Path] = []
# Iterate through list items
while not progress.finished:
with futures.ThreadPoolExecutor(max_workers=self.settings.data.downloads_concurrent_max) as executor:
# Dispatch all download tasks to worker threads
- l_futures: [futures.Future] = [
+ download_futures: list[futures.Future] = [
executor.submit(
self.item,
media=item_media,
@@ -787,39 +1371,114 @@ class Download:
for count, item_media in enumerate(items)
]
- # Report results as they become available
- for future in futures.as_completed(l_futures):
- # Retrieve result
- status, result_path_file = future.result()
+ # Process download results
+ result_dirs = self._process_download_futures(download_futures, progress, progress_task, progress_stdout)
- if result_path_file:
- result_dirs.append(result_path_file.parent)
+ # Check for abort signal
+ if self.event_abort.is_set():
+ return result_dirs
- # Advance progress bar.
- progress.advance(p_task1)
+ return result_dirs
- if not progress_stdout:
- self.progress_gui.list_item.emit(progress.tasks[p_task1].percentage)
+ def _create_download_futures(
+ self,
+ items: list,
+ file_name_relative: str,
+ quality_audio: Quality | None,
+ quality_video: QualityVideo | None,
+ download_delay: bool,
+ is_album: bool,
+ list_total: int,
+ ) -> list[futures.Future]:
+ """Create download futures for all items in the collection.
- # If app is terminated (CTRL+C)
- if self.event_abort.is_set():
- # Cancel all not yet started tasks
- for f in l_futures:
- f.cancel()
+ Args:
+ items (list): List of media items to download.
+ file_name_relative (str): Relative file name template.
+ quality_audio (Quality | None): Audio quality setting.
+ quality_video (QualityVideo | None): Video quality setting.
+ download_delay (bool): Whether to apply download delay.
+ is_album (bool): Whether this is an album.
+ list_total (int): Total number of items.
- # End method here.
- return
+ Returns:
+ list[futures.Future]: List of download futures.
+ """
+ with futures.ThreadPoolExecutor(max_workers=self.settings.data.downloads_concurrent_max) as executor:
+ return [
+ executor.submit(
+ self.item,
+ media=item_media,
+ file_template=file_name_relative,
+ quality_audio=quality_audio,
+ quality_video=quality_video,
+ download_delay=download_delay,
+ is_parent_album=is_album,
+ list_position=count + 1,
+ list_total=list_total,
+ )
+ for count, item_media in enumerate(items)
+ ]
- # Create playlist file
- if self.settings.data.playlist_create:
- self.playlist_populate(set(result_dirs), list_media_name, is_album, sort_by_track_num)
+ def _process_download_futures(
+ self,
+ futures_list: list[futures.Future],
+ progress: Progress,
+ progress_task: TaskID,
+ progress_stdout: bool,
+ ) -> list[pathlib.Path]:
+ """Process download futures and collect results.
- self.fn_logger.info(f"Finished list '{list_media_name}'.")
+ Args:
+ futures_list (list[futures.Future]): List of download futures.
+ progress (Progress): Progress bar instance.
+ progress_task (TaskID): Progress task ID.
+ progress_stdout (bool): Whether to show progress in stdout.
+
+ Returns:
+ list[pathlib.Path]: List of result directories.
+ """
+ result_dirs: list[pathlib.Path] = []
+
+ # Report results as they become available
+ for future in futures.as_completed(futures_list):
+ # Retrieve result
+ status, result_path_file = future.result()
+
+ if result_path_file:
+ result_dirs.append(result_path_file.parent)
+
+ # Advance progress bar.
+ progress.advance(progress_task)
+
+ if not progress_stdout:
+ self.progress_gui.list_item.emit(progress.tasks[progress_task].percentage)
+
+ # If app is terminated (CTRL+C)
+ if self.event_abort.is_set():
+ # Cancel all not yet started tasks
+ for f in futures_list:
+ f.cancel()
+
+ break
+
+ return result_dirs
def playlist_populate(
- self, dirs_scoped: [pathlib.Path], name_list: str, is_album: bool, sort_alphabetically
- ) -> [pathlib.Path]:
- result: [pathlib.Path] = []
+ self, dirs_scoped: set[pathlib.Path], name_list: str, is_album: bool, sort_alphabetically: bool
+ ) -> list[pathlib.Path]:
+ """Create playlist files (m3u) for downloaded tracks in each directory.
+
+ Args:
+ dirs_scoped (set[pathlib.Path]): Set of directories containing tracks.
+ name_list (str): Name of the playlist.
+ is_album (bool): Whether this is an album.
+ sort_alphabetically (bool): Whether to sort tracks alphabetically.
+
+ Returns:
+ list[pathlib.Path]: List of created playlist file paths.
+ """
+ result: list[pathlib.Path] = []
# For each dir, which contains tracks
for dir_scoped in dirs_scoped:
@@ -830,7 +1489,7 @@ class Download:
self.fn_logger.debug(f"Playlist: Creating {path_playlist}")
# Get all tracks in the directory
- path_tracks: [pathlib.Path] = []
+ path_tracks: list[pathlib.Path] = []
for extension_audio in AudioExtensions:
path_tracks = path_tracks + list(dir_scoped.glob(f"*{extension_audio!s}"))
@@ -858,6 +1517,14 @@ class Download:
return result
def _video_convert(self, path_file: pathlib.Path) -> pathlib.Path:
+ """Convert a TS video file to MP4 using ffmpeg.
+
+ Args:
+ path_file (pathlib.Path): Path to the TS file.
+
+ Returns:
+ pathlib.Path: Path to the converted MP4 file.
+ """
path_file_out: pathlib.Path = path_file.with_suffix(AudioExtensions.MP4)
ffmpeg = (
FFmpeg(executable=self.settings.data.path_binary_ffmpeg)
@@ -871,6 +1538,14 @@ class Download:
return path_file_out
def _extract_flac(self, path_media_src: pathlib.Path) -> pathlib.Path:
+ """Extract FLAC audio from a media file using ffmpeg.
+
+ Args:
+ path_media_src (pathlib.Path): Path to the source media file.
+
+ Returns:
+ pathlib.Path: Path to the extracted FLAC file.
+ """
path_media_out = path_media_src.with_suffix(AudioExtensions.FLAC)
ffmpeg = (
FFmpeg(executable=self.settings.data.path_binary_ffmpeg)
@@ -889,7 +1564,16 @@ class Download:
return path_media_out
- def _extract_video_stream(self, m3u8_variant: m3u8.M3U8, quality: int) -> (m3u8.M3U8 | bool, str):
+ def _extract_video_stream(self, m3u8_variant: m3u8.M3U8, quality: int) -> tuple[m3u8.M3U8 | bool, str]:
+ """Extract the best matching video stream from an m3u8 variant playlist.
+
+ Args:
+ m3u8_variant (m3u8.M3U8): The m3u8 variant playlist.
+ quality (int): Desired video quality (vertical resolution).
+
+ Returns:
+ tuple[m3u8.M3U8 | bool, str]: (Selected m3u8 playlist or False, codecs string)
+ """
m3u8_playlist: m3u8.M3U8 | bool = False
resolution_best: int = 0
mime_type: str = ""
diff --git a/tidal_dl_ng/gui.py b/tidal_dl_ng/gui.py
index 73da806..80dc07e 100644
--- a/tidal_dl_ng/gui.py
+++ b/tidal_dl_ng/gui.py
@@ -47,7 +47,8 @@
import math
import sys
import time
-from collections.abc import Callable, Sequence
+from collections.abc import Callable, Iterable, Sequence
+from typing import Any
from requests.exceptions import HTTPError
from tidalapi.session import LinkLogin
@@ -106,6 +107,11 @@ from tidal_dl_ng.worker import Worker
# TODO: Make more use of Exceptions
class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
+ """Main application window for TIDAL Downloader Next Generation.
+
+ Handles GUI setup, user interactions, and download logic.
+ """
+
settings: Settings
tidal: Tidal
dl: Download
@@ -137,7 +143,12 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
s_queue_download_item_failed: QtCore.Signal = QtCore.Signal(object)
s_queue_download_item_skipped: QtCore.Signal = QtCore.Signal(object)
- def __init__(self, tidal: Tidal | None = None):
+ def __init__(self, tidal: Tidal | None = None) -> None:
+ """Initialize the main window and all components.
+
+ Args:
+ tidal (Tidal | None): Optional Tidal session object.
+ """
super().__init__()
self.setupUi(self)
# self.setGeometry(50, 50, 500, 300)
@@ -167,10 +178,16 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
logger_gui.debug("All setup.")
- def _init_gui(self):
- self.spinners = {}
+ def _init_gui(self) -> None:
+ """Initialize GUI-specific variables and state."""
+ self.spinners: dict[QtWidgets.QWidget, QtWaitingSpinner] = {}
+
+ def init_tidal(self, tidal: Tidal | None = None):
+ """Initialize Tidal session and handle login flow.
- def init_tidal(self, tidal: Tidal = None):
+ Args:
+ tidal (Tidal, optional): Existing Tidal session. Defaults to None.
+ """
result: bool = False
if tidal:
@@ -185,10 +202,11 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
while not result:
link_login: LinkLogin = self.tidal.session.get_link_login()
+ expires_in = int(link_login.expires_in) if hasattr(link_login, "expires_in") else 0
d_login: DialogLogin = DialogLogin(
url_login=link_login.verification_uri_complete,
hint=hint,
- expires_in=link_login.expires_in,
+ expires_in=expires_in,
parent=self,
)
@@ -211,10 +229,12 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
self.thread_it(self.tidal_user_lists)
def _init_threads(self):
+ """Initialize thread pool and start background workers."""
self.threadpool = QtCore.QThreadPool()
self.thread_it(self.watcher_queue_download)
def _init_dl(self):
+ """Initialize Download object and related progress bars."""
# Init `Download` object.
data_pb: ProgressBars = ProgressBars(
item=self.s_item_advance,
@@ -236,6 +256,7 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
)
def _init_progressbar(self):
+ """Initialize and add progress bars to the status bar."""
self.pb_list = QtWidgets.QProgressBar()
self.pb_item = QtWidgets.QProgressBar()
pbs = [self.pb_list, self.pb_item]
@@ -246,18 +267,30 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
self.statusbar.addPermanentWidget(pb)
def _init_info(self):
+ """Set default album cover image in the GUI."""
path_image: str = resource_path("tidal_dl_ng/ui/default_album_image.png")
self.l_pm_cover.setPixmap(QtGui.QPixmap(path_image))
def on_progress_reset(self):
+ """Reset progress bars to zero."""
self.pb_list.setValue(0)
self.pb_item.setValue(0)
def on_statusbar_message(self, data: StatusbarMessage):
+ """Show a message in the status bar.
+
+ Args:
+ data (StatusbarMessage): Message and timeout.
+ """
self.statusbar.showMessage(data.message, data.timeout)
- def _log_output(self, text):
+ def _log_output(self, text: str) -> None:
+ """Redirect log output to the debug text area.
+
+ Args:
+ text (str): Log message.
+ """
display_msg = coloredlogs.converter.convert(text)
cursor: QtGui.QTextCursor = self.te_debug.textCursor()
@@ -267,21 +300,33 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
self.te_debug.setTextCursor(cursor)
self.te_debug.ensureCursorVisible()
- def _populate_quality(self, ui_target: QtWidgets.QComboBox, options: type[Quality | QualityVideo]):
+ def _populate_quality(self, ui_target: QtWidgets.QComboBox, options: Iterable[Any]) -> None:
+ """Populate a combo box with quality options.
+
+ Args:
+ ui_target (QComboBox): Target combo box.
+ options (Iterable): Enum of quality options.
+ """
for item in options:
ui_target.addItem(item.name, item)
- def _populate_search_types(self, ui_target: QtWidgets.QComboBox, options: SearchTypes):
+ def _populate_search_types(self, ui_target: QtWidgets.QComboBox, options: Iterable[Any]) -> None:
+ """Populate a combo box with search type options.
+
+ Args:
+ ui_target (QComboBox): Target combo box.
+ options (Iterable): Enum of search types.
+ """
for item in options:
if item:
ui_target.addItem(item.__name__, item)
self.cb_search_type.setCurrentIndex(2)
- def handle_filter_activated(self):
- header: FilterHeader = self.tr_results.header()
- filters = []
-
+ def handle_filter_activated(self) -> None:
+ """Handle activation of filter headers in the results tree."""
+ header = self.tr_results.header()
+ filters: list[tuple[int, str]] = []
for i in range(header.count()):
text: str = header.filter_text(i)
@@ -292,6 +337,12 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
proxy_model.filters = filters
def _init_tree_results(self, tree: QtWidgets.QTreeView, model: QtGui.QStandardItemModel) -> None:
+ """Initialize the results tree view and its model.
+
+ Args:
+ tree (QTreeView): The tree view widget.
+ model (QStandardItemModel): The model for the tree.
+ """
header: FilterHeader = FilterHeader(tree)
self.proxy_tr_results: HumanProxyModel = HumanProxyModel(self)
@@ -313,31 +364,48 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
tree.customContextMenuRequested.connect(self.menu_context_tree_results)
def _init_tree_results_model(self, model: QtGui.QStandardItemModel) -> None:
- labels_column: [str] = ["#", "obj", "Artist", "Title", "Album", "Duration", "Quality", "Date"]
+ """Initialize the model for the results tree view.
+
+ Args:
+ model (QStandardItemModel): The model to initialize.
+ """
+ labels_column: list[str] = ["#", "obj", "Artist", "Title", "Album", "Duration", "Quality", "Date"]
model.setColumnCount(len(labels_column))
model.setRowCount(0)
model.setHorizontalHeaderLabels(labels_column)
- def _init_tree_queue(self, tree: QtWidgets.QTableWidget):
+ def _init_tree_queue(self, tree: QtWidgets.QTableWidget) -> None:
+ """Initialize the download queue table widget.
+
+ Args:
+ tree (QTableWidget): The table widget.
+ """
tree.setColumnHidden(1, True)
tree.setColumnWidth(2, 200)
header = tree.header()
- header.setSectionResizeMode(0, QtWidgets.QHeaderView.ResizeToContents)
- header.setStretchLastSection(False)
+ if hasattr(header, "setSectionResizeMode"):
+ header.setSectionResizeMode(0, QtWidgets.QHeaderView.ResizeToContents)
+ tree.setContextMenuPolicy(QtCore.Qt.CustomContextMenu)
- def tidal_user_lists(self):
+ def tidal_user_lists(self) -> None:
+ """Fetch and emit user playlists, mixes, and favorites from Tidal."""
# Start loading spinner
self.s_spinner_start.emit(self.tr_lists_user)
self.s_pb_reload_status.emit(False)
- user_all: [Playlist | UserPlaylist | Mix] = user_media_lists(self.tidal.session)
+ user_all: list[Any] = user_media_lists(self.tidal.session)
self.s_populate_tree_lists.emit(user_all)
- def on_populate_tree_lists(self, user_lists: [Playlist | UserPlaylist | Mix]):
+ def on_populate_tree_lists(self, user_lists: list[Playlist | UserPlaylist | Mix]) -> None:
+ """Populate the user lists tree with playlists, mixes, and favorites.
+
+ Args:
+ user_lists (list[Playlist | UserPlaylist | Mix]): List of user playlists, mixes, and favorites.
+ """
twi_playlists: QtWidgets.QTreeWidgetItem = self.tr_lists_user.findItems(
TidalLists.Playlists, QtCore.Qt.MatchExactly, 0
)[0]
@@ -357,7 +425,7 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
for item in user_lists:
if isinstance(item, UserPlaylist | Playlist):
twi_child = QtWidgets.QTreeWidgetItem(twi_playlists)
- name: str = item.name
+ name: str = item.name if getattr(item, "name", None) is not None else ""
description: str = f" {item.description}" if item.description else ""
info: str = f"({item.num_tracks + item.num_videos} Tracks){description}"
elif isinstance(item, Mix):
@@ -389,7 +457,12 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
self.s_spinner_stop.emit()
self.s_pb_reload_status.emit(True)
- def _init_tree_lists(self, tree: QtWidgets.QTreeWidget):
+ def _init_tree_lists(self, tree: QtWidgets.QTreeWidget) -> None:
+ """Initialize the user lists tree widget.
+
+ Args:
+ tree (QTreeWidget): The tree widget.
+ """
# Adjust Tree.
tree.setColumnWidth(0, 200)
tree.setColumnHidden(1, True)
@@ -400,19 +473,31 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
tree.setContextMenuPolicy(QtCore.Qt.CustomContextMenu)
tree.customContextMenuRequested.connect(self.menu_context_tree_lists)
- def on_update_check(self, on_startup: bool = True):
+ def on_update_check(self, on_startup: bool = True) -> None:
+ """Check for application updates and emit update signals.
+
+ Args:
+ on_startup (bool, optional): Whether this is called on startup. Defaults to True.
+ """
is_available, info = update_available()
if (on_startup and is_available) or not on_startup:
self.s_update_show.emit(True, is_available, info)
- def apply_settings(self, settings: Settings):
- l_cb = [
- {"element": self.cb_quality_audio, "setting": settings.data.quality_audio, "default_id": 1},
- {"element": self.cb_quality_video, "setting": settings.data.quality_video, "default_id": 0},
+ def apply_settings(self, settings: Settings) -> None:
+ """Apply user settings to the GUI.
+
+ Args:
+ settings (Settings): The settings object.
+ """
+ quality_audio = getattr(getattr(settings, "data", None), "quality_audio", 1)
+ quality_video = getattr(getattr(settings, "data", None), "quality_video", 0)
+ elements = [
+ {"element": self.cb_quality_audio, "setting": quality_audio, "default_id": 1},
+ {"element": self.cb_quality_video, "setting": quality_video, "default_id": 0},
]
- for item in l_cb:
+ for item in elements:
idx = item["element"].findData(item["setting"])
if idx > -1:
@@ -420,7 +505,12 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
else:
item["element"].setCurrentIndex(item["default_id"])
- def on_spinner_start(self, parent: QtWidgets.QWidget):
+ def on_spinner_start(self, parent: QtWidgets.QWidget) -> None:
+ """Start a loading spinner on the given parent widget.
+
+ Args:
+ parent (QWidget): The parent widget.
+ """
# Stop any existing spinner for this parent
if parent in self.spinners:
spinner = self.spinners[parent]
@@ -437,7 +527,8 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
self.spinners[parent] = spinner
- def on_spinner_stop(self):
+ def on_spinner_stop(self) -> None:
+ """Stop all active loading spinners."""
# Stop all spinners
for spinner in list(self.spinners.values()):
spinner.stop()
@@ -445,7 +536,12 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
self.spinners.clear()
- def menu_context_tree_lists(self, point: QtCore.QPoint):
+ def menu_context_tree_lists(self, point: QtCore.QPoint) -> None:
+ """Show context menu for user lists tree.
+
+ Args:
+ point (QPoint): The point where the menu is requested.
+ """
# Infos about the node selected.
index = self.tr_lists_user.indexAt(point)
@@ -460,7 +556,12 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
menu.exec(self.tr_lists_user.mapToGlobal(point))
- def menu_context_tree_results(self, point: QtCore.QPoint):
+ def menu_context_tree_results(self, point: QtCore.QPoint) -> None:
+ """Show context menu for results tree.
+
+ Args:
+ point (QPoint): The point where the menu is requested.
+ """
# Infos about the node selected.
index = self.tr_results.indexAt(point)
@@ -474,10 +575,23 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
menu.exec(self.tr_results.mapToGlobal(point))
- def thread_download_list_media(self, point: QtCore.QPoint):
+ def thread_download_list_media(self, point: QtCore.QPoint) -> None:
+ """Start download of a list media item in a thread.
+
+ Args:
+ point (QPoint): The point in the tree.
+ """
self.thread_it(self.on_download_list_media, point)
- def on_copy_url_share(self, tree_target: QtWidgets.QTreeWidget | QtWidgets.QTreeView, point: QtCore.QPoint = None):
+ def on_copy_url_share(
+ self, tree_target: QtWidgets.QTreeWidget | QtWidgets.QTreeView, point: QtCore.QPoint = None
+ ) -> None:
+ """Copy the share URL of a media item to the clipboard.
+
+ Args:
+ tree_target (QTreeWidget | QTreeView): The tree widget.
+ point (QPoint, optional): The point in the tree. Defaults to None.
+ """
if isinstance(tree_target, QtWidgets.QTreeWidget):
item: QtWidgets.QTreeWidgetItem = tree_target.itemAt(point)
@@ -494,8 +608,13 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
clipboard.clear()
clipboard.setText(url_share)
- def on_download_list_media(self, point: QtCore.QPoint = None):
- items: [QtWidgets.QTreeWidgetItem]
+ def on_download_list_media(self, point: QtCore.QPoint | None = None) -> None:
+ """Download all media items in a selected list.
+
+ Args:
+ point (QPoint | None, optional): The point in the tree. Defaults to None.
+ """
+ items: list[QtWidgets.QTreeWidgetItem] = []
if point:
items = [self.tr_lists_user.itemAt(point)]
@@ -507,19 +626,29 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
for item in items:
media = get_user_list_media_item(item)
- queue_dl_item: QueueDownloadItem | False = self.media_to_queue_download_model(media)
+ queue_dl_item: QueueDownloadItem | None = self.media_to_queue_download_model(media)
if queue_dl_item:
self.queue_download_media(queue_dl_item)
- def search_populate_results(self, query: str, type_media: SearchTypes):
- self.model_tr_results.removeRows(0, self.model_tr_results.rowCount())
+ def search_populate_results(self, query: str, type_media: Any) -> None:
+ """Populate the results tree with search results.
- results: [ResultItem] = self.search(query, [type_media])
+ Args:
+ query (str): The search query.
+ type_media (SearchTypes): The type of media to search for.
+ """
+ results: list[ResultItem] = self.search(query, [type_media])
self.populate_tree_results(results)
- def populate_tree_results(self, results: [ResultItem], parent: QtGui.QStandardItem = None):
+ def populate_tree_results(self, results: list[ResultItem], parent: QtGui.QStandardItem | None = None) -> None:
+ """Populate the results tree with ResultItem objects.
+
+ Args:
+ results (list[ResultItem]): The results to display.
+ parent (QStandardItem, optional): Parent item for nested results. Defaults to None.
+ """
if not parent:
self.model_tr_results.removeRows(0, self.model_tr_results.rowCount())
@@ -535,6 +664,15 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
self.s_tr_results_add_top_level_item.emit(child)
def populate_tree_result_child(self, item: ResultItem, index_count_digits: int) -> Sequence[QtGui.QStandardItem]:
+ """Create a row of QStandardItems for a ResultItem.
+
+ Args:
+ item (ResultItem): The result item.
+ index_count_digits (int): Number of digits for index formatting.
+
+ Returns:
+ Sequence[QStandardItem]: The row of items.
+ """
duration: str = ""
# TODO: Duration needs to be calculated later to properly fill with zeros.
@@ -582,14 +720,29 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
)
def on_tr_results_add_top_level_item(self, item_child: Sequence[QtGui.QStandardItem]):
+ """Add a top-level item to the results tree model.
+
+ Args:
+ item_child (Sequence[QStandardItem]): The row to add.
+ """
self.model_tr_results.appendRow(item_child)
- def on_settings_save(self):
+ def on_settings_save(self) -> None:
+ """Save settings and re-apply them to the GUI."""
self.settings.save()
self.apply_settings(self.settings)
self._init_dl()
- def search(self, query: str, types_media: SearchTypes) -> [ResultItem]:
+ def search(self, query: str, types_media: list[Any]) -> list[ResultItem]:
+ """Perform a search and return a list of ResultItems.
+
+ Args:
+ query (str): The search query.
+ types_media (list[Any]): The types of media to search for.
+
+ Returns:
+ list[ResultItem]: The search results.
+ """
query = query.strip()
# If a direct link was searched for, skip search and create the object from the link directly.
@@ -606,11 +759,11 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
result_search = {"direct": [media]}
else:
- result_search: dict[str, [SearchTypes]] = search_results_all(
+ result_search: dict[str, list[SearchTypes]] = search_results_all(
session=self.tidal.session, needle=query, types_media=types_media
)
- result: [ResultItem] = []
+ result: list[ResultItem] = []
for _media_type, l_media in result_search.items():
if isinstance(l_media, list):
@@ -618,129 +771,257 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
return result
- def search_result_to_model(self, items: [*SearchTypes]) -> [ResultItem]:
- result = []
+ def search_result_to_model(self, items: list[SearchTypes]) -> list[ResultItem]:
+ """Convert search results to ResultItem models.
+
+ Args:
+ items (list[SearchTypes]): List of search result items.
+
+ Returns:
+ list[ResultItem]: List of ResultItem models.
+ """
+ result: list[ResultItem] = []
for idx, item in enumerate(items):
- if not item:
- continue
+ result_item = self._to_result_item(idx, item)
- explicit: str = ""
- # Check if item is available on TIDAL.
- if hasattr(item, "available") and not item.available:
- continue
+ if result_item is not None:
+ result.append(result_item)
- if isinstance(item, Track | Video | Album):
- explicit = " šŸ…“" if item.explicit else ""
+ return result
- date_user_added: str = item.user_date_added.strftime("%Y-%m-%d_%H:%M") if item.user_date_added else ""
- date_release: str = (
- item.album.release_date.strftime("%Y-%m-%d_%H:%M")
- if hasattr(item, "album") and item.album and item.album.release_date
- else (
- item.release_date.strftime("%Y-%m-%d_%H:%M")
- if hasattr(item, "release_date") and item.release_date
- else ""
- )
- )
+ def _to_result_item(self, idx: int, item) -> ResultItem | None:
+ """Helper to convert a single item to ResultItem, or None if not valid.
- if isinstance(item, Track):
- result_item: ResultItem = ResultItem(
- position=idx,
- artist=name_builder_artist(item),
- title=f"{name_builder_title(item)}{explicit}",
- album=item.album.name,
- duration_sec=item.duration,
- obj=item,
- quality=quality_audio_highest(item),
- explicit=bool(item.explicit),
- date_user_added=date_user_added,
- date_release=date_release,
- )
+ Args:
+ idx (int): Index of the item.
+ item: The item to convert.
- result.append(result_item)
- elif isinstance(item, Video):
- result_item: ResultItem = ResultItem(
- position=idx,
- artist=name_builder_artist(item),
- title=f"{name_builder_title(item)}{explicit}",
- album=item.album.name if item.album else "",
- duration_sec=item.duration,
- obj=item,
- quality=item.video_quality,
- explicit=bool(item.explicit),
- date_user_added=date_user_added,
- date_release=date_release,
- )
+ Returns:
+ ResultItem | None: The converted ResultItem or None if not valid.
+ """
+ if not item:
+ return None
- result.append(result_item)
- elif isinstance(item, Playlist):
- result_item: ResultItem = ResultItem(
- position=idx,
- artist=", ".join(artist.name for artist in item.promoted_artists) if item.promoted_artists else "",
- title=item.name,
- album="",
- duration_sec=item.duration,
- obj=item,
- quality="",
- explicit=False,
- date_user_added=date_user_added,
- date_release=date_release,
- )
+ if hasattr(item, "available") and not item.available:
+ return None
- result.append(result_item)
- elif isinstance(item, Album):
- result_item: ResultItem = ResultItem(
- position=idx,
- artist=name_builder_artist(item),
- title="",
- album=f"{item.name}{explicit}",
- duration_sec=item.duration,
- obj=item,
- quality=quality_audio_highest(item),
- explicit=bool(item.explicit),
- date_user_added=date_user_added,
- date_release=date_release,
- )
+ explicit: str = ""
+ if isinstance(item, Track | Video | Album):
+ explicit = " šŸ…“" if item.explicit else ""
- result.append(result_item)
- elif isinstance(item, Mix):
- result_item: ResultItem = ResultItem(
- position=idx,
- artist=item.sub_title,
- title=item.title,
- album="",
- # TODO: Calculate total duration.
- duration_sec=-1,
- obj=item,
- quality="",
- explicit=False,
- date_user_added=date_user_added,
- date_release=date_release,
- )
+ date_user_added: str = (
+ item.user_date_added.strftime("%Y-%m-%d_%H:%M") if getattr(item, "user_date_added", None) else ""
+ )
+ date_release: str = self._get_date_release(item)
- result.append(result_item)
- elif isinstance(item, Artist):
- result_item: ResultItem = ResultItem(
- position=idx,
- artist=item.name,
- title="",
- album="",
- duration_sec=-1,
- obj=item,
- quality="",
- explicit=False,
- date_user_added=date_user_added,
- date_release=date_release,
- )
+ if isinstance(item, Track):
+ return self._result_item_from_track(idx, item, explicit, date_user_added, date_release)
- result.append(result_item)
+ if isinstance(item, Video):
+ return self._result_item_from_video(idx, item, explicit, date_user_added, date_release)
- return result
+ if isinstance(item, Playlist):
+ return self._result_item_from_playlist(idx, item, date_user_added, date_release)
+
+ if isinstance(item, Album):
+ return self._result_item_from_album(idx, item, explicit, date_user_added, date_release)
+
+ if isinstance(item, Mix):
+ return self._result_item_from_mix(idx, item, date_user_added, date_release)
+
+ if isinstance(item, Artist):
+ return self._result_item_from_artist(idx, item, date_user_added, date_release)
+
+ return None
+
+ def _get_date_release(self, item) -> str:
+ """Get the release date string for an item.
+
+ Args:
+ item: The item to extract the release date from.
+
+ Returns:
+ str: The formatted release date or empty string.
+ """
+ if hasattr(item, "album") and item.album and getattr(item.album, "release_date", None):
+ return item.album.release_date.strftime("%Y-%m-%d_%H:%M")
+
+ if hasattr(item, "release_date") and item.release_date:
+ return item.release_date.strftime("%Y-%m-%d_%H:%M")
+
+ return ""
+
+ def _result_item_from_track(
+ self, idx: int, item, explicit: str, date_user_added: str, date_release: str
+ ) -> ResultItem:
+ """Create a ResultItem from a Track.
+
+ Args:
+ idx (int): Index of the item.
+ item: The Track item.
+ explicit (str): Explicit tag.
+ date_user_added (str): Date user added.
+ date_release (str): Release date.
+
+ Returns:
+ ResultItem: The constructed ResultItem.
+ """
+ return ResultItem(
+ position=idx,
+ artist=name_builder_artist(item),
+ title=f"{name_builder_title(item)}{explicit}",
+ album=item.album.name,
+ duration_sec=item.duration,
+ obj=item,
+ quality=quality_audio_highest(item),
+ explicit=bool(item.explicit),
+ date_user_added=date_user_added,
+ date_release=date_release,
+ )
+
+ def _result_item_from_video(
+ self, idx: int, item, explicit: str, date_user_added: str, date_release: str
+ ) -> ResultItem:
+ """Create a ResultItem from a Video.
+
+ Args:
+ idx (int): Index of the item.
+ item: The Video item.
+ explicit (str): Explicit tag.
+ date_user_added (str): Date user added.
+ date_release (str): Release date.
+
+ Returns:
+ ResultItem: The constructed ResultItem.
+ """
+ return ResultItem(
+ position=idx,
+ artist=name_builder_artist(item),
+ title=f"{name_builder_title(item)}{explicit}",
+ album=item.album.name if item.album else "",
+ duration_sec=item.duration,
+ obj=item,
+ quality=item.video_quality,
+ explicit=bool(item.explicit),
+ date_user_added=date_user_added,
+ date_release=date_release,
+ )
+
+ def _result_item_from_playlist(self, idx: int, item, date_user_added: str, date_release: str) -> ResultItem:
+ """Create a ResultItem from a Playlist.
+
+ Args:
+ idx (int): Index of the item.
+ item: The Playlist item.
+ date_user_added (str): Date user added.
+ date_release (str): Release date.
+
+ Returns:
+ ResultItem: The constructed ResultItem.
+ """
+ return ResultItem(
+ position=idx,
+ artist=", ".join(artist.name for artist in item.promoted_artists) if item.promoted_artists else "",
+ title=item.name,
+ album="",
+ duration_sec=item.duration,
+ obj=item,
+ quality="",
+ explicit=False,
+ date_user_added=date_user_added,
+ date_release=date_release,
+ )
+
+ def _result_item_from_album(
+ self, idx: int, item, explicit: str, date_user_added: str, date_release: str
+ ) -> ResultItem:
+ """Create a ResultItem from an Album.
+
+ Args:
+ idx (int): Index of the item.
+ item: The Album item.
+ explicit (str): Explicit tag.
+ date_user_added (str): Date user added.
+ date_release (str): Release date.
+
+ Returns:
+ ResultItem: The constructed ResultItem.
+ """
+ return ResultItem(
+ position=idx,
+ artist=name_builder_artist(item),
+ title="",
+ album=f"{item.name}{explicit}",
+ duration_sec=item.duration,
+ obj=item,
+ quality=quality_audio_highest(item),
+ explicit=bool(item.explicit),
+ date_user_added=date_user_added,
+ date_release=date_release,
+ )
+
+ def _result_item_from_mix(self, idx: int, item, date_user_added: str, date_release: str) -> ResultItem:
+ """Create a ResultItem from a Mix.
+
+ Args:
+ idx (int): Index of the item.
+ item: The Mix item.
+ date_user_added (str): Date user added.
+ date_release (str): Release date.
+
+ Returns:
+ ResultItem: The constructed ResultItem.
+ """
+ return ResultItem(
+ position=idx,
+ artist=item.sub_title,
+ title=item.title,
+ album="",
+ duration_sec=-1, # TODO: Calculate total duration.
+ obj=item,
+ quality="",
+ explicit=False,
+ date_user_added=date_user_added,
+ date_release=date_release,
+ )
+
+ def _result_item_from_artist(self, idx: int, item, date_user_added: str, date_release: str) -> ResultItem:
+ """Create a ResultItem from an Artist.
+
+ Args:
+ idx (int): Index of the item.
+ item: The Artist item.
+ date_user_added (str): Date user added.
+ date_release (str): Release date.
+
+ Returns:
+ ResultItem: The constructed ResultItem.
+ """
+ return ResultItem(
+ position=idx,
+ artist=item.name,
+ title="",
+ album="",
+ duration_sec=-1,
+ obj=item,
+ quality="",
+ explicit=False,
+ date_user_added=date_user_added,
+ date_release=date_release,
+ )
def media_to_queue_download_model(
self, media: Artist | Track | Video | Album | Playlist | Mix
) -> QueueDownloadItem | bool:
+ """Convert a media object to a QueueDownloadItem for the download queue.
+
+ Args:
+ media (Artist | Track | Video | Album | Playlist | Mix): The media object.
+
+ Returns:
+ QueueDownloadItem | bool: The queue item or False if not available.
+ """
result: QueueDownloadItem | False
name: str = ""
quality_audio: Quality = self.settings.data.quality_audio
@@ -789,7 +1070,8 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
return result
- def _init_signals(self):
+ def _init_signals(self) -> None:
+ """Connect signals to their respective slots."""
self.pb_download.clicked.connect(lambda: self.thread_it(self.on_download_results))
self.pb_download_list.clicked.connect(lambda: self.thread_it(self.on_download_list_media))
self.pb_reload_user_lists.clicked.connect(lambda: self.thread_it(self.tidal_user_lists))
@@ -839,45 +1121,87 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
self.s_queue_download_item_failed.connect(self.on_queue_download_item_failed)
self.s_queue_download_item_skipped.connect(self.on_queue_download_item_skipped)
- def _init_buttons(self):
+ def _init_buttons(self) -> None:
+ """Initialize the state of the download buttons."""
self.pb_queue_download_run()
- def on_logout(self):
+ def on_logout(self) -> None:
+ """Log out from TIDAL and close the application."""
result: bool = self.tidal.logout()
if result:
sys.exit(0)
- def on_progress_list(self, value: float):
+ def on_progress_list(self, value: float) -> None:
+ """Update the progress of the list progress bar.
+
+ Args:
+ value (float): The progress value as a percentage.
+ """
self.pb_list.setValue(int(math.ceil(value)))
- def on_progress_item(self, value: float):
+ def on_progress_item(self, value: float) -> None:
+ """Update the progress of the item progress bar.
+
+ Args:
+ value (float): The progress value as a percentage.
+ """
self.pb_item.setValue(int(math.ceil(value)))
- def on_progress_item_name(self, value: str):
+ def on_progress_item_name(self, value: str) -> None:
+ """Set the format of the item progress bar.
+
+ Args:
+ value (str): The item name.
+ """
self.pb_item.setFormat(f"%p% {value}")
- def on_progress_list_name(self, value: str):
+ def on_progress_list_name(self, value: str) -> None:
+ """Set the format of the list progress bar.
+
+ Args:
+ value (str): The list name.
+ """
self.pb_list.setFormat(f"%p% {value}")
- def on_quality_set_audio(self, index):
+ def on_quality_set_audio(self, index: int) -> None:
+ """Set the audio quality for downloads.
+
+ Args:
+ index: The index of the selected quality in the combo box.
+ """
self.settings.data.quality_audio = Quality(self.cb_quality_audio.itemData(index))
self.settings.save()
if self.tidal:
self.tidal.settings_apply()
- def on_quality_set_video(self, index):
+ def on_quality_set_video(self, index: int) -> None:
+ """Set the video quality for downloads.
+
+ Args:
+ index: The index of the selected quality in the combo box.
+ """
self.settings.data.quality_video = QualityVideo(self.cb_quality_video.itemData(index))
self.settings.save()
if self.tidal:
self.tidal.settings_apply()
- def on_list_items_show(self, item: QtWidgets.QTreeWidgetItem):
+ def on_list_items_show(self, item: QtWidgets.QTreeWidgetItem) -> None:
+ """Show the items in the selected playlist or mix.
+
+ Args:
+ item (QtWidgets.QTreeWidgetItem): The selected tree widget item.
+ """
self.thread_it(self.list_items_show, item)
- def list_items_show(self, item: QtWidgets.QTreeWidgetItem):
+ def list_items_show(self, item: QtWidgets.QTreeWidgetItem) -> None:
+ """Fetch and display the items in a playlist or mix.
+
+ Args:
+ item (QtWidgets.QTreeWidgetItem): The tree widget item representing a playlist or mix.
+ """
media_list: Album | Playlist | str = get_user_list_media_item(item)
# Only if clicked item is not a top level item.
@@ -896,6 +1220,11 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
self.s_spinner_stop.emit()
def on_result_item_clicked(self, index: QtCore.QModelIndex) -> None:
+ """Handle the event when a result item is clicked.
+
+ Args:
+ index (QtCore.QModelIndex): The index of the clicked item.
+ """
media: Track | Video | Album | Artist = get_results_media_item(
index, self.proxy_tr_results, self.model_tr_results
)
@@ -904,12 +1233,23 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
self.thread_it(self.cover_show, media)
def on_queue_download_item_clicked(self, item: QtWidgets.QTreeWidgetItem, column: int) -> None:
+ """Handle the event when a queue download item is clicked.
+
+ Args:
+ item (QtWidgets.QTreeWidgetItem): The clicked tree widget item.
+ column (int): The column index of the clicked item.
+ """
media: Track | Video | Album | Artist | Mix | Playlist = get_queue_download_media(item)
# Load cover asynchronously to avoid blocking the GUI
self.thread_it(self.cover_show, media)
def cover_show(self, media: Album | Playlist | Track | Video | Album | Artist) -> None:
+ """Show the cover image of the selected media item.
+
+ Args:
+ media (Album | Playlist | Track | Video | Album | Artist): The media item.
+ """
cover_url: str = ""
# Show spinner in the cover label itself
parent_widget = self.l_pm_cover
@@ -948,9 +1288,17 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
self,
media_list: Album | Playlist | Mix | Artist | None = None,
point: QtCore.QPoint | None = None,
- parent: QtGui.QStandardItem = None,
- favorite_function: Callable = None,
+ parent: QtGui.QStandardItem | None = None,
+ favorite_function: Callable | None = None,
) -> None:
+ """Populate the results tree with the items of a media list.
+
+ Args:
+ media_list (Album | Playlist | Mix | Artist | None, optional): The media list to show. Defaults to None.
+ point (QPoint | None, optional): The point in the tree. Defaults to None.
+ parent (QStandardItem | None, optional): Parent item for nested results. Defaults to None.
+ favorite_function (Callable | None, optional): Function to fetch favorite items. Defaults to None.
+ """
if point:
item = self.tr_lists_user.itemAt(point)
media_list = get_user_list_media_item(item)
@@ -960,39 +1308,54 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
if isinstance(media_list, str):
favorite_function = favorite_function_factory(self.tidal, media_list)
- media_items: [Track | Video | Album] = favorite_function()
+ media_items: list[Track | Video | Album] = favorite_function()
else:
- media_items: [Track | Video | Album] = items_results_all(media_list)
+ media_items: list[Track | Video | Album] = items_results_all(media_list)
- result: [ResultItem] = self.search_result_to_model(media_items)
+ result: list[ResultItem] = self.search_result_to_model(media_items)
self.populate_tree_results(result, parent=parent)
- def thread_it(self, fn: Callable, *args, **kwargs):
+ def thread_it(self, fn: Callable, *args, **kwargs) -> None:
+ """Run a function in a separate thread.
+
+ Args:
+ fn (Callable): The function to run.
+ *args: Positional arguments for the function.
+ **kwargs: Keyword arguments for the function.
+ """
# Any other args, kwargs are passed to the run function
worker = Worker(fn, *args, **kwargs)
# Execute
self.threadpool.start(worker)
- def on_queue_download_clear_all(self):
+ 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):
+ 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):
- items: [QtWidgets.QTreeWidgetItem | None] = self.tr_queue_download.findItems(
+ 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.tr_queue_download.findItems(
regex, QtCore.Qt.MatchFlag.MatchRegularExpression, column=0
)
for item in items:
self.tr_queue_download.takeTopLevelItem(self.tr_queue_download.indexOfTopLevelItem(item))
- def on_queue_download_remove(self):
- items: [QtWidgets.QTreeWidgetItem | None] = self.tr_queue_download.selectedItems()
+ def on_queue_download_remove(self) -> None:
+ """Remove selected items from the download queue."""
+ items: list[QtWidgets.QTreeWidgetItem | None] = self.tr_queue_download.selectedItems()
if len(items) == 0:
logger_gui.error("Please select an item from the queue first.")
@@ -1017,7 +1380,8 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
else:
self.pb_queue_download_run()
- def pb_queue_download_run(self):
+ 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()
@@ -1026,7 +1390,8 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
self.pb_queue_download_toggle.setIcon(icon)
self.pb_queue_download_toggle.setStyleSheet("background-color: #218838; color: #fff")
- def pb_queue_download_pause(self):
+ 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()
@@ -1037,6 +1402,7 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
# TODO: Must happen in main thread. Do not thread this.
def on_download_results(self) -> None:
+ """Download the selected results in the results tree."""
items: [HumanProxyModel | None] = self.tr_results.selectionModel().selectedRows()
if len(items) == 0:
@@ -1052,6 +1418,11 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
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.
+
+ Args:
+ queue_dl_item (QueueDownloadItem): The item to add to the queue.
+ """
# Populate child
child: QtWidgets.QTreeWidgetItem = QtWidgets.QTreeWidgetItem()
@@ -1064,10 +1435,11 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
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: [QtWidgets.QTreeWidgetItem | None] = self.tr_queue_download.findItems(
+ items: list[QtWidgets.QTreeWidgetItem | None] = self.tr_queue_download.findItems(
QueueDownloadStatus.Waiting, QtCore.Qt.MatchFlag.MatchExactly, column=0
)
@@ -1093,18 +1465,44 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
time.sleep(2)
def on_queue_download_item_downloading(self, item: QtWidgets.QTreeWidgetItem) -> None:
+ """Update the status of a queue download item to 'Downloading'.
+
+ Args:
+ item (QtWidgets.QTreeWidgetItem): The item to update.
+ """
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'.
+
+ Args:
+ item (QtWidgets.QTreeWidgetItem): The item to update.
+ """
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'.
+
+ Args:
+ item (QtWidgets.QTreeWidgetItem): The item to update.
+ """
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'.
+
+ Args:
+ item (QtWidgets.QTreeWidgetItem): The item to update.
+ """
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.
+
+ Args:
+ item (QtWidgets.QTreeWidgetItem): The item to update.
+ status (str): The status text.
+ """
item.setText(0, status)
def on_queue_download(
@@ -1113,6 +1511,16 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
quality_audio: Quality | None = None,
quality_video: QualityVideo | None = None,
) -> QueueDownloadStatus:
+ """Download the specified media item(s) and return the result status.
+
+ Args:
+ media (Track | Album | Playlist | Video | Mix | Artist): The media item(s) to download.
+ quality_audio (Quality | None, optional): Desired audio quality. Defaults to None.
+ quality_video (QualityVideo | None, optional): Desired video quality. Defaults to None.
+
+ Returns:
+ QueueDownloadStatus: The status of the download operation.
+ """
result: QueueDownloadStatus
items_media: [Track | Album | Playlist | Video | Mix | Artist]
@@ -1142,6 +1550,18 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
quality_audio: Quality | None = None,
quality_video: QualityVideo | None = None,
) -> QueueDownloadStatus:
+ """Download a media item and return the result status.
+
+ Args:
+ media (Track | Album | Playlist | Video | Mix | Artist): The media item to download.
+ dl (Download): The Download object to use.
+ delay_track (bool, optional): Whether to apply download delay. Defaults to False.
+ quality_audio (Quality | None, optional): Desired audio quality. Defaults to None.
+ quality_video (QualityVideo | None, optional): Desired video quality. Defaults to None.
+
+ Returns:
+ QueueDownloadStatus: The status of the download operation.
+ """
result_dl: bool
path_file: str
result: QueueDownloadStatus
@@ -1184,24 +1604,42 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
return result
def on_version(
- self, update_check: bool = False, update_available: bool = False, update_info: ReleaseLatest = None
+ self, update_check: bool = False, update_available: bool = False, update_info: ReleaseLatest | None = None
) -> None:
+ """Show the version information dialog.
+
+ Args:
+ update_check (bool, optional): Whether to check for updates. Defaults to False.
+ update_available (bool, optional): Whether an update is available. Defaults to False.
+ update_info (ReleaseLatest | None, optional): Information about the latest release. Defaults to None.
+ """
DialogVersion(self, update_check, update_available, update_info)
def on_preferences(self) -> None:
+ """Open the preferences dialog."""
DialogPreferences(settings=self.settings, settings_save=self.s_settings_save, parent=self)
def on_tr_results_expanded(self, index: QtCore.QModelIndex) -> None:
+ """Handle the event when a result item group is expanded.
+
+ Args:
+ index (QtCore.QModelIndex): The index of the expanded item.
+ """
self.thread_it(self.tr_results_expanded, index)
def tr_results_expanded(self, index: QtCore.QModelIndex) -> None:
+ """Load and display the children of an expanded result item.
+
+ Args:
+ index (QtCore.QModelIndex): The index of the expanded item.
+ """
# If the child is a dummy the list_item has not been expanded before
item: QtGui.QStandardItem = self.model_tr_results.itemFromIndex(self.proxy_tr_results.mapToSource(index))
load_children: bool = not item.child(0, 0).isEnabled()
if load_children:
item.removeRow(0)
- media_list: [Mix | Album | Playlist | Artist] = get_results_media_item(
+ media_list: list[Mix | Album | Playlist | Artist] = get_results_media_item(
index, self.proxy_tr_results, self.model_tr_results
)
@@ -1213,7 +1651,12 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
finally:
self.s_spinner_stop.emit()
- def button_reload_status(self, status: bool):
+ def button_reload_status(self, status: bool) -> None:
+ """Update the reload button's state and text.
+
+ Args:
+ status (bool): The new status.
+ """
button_text: str = "Reloading..."
if status:
@@ -1222,7 +1665,12 @@ class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
self.pb_reload_user_lists.setEnabled(status)
self.pb_reload_user_lists.setText(button_text)
- def closeEvent(self, event):
+ def closeEvent(self, event: QtGui.QCloseEvent) -> None:
+ """Handle the close event of the main window.
+
+ Args:
+ event (QtGui.QCloseEvent): The close event.
+ """
self.shutdown = True
handling_app: HandlingApp = HandlingApp()
diff --git a/tidal_dl_ng/helper/decryption.py b/tidal_dl_ng/helper/decryption.py
index 17ea19b..2f38fc5 100644
--- a/tidal_dl_ng/helper/decryption.py
+++ b/tidal_dl_ng/helper/decryption.py
@@ -7,9 +7,17 @@ from Crypto.Util import Counter
def decrypt_security_token(security_token: str) -> (str, str):
"""
- Decrypts security token into key and nonce pair
+ The `decrypt_security_token` function decrypts a security token into a key and nonce pair using AES
+ encryption.
- security_token should match the securityToken value from the web response
+ Args:
+ security_token (str): The `security_token` parameter in the `decrypt_security_token` function is a
+ string that represents an encrypted security token. This function decrypts the security token into a
+ key and nonce pair using AES encryption. security_token should match the securityToken value from the web response.
+
+ Returns:
+ The `decrypt_security_token` function returns a tuple containing the key and nonce extracted from
+ the decrypted security token.
"""
# Do not change this
diff --git a/tidal_dl_ng/helper/path.py b/tidal_dl_ng/helper/path.py
index 29347cb..12ffdce 100644
--- a/tidal_dl_ng/helper/path.py
+++ b/tidal_dl_ng/helper/path.py
@@ -24,6 +24,11 @@ from tidal_dl_ng.helper.tidal import name_builder_album_artist, name_builder_art
def path_home() -> str:
+ """Get the home directory path.
+
+ Returns:
+ str: The home directory path.
+ """
if "XDG_CONFIG_HOME" in os.environ:
return os.environ["XDG_CONFIG_HOME"]
elif "HOME" in os.environ:
@@ -35,6 +40,11 @@ def path_home() -> str:
def path_config_base() -> str:
+ """Get the base configuration path.
+
+ Returns:
+ str: The base configuration path.
+ """
# https://wiki.archlinux.org/title/XDG_Base_Directory
# X11 workaround: If user specified config path is set, do not point to "~/.config"
path_user_custom: str = os.environ.get("XDG_CONFIG_HOME", "")
@@ -45,14 +55,29 @@ def path_config_base() -> str:
def path_file_log() -> str:
+ """Get the path to the log file.
+
+ Returns:
+ str: The log file path.
+ """
return os.path.join(path_config_base(), "app.log")
def path_file_token() -> str:
+ """Get the path to the token file.
+
+ Returns:
+ str: The token file path.
+ """
return os.path.join(path_config_base(), "token.json")
def path_file_settings() -> str:
+ """Get the path to the settings file.
+
+ Returns:
+ str: The settings file path.
+ """
return os.path.join(path_config_base(), "settings.json")
@@ -91,130 +116,300 @@ def format_str_media(
list_pos: int = 0,
list_total: int = 0,
) -> str:
- result: str = name
+ """Format a string based on media attributes.
+ Args:
+ name (str): The format template name.
+ media (Track | Album | Playlist | UserPlaylist | Video | Mix): The media object.
+ album_track_num_pad_min (int): Minimum padding for track numbers. Defaults to 0.
+ list_pos (int): Position in a list. Defaults to 0.
+ list_total (int): Total items in a list. Defaults to 0.
+
+ Returns:
+ str: The formatted string.
+ """
try:
- match name:
- case "artist_name":
- if isinstance(media, Track | Video):
- if hasattr(media, "artists"):
- result = name_builder_artist(media)
- elif hasattr(media, "artist"):
- result = media.artist.name
- case "album_artist":
- result = name_builder_album_artist(media, first_only=True)
- case "album_artists":
- result = name_builder_album_artist(media)
- case "track_title":
- if isinstance(media, Track | Video):
- result = name_builder_title(media)
- case "mix_name":
- if isinstance(media, Mix):
- result = media.title
- case "playlist_name":
- if isinstance(media, Playlist | UserPlaylist):
- result = media.name
- case "album_title":
- if isinstance(media, Album):
- result = media.name
- elif isinstance(media, Track):
- result = media.album.name
- case "album_track_num":
- if isinstance(media, Track | Video):
- result = calculate_number_padding(
- album_track_num_pad_min,
- media.track_num,
- media.album.num_tracks if hasattr(media, "album") else 1,
- )
- case "album_num_tracks":
- if isinstance(media, Track | Video):
- result = str(media.album.num_tracks if hasattr(media, "album") else 1)
- case "track_id":
- if isinstance(media, Track | Video):
- result = str(media.id)
- case "playlist_id":
- if isinstance(media, Playlist):
- result = str(media.id)
- case "album_id":
- if isinstance(media, Album):
- result = str(media.id)
- elif isinstance(media, Track):
- result = str(media.album.id)
- case "track_duration_seconds":
- if isinstance(media, Track | Video):
- result = str(media.duration)
- case "track_duration_minutes":
- if isinstance(media, Track | Video):
- m, s = divmod(media.duration, 60)
- result = f"{m:01d}:{s:02d}"
- case "album_duration_seconds":
- if isinstance(media, Album):
- result = str(media.duration)
- case "album_duration_minutes":
- if isinstance(media, Album):
- m, s = divmod(media.duration, 60)
- result = f"{m:01d}:{s:02d}"
- case "playlist_duration_seconds":
- if isinstance(media, Album):
- result = str(media.duration)
- case "playlist_duration_minutes":
- if isinstance(media, Album):
- m, s = divmod(media.duration, 60)
- result = f"{m:01d}:{s:02d}"
- case "album_year":
- if isinstance(media, Album):
- result = str(media.year)
- elif isinstance(media, Track):
- result = str(media.album.year)
- case "video_quality":
- if isinstance(media, Video):
- result = media.video_quality
- case "track_quality":
- if isinstance(media, Track):
- result = ", ".join(tag for tag in media.media_metadata_tags)
- case "track_explicit":
- if isinstance(media, Track | Video):
- result = FORMAT_TEMPLATE_EXPLICIT if media.explicit else ""
- case "album_explicit":
- if isinstance(media, Album):
- result = FORMAT_TEMPLATE_EXPLICIT if media.explicit else ""
- case "album_num_volumes":
- if isinstance(media, Album):
- result = str(media.num_volumes)
- case "track_volume_num":
- if isinstance(media, Track | Video):
- result = str(media.volume_num)
- case "track_volume_num_optional":
- if isinstance(media, Track | Video):
- num_volumes: int = media.album.num_volumes if hasattr(media, "album") else 1
- result = "" if num_volumes == 1 else str(media.volume_num)
- case "track_volume_num_optional_CD":
- if isinstance(media, Track | Video):
- num_volumes: int = media.album.num_volumes if hasattr(media, "album") else 1
- result = "" if num_volumes == 1 else f"CD{media.volume_num!s}"
- case "isrc":
- if isinstance(media, Track):
- result = media.isrc
- case "list_pos":
- if isinstance(media, Track | Video):
- # TODO: Rename `album_track_num_pad_min` globally.
- result = calculate_number_padding(album_track_num_pad_min, list_pos, list_total)
+ # Try each formatter function in sequence
+ for formatter in (
+ _format_names,
+ _format_numbers,
+ _format_ids,
+ _format_durations,
+ _format_metadata,
+ _format_volumes,
+ ):
+ result = formatter(name, media, album_track_num_pad_min, list_pos, list_total)
+ if result is not None:
+ return result
except Exception as e:
# TODO: Implement better exception logging.
print(e)
- pass
+ return name
+
+
+def _format_artist_names(
+ name: str,
+ media: Track | Album | Playlist | UserPlaylist | Video | Mix,
+ *_args,
+) -> str | None:
+ """Handle artist name-related format strings.
+
+ Args:
+ name (str): The format string name to check.
+ media (Track | Album | Playlist | UserPlaylist | Video | Mix): The media object to extract artist information from.
+ *_args (Any): Additional arguments (not used).
+
+ Returns:
+ str | None: The formatted artist name or None if the format string is not artist-related.
+ """
+ if name == "artist_name" and isinstance(media, Track | Video):
+ if hasattr(media, "artists"):
+ return name_builder_artist(media)
+ elif hasattr(media, "artist"):
+ return media.artist.name
+ elif name == "album_artist":
+ return name_builder_album_artist(media, first_only=True)
+ elif name == "album_artists":
+ return name_builder_album_artist(media)
+ return None
+
+
+def _format_titles(
+ name: str,
+ media: Track | Album | Playlist | UserPlaylist | Video | Mix,
+ *_args,
+) -> str | None:
+ """Handle title-related format strings.
+
+ Args:
+ name (str): The format string name to check.
+ media (Track | Album | Playlist | UserPlaylist | Video | Mix): The media object to extract title information from.
+ *_args (Any): Additional arguments (not used).
+
+ Returns:
+ str | None: The formatted title or None if the format string is not title-related.
+ """
+ if name == "track_title" and isinstance(media, Track | Video):
+ return name_builder_title(media)
+ elif name == "mix_name" and isinstance(media, Mix):
+ return media.title
+ elif name == "playlist_name" and isinstance(media, Playlist | UserPlaylist):
+ return media.name
+ elif name == "album_title":
+ if isinstance(media, Album):
+ return media.name
+ elif isinstance(media, Track):
+ return media.album.name
+ return None
+
+
+def _format_names(
+ name: str,
+ media: Track | Album | Playlist | UserPlaylist | Video | Mix,
+ *_args,
+) -> str | None:
+ """Handle name-related format strings.
+
+ Args:
+ name (str): The format string name to check.
+ media (Track | Album | Playlist | UserPlaylist | Video | Mix): The media object to extract name information from.
+ *_args (Any): Additional arguments (not used).
+
+ Returns:
+ str | None: The formatted name or None if the format string is not name-related.
+ """
+ # First try artist name formats
+ result = _format_artist_names(name, media)
+ if result is not None:
+ return result
+
+ # Then try title formats
+ return _format_titles(name, media)
+
+
+def _format_numbers(
+ name: str,
+ media: Track | Album | Playlist | UserPlaylist | Video | Mix,
+ album_track_num_pad_min: int,
+ list_pos: int,
+ list_total: int,
+ *_args,
+) -> str | None:
+ """Handle number-related format strings.
+
+ Args:
+ name (str): The format string name to check.
+ media (Track | Album | Playlist | UserPlaylist | Video | Mix): The media object to extract number information from.
+ album_track_num_pad_min (int): Minimum padding for track numbers.
+ list_pos (int): Position in a list.
+ list_total (int): Total items in a list.
+ *_args (Any): Additional arguments (not used).
+
+ Returns:
+ str | None: The formatted number or None if the format string is not number-related.
+ """
+ if name == "album_track_num" and isinstance(media, Track | Video):
+ return calculate_number_padding(
+ album_track_num_pad_min,
+ media.track_num,
+ media.album.num_tracks if hasattr(media, "album") else 1,
+ )
+ elif name == "album_num_tracks" and isinstance(media, Track | Video):
+ return str(media.album.num_tracks if hasattr(media, "album") else 1)
+ elif name == "list_pos" and isinstance(media, Track | Video):
+ # TODO: Rename `album_track_num_pad_min` globally.
+ return calculate_number_padding(album_track_num_pad_min, list_pos, list_total)
+ return None
+
+
+def _format_ids(
+ name: str,
+ media: Track | Album | Playlist | UserPlaylist | Video | Mix,
+ *_args,
+) -> str | None:
+ """Handle ID-related format strings.
+
+ Args:
+ name (str): The format string name to check.
+ media (Track | Album | Playlist | UserPlaylist | Video | Mix): The media object to extract ID information from.
+ *_args (Any): Additional arguments (not used).
+
+ Returns:
+ str | None: The formatted ID or None if the format string is not ID-related.
+ """
+ # Handle track and playlist IDs
+ if (name == "track_id" and isinstance(media, Track | Video)) or (
+ name == "playlist_id" and isinstance(media, Playlist)
+ ):
+ return str(media.id)
+ # Handle album IDs
+ elif name == "album_id":
+ if isinstance(media, Album):
+ return str(media.id)
+ elif isinstance(media, Track):
+ return str(media.album.id)
+ # Handle ISRC
+ elif name == "isrc" and isinstance(media, Track):
+ return media.isrc
+ return None
+
+
+def _format_durations(
+ name: str,
+ media: Track | Album | Playlist | UserPlaylist | Video | Mix,
+ *_args,
+) -> str | None:
+ """Handle duration-related format strings.
+
+ Args:
+ name (str): The format string name to check.
+ media (Track | Album | Playlist | UserPlaylist | Video | Mix): The media object to extract duration information from.
+ *_args (Any): Additional arguments (not used).
+
+ Returns:
+ str | None: The formatted duration or None if the format string is not duration-related.
+ """
+ # Format track durations
+ if name == "track_duration_seconds" and isinstance(media, Track | Video):
+ return str(media.duration)
+ elif name == "track_duration_minutes" and isinstance(media, Track | Video):
+ m, s = divmod(media.duration, 60)
+ return f"{m:01d}:{s:02d}"
+
+ # Format album durations
+ elif name == "album_duration_seconds" and isinstance(media, Album):
+ return str(media.duration)
+ elif name == "album_duration_minutes" and isinstance(media, Album):
+ m, s = divmod(media.duration, 60)
+ return f"{m:01d}:{s:02d}"
+
+ # Format playlist durations
+ elif name == "playlist_duration_seconds" and isinstance(media, Album):
+ return str(media.duration)
+ elif name == "playlist_duration_minutes" and isinstance(media, Album):
+ m, s = divmod(media.duration, 60)
+ return f"{m:01d}:{s:02d}"
+
+ # Format year
+ elif name == "album_year":
+ if isinstance(media, Album):
+ return str(media.year)
+ elif isinstance(media, Track):
+ return str(media.album.year)
+ return None
+
+
+def _format_metadata(
+ name: str,
+ media: Track | Album | Playlist | UserPlaylist | Video | Mix,
+ *_args,
+) -> str | None:
+ """Handle metadata-related format strings.
+
+ Args:
+ name (str): The format string name to check.
+ media (Track | Album | Playlist | UserPlaylist | Video | Mix): The media object to extract metadata information from.
+ *_args (Any): Additional arguments (not used).
+
+ Returns:
+ str | None: The formatted metadata or None if the format string is not metadata-related.
+ """
+ if name == "video_quality" and isinstance(media, Video):
+ return media.video_quality
+ elif name == "track_quality" and isinstance(media, Track):
+ return ", ".join(tag for tag in media.media_metadata_tags if tag is not None)
+ elif (name == "track_explicit" and isinstance(media, Track | Video)) or (
+ name == "album_explicit" and isinstance(media, Album)
+ ):
+ return FORMAT_TEMPLATE_EXPLICIT if media.explicit else ""
+ return None
- return result
+
+def _format_volumes(
+ name: str,
+ media: Track | Album | Playlist | UserPlaylist | Video | Mix,
+ *_args,
+) -> str | None:
+ """Handle volume-related format strings.
+
+ Args:
+ name (str): The format string name to check.
+ media (Track | Album | Playlist | UserPlaylist | Video | Mix): The media object to extract volume information from.
+ *_args (Any): Additional arguments (not used).
+
+ Returns:
+ str | None: The formatted volume information or None if the format string is not volume-related.
+ """
+ if name == "album_num_volumes" and isinstance(media, Album):
+ return str(media.num_volumes)
+ elif name == "track_volume_num" and isinstance(media, Track | Video):
+ return str(media.volume_num)
+ elif name == "track_volume_num_optional" and isinstance(media, Track | Video):
+ num_volumes: int = media.album.num_volumes if hasattr(media, "album") else 1
+ return "" if num_volumes == 1 else str(media.volume_num)
+ elif name == "track_volume_num_optional_CD" and isinstance(media, Track | Video):
+ num_volumes: int = media.album.num_volumes if hasattr(media, "album") else 1
+ return "" if num_volumes == 1 else f"CD{media.volume_num!s}"
+ return None
def calculate_number_padding(padding_minimum: int, item_position: int, items_max: int) -> str:
+ """Calculate the padded number string for an item.
+
+ Args:
+ padding_minimum (int): Minimum number of digits for padding.
+ item_position (int): The position of the item.
+ items_max (int): The maximum number of items.
+
+ Returns:
+ str: The padded number string.
+ """
result: str
if items_max > 0:
- count_digits: int = int(math.log10(items_max)) + 1
- count_digits_computed: int = count_digits if count_digits > padding_minimum else padding_minimum
- result = str(item_position).zfill(count_digits_computed)
+ count_digits = max(int(math.log10(items_max)) + 1, padding_minimum)
+ result = str(item_position).zfill(count_digits)
else:
result = str(item_position)
@@ -224,6 +419,15 @@ def calculate_number_padding(padding_minimum: int, item_position: int, items_max
def get_format_template(
media: Track | Album | Playlist | UserPlaylist | Video | Mix | MediaType, settings
) -> str | bool:
+ """Get the format template for a given media type.
+
+ Args:
+ media (Track | Album | Playlist | UserPlaylist | Video | Mix | MediaType): The media object or type.
+ settings: The settings object containing format templates.
+
+ Returns:
+ str | bool: The format template string or False if not found.
+ """
result = False
if isinstance(media, Track) or media == MediaType.TRACK:
@@ -241,83 +445,59 @@ def get_format_template(
def path_file_sanitize(path_file: pathlib.Path, adapt: bool = False, uniquify: bool = False) -> pathlib.Path:
- sanitized_filename: str = path_file.name
- sanitized_path: pathlib.Path = path_file.parent
- result: pathlib.Path
+ """Sanitize a file path to ensure it is valid and optionally make it unique.
- # Sanitize filename and make sure it does not exceed FILENAME_LENGTH_MAX
- try:
- # sanitize_filename can shorten the file name actually
- sanitized_filename = sanitize_filename(
- sanitized_filename, replacement_text="_", validate_after_sanitize=True, platform="auto"
- )
+ Args:
+ path_file (pathlib.Path): The file path to sanitize.
+ adapt (bool, optional): Whether to adapt the path in case of errors. Defaults to False.
+ uniquify (bool, optional): Whether to make the file name unique. Defaults to False.
- # Check if the file extension was removed by shortening the filename length
- if not sanitized_filename.endswith(path_file.suffix):
- # Add the original file extension
- file_suffix: str = FILENAME_SANITIZE_PLACEHOLDER + path_file.suffix
- sanitized_filename = sanitized_filename[: -len(file_suffix)] + file_suffix
- except ValidationError as e:
- if adapt:
- # TODO: Implement proper exception handling and logging.
- # Hacky stuff, since the sanitizing function does not shorten the filename (filename too long)
- if str(e).startswith("[PV1101]"):
- byte_ct: int = len(sanitized_filename.encode(sys.getfilesystemencoding())) - FILENAME_LENGTH_MAX
- sanitized_filename = (
- sanitized_filename[: -byte_ct - len(FILENAME_SANITIZE_PLACEHOLDER) - len(path_file.suffix)]
- + FILENAME_SANITIZE_PLACEHOLDER
- + path_file.suffix
- )
- else:
- raise
- else:
- raise
+ Returns:
+ pathlib.Path: The sanitized file path.
+ """
+ sanitized_filename = sanitize_filename(
+ path_file.name, replacement_text="_", validate_after_sanitize=True, platform="auto"
+ )
- # Sanitize the path.
- # First sanitize sanitize each part of the path. Each part of the path is not allowed to be longer then 'PC_NAME_MAX'.
- sanitized_parts = []
+ if not sanitized_filename.endswith(path_file.suffix):
+ sanitized_filename = (
+ sanitized_filename[: -len(path_file.suffix)] + FILENAME_SANITIZE_PLACEHOLDER + path_file.suffix
+ )
- for part in sanitized_path.parts:
- if part in sanitized_path.anchor:
- sanitized_parts.append(part)
- else:
- sanitized_parts.append(
+ sanitized_path = pathlib.Path(
+ *[
+ (
sanitize_filename(part, replacement_text="_", validate_after_sanitize=True, platform="auto")
+ if part not in path_file.anchor
+ else part
)
+ for part in path_file.parent.parts
+ ]
+ )
- sanitized_path = pathlib.Path(*sanitized_parts)
-
- # Then sanitize the whole path itself. The whole path is not allowed to be longer than 'PC_NAME_MAX'.
try:
- sanitized_path: pathlib.Path = sanitize_filepath(
+ sanitized_path = sanitize_filepath(
sanitized_path, replacement_text="_", validate_after_sanitize=True, platform="auto"
)
except ValidationError as e:
- # If adaption of path is allowed in case of an error set path to HOME.
- if adapt:
- if str(e).startswith("[PV1101]"):
- sanitized_path = pathlib.Path.home()
- else:
- raise
+ if adapt and str(e).startswith("[PV1101]"):
+ sanitized_path = pathlib.Path.home()
else:
raise
result = sanitized_path / sanitized_filename
- # Uniquify
- if uniquify:
- result = path_file_uniquify(result)
-
- return result
+ return path_file_uniquify(result) if uniquify else result
def path_file_uniquify(path_file: pathlib.Path) -> pathlib.Path:
- """Checks whether the file exists, if so it tries to return an unique name suffix.
+ """Ensure a file path is unique by appending a suffix if necessary.
+
+ Args:
+ path_file (pathlib.Path): The file path to uniquify.
- :param path_file: Path to file name which shall be unique.
- :type path_file: pathlib.Path
- :return: Unique file name with path for given input.
- :rtype: pathlib.Path
+ Returns:
+ pathlib.Path: The unique file path.
"""
unique_suffix: str = file_unique_suffix(path_file)
@@ -333,7 +513,16 @@ def path_file_uniquify(path_file: pathlib.Path) -> pathlib.Path:
return path_file
-def file_unique_suffix(path_file: pathlib.Path, seperator: str = "_") -> str:
+def file_unique_suffix(path_file: pathlib.Path, separator: str = "_") -> str:
+ """Generate a unique suffix for a file path.
+
+ Args:
+ path_file (pathlib.Path): The file path to check for uniqueness.
+ separator (str, optional): The separator to use for the suffix. Defaults to "_".
+
+ Returns:
+ str: The unique suffix, or an empty string if not needed.
+ """
threshold_zfill: int = len(str(UNIQUIFY_THRESHOLD))
count: int = 0
path_file_tmp: pathlib.Path = deepcopy(path_file)
@@ -341,46 +530,60 @@ def file_unique_suffix(path_file: pathlib.Path, seperator: str = "_") -> str:
while check_file_exists(path_file_tmp) and count < UNIQUIFY_THRESHOLD:
count += 1
- unique_suffix = seperator + str(count).zfill(threshold_zfill)
+ unique_suffix = separator + str(count).zfill(threshold_zfill)
path_file_tmp = path_file.parent / (path_file.stem + unique_suffix + path_file.suffix)
return unique_suffix
def check_file_exists(path_file: pathlib.Path, extension_ignore: bool = False) -> bool:
+ """Check if a file exists.
+
+ Args:
+ path_file (pathlib.Path): The file path to check.
+ extension_ignore (bool, optional): Whether to ignore the file extension. Defaults to False.
+
+ Returns:
+ bool: True if the file exists, False otherwise.
+ """
if extension_ignore:
path_file_stem: str = pathlib.Path(path_file).stem
path_parent: pathlib.Path = pathlib.Path(path_file).parent
- path_files: [str] = []
+ path_files: list[str] = []
- for extension in AudioExtensions:
- path_files.append(str(path_parent.joinpath(path_file_stem + extension)))
+ path_files.extend(str(path_parent.joinpath(path_file_stem + extension)) for extension in AudioExtensions)
else:
- path_files: [str] = [path_file]
+ path_files: list[str] = [str(path_file)]
- result = bool(sum([[True] if os.path.isfile(_file) else [] for _file in path_files], []))
+ return any(os.path.isfile(_file) for _file in path_files)
- return result
+def resource_path(relative_path: str) -> str:
+ """Get the absolute path to a resource.
-def resource_path(relative_path):
- try:
- base_path = sys._MEIPASS
- except Exception:
- base_path = os.path.abspath(".")
+ Args:
+ relative_path: The relative path to the resource.
+
+ Returns:
+ str: The absolute path to the resource.
+ """
+ # PyInstaller creates a temp folder and stores path in _MEIPASS
+ base_path = getattr(sys, "_MEIPASS", os.path.abspath("."))
return os.path.join(base_path, relative_path)
def url_to_filename(url: str) -> str:
- """Return basename corresponding to url.
- >>> print(url_to_filename('http://example.com/path/to/file%C3%80?opt=1'))
- fileƀ
- >>> print(url_to_filename('http://example.com/slash%2fname')) # '/' in name
- Taken from https://gist.github.com/zed/c2168b9c52b032b5fb7d
- Traceback (most recent call last):
- ...
- ValueError
+ """Convert a URL to a valid filename.
+
+ Args:
+ url (str): The URL to convert.
+
+ Returns:
+ str: The corresponding filename.
+
+ Raises:
+ ValueError: If the URL contains invalid characters for a filename.
"""
urlpath: str = urlsplit(url).path
basename: str = posixpath.basename(unquote(urlpath))