131 lines
3.9 KiB
Python
131 lines
3.9 KiB
Python
"""
|
|
Qobuz search and download via the qobuz-dl library.
|
|
|
|
search_by_type returns: [{"text": "display string", "url": "https://play.qobuz.com/TYPE/ID"}]
|
|
handle_url downloads an album or artist URL directly.
|
|
For top-N artist albums we fetch artist meta and pick N most recent albums.
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
from dataclasses import dataclass
|
|
|
|
from qobuz_dl.core import QobuzDL
|
|
|
|
import config
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class SearchItem:
|
|
text: str # formatted display string from qobuz-dl
|
|
url: str # https://play.qobuz.com/TYPE/ID
|
|
|
|
|
|
@dataclass
|
|
class AlbumItem:
|
|
id: str
|
|
title: str
|
|
artist: str
|
|
released_at: int # unix timestamp, 0 if unknown
|
|
|
|
|
|
@dataclass
|
|
class SearchResults:
|
|
artists: list[SearchItem]
|
|
albums: list[SearchItem]
|
|
|
|
|
|
def _make_client() -> QobuzDL:
|
|
client = QobuzDL(
|
|
directory=str(config.DOWNLOAD_DIR),
|
|
quality=config.QOBUZ_QUALITY,
|
|
smart_discography=True, # skip singles/EPs for artist downloads
|
|
)
|
|
client.get_tokens() # fetches app_id and secrets from the Qobuz web app
|
|
client.initialize_client(
|
|
email=config.QOBUZ_EMAIL,
|
|
pwd=config.QOBUZ_PASSWORD,
|
|
app_id=client.app_id,
|
|
secrets=client.secrets,
|
|
)
|
|
return client
|
|
|
|
|
|
_client: QobuzDL | None = None
|
|
_client_lock = asyncio.Lock()
|
|
|
|
|
|
async def get_client() -> QobuzDL:
|
|
global _client
|
|
async with _client_lock:
|
|
if _client is None:
|
|
loop = asyncio.get_running_loop()
|
|
_client = await loop.run_in_executor(None, _make_client)
|
|
return _client
|
|
|
|
|
|
def _search_sync(client: QobuzDL, query: str) -> SearchResults:
|
|
raw_artists = client.search_by_type(query, "artist", limit=config.SEARCH_ARTIST_LIMIT) or []
|
|
raw_albums = client.search_by_type(query, "album", limit=config.SEARCH_ALBUM_LIMIT) or []
|
|
return SearchResults(
|
|
artists=[SearchItem(text=r["text"], url=r["url"]) for r in raw_artists],
|
|
albums=[SearchItem(text=r["text"], url=r["url"]) for r in raw_albums],
|
|
)
|
|
|
|
|
|
async def search(query: str) -> SearchResults:
|
|
client = await get_client()
|
|
loop = asyncio.get_running_loop()
|
|
return await loop.run_in_executor(None, _search_sync, client, query)
|
|
|
|
|
|
def _artist_id_from_url(url: str) -> str:
|
|
"""Extract numeric ID from https://play.qobuz.com/artist/ID"""
|
|
return url.rstrip("/").split("/")[-1]
|
|
|
|
|
|
def _get_artist_albums_sync(client: QobuzDL, artist_url: str) -> list[AlbumItem]:
|
|
artist_id = _artist_id_from_url(artist_url)
|
|
albums: list[AlbumItem] = []
|
|
for chunk in client.client.get_artist_meta(artist_id):
|
|
for item in chunk.get("albums", {}).get("items", []):
|
|
albums.append(AlbumItem(
|
|
id=str(item.get("id", "")),
|
|
title=item.get("title", "Unknown"),
|
|
artist=item.get("artist", {}).get("name", "Unknown"),
|
|
released_at=int(item.get("released_at", 0) or 0),
|
|
))
|
|
# Sort newest first
|
|
albums.sort(key=lambda a: a.released_at, reverse=True)
|
|
return albums
|
|
|
|
|
|
async def get_artist_albums(artist_url: str) -> list[AlbumItem]:
|
|
client = await get_client()
|
|
loop = asyncio.get_running_loop()
|
|
return await loop.run_in_executor(None, _get_artist_albums_sync, client, artist_url)
|
|
|
|
|
|
def _download_url_sync(client: QobuzDL, url: str) -> None:
|
|
client.handle_url(url)
|
|
|
|
|
|
def _download_album_id_sync(client: QobuzDL, album_id: str) -> None:
|
|
client.download_from_id(album_id, album=True)
|
|
|
|
|
|
async def download_url(url: str) -> None:
|
|
"""Download an album or artist discography by Qobuz URL."""
|
|
client = await get_client()
|
|
loop = asyncio.get_running_loop()
|
|
await loop.run_in_executor(None, _download_url_sync, client, url)
|
|
|
|
|
|
async def download_album_id(album_id: str) -> None:
|
|
"""Download a single album by its Qobuz ID."""
|
|
client = await get_client()
|
|
loop = asyncio.get_running_loop()
|
|
await loop.run_in_executor(None, _download_album_id_sync, client, album_id)
|