Lyra/worker/lyra_worker/main.py

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_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"}


def _set_config(conn, key: str, value: str) -> None:
    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 _run_scan(conn, resolver, probe, browser, dest_root: str = DEST_ROOT) -> None:
    result = scan_library(conn, resolver, probe, browser, dest_root)
    _set_config(conn, "scan.result", f"imported {result.imported}, skipped {result.skipped}")
    _set_config(conn, "scan.requested", "false")
    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
    adapters = build_adapters(get_config(conn))
    resolver = build_resolver()
    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)
            mcfg = MonitorConfig.from_config(config)
            now = time.monotonic()
            if mcfg.enabled and now - last_tick >= MONITOR_TICK_SECONDS:
                try:
                    sweep(conn, browser, mcfg)
                except Exception as e:  # a monitor error must never kill the worker
                    print(f"worker: monitor sweep failed: {e}", flush=True)
                    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)
                except Exception as e:  # a scan error must never kill the worker
                    print(f"worker: library scan failed: {e}", flush=True)
                    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)
                continue
            print(f"worker: claimed job {job_id}", flush=True)
            try:
                run_pipeline(conn, job_id, adapters, resolver=resolver, tagger=tagger,
                             upgrade_cutoff=mcfg.quality_cutoff)
            except Exception as e:  # one bad job must not take down the worker loop
                print(f"worker: pipeline failed for job {job_id}: {e}", flush=True)
                conn.rollback()
            if mcfg.enabled:
                try:
                    reconcile(conn, job_id, mcfg)
                except Exception as e:
                    print(f"worker: reconcile failed for job {job_id}: {e}", flush=True)
                    conn.rollback()
            print(f"worker: finished job {job_id}", flush=True)
    finally:
        conn.close()


if __name__ == "__main__":
    run_forever()