Jonathan
84 lines
3.2 KiB
Python
84 lines
3.2 KiB
Python
import asyncio
|
|
import os
|
|
from typing import Callable
|
|
|
|
|
|
class StreamripClient:
|
|
"""Real Qobuz client backed by streamrip (v2, async → sync via asyncio.run).
|
|
|
|
NOT unit-tested offline; see test_qobuz_live.py. streamrip is imported lazily
|
|
inside the async helpers so importing/constructing this class stays offline.
|
|
"""
|
|
|
|
def __init__(self, config: dict):
|
|
self._email = config.get("qobuz.email", "")
|
|
self._password = config.get("qobuz.password", "")
|
|
|
|
def is_configured(self) -> bool:
|
|
return bool(self._email and self._password)
|
|
|
|
def _make_config(self, download_folder: str | None = None):
|
|
from streamrip.config import Config
|
|
|
|
cfg = Config.defaults()
|
|
cfg.session.qobuz.use_auth_token = False
|
|
cfg.session.qobuz.email_or_userid = self._email
|
|
cfg.session.qobuz.password_or_token = self._password
|
|
cfg.session.qobuz.quality = 3
|
|
if download_folder is not None:
|
|
cfg.session.downloads.folder = download_folder
|
|
return cfg
|
|
|
|
def search_album(self, artist: str, album: str) -> list[dict]:
|
|
return asyncio.run(self._search(f"{artist} {album}"))
|
|
|
|
async def _search(self, query: str) -> list[dict]:
|
|
from streamrip.client import QobuzClient
|
|
|
|
client = QobuzClient(self._make_config())
|
|
try:
|
|
await client.login()
|
|
pages = await client.search("album", query, limit=5)
|
|
finally:
|
|
await client.session.close()
|
|
results: list[dict] = []
|
|
for page in pages:
|
|
for item in page.get("albums", {}).get("items", []):
|
|
results.append(
|
|
{
|
|
"source_ref": str(item.get("id", "")),
|
|
"title": item.get("title", ""),
|
|
"artist": (item.get("artist") or {}).get("name", ""),
|
|
"track_count": item.get("tracks_count", 0) or 0,
|
|
"bit_depth": item.get("maximum_bit_depth", 16) or 16,
|
|
"sampling_rate": item.get("maximum_sampling_rate", 44.1) or 44.1,
|
|
}
|
|
)
|
|
return results
|
|
|
|
def download(self, source_ref: str, dest: str, on_progress: Callable[[float], None]) -> dict:
|
|
return asyncio.run(self._download(source_ref, dest, on_progress))
|
|
|
|
async def _download(self, album_id: str, dest: str, on_progress: Callable[[float], None]) -> dict:
|
|
os.makedirs(dest, exist_ok=True)
|
|
from streamrip.client import QobuzClient
|
|
from streamrip.db import Database, Dummy
|
|
from streamrip.media import PendingAlbum
|
|
|
|
cfg = self._make_config(download_folder=dest)
|
|
client = QobuzClient(cfg)
|
|
try:
|
|
await client.login()
|
|
on_progress(0.0)
|
|
db = Database(downloads=Dummy(), failed=Dummy())
|
|
pending = PendingAlbum(album_id, client, cfg, db)
|
|
album = await pending.resolve()
|
|
if album is None:
|
|
raise RuntimeError(f"could not resolve Qobuz album {album_id}")
|
|
await album.rip()
|
|
track_count = len(getattr(album, "tracks", []) or [])
|
|
on_progress(1.0)
|
|
finally:
|
|
await client.session.close()
|
|
return {"track_count": track_count, "path": dest}
|