k

Provider/tidal_shared.py

@@ -0,0 +1,109 @@
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional, Set
+
+
+def stringify(value: Any) -> str:
+    text = str(value or "").strip()
+    return text
+
+
+def extract_artists(item: Dict[str, Any]) -> List[str]:
+    names: List[str] = []
+    artists = item.get("artists")
+    if isinstance(artists, list):
+        for artist in artists:
+            if isinstance(artist, dict):
+                name = stringify(artist.get("name"))
+                if name and name not in names:
+                    names.append(name)
+    if not names:
+        primary = item.get("artist")
+        if isinstance(primary, dict):
+            name = stringify(primary.get("name"))
+            if name:
+                names.append(name)
+    return names
+
+
+def build_track_tags(metadata: Dict[str, Any]) -> Set[str]:
+    tags: Set[str] = {"tidal"}
+
+    audio_quality = stringify(metadata.get("audioQuality"))
+    if audio_quality:
+        tags.add(f"quality:{audio_quality.lower()}")
+
+    media_md = metadata.get("mediaMetadata")
+    if isinstance(media_md, dict):
+        tag_values = media_md.get("tags") or []
+        for tag in tag_values:
+            if isinstance(tag, str):
+                candidate = tag.strip()
+                if candidate:
+                    tags.add(candidate.lower())
+
+    title_text = stringify(metadata.get("title"))
+    if title_text:
+        tags.add(f"title:{title_text}")
+
+    artists = extract_artists(metadata)
+    for artist in artists:
+        artist_clean = stringify(artist)
+        if artist_clean:
+            tags.add(f"artist:{artist_clean}")
+
+    album_title = ""
+    album_obj = metadata.get("album")
+    if isinstance(album_obj, dict):
+        album_title = stringify(album_obj.get("title"))
+    else:
+        album_title = stringify(metadata.get("album"))
+    if album_title:
+        tags.add(f"album:{album_title}")
+
+    track_no_val = metadata.get("trackNumber") or metadata.get("track_number")
+    if track_no_val is not None:
+        try:
+            track_int = int(track_no_val)
+            if track_int > 0:
+                tags.add(f"track:{track_int}")
+        except Exception:
+            track_text = stringify(track_no_val)
+            if track_text:
+                tags.add(f"track:{track_text}")
+
+    return tags
+
+
+def coerce_duration_seconds(value: Any) -> Optional[int]:
+    candidates = [value]
+    try:
+        if isinstance(value, dict):
+            for key in (
+                "duration",
+                "durationSeconds",
+                "duration_sec",
+                "duration_ms",
+                "durationMillis",
+            ):
+                if key in value:
+                    candidates.append(value.get(key))
+    except Exception:
+        pass
+
+    for cand in candidates:
+        try:
+            if cand is None:
+                continue
+            text = str(cand).strip()
+            if text.lower().endswith("ms"):
+                text = text[:-2].strip()
+            num = float(text)
+            if num <= 0:
+                continue
+            if num > 10_000:
+                num = num / 1000.0
+            return int(round(num))
+        except Exception:
+            continue
+    return None