From a15f4ae8bbe0850e87bd30ecab01a64155924887 Mon Sep 17 00:00:00 2001 From: Jonathan Date: Sun, 12 Jul 2026 00:22:34 +0200 Subject: [PATCH] feat(discovery): build_similarity_sources + worker sweep/trigger wiring Adds registry.build_similarity_sources() (always constructs ListenBrainzSource; reachability is a per-run health() concern, not a startup gate) and wires discovery into the worker's main loop: a scheduled sweep gated on discover.enabled + DISCOVER_TICK_SECONDS, and a one-shot discover.requested trigger mirroring the existing scan.requested pattern. Also hardens run_discovery's health-check pass: a source whose health() raises is now logged and skipped instead of aborting the whole sweep (accepted finding from the Task-3 review). Co-Authored-By: Claude Opus 4.8 (1M context) --- worker/lyra_worker/discovery.py | 15 +++++++++- worker/lyra_worker/main.py | 34 ++++++++++++++++++++++- worker/lyra_worker/registry.py | 8 ++++++ worker/tests/test_discovery.py | 15 ++++++++++ worker/tests/test_discovery_trigger.py | 35 ++++++++++++++++++++++++ worker/tests/test_similarity_registry.py | 14 ++++++++++ 6 files changed, 119 insertions(+), 2 deletions(-) create mode 100644 worker/tests/test_discovery_trigger.py create mode 100644 worker/tests/test_similarity_registry.py diff --git a/worker/lyra_worker/discovery.py b/worker/lyra_worker/discovery.py index a32e2fe..d4d153b 100644 --- a/worker/lyra_worker/discovery.py +++ b/worker/lyra_worker/discovery.py @@ -48,6 +48,19 @@ class DiscoveryResult: albums: int +def _healthy_sources(sources): + """Return sources whose health() is truthy; a source whose health() raises + is logged and skipped, never aborting the sweep.""" + live = [] + for s in sources: + try: + if s.health(): + live.append(s) + except Exception as e: # a bad health check must not abort the sweep + print(f"worker: discovery source {getattr(s, 'name', '?')} health check failed: {e}", flush=True) + return live + + def _seed_artists(conn: psycopg.Connection, cfg: DiscoveryConfig): with conn.cursor() as cur: cur.execute( @@ -134,7 +147,7 @@ def run_discovery(conn: psycopg.Connection, sources: list[SimilaritySource], """Aggregate similar artists across the library's seed artists; upsert suggestions.""" seeds = _seed_artists(conn, cfg) followed = _followed_mbids(conn) - live = [s for s in sources if s.health()] + live = _healthy_sources(sources) agg: dict[str, dict] = {} for seed_mbid, _seed_name in seeds: diff --git a/worker/lyra_worker/main.py b/worker/lyra_worker/main.py index 71e13c3..bdf827f 100644 --- a/worker/lyra_worker/main.py +++ b/worker/lyra_worker/main.py @@ -3,14 +3,19 @@ import time from lyra_worker.claim import claim_next from lyra_worker.config import get_config from lyra_worker.db import wait_for_db +from lyra_worker.discovery import DiscoveryConfig, run_discovery from lyra_worker.library import clear_staging_root from lyra_worker.monitor import MonitorConfig, reconcile, sweep from lyra_worker.pipeline import run_pipeline -from lyra_worker.registry import build_adapters, build_browser, build_probe, build_resolver, build_tagger +from lyra_worker.registry import ( + build_adapters, build_browser, build_probe, build_resolver, + build_similarity_sources, build_tagger, +) from lyra_worker.scan import scan_library IDLE_SLEEP = 2.0 MONITOR_TICK_SECONDS = 60.0 +DISCOVER_TICK_SECONDS = 300.0 # check the discovery interval every 5 min (interval itself is hours) DEST_ROOT = "/music" # must match run_pipeline's default library root _TRUE = {"1", "true", "yes", "on"} @@ -32,6 +37,14 @@ def _run_scan(conn, resolver, probe, browser, dest_root: str = DEST_ROOT) -> Non print(f"worker: library scan done — {result.imported} imported, {result.skipped} skipped", flush=True) +def _run_discovery(conn, sources, browser, config=None) -> None: + cfg = DiscoveryConfig.from_config(config if config is not None else get_config(conn)) + result = run_discovery(conn, sources, browser, cfg) + _set_config(conn, "discover.result", f"artists {result.artists}, albums {result.albums}") + _set_config(conn, "discover.requested", "false") + print(f"worker: discovery done — {result.artists} artists, {result.albums} albums", flush=True) + + def run_forever() -> None: conn = wait_for_db() clear_staging_root(DEST_ROOT) # sweep any staging dirs orphaned by a prior crash @@ -40,9 +53,11 @@ def run_forever() -> None: tagger = build_tagger() browser = build_browser() probe = build_probe() + similarity_sources = build_similarity_sources(get_config(conn)) print(f"worker: {len(adapters)} adapter(s) enabled: {[a.name for a in adapters]}", flush=True) print("worker: waiting for jobs", flush=True) last_tick = 0.0 + last_discover_tick = 0.0 try: while True: config = get_config(conn) @@ -56,6 +71,15 @@ def run_forever() -> None: conn.rollback() last_tick = now + dcfg = DiscoveryConfig.from_config(config) + if dcfg.enabled and now - last_discover_tick >= DISCOVER_TICK_SECONDS: + try: + _run_discovery(conn, similarity_sources, browser, config) + except Exception as e: # a discovery error must never kill the worker + print(f"worker: discovery sweep failed: {e}", flush=True) + conn.rollback() + last_discover_tick = now + if str(config.get("scan.requested", "")).strip().lower() in _TRUE: try: _run_scan(conn, resolver, probe, browser) @@ -64,6 +88,14 @@ def run_forever() -> None: conn.rollback() _set_config(conn, "scan.requested", "false") + if str(config.get("discover.requested", "")).strip().lower() in _TRUE: + try: + _run_discovery(conn, similarity_sources, browser, config) + except Exception as e: # a discovery error must never kill the worker + print(f"worker: discovery run failed: {e}", flush=True) + conn.rollback() + _set_config(conn, "discover.requested", "false") + job_id = claim_next(conn) if job_id is None: time.sleep(IDLE_SLEEP) diff --git a/worker/lyra_worker/registry.py b/worker/lyra_worker/registry.py index 6eed9b5..f6ec332 100644 --- a/worker/lyra_worker/registry.py +++ b/worker/lyra_worker/registry.py @@ -12,6 +12,8 @@ from lyra_worker.adapters.youtube import YouTubeAdapter from lyra_worker.browser import MbBrowser from lyra_worker.probe import AudioProbe from lyra_worker.resolver import MbResolver +from lyra_worker.similarity._listenbrainz import ListenBrainzSource +from lyra_worker.similarity.base import SimilaritySource from lyra_worker.tagger import Tagger @@ -47,3 +49,9 @@ def build_browser() -> MbBrowser: def build_probe() -> AudioProbe: """The audio probe used by the library scan.""" return MutagenProbe() + + +def build_similarity_sources(config: dict) -> list[SimilaritySource]: + """Similarity sources for discovery. ListenBrainz needs no credentials, so it is + always constructed; per-run reachability is checked via each source's health().""" + return [ListenBrainzSource(base_url=config.get("discover.listenBrainzUrl") or "")] diff --git a/worker/tests/test_discovery.py b/worker/tests/test_discovery.py index 47ae187..991d59f 100644 --- a/worker/tests/test_discovery.py +++ b/worker/tests/test_discovery.py @@ -72,6 +72,21 @@ def test_unhealthy_source_contributes_nothing(conn): assert _artist_rows(conn) == [] +def test_source_with_raising_health_is_skipped_not_fatal(conn): + insert_watched_artist(conn, mbid="s1", name="Seed") + + class BoomSource: + name = "boom" + def health(self): + raise RuntimeError("boom") + def similar_artists(self, mbid): + return [] + + good = FakeSimilaritySource(similar={"s1": [SimilarArtist("c1", "X", 0.9)]}) + result = run_discovery(conn, [BoomSource(), good], FakeMbBrowser(), DiscoveryConfig()) + assert result.artists == 1 # good source still processed; boom skipped, no crash + + def test_config_from_config_parses_and_defaults(): cfg = DiscoveryConfig.from_config({"discover.enabled": "true", "discover.maxSeeds": "10", "discover.minScore": "0.3"}) diff --git a/worker/tests/test_discovery_trigger.py b/worker/tests/test_discovery_trigger.py new file mode 100644 index 0000000..c0c7c9e --- /dev/null +++ b/worker/tests/test_discovery_trigger.py @@ -0,0 +1,35 @@ +from lyra_worker.adapters.fakes import FakeMbBrowser, FakeSimilaritySource +from lyra_worker.main import _run_discovery +from lyra_worker.similarity.base import SimilarArtist +from tests.conftest import insert_watched_artist + + +def _set(conn, key, value): + with conn.cursor() as cur: + cur.execute( + 'INSERT INTO "Config"(key, value, secret, "updatedAt") VALUES (%s, %s, false, now()) ' + 'ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, "updatedAt" = now()', + (key, value), + ) + conn.commit() + + +def _get(conn, key): + with conn.cursor() as cur: + cur.execute('SELECT value FROM "Config" WHERE key = %s', (key,)) + row = cur.fetchone() + return row[0] if row else None + + +def test_run_discovery_writes_result_and_clears_flag(conn): + insert_watched_artist(conn, mbid="s1", name="Seed") + _set(conn, "discover.requested", "true") + src = FakeSimilaritySource(similar={"s1": [SimilarArtist("c1", "Cand", 0.9)]}) + + _run_discovery(conn, [src], FakeMbBrowser()) + + assert _get(conn, "discover.requested") == "false" # flag cleared + assert "artists" in (_get(conn, "discover.result") or "") # summary written + with conn.cursor() as cur: + cur.execute('SELECT count(*) FROM "DiscoverySuggestion" WHERE kind = \'artist\'') + assert cur.fetchone()[0] == 1 diff --git a/worker/tests/test_similarity_registry.py b/worker/tests/test_similarity_registry.py new file mode 100644 index 0000000..cb805e9 --- /dev/null +++ b/worker/tests/test_similarity_registry.py @@ -0,0 +1,14 @@ +from lyra_worker.registry import build_similarity_sources +from lyra_worker.similarity._listenbrainz import ListenBrainzSource + + +def test_build_returns_listenbrainz_by_default(): + sources = build_similarity_sources({}) + assert len(sources) == 1 + assert isinstance(sources[0], ListenBrainzSource) + assert sources[0].name == "listenbrainz" + + +def test_base_url_override_from_config(): + src = build_similarity_sources({"discover.listenBrainzUrl": "http://lb.local"})[0] + assert src._base == "http://lb.local"