Jonathan
00b3dddca6
Qobuz's email/password API login returns 401 even with valid credentials (a known Qobuz limitation, unaffected by streamrip version). Add settings fields for a Qobuz user ID + auth token (token stored encrypted); when both are present StreamripClient uses use_auth_token=True, else falls back to email/password. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
106 lines
4.4 KiB
Python
106 lines
4.4 KiB
Python
import asyncio
|
|
import hashlib
|
|
import os
|
|
from typing import Callable
|
|
|
|
|
|
def _hashed_password(password: str) -> str:
|
|
"""streamrip's Qobuz email/password login expects the password as an MD5 hex digest."""
|
|
return hashlib.md5(password.encode("utf-8")).hexdigest()
|
|
|
|
|
|
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", "")
|
|
self._user_id = config.get("qobuz.user_id", "")
|
|
self._token = config.get("qobuz.auth_token", "")
|
|
|
|
def is_configured(self) -> bool:
|
|
# either a user_id + auth token, or an email + password
|
|
return bool((self._user_id and self._token) or (self._email and self._password))
|
|
|
|
def _make_config(self, download_folder: str | None = None):
|
|
from streamrip.config import Config
|
|
|
|
cfg = Config.defaults()
|
|
q = cfg.session.qobuz
|
|
if self._user_id and self._token:
|
|
# token auth: more reliable than email/password, which Qobuz often rejects
|
|
q.use_auth_token = True
|
|
q.email_or_userid = self._user_id
|
|
q.password_or_token = self._token
|
|
else:
|
|
q.use_auth_token = False
|
|
q.email_or_userid = self._email
|
|
q.password_or_token = _hashed_password(self._password)
|
|
q.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:
|
|
try:
|
|
await client.login()
|
|
except Exception as e: # never let streamrip's credential-bearing error escape
|
|
raise RuntimeError(f"Qobuz login failed ({type(e).__name__}) — check credentials") from None
|
|
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:
|
|
try:
|
|
await client.login()
|
|
except Exception as e: # never let streamrip's credential-bearing error escape
|
|
raise RuntimeError(f"Qobuz login failed ({type(e).__name__}) — check credentials") from None
|
|
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}
|