Jonathan
66 lines
2.1 KiB
Python
66 lines
2.1 KiB
Python
from typing import Callable, Protocol
|
|
|
|
from lyra_worker.types import Candidate, DownloadResult, MBTarget, Quality
|
|
|
|
|
|
class QobuzClient(Protocol):
|
|
def is_configured(self) -> bool:
|
|
...
|
|
|
|
def search_album(self, artist: str, album: str) -> list[dict]:
|
|
"""Album matches: dicts with source_ref, title, artist, track_count, bit_depth, sampling_rate(kHz)."""
|
|
...
|
|
|
|
def download(self, source_ref: str, dest: str, on_progress: Callable[[float], None]) -> dict:
|
|
"""Download into dest; return {track_count, path}. Raise on failure."""
|
|
...
|
|
|
|
|
|
def _quality(bit_depth: int, sampling_rate_khz: float) -> Quality:
|
|
return Quality(
|
|
fmt="FLAC",
|
|
lossless=True,
|
|
bit_depth=bit_depth,
|
|
sample_rate=int(sampling_rate_khz * 1000),
|
|
)
|
|
|
|
|
|
class QobuzAdapter:
|
|
name = "qobuz"
|
|
tier = 0
|
|
|
|
def __init__(self, client: QobuzClient):
|
|
self._client = client
|
|
|
|
def health(self) -> bool:
|
|
return self._client.is_configured()
|
|
|
|
def search(self, target: MBTarget) -> list[Candidate]:
|
|
candidates: list[Candidate] = []
|
|
for r in self._client.search_album(target.artist, target.album):
|
|
candidates.append(
|
|
Candidate(
|
|
source="qobuz",
|
|
source_ref=r["source_ref"],
|
|
matched_artist=r.get("artist", ""),
|
|
matched_album=r.get("title", ""),
|
|
quality=_quality(r.get("bit_depth", 16), r.get("sampling_rate", 44.1)),
|
|
track_count=r.get("track_count", 0) or 0,
|
|
source_tier=self.tier,
|
|
)
|
|
)
|
|
return candidates
|
|
|
|
def download(
|
|
self, candidate: Candidate, dest: str, on_progress: Callable[[float], None]
|
|
) -> DownloadResult:
|
|
try:
|
|
info = self._client.download(candidate.source_ref, dest, on_progress)
|
|
except Exception as e:
|
|
return DownloadResult(ok=False, error=str(e))
|
|
return DownloadResult(
|
|
ok=True,
|
|
path=info.get("path", dest),
|
|
track_count=info.get("track_count", candidate.track_count),
|
|
)
|