153 lines
4.8 KiB
Python
153 lines
4.8 KiB
Python
import os
|
|
from typing import Any, Dict, Optional
|
|
import threading
|
|
|
|
from deezspot.libutils import LibrespotClient
|
|
|
|
# Config helpers to resolve active credentials
|
|
from routes.utils.celery_config import get_config_params
|
|
from routes.utils.credentials import get_spotify_blob_path
|
|
|
|
|
|
# -------- Shared Librespot client (process-wide) --------
|
|
|
|
_shared_client: Optional[LibrespotClient] = None
|
|
_shared_blob_path: Optional[str] = None
|
|
_client_lock = threading.RLock()
|
|
|
|
|
|
def _resolve_blob_path() -> str:
|
|
cfg = get_config_params() or {}
|
|
active_account = cfg.get("spotify")
|
|
if not active_account:
|
|
raise RuntimeError("Active Spotify account not set in configuration.")
|
|
blob_path = get_spotify_blob_path(active_account)
|
|
abs_path = os.path.abspath(str(blob_path))
|
|
if not os.path.isfile(abs_path):
|
|
raise FileNotFoundError(
|
|
f"Spotify credentials blob not found for account '{active_account}' at {abs_path}"
|
|
)
|
|
return abs_path
|
|
|
|
|
|
def get_client() -> LibrespotClient:
|
|
"""
|
|
Return a shared LibrespotClient instance initialized from the active account blob.
|
|
Re-initializes if the active account changes.
|
|
"""
|
|
global _shared_client, _shared_blob_path
|
|
with _client_lock:
|
|
desired_blob = _resolve_blob_path()
|
|
if _shared_client is None or _shared_blob_path != desired_blob:
|
|
try:
|
|
if _shared_client is not None:
|
|
_shared_client.close()
|
|
except Exception:
|
|
pass
|
|
_shared_client = LibrespotClient(stored_credentials_path=desired_blob)
|
|
_shared_blob_path = desired_blob
|
|
return _shared_client
|
|
|
|
|
|
# -------- Thin wrapper API (programmatic use) --------
|
|
|
|
|
|
def create_client(credentials_path: str) -> LibrespotClient:
|
|
"""
|
|
Create a LibrespotClient from a librespot-generated credentials.json file.
|
|
"""
|
|
abs_path = os.path.abspath(credentials_path)
|
|
if not os.path.isfile(abs_path):
|
|
raise FileNotFoundError(f"Credentials file not found: {abs_path}")
|
|
return LibrespotClient(stored_credentials_path=abs_path)
|
|
|
|
|
|
def close_client(client: LibrespotClient) -> None:
|
|
"""
|
|
Dispose a LibrespotClient instance.
|
|
"""
|
|
client.close()
|
|
|
|
|
|
def get_track(client: LibrespotClient, track_in: str) -> Dict[str, Any]:
|
|
"""Fetch a track object."""
|
|
return client.get_track(track_in)
|
|
|
|
|
|
def get_album(
|
|
client: LibrespotClient, album_in: str, include_tracks: bool = False
|
|
) -> Dict[str, Any]:
|
|
"""Fetch an album object; optionally include expanded tracks."""
|
|
return client.get_album(album_in, include_tracks=include_tracks)
|
|
|
|
|
|
def get_artist(client: LibrespotClient, artist_in: str) -> Dict[str, Any]:
|
|
"""Fetch an artist object."""
|
|
return client.get_artist(artist_in)
|
|
|
|
|
|
def get_playlist(
|
|
client: LibrespotClient, playlist_in: str, expand_items: bool = False
|
|
) -> Dict[str, Any]:
|
|
"""Fetch a playlist object; optionally expand track items to full track objects."""
|
|
return client.get_playlist(playlist_in, expand_items=expand_items)
|
|
|
|
|
|
def get_spotify_info(
|
|
spotify_id: str,
|
|
info_type: str,
|
|
limit: int = 50,
|
|
offset: int = 0,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Thin, typed wrapper around common Spotify info lookups using the shared client.
|
|
|
|
Currently supports:
|
|
- "artist_discography": returns a paginated view over the artist's releases
|
|
combined across album_group/single_group/compilation_group/appears_on_group.
|
|
|
|
Returns a mapping with at least: items, total, limit, offset.
|
|
Also includes a truthy "next" key when more pages are available.
|
|
"""
|
|
client = get_client()
|
|
|
|
if info_type == "artist_discography":
|
|
artist = client.get_artist(spotify_id)
|
|
all_items = []
|
|
for key in (
|
|
"album_group",
|
|
"single_group",
|
|
"compilation_group",
|
|
"appears_on_group",
|
|
):
|
|
grp = artist.get(key)
|
|
if isinstance(grp, list):
|
|
all_items.extend(grp)
|
|
elif isinstance(grp, dict):
|
|
items = grp.get("items") or grp.get("releases") or []
|
|
if isinstance(items, list):
|
|
all_items.extend(items)
|
|
total = len(all_items)
|
|
start = max(0, offset or 0)
|
|
page_limit = max(1, limit or 50)
|
|
end = min(total, start + page_limit)
|
|
page_items = all_items[start:end]
|
|
has_more = end < total
|
|
return {
|
|
"items": page_items,
|
|
"total": total,
|
|
"limit": page_limit,
|
|
"offset": start,
|
|
"next": bool(has_more),
|
|
}
|
|
|
|
raise ValueError(f"Unsupported info_type: {info_type}")
|
|
|
|
|
|
def get_playlist_metadata(playlist_id: str) -> Dict[str, Any]:
|
|
"""
|
|
Fetch playlist metadata using the shared client without expanding items.
|
|
"""
|
|
client = get_client()
|
|
return get_playlist(client, playlist_id, expand_items=False)
|