Jonathan
4907d0d5a4
Adds the AudioProbe seam (probe.py, _probe.py MutagenProbe with a lazy mutagen import, FakeAudioProbe) and scan_library, which walks /music, resolves each album against MusicBrainz (falling back to embedded tags when the folder name doesn't resolve), and idempotently records matched albums as have (LibraryItem), monitored (MonitoredRelease, fulfilled), and followed (WatchedArtist, autoMonitorFuture=false). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
93 lines
2.8 KiB
Python
93 lines
2.8 KiB
Python
from typing import Callable
|
|
|
|
from lyra_worker.browser import ArtistHit, ReleaseGroupInfo
|
|
from lyra_worker.probe import ProbeResult
|
|
from lyra_worker.types import Candidate, DownloadResult, MBTarget, Quality
|
|
|
|
_QUALITIES = {
|
|
"qobuz": Quality(fmt="FLAC", lossless=True, bit_depth=24, sample_rate=96000),
|
|
"soulseek": Quality(fmt="FLAC", lossless=True, bit_depth=16, sample_rate=44100),
|
|
"youtube": Quality(fmt="OPUS", lossless=False, bitrate_kbps=160),
|
|
}
|
|
|
|
|
|
class _BaseFake:
|
|
name = "fake"
|
|
tier = 99
|
|
|
|
def __init__(self, matches: bool = True):
|
|
self._matches = matches
|
|
|
|
def health(self) -> bool:
|
|
return True
|
|
|
|
def search(self, target: MBTarget) -> list[Candidate]:
|
|
if not self._matches:
|
|
return []
|
|
tracks = target.track_count or 10
|
|
return [
|
|
Candidate(
|
|
source=self.name,
|
|
source_ref=f"{self.name}:{target.artist}:{target.album}",
|
|
matched_artist=target.artist,
|
|
matched_album=target.album,
|
|
quality=_QUALITIES.get(self.name, _QUALITIES["youtube"]),
|
|
track_count=tracks,
|
|
source_tier=self.tier,
|
|
)
|
|
]
|
|
|
|
def download(
|
|
self, candidate: Candidate, dest: str, on_progress: Callable[[float], None]
|
|
) -> DownloadResult:
|
|
for pct in (0.25, 0.5, 0.75, 1.0):
|
|
on_progress(pct)
|
|
return DownloadResult(ok=True, path=dest, track_count=candidate.track_count)
|
|
|
|
|
|
class FakeQobuz(_BaseFake):
|
|
name = "qobuz"
|
|
tier = 0
|
|
|
|
|
|
class FakeSoulseek(_BaseFake):
|
|
name = "soulseek"
|
|
tier = 1
|
|
|
|
|
|
class FakeYouTube(_BaseFake):
|
|
name = "youtube"
|
|
tier = 2
|
|
|
|
|
|
class FailingAdapter(_BaseFake):
|
|
name = "qobuz" # pretends to be a real source that then fails to download
|
|
tier = 0
|
|
|
|
def download(
|
|
self, candidate: Candidate, dest: str, on_progress: Callable[[float], None]
|
|
) -> DownloadResult:
|
|
return DownloadResult(ok=False, error="simulated download failure")
|
|
|
|
|
|
class FakeMbBrowser:
|
|
"""In-memory MbBrowser for tests. `releases` maps artist mbid -> [ReleaseGroupInfo]."""
|
|
|
|
def __init__(self, artists=None, releases=None):
|
|
self._artists = list(artists or [])
|
|
self._releases = dict(releases or {})
|
|
|
|
def search_artist(self, name: str) -> list[ArtistHit]:
|
|
return list(self._artists)
|
|
|
|
def browse_release_groups(self, artist_mbid: str) -> list[ReleaseGroupInfo]:
|
|
return list(self._releases.get(artist_mbid, []))
|
|
|
|
|
|
class FakeAudioProbe:
|
|
"""In-memory AudioProbe for tests."""
|
|
def __init__(self, artist="", album="", fmt="FLAC", quality_class=3):
|
|
self._r = ProbeResult(artist=artist, album=album, fmt=fmt, quality_class=quality_class)
|
|
def probe(self, path: str) -> ProbeResult:
|
|
return self._r
|