huge refactor of plugin system
This commit is contained in:
File diff suppressed because it is too large
Load Diff
-327
@@ -1,327 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Dict, List, Optional, Set
|
||||
|
||||
from .base import API, ApiError
|
||||
from SYS.logger import debug, debug_panel
|
||||
|
||||
DEFAULT_BASE_URL = "https://tidal-api.binimum.org"
|
||||
|
||||
|
||||
def _debug_payload_summary(title: str, payload: Any) -> None:
|
||||
try:
|
||||
keys = list(payload.keys()) if isinstance(payload, dict) else []
|
||||
except Exception:
|
||||
keys = []
|
||||
|
||||
preview = ", ".join(str(key) for key in keys[:8]) if keys else "<none>"
|
||||
if keys and len(keys) > 8:
|
||||
preview = f"{preview}, ..."
|
||||
|
||||
try:
|
||||
payload_size = len(str(payload))
|
||||
except Exception:
|
||||
payload_size = "<unknown>"
|
||||
|
||||
debug_panel(
|
||||
title,
|
||||
[
|
||||
("type", type(payload).__name__),
|
||||
("keys", preview),
|
||||
("size", payload_size),
|
||||
],
|
||||
border_style="cyan",
|
||||
)
|
||||
|
||||
|
||||
def stringify(value: Any) -> str:
|
||||
"""Helper to ensure we have a stripped string or empty."""
|
||||
return str(value or "").strip()
|
||||
|
||||
|
||||
def extract_artists(item: Dict[str, Any]) -> List[str]:
|
||||
"""Extract list of artist names from a Tidal-style metadata dict."""
|
||||
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]:
|
||||
"""Create a set of searchable tags from track metadata."""
|
||||
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 parse_track_item(item: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Parse raw Tidal track data into a clean, flat dictionary.
|
||||
|
||||
Extracts core fields: id, title, duration, Track:, url, artist name, and album title.
|
||||
"""
|
||||
if not isinstance(item, dict):
|
||||
return {}
|
||||
|
||||
# Handle the "data" wrapper if present
|
||||
data = item.get("data") if isinstance(item.get("data"), dict) else item
|
||||
|
||||
artist_name = ""
|
||||
artist_obj = data.get("artist")
|
||||
if isinstance(artist_obj, dict):
|
||||
artist_name = stringify(artist_obj.get("name"))
|
||||
if not artist_name:
|
||||
artists = extract_artists(data)
|
||||
if artists:
|
||||
artist_name = artists[0]
|
||||
|
||||
album_title = ""
|
||||
album_obj = data.get("album")
|
||||
if isinstance(album_obj, dict):
|
||||
album_title = stringify(album_obj.get("title"))
|
||||
if not album_title and isinstance(data.get("album"), str):
|
||||
album_title = stringify(data.get("album"))
|
||||
|
||||
return {
|
||||
"id": data.get("id"),
|
||||
"title": stringify(data.get("title")),
|
||||
"duration": data.get("duration"),
|
||||
"Track:": data.get("trackNumber"),
|
||||
"url": stringify(data.get("url")),
|
||||
"artist": artist_name,
|
||||
"album": album_title,
|
||||
}
|
||||
|
||||
|
||||
def coerce_duration_seconds(value: Any) -> Optional[int]:
|
||||
"""Attempt to extracts seconds from various Tidal duration formats."""
|
||||
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:
|
||||
# Suspect milliseconds
|
||||
num = num / 1000.0
|
||||
return int(round(num))
|
||||
except Exception:
|
||||
continue
|
||||
return None
|
||||
|
||||
|
||||
class TidalApiError(ApiError):
|
||||
"""Raised when the Tidal API returns an error or malformed response."""
|
||||
|
||||
|
||||
class Tidal(API):
|
||||
"""Client for the Tidal (Tidal) API endpoints.
|
||||
|
||||
This client communicates with the configured Tidal backend to retrieve
|
||||
track metadata, manifests, search results, and lyrics.
|
||||
"""
|
||||
|
||||
def __init__(self, base_url: str = DEFAULT_BASE_URL, *, timeout: float = 10.0) -> None:
|
||||
super().__init__(base_url, timeout)
|
||||
|
||||
def search(self, params: Dict[str, str]) -> Dict[str, Any]:
|
||||
usable = {k: v for k, v in (params or {}).items() if v}
|
||||
search_keys = [key for key in ("s", "a", "v", "p") if usable.get(key)]
|
||||
if not search_keys:
|
||||
raise TidalApiError("One of s/a/v/p is required for /search/")
|
||||
if len(search_keys) > 1:
|
||||
first = search_keys[0]
|
||||
usable = {first: usable[first]}
|
||||
return self._get_json("search/", params=usable)
|
||||
|
||||
def track(self, track_id: int, *, quality: Optional[str] = None) -> Dict[str, Any]:
|
||||
try:
|
||||
track_int = int(track_id)
|
||||
except Exception as exc:
|
||||
raise TidalApiError(f"track_id must be int-compatible: {exc}") from exc
|
||||
if track_int <= 0:
|
||||
raise TidalApiError("track_id must be positive")
|
||||
|
||||
p: Dict[str, Any] = {"id": track_int}
|
||||
if quality:
|
||||
p["quality"] = str(quality)
|
||||
return self._get_json("track/", params=p)
|
||||
|
||||
def info(self, track_id: int) -> Dict[str, Any]:
|
||||
"""Fetch and parse core track metadata (id, title, artist, album, duration, etc)."""
|
||||
try:
|
||||
track_int = int(track_id)
|
||||
except Exception as exc:
|
||||
raise TidalApiError(f"track_id must be int-compatible: {exc}") from exc
|
||||
if track_int <= 0:
|
||||
raise TidalApiError("track_id must be positive")
|
||||
|
||||
raw = self._get_json("info/", params={"id": track_int})
|
||||
return parse_track_item(raw)
|
||||
|
||||
def album(self, album_id: int) -> Dict[str, Any]:
|
||||
"""Fetch album details, including track list when provided by the backend."""
|
||||
try:
|
||||
album_int = int(album_id)
|
||||
except Exception as exc:
|
||||
raise TidalApiError(f"album_id must be int-compatible: {exc}") from exc
|
||||
if album_int <= 0:
|
||||
raise TidalApiError("album_id must be positive")
|
||||
|
||||
return self._get_json("album/", params={"id": album_int})
|
||||
|
||||
def lyrics(self, track_id: int) -> Dict[str, Any]:
|
||||
"""Fetch lyrics (including subtitles/LRC) for a track."""
|
||||
try:
|
||||
track_int = int(track_id)
|
||||
except Exception as exc:
|
||||
raise TidalApiError(f"track_id must be int-compatible: {exc}") from exc
|
||||
if track_int <= 0:
|
||||
raise TidalApiError("track_id must be positive")
|
||||
|
||||
return self._get_json("lyrics/", params={"id": track_int})
|
||||
|
||||
def get_full_track_metadata(self, track_id: int) -> Dict[str, Any]:
|
||||
"""
|
||||
Orchestrate fetching all details for a track:
|
||||
1. Base info (/info/)
|
||||
2. Playback/Quality info (/track/)
|
||||
3. Lyrics (/lyrics/)
|
||||
4. Derived tags
|
||||
"""
|
||||
try:
|
||||
track_int = int(track_id)
|
||||
except Exception as exc:
|
||||
raise TidalApiError(f"track_id must be int-compatible: {exc}") from exc
|
||||
|
||||
# 1. Fetch info (metadata) - fetch raw to ensure all fields are available for merging
|
||||
info_resp = self._get_json("info/", params={"id": track_int})
|
||||
_debug_payload_summary("API.Tidal info", info_resp)
|
||||
info_data = info_resp.get("data") if isinstance(info_resp, dict) else info_resp
|
||||
if not isinstance(info_data, dict) or "id" not in info_data:
|
||||
info_data = info_resp if isinstance(info_resp, dict) and "id" in info_resp else {}
|
||||
|
||||
# 2. Fetch track (manifest/bit depth)
|
||||
track_resp = self.track(track_id)
|
||||
_debug_payload_summary("API.Tidal track", track_resp)
|
||||
# Note: track() method in this class currently returns raw JSON, so we handle it similarly.
|
||||
track_data = track_resp.get("data") if isinstance(track_resp, dict) else track_resp
|
||||
if not isinstance(track_data, dict):
|
||||
track_data = track_resp if isinstance(track_resp, dict) else {}
|
||||
|
||||
# 3. Fetch lyrics
|
||||
lyrics_data = {}
|
||||
try:
|
||||
lyr_resp = self.lyrics(track_id)
|
||||
_debug_payload_summary("API.Tidal lyrics", lyr_resp)
|
||||
lyrics_data = lyr_resp.get("lyrics") or lyr_resp if isinstance(lyr_resp, dict) else {}
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Merged data for tags and parsing
|
||||
merged_md = {}
|
||||
if isinstance(info_data, dict):
|
||||
merged_md.update(info_data)
|
||||
if isinstance(track_data, dict):
|
||||
merged_md.update(track_data)
|
||||
|
||||
# Derived tags and normalized/parsed info
|
||||
tags = build_track_tags(merged_md)
|
||||
parsed_info = parse_track_item(merged_md)
|
||||
|
||||
# Structure for return
|
||||
res = {
|
||||
"metadata": merged_md,
|
||||
"parsed": parsed_info,
|
||||
"tags": list(tags),
|
||||
"lyrics": lyrics_data,
|
||||
}
|
||||
debug_panel(
|
||||
"API.Tidal full track metadata",
|
||||
[
|
||||
("track_id", track_int),
|
||||
("metadata_keys", len(merged_md)),
|
||||
("tags", len(tags)),
|
||||
("has_lyrics", bool(lyrics_data)),
|
||||
("result_keys", ", ".join(res.keys())),
|
||||
],
|
||||
border_style="cyan",
|
||||
)
|
||||
return res
|
||||
|
||||
|
||||
# Legacy alias for TidalApiClient
|
||||
TidalApiClient = Tidal
|
||||
HifiApiClient = Tidal
|
||||
-1165
File diff suppressed because it is too large
Load Diff
-71
@@ -1,71 +0,0 @@
|
||||
"""Library of Congress (LoC) API helpers.
|
||||
|
||||
This module currently focuses on the LoC JSON API endpoint for the
|
||||
Chronicling America collection.
|
||||
|
||||
Docs:
|
||||
- https://www.loc.gov/apis/
|
||||
- https://www.loc.gov/apis/json-and-yaml/
|
||||
|
||||
The LoC JSON API does not require an API key.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
from .base import API, ApiError
|
||||
|
||||
|
||||
class LOCError(ApiError):
|
||||
pass
|
||||
|
||||
|
||||
class LOCClient(API):
|
||||
"""Minimal client for the public LoC JSON API."""
|
||||
|
||||
def __init__(self, *, base_url: str = "https://www.loc.gov", timeout: float = 20.0):
|
||||
super().__init__(base_url=base_url, timeout=timeout)
|
||||
|
||||
def search_chronicling_america(
|
||||
self,
|
||||
query: str,
|
||||
*,
|
||||
start: int = 1,
|
||||
count: int = 25,
|
||||
extra_params: Optional[Dict[str,
|
||||
Any]] = None,
|
||||
) -> Dict[str,
|
||||
Any]:
|
||||
"""Search the Chronicling America collection via LoC JSON API.
|
||||
|
||||
Args:
|
||||
query: Free-text query.
|
||||
start: 1-based start index (LoC uses `sp`).
|
||||
count: Results per page (LoC uses `c`).
|
||||
extra_params: Additional LoC API params (facets, filters, etc.).
|
||||
|
||||
Returns:
|
||||
Parsed JSON response.
|
||||
"""
|
||||
|
||||
q = str(query or "").strip()
|
||||
if not q:
|
||||
return {
|
||||
"results": []
|
||||
}
|
||||
|
||||
params: Dict[str,
|
||||
Any] = {
|
||||
"q": q,
|
||||
"fo": "json",
|
||||
"c": int(count) if int(count) > 0 else 25,
|
||||
"sp": int(start) if int(start) > 0 else 1,
|
||||
}
|
||||
if extra_params:
|
||||
for k, v in extra_params.items():
|
||||
if v is None:
|
||||
continue
|
||||
params[str(k)] = v
|
||||
|
||||
return self._get_json("/collections/chronicling-america/", params)
|
||||
@@ -1,158 +0,0 @@
|
||||
"""PodcastIndex.org API integration.
|
||||
|
||||
Docs: https://podcastindex-org.github.io/docs-api/
|
||||
|
||||
Authentication headers required for most endpoints:
|
||||
- User-Agent
|
||||
- X-Auth-Key
|
||||
- X-Auth-Date
|
||||
- Authorization (sha1(apiKey + apiSecret + unixTime))
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import time
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from .base import API, ApiError
|
||||
|
||||
|
||||
class PodcastIndexError(ApiError):
|
||||
pass
|
||||
|
||||
|
||||
def build_auth_headers(
|
||||
api_key: str,
|
||||
api_secret: str,
|
||||
*,
|
||||
unix_time: Optional[int] = None,
|
||||
user_agent: str = "downlow/1.0",
|
||||
) -> Dict[str, str]:
|
||||
"""Build PodcastIndex auth headers.
|
||||
|
||||
The API expects X-Auth-Date to be the current UTC unix epoch time
|
||||
(integer string), and Authorization to be the SHA-1 hex digest of
|
||||
`api_key + api_secret + X-Auth-Date`.
|
||||
"""
|
||||
|
||||
key = str(api_key or "").strip()
|
||||
secret = str(api_secret or "").strip()
|
||||
if not key or not secret:
|
||||
raise PodcastIndexError("PodcastIndex api key/secret are required")
|
||||
|
||||
ts = int(unix_time if unix_time is not None else time.time())
|
||||
ts_str = str(ts)
|
||||
|
||||
token = hashlib.sha1((key + secret + ts_str).encode("utf-8")).hexdigest()
|
||||
|
||||
return {
|
||||
"User-Agent": str(user_agent or "downlow/1.0"),
|
||||
"X-Auth-Key": key,
|
||||
"X-Auth-Date": ts_str,
|
||||
"Authorization": token,
|
||||
}
|
||||
|
||||
|
||||
class PodcastIndexClient(API):
|
||||
def __init__(
|
||||
self,
|
||||
api_key: str,
|
||||
api_secret: str,
|
||||
*,
|
||||
base_url: str = "https://api.podcastindex.org/api/1.0",
|
||||
user_agent: str = "downlow/1.0",
|
||||
timeout: float = 30.0,
|
||||
):
|
||||
super().__init__(base_url=base_url, timeout=timeout)
|
||||
self.api_key = str(api_key or "").strip()
|
||||
self.api_secret = str(api_secret or "").strip()
|
||||
self.user_agent = str(user_agent or "downlow/1.0")
|
||||
|
||||
if not self.api_key or not self.api_secret:
|
||||
raise PodcastIndexError("PodcastIndex api key/secret are required")
|
||||
|
||||
def _get(self, path: str, *, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
||||
headers = build_auth_headers(
|
||||
self.api_key,
|
||||
self.api_secret,
|
||||
user_agent=self.user_agent,
|
||||
)
|
||||
return self._get_json(path, params=params, headers=headers)
|
||||
|
||||
def search_byterm(self, query: str, *, max_results: int = 10) -> List[Dict[str, Any]]:
|
||||
q = str(query or "").strip()
|
||||
if not q:
|
||||
return []
|
||||
|
||||
max_int = int(max_results)
|
||||
if max_int < 1:
|
||||
max_int = 1
|
||||
if max_int > 1000:
|
||||
max_int = 1000
|
||||
|
||||
data = self._get(
|
||||
"search/byterm",
|
||||
params={
|
||||
"q": q,
|
||||
"max": max_int,
|
||||
},
|
||||
)
|
||||
|
||||
feeds = data.get("feeds")
|
||||
return feeds if isinstance(feeds, list) else []
|
||||
|
||||
def episodes_byfeedid(self, feed_id: int | str, *, max_results: int = 50) -> List[Dict[str, Any]]:
|
||||
"""List recent episodes for a feed by its PodcastIndex feed id."""
|
||||
try:
|
||||
feed_id_int = int(feed_id)
|
||||
except Exception:
|
||||
feed_id_int = None
|
||||
if feed_id_int is None or feed_id_int <= 0:
|
||||
return []
|
||||
|
||||
max_int = int(max_results)
|
||||
if max_int < 1:
|
||||
max_int = 1
|
||||
if max_int > 1000:
|
||||
max_int = 1000
|
||||
|
||||
data = self._get(
|
||||
"episodes/byfeedid",
|
||||
params={
|
||||
"id": feed_id_int,
|
||||
"max": max_int,
|
||||
},
|
||||
)
|
||||
|
||||
items = data.get("items")
|
||||
if isinstance(items, list):
|
||||
return items
|
||||
episodes = data.get("episodes")
|
||||
return episodes if isinstance(episodes, list) else []
|
||||
|
||||
def episodes_byfeedurl(self, feed_url: str, *, max_results: int = 50) -> List[Dict[str, Any]]:
|
||||
"""List recent episodes for a feed by its RSS URL."""
|
||||
url = str(feed_url or "").strip()
|
||||
if not url:
|
||||
return []
|
||||
|
||||
max_int = int(max_results)
|
||||
if max_int < 1:
|
||||
max_int = 1
|
||||
if max_int > 1000:
|
||||
max_int = 1000
|
||||
|
||||
data = self._get(
|
||||
"episodes/byfeedurl",
|
||||
params={
|
||||
"url": url,
|
||||
"max": max_int,
|
||||
},
|
||||
)
|
||||
|
||||
items = data.get("items")
|
||||
if isinstance(items, list):
|
||||
return items
|
||||
episodes = data.get("episodes")
|
||||
return episodes if isinstance(episodes, list) else []
|
||||
Reference in New Issue
Block a user