diff options
Diffstat (limited to 'tidal_dl_ng/helper')
| -rw-r--r-- | tidal_dl_ng/helper/hover_manager.py | 187 | ||||
| -rw-r--r-- | tidal_dl_ng/helper/metadata_utils.py | 227 | ||||
| -rw-r--r-- | tidal_dl_ng/helper/tidal.py | 340 |
3 files changed, 749 insertions, 5 deletions
diff --git a/tidal_dl_ng/helper/hover_manager.py b/tidal_dl_ng/helper/hover_manager.py new file mode 100644 index 0000000..3bda020 --- /dev/null +++ b/tidal_dl_ng/helper/hover_manager.py @@ -0,0 +1,187 @@ +"""Hover Event Manager with Debounce Logic. + +This module provides a robust hover event management system for the track list, +implementing debouncing to prevent UI flickering and excessive updates during +rapid mouse movements. + +The HoverManager intercepts mouse movement events and triggers callbacks only +after the mouse has remained stationary over an item for a configurable delay. + +Architecture: + - HoverManager: Main coordinator with debounce timer + - Event filter for capturing mouse movements + - Signal-based callbacks for decoupled communication + +Performance: + - Configurable debounce delay (default: 350ms) + - Automatic cleanup of pending timers + - Thread-safe signal emissions +""" + +from PySide6 import QtCore, QtGui, QtWidgets +from tidalapi import Album, Mix, Playlist, Track, Video +from tidalapi.artist import Artist + +from tidal_dl_ng.helper.gui import get_results_media_item +from tidal_dl_ng.logger import logger_gui + + +class HoverManager(QtCore.QObject): + """Manages hover events with debouncing to prevent UI flickering. + + This class implements an event filter that intercepts mouse movements over + a QTreeView and emits signals after a configurable delay to prevent rapid + updates when the user is just scanning through the list. + + Signals: + s_hover_confirmed (object): Emitted when hover is confirmed after debounce delay. + s_hover_left: Emitted when mouse leaves the tracked widget. + + Attributes: + debounce_delay_ms (int): Delay in milliseconds before confirming hover. + tree_view (QTreeView): The tree view to monitor. + proxy_model (QSortFilterProxyModel): Proxy model for the tree view. + source_model (QStandardItemModel): Source model for the tree view. + debounce_timer (QTimer): Timer for debouncing hover events. + last_hovered_media (Track | Video | Album | None): Last media item hovered. + """ + + # Signals + s_hover_confirmed: QtCore.Signal = QtCore.Signal(object) # Emits media object + s_hover_left: QtCore.Signal = QtCore.Signal() # Emits when hover leaves + + def __init__( + self, + tree_view: QtWidgets.QTreeView, + proxy_model: QtCore.QSortFilterProxyModel, + source_model: QtGui.QStandardItemModel, + debounce_delay_ms: int = 350, + parent: QtCore.QObject | None = None, + ) -> None: + """Initialize the HoverManager. + + Args: + tree_view (QTreeView): The tree view to monitor for hover events. + proxy_model (QSortFilterProxyModel): Proxy model wrapping the source model. + source_model (QStandardItemModel): Source model containing the data. + debounce_delay_ms (int, optional): Debounce delay in milliseconds. Defaults to 350. + parent (QObject | None, optional): Parent QObject. Defaults to None. + """ + super().__init__(parent) + + self.debounce_delay_ms: int = debounce_delay_ms + self.tree_view: QtWidgets.QTreeView = tree_view + self.proxy_model: QtCore.QSortFilterProxyModel = proxy_model + self.source_model: QtGui.QStandardItemModel = source_model + self.last_hovered_media: Track | Video | Album | Mix | Playlist | Artist | None = None + + # Debounce timer (single-shot) + self.debounce_timer: QtCore.QTimer = QtCore.QTimer(self) + self.debounce_timer.setSingleShot(True) + self.debounce_timer.setInterval(self.debounce_delay_ms) + self.debounce_timer.timeout.connect(self._on_debounce_timeout) + + # Track hover state + self.pending_media: Track | Video | Album | Mix | Playlist | Artist | None = None + + # Install event filter on viewport (where mouse events occur) + self.tree_view.viewport().installEventFilter(self) + self.tree_view.viewport().setMouseTracking(True) + + def stop(self) -> None: + """Stop the hover manager and clean up event filters.""" + try: + if self.tree_view: + self.tree_view.viewport().removeEventFilter(self) + self.tree_view = None + except RuntimeError: + pass + if self.timer and self.timer.isActive(): + self.timer.stop() + + def eventFilter(self, watched: QtCore.QObject, event: QtCore.QEvent) -> bool: + """Filter events to detect mouse movements over tree items. + + Args: + watched (QObject): The object being watched. + event (QEvent): The event to filter. + + Returns: + bool: True if event is handled, False otherwise. + """ + if watched == self.tree_view.viewport(): + if event.type() == QtCore.QEvent.Type.MouseMove: + # Cast to QMouseEvent for type safety + mouse_event = QtGui.QMouseEvent(event) + self._handle_mouse_move(mouse_event) + elif event.type() == QtCore.QEvent.Type.Leave: + self._handle_mouse_leave() + + # Don't consume the event - let it propagate + return False + + def _handle_mouse_move(self, event: QtGui.QMouseEvent) -> None: + """Handle mouse move events over the tree view. + + Args: + event (QMouseEvent): The mouse move event. + """ + # Get the index under the cursor + pos = event.position().toPoint() + index = self.tree_view.indexAt(pos) + + if not index.isValid(): + # Mouse is not over a valid item + self._cancel_pending_hover() + return + + try: + # Extract media from the index + media = get_results_media_item(index, self.proxy_model, self.source_model) + + if not media: + self._cancel_pending_hover() + return + + # Check if we're hovering over a different item + if media != self.pending_media: + # Cancel previous timer and start new one + self.pending_media = media + self.debounce_timer.stop() + self.debounce_timer.start() + + except Exception as e: + logger_gui.debug(f"Error extracting media from hover: {e}") + self._cancel_pending_hover() + + def _handle_mouse_leave(self) -> None: + """Handle mouse leaving the tree view.""" + self._cancel_pending_hover() + self.s_hover_left.emit() + + def _cancel_pending_hover(self) -> None: + """Cancel any pending hover confirmation.""" + self.debounce_timer.stop() + self.pending_media = None + + def _on_debounce_timeout(self) -> None: + """Handle debounce timer timeout - hover is confirmed.""" + if self.pending_media: + self.last_hovered_media = self.pending_media + self.s_hover_confirmed.emit(self.pending_media) + # logger_gui.debug(f"Hover confirmed: {getattr(self.pending_media, 'name', 'Unknown')}") + + def reset(self) -> None: + """Reset the hover manager state.""" + self._cancel_pending_hover() + self.last_hovered_media = None + + def set_debounce_delay(self, delay_ms: int) -> None: + """Change the debounce delay. + + Args: + delay_ms (int): New delay in milliseconds. + """ + self.debounce_delay_ms = delay_ms + self.debounce_timer.setInterval(delay_ms) + logger_gui.debug(f"Hover debounce delay set to {delay_ms}ms") diff --git a/tidal_dl_ng/helper/metadata_utils.py b/tidal_dl_ng/helper/metadata_utils.py new file mode 100644 index 0000000..eac69bd --- /dev/null +++ b/tidal_dl_ng/helper/metadata_utils.py @@ -0,0 +1,227 @@ +"""Utilities for extracting and formatting metadata from TIDAL objects.""" + +from contextlib import suppress + + +def _convert_list_to_str(value: list | tuple) -> str: + """Convert list/tuple to comma-separated string.""" + if not value: + return "—" + with suppress(Exception): + return ", ".join([str(x) for x in value]) + return str(value) + + +def _convert_dict_to_str(value: dict) -> str: + """Extract meaningful string from dict.""" + if "name" in value and value["name"]: + return str(value["name"]) + for k in ("label", "title", "genre", "name"): + if k in value and value[k]: + return str(value[k]) + with suppress(Exception): + vals = [str(v) for v in value.values() if v is not None] + if vals: + return ", ".join(vals) + return str(value) + + +def safe_str(value: object) -> str: + """Convert a potentially non-str value into a safe string for display. + + Args: + value: The value to convert. + + Returns: + str: Safe string representation, '—' for None/empty. + """ + with suppress(Exception): + if value is None: + return "—" + if isinstance(value, str): + return value if value != "" else "—" + if isinstance(value, list | tuple): + return _convert_list_to_str(value) + if isinstance(value, dict): + return _convert_dict_to_str(value) + return str(value) + return "—" + + +def _find_in_dict_container(container: dict, names: tuple[str, ...]) -> object | None: + """Search for names in a dict container.""" + for n in names: + if n in container and container[n] is not None: + return container[n] + # Fuzzy key match + keys = list(container.keys()) + for n in names: + for k in keys: + if n.lower() in str(k).lower(): + return container[k] + return None + + +def _fuzzy_scan_attrs(obj: object, names: tuple[str, ...]) -> object | None: + """Fuzzy scan object attributes for matching names.""" + with suppress(Exception): + for k in dir(obj): + kl = k.lower() + for n in names: + if n.lower() in kl: + with suppress(Exception): + val = getattr(obj, k) + if val is not None: + return val + return None + + +def find_attr(obj: object, *names: str) -> object | None: + """Attempt to find an attribute or data key from an object or its internals. + + Args: + obj: The object to inspect. + *names: Attribute/key names to search for. + + Returns: + The found value or None. + """ + # Direct attributes + for n in names: + with suppress(Exception): + if hasattr(obj, n): + val = getattr(obj, n) + if val is not None: + return val + + # Inspect common dict-like internals + for container_name in ("_data", "data", "__dict__"): + with suppress(Exception): + container = getattr(obj, container_name, None) + if isinstance(container, dict): + result = _find_in_dict_container(container, names) + if result is not None: + return result + + # Fuzzy scan of attributes + return _fuzzy_scan_attrs(obj, names) + + +def _scan_dict_recursive(container: dict, key_substrings: list[str]) -> object | None: + """Recursively scan a dict for keys matching any substring.""" + for k, v in container.items(): + kl = str(k).lower() + for s in key_substrings: + if s.lower() in kl and v is not None: + return v + # recurse into nested structures + if isinstance(v, dict): + found = _scan_dict_recursive(v, key_substrings) + if found is not None: + return found + if isinstance(v, list | tuple): + for item in v: + if isinstance(item, dict): + found = _scan_dict_recursive(item, key_substrings) + if found is not None: + return found + return None + + +def search_in_data(obj: object, key_substrings: list[str]) -> object | None: + """Recursively search dict-like internals for keys containing any substring. + + Args: + obj: The object to search. + key_substrings: List of key substrings to search for. + + Returns: + The first matching value found, or None. + """ + # check common containers + for container_name in ("_data", "data", "__dict__"): + with suppress(Exception): + container = getattr(obj, container_name, None) + if isinstance(container, dict): + found = _scan_dict_recursive(container, key_substrings) + if found is not None: + return found + + # as last resort, try obj.__dict__ if available + with suppress(Exception): + d = getattr(obj, "__dict__", None) + if isinstance(d, dict): + return _scan_dict_recursive(d, key_substrings) + + return None + + +def _extract_name_from_dict(item: dict, match_types: tuple[str, ...] | None) -> str | None: + """Extract name from a dict if it matches type filters.""" + if "name" not in item or not item["name"]: + return None + if match_types: + t = item.get("type") or item.get("role") or item.get("credit_type") + if t and any(mt in str(t).lower() for mt in match_types): + return str(item["name"]) + return None + return str(item["name"]) + + +def _extract_name_from_item(item: object) -> str | None: + """Extract name from various item formats.""" + if item is None: + return None + if isinstance(item, str): + return item + if isinstance(item, dict): + # Try common name keys + for k in ("name", "artist", "person"): + if k in item and item[k]: + return str(item[k]) + return None + # Try object attributes + nm = getattr(item, "name", None) or getattr(item, "title", None) + if nm: + return str(nm) + with suppress(Exception): + return str(item) + return None + + +def extract_names_from_mixed(value: object, match_types: tuple[str, ...] | None = None) -> list[str]: + """Normalize various credit-like structures into a list of names. + + Accepts lists of dicts, dicts, strings, or objects. + If match_types provided, only include entries where the type/role matches one of them. + + Args: + value: The value to extract names from. + match_types: Optional tuple of role types to filter by. + + Returns: + List of extracted names. + """ + if value is None: + return [] + if isinstance(value, str): + return [value] + if isinstance(value, dict): + name = _extract_name_from_dict(value, match_types) + if name: + return [name] + vals = [str(v) for v in value.values() if v is not None] + return vals + if isinstance(value, list | tuple): + names: list[str] = [] + for item in value: + if isinstance(item, dict): + name = _extract_name_from_dict(item, match_types) + if name: + names.append(name) + else: + name = _extract_name_from_item(item) + if name: + names.append(name) + return names + return [] diff --git a/tidal_dl_ng/helper/tidal.py b/tidal_dl_ng/helper/tidal.py index 586e196..22ae8b0 100644 --- a/tidal_dl_ng/helper/tidal.py +++ b/tidal_dl_ng/helper/tidal.py @@ -1,5 +1,8 @@ +import contextlib +import os from collections.abc import Callable +import requests from tidalapi import Album, Mix, Playlist, Session, Track, UserPlaylist, Video from tidalapi.artist import Artist, Role from tidalapi.media import MediaMetadataTags, Quality @@ -135,7 +138,7 @@ def search_results_all(session: Session, needle: str, types_media: SearchTypes = def items_results_all( - media_list: [Mix | Playlist | Album | Artist], videos_include: bool = True + session: Session, media_list: [Mix | Playlist | Album | Artist], videos_include: bool = True ) -> [Track | Video | Album]: result: [Track | Video | Album] = [] @@ -149,7 +152,7 @@ def items_results_all( func_get_items_media.append(media_list.items) else: func_get_items_media.append(media_list.tracks) - else: + elif isinstance(media_list, Artist): func_get_items_media.append(media_list.get_albums) func_get_items_media.append(media_list.get_ep_singles) @@ -257,12 +260,20 @@ def instantiate_media( def quality_audio_highest(media: Track | Album) -> Quality: quality: Quality - if MediaMetadataTags.hi_res_lossless in media.media_metadata_tags: + # media_metadata_tags may be missing (Mock objects) or None; use safe getter + tags = getattr(media, "media_metadata_tags", None) + try: + iterable_tags = set(tags) if tags is not None else set() + except Exception: + # If tags is a Mock or non-iterable, fall back to empty set + iterable_tags = set() + + if MediaMetadataTags.hi_res_lossless in iterable_tags: quality = Quality.hi_res_lossless - elif MediaMetadataTags.lossless in media.media_metadata_tags: + elif MediaMetadataTags.lossless in iterable_tags: quality = Quality.high_lossless else: - quality = media.audio_quality + quality = getattr(media, "audio_quality", Quality.low_320k) return quality @@ -272,3 +283,322 @@ def favorite_function_factory(tidal, favorite_item: str): function_list: Callable = getattr(tidal.session.user.favorites, function_name) return function_list + + +def fetch_raw_media_json( + session: Session, media_type: str, media_id: str, country_code: str | None = None, extra_params: dict | None = None +) -> dict | None: + """Fetch raw JSON for a media resource using tidalapi's session.request. + + Args: + session (Session): the tidalapi Session + media_type (str): 'tracks' or 'albums' + media_id (str): id of media + country_code (str | None): optional countryCode param + + Returns: + dict | None: parsed JSON or None if fetch fails + """ + try: + params = {} + # If caller didn't provide a country code, check environment variable + cc = country_code or os.environ.get("TIDAL_COUNTRY") + if cc: + params["countryCode"] = cc + # merge extra params if provided (do not overwrite existing keys unless provided) + if extra_params and isinstance(extra_params, dict): + for k, v in extra_params.items(): + params[k] = v + + # Use session.request.request to call the internal API endpoint + resp = session.request.request("GET", f"{media_type}/{media_id}", params=params) + resp.raise_for_status() + return resp.json() + except requests.exceptions.HTTPError: + return None # Silently ignore HTTP errors + except Exception: + return None # Silently ignore other errors + + +def fetch_raw_track_and_album( + session: Session, + track_id: str, + country_code: str | None = None, + extra_params: dict | None = None, +) -> tuple[dict | None, dict | None]: + """Convenience to fetch raw track JSON and its album JSON (if available). + + Returns a tuple (track_json, album_json). + + Uses TIDAL API parameters to fetch extended metadata. + Note: 'credits' and 'contributors' are NOT available in TIDAL API v2. + Available include values: albums, artists, genres, lyrics, owners, providers, + radio, shares, similarTracks, sourceFile, trackStatistics + """ + # Use valid include parameters according to TIDAL API v2 spec + # https://tidal-music.github.io/tidal-api-reference/ + default_track_params = { + "include": "albums,artists,genres", # Valid parameters per API spec + } + merged_track_params = {**default_track_params, **(extra_params or {})} + + track_json = fetch_raw_media_json( + session, + "tracks", + str(track_id), + country_code=country_code, + extra_params=merged_track_params, + ) + + album_json = None + try: + if isinstance(track_json, dict): + album = track_json.get("album") + album_id = album.get("id") if isinstance(album, dict) else album + if album_id: + # Request extended album metadata + default_album_params = { + "include": "artists,genres", # Valid parameters for albums + } + merged_album_params = {**default_album_params, **(extra_params or {})} + + album_json = fetch_raw_media_json( + session, + "albums", + str(album_id), + country_code=country_code, + extra_params=merged_album_params, + ) + except Exception: + album_json = None + + return track_json, album_json + + +def _normalize_dict_contributors(raw_contributors: dict) -> dict[str, list[str]]: + """Process contributors in dict format: role -> list[{name, ...}].""" + result: dict[str, list[str]] = {} + for role, people in raw_contributors.items(): + if not isinstance(people, list): + continue + names: list[str] = [] + for person in people: + if isinstance(person, dict): + name = person.get("name") + if isinstance(name, str) and name: + names.append(name) + if names: + result[role] = names + return result + + +def _normalize_list_contributors(raw_contributors: list) -> dict[str, list[str]]: + """Process contributors in list format: [{name, role, ...}, ...].""" + result: dict[str, list[str]] = {} + for person in raw_contributors: + if not isinstance(person, dict): + continue + name = person.get("name") + role = person.get("role") + if isinstance(name, str) and name and isinstance(role, str) and role: + result.setdefault(role, []).append(name) + return result + + +def _normalize_contributors(raw_contributors: object) -> dict[str, list[str]]: + """Normalize various possible contributor JSON shapes into role -> list[str] names. + + The TIDAL API has used at least two shapes historically: + - dict role -> list[ {"name": str, ...} ] + - list[ {"name": str, "role": str, ...} ] + + We accept both and ignore malformed entries. + """ + if isinstance(raw_contributors, dict): + return _normalize_dict_contributors(raw_contributors) + if isinstance(raw_contributors, list): + return _normalize_list_contributors(raw_contributors) + return {} + + +def _extract_bpm_from_track(track_json: dict) -> int | None: + """Extract BPM from track JSON.""" + bpm = track_json.get("bpm") + if isinstance(bpm, int | float): + return int(round(bpm)) + if isinstance(bpm, str): + with contextlib.suppress(ValueError): + return int(round(float(bpm))) + return None + + +def _process_credits_contributors(credits_list: list) -> dict[str, list[str]]: + """Process credits API v2 format and return contributors by role.""" + role_mapping = { + "producer": "producer", + "producers": "producer", + "composer": "composer", + "composers": "composer", + "lyricist": "lyricist", + "lyricists": "lyricist", + "writer": "composer", + "writers": "composer", + } + result: dict[str, list[str]] = {} + for credit in credits_list: + if not isinstance(credit, dict): + continue + credit_type = credit.get("type", "").lower() + contributors = credit.get("contributors", []) + role = role_mapping.get(credit_type, credit_type) + if isinstance(contributors, list): + for contributor in contributors: + if isinstance(contributor, dict): + name = contributor.get("name") + if name: + result.setdefault(role, []).append(name) + return result + + +def _extract_track_contributors(track_json: dict) -> dict[str, list[str]]: + """Extract contributors from track JSON.""" + # Try credits first (API v2) + track_credits = track_json.get("credits") + if track_credits and isinstance(track_credits, list): + contributors = _process_credits_contributors(track_credits) + if contributors: + return contributors + # Fallback to old format + raw_contributors = track_json.get("contributors") + if raw_contributors: + return _normalize_contributors(raw_contributors) + return {} + + +def _process_genre_item(g: object) -> str | None: + """Extract genre name from various formats.""" + if isinstance(g, str) and g: + return g + if isinstance(g, dict): + name = g.get("name") + if isinstance(name, str) and name: + return name + return None + + +def _deduplicate_genres(genres: list[str]) -> list[str]: + """Deduplicate genres while preserving order.""" + seen: set[str] = set() + unique: list[str] = [] + for g in genres: + if g not in seen: + seen.add(g) + unique.append(g) + return unique + + +def _extract_album_label_genres(album_json: dict) -> tuple[str, list[str]]: + """Extract label and genres from album JSON.""" + # Label + label = album_json.get("label") or album_json.get("recordLabel") + label_str = label if isinstance(label, str) else "" + + # Genres + raw_genres = album_json.get("genres") or album_json.get("genre") + genres: list[str] = [] + + if isinstance(raw_genres, list): + for g in raw_genres: + genre = _process_genre_item(g) + if genre: + genres.append(genre) + elif isinstance(raw_genres, str) and raw_genres: + genres.append(raw_genres) + else: + genre = _process_genre_item(raw_genres) + if genre: + genres.append(genre) + + # Deduplicate while preserving order + if genres: + return label_str, _deduplicate_genres(genres) + return label_str, [] + + +def _extract_album_contributors(album_json: dict) -> dict[str, list[str]]: + """Extract contributors from album JSON.""" + # Try credits first (API v2) + album_credits = album_json.get("credits") + if album_credits and isinstance(album_credits, list): + contributors = _process_credits_contributors(album_credits) + if contributors: + return contributors + # Fallback to old format + raw_contributors = album_json.get("contributors") + if raw_contributors: + return _normalize_contributors(raw_contributors) + return {} + + +def parse_track_and_album_extras( + track_json: dict | None, + album_json: dict | None, +) -> dict: + """Extract extra metadata from raw TIDAL JSON for a track and its album. + + Returned dict keys (all optional, may be missing or empty): + - bpm: int | None + - label: str + - genres: list[str] + - contributors_by_role: dict[str, list[str]] + """ + + extras: dict = { + "bpm": None, + "label": "", + "genres": [], + "contributors_by_role": {}, + } + + # Extract from track + if isinstance(track_json, dict): + extras["bpm"] = _extract_bpm_from_track(track_json) + extras["contributors_by_role"] = _extract_track_contributors(track_json) + + # Extract from album + if isinstance(album_json, dict): + label, genres = _extract_album_label_genres(album_json) + extras["label"] = label + extras["genres"] = genres + + # If we did not get track-level contributors, try album-level + if not extras["contributors_by_role"]: + extras["contributors_by_role"] = _extract_album_contributors(album_json) + + return extras + + +def extract_contributor_names( + contributors_by_role: dict[str, list[str]] | None, + role: str, + delimiter: str = ", ", +) -> str: + """Return a delimited string of contributor names for a given role. + + If the role is not present or has no names, returns an empty string. + Role matching is case-insensitive. + """ + if not contributors_by_role: + return "" + + # Normalise keys to lowercase for robust lookups. + role_lc = role.lower() + for r, names in contributors_by_role.items(): + if not isinstance(r, str): + continue + if r.lower() == role_lc and isinstance(names, list): + filtered = [n for n in names if isinstance(n, str) and n] + if filtered: + return delimiter.join(filtered) + + return "" |