conjurer/navidrome_bridge/bridge.py

#!/usr/bin/env python3
"""Navidrome -> betoniarka request bridge.

EXTERNAL COMPONENT. This does not import anything from Conjurer and does not
talk to the Discord bot ("Kondziu") or its API. It speaks two protocols only:

    Navidrome  <--(Subsonic API, read-only + optional unstar)--  bridge
    betoniarka <--(/request_radio_file, or the request playlist)--  bridge

The one-click trigger is Navidrome's own star/love button: star the track you
are listening to and it lands in the radio's request queue. Navidrome's plugin
system cannot add a button to its UI (see README), so the built-in star is used
as the trigger instead - which has the pleasant side effect of working in every
Subsonic client, phone apps included.

Both sides index the same mp3 files under different mount points, so every path
is translated between the two roots before being handed to the radio.
"""
import hashlib
import json
import logging
import os
import posixpath
import re
import secrets
import sys
import time
from pathlib import Path

import requests

LOG = logging.getLogger("navidrome-bridge")


def _env(name: str, default: str = "") -> str:
    return os.getenv(name, default)


def _env_bool(name: str, default: bool = False) -> bool:
    raw = os.getenv(name)
    if raw is None:
        return default
    return raw.strip().lower() in ("1", "true", "yes", "on")


# ----------------------------------------------------------------- config
NAVIDROME_URL = _env("NAVIDROME_URL", "http://localhost:4533").rstrip("/")
NAVIDROME_USER = _env("NAVIDROME_USER")
NAVIDROME_PASSWORD = _env("NAVIDROME_PASSWORD")

# The same library seen from two places. Navidrome reports a path; whatever
# prefix it carries is swapped for the radio's mount point.
NAVIDROME_LIBRARY_ROOT = _env("NAVIDROME_LIBRARY_ROOT", "/music").rstrip("/")
BETONIARKA_LIBRARY_ROOT = _env("BETONIARKA_LIBRARY_ROOT", "/srv/betoniarka/music").rstrip("/")

BETONIARKA_URL = _env("BETONIARKA_URL", "http://localhost:5005").rstrip("/")
CONJURER_API_KEY = _env("CONJURER_API_KEY")

# "api"     - POST /request_radio_file. Works over the network, but betoniarka
#             re-finds the track from keywords, so it is a fuzzy match.
# "dropbox" - write the exact path straight into the spool file that
#             radio_conjurer.liq drains. Exact, and involves neither betoniarka
#             nor the bot: just this file and Liquidsoap. ("playlist" is
#             accepted as an alias for this mode.)
MODE = _env("BRIDGE_MODE", "api").strip().lower()
if MODE == "playlist":
    MODE = "dropbox"

# The editable drop box itself. Point this at the exact file the liq reads -
# note the name is singular, request.playlist.
REQUEST_PLAYLIST = Path(_env("BETONIARKA_REQUEST_PLAYLIST", "/srv/betoniarka/data/request.playlist"))
# Whether to confirm the translated path really exists before queueing it.
# "auto" (default) checks only when the library happens to be mounted here, so
# a drop-box-only setup is not forced to mount the whole music collection.
VERIFY_FILE_EXISTS = _env("BRIDGE_VERIFY_FILE_EXISTS", "auto").strip().lower()

POLL_SECONDS = int(_env("BRIDGE_POLL_SECONDS", "15"))
STATE_FILE = Path(_env("BRIDGE_STATE_FILE", "/data/queued.json"))
# Un-star after queueing, turning the heart into a fire-and-forget "send to
# radio" button. Off by default: most people use stars as actual favourites.
UNSTAR_AFTER_QUEUE = _env_bool("BRIDGE_UNSTAR_AFTER_QUEUE", False)
# Only act on tracks starred from now on, so enabling the bridge does not
# dump a lifetime of favourites into the radio on first run.
SKIP_EXISTING_ON_FIRST_RUN = _env_bool("BRIDGE_SKIP_EXISTING_ON_FIRST_RUN", True)
DRY_RUN = _env_bool("BRIDGE_DRY_RUN", False)

SUBSONIC_CLIENT = "conjurer-navidrome-bridge"
SUBSONIC_VERSION = "1.16.1"
HTTP_TIMEOUT = 30


# ------------------------------------------------------------- subsonic io
def _auth_params() -> dict:
    """Subsonic salted-token auth - the password itself never goes on the wire."""
    salt = secrets.token_hex(8)
    token = hashlib.md5(  # noqa: S324 - mandated by the Subsonic protocol
        (NAVIDROME_PASSWORD + salt).encode("utf-8")
    ).hexdigest()
    return {
        "u": NAVIDROME_USER,
        "t": token,
        "s": salt,
        "v": SUBSONIC_VERSION,
        "c": SUBSONIC_CLIENT,
        "f": "json",
    }


def _subsonic(endpoint: str, extra: dict = None) -> dict:
    params = _auth_params()
    if extra:
        params.update(extra)
    response = requests.get(
        f"{NAVIDROME_URL}/rest/{endpoint}", params=params, timeout=HTTP_TIMEOUT
    )
    response.raise_for_status()
    payload = response.json().get("subsonic-response", {})
    if payload.get("status") != "ok":
        error = payload.get("error", {})
        raise RuntimeError(
            f"Subsonic {endpoint} failed: {error.get('code')} {error.get('message')}"
        )
    return payload


def fetch_starred() -> list:
    """Return the starred songs as a list of dicts (may be empty)."""
    payload = _subsonic("getStarred2")
    songs = payload.get("starred2", {}).get("song", [])
    # Subsonic collapses a single-element list into an object in some clients;
    # normalise so callers can always iterate.
    if isinstance(songs, dict):
        songs = [songs]
    return songs


def unstar(song_id: str) -> None:
    _subsonic("unstar", {"id": song_id})


# ------------------------------------------------------- path translation
def translate_path(navidrome_path: str) -> str:
    """Map a Navidrome-side path onto the radio's mount point.

    Navidrome may report either an absolute path inside its own library root or
    a path relative to it, so both are accepted: any leading library root is
    stripped and the remainder is re-rooted at the radio's mount point.
    """
    path = navidrome_path.replace("\\", "/").strip()
    if NAVIDROME_LIBRARY_ROOT and path.startswith(NAVIDROME_LIBRARY_ROOT):
        path = path[len(NAVIDROME_LIBRARY_ROOT):]
    return posixpath.join(BETONIARKA_LIBRARY_ROOT, path.lstrip("/"))


def keywords_for(radio_path: str) -> list:
    """Distinctive path tokens for betoniarka's keyword search.

    Only word characters are kept: the radio feeds each token straight into a
    regex, so punctuation would either break the pattern or fail to match its
    own sanitised library tokens.
    """
    relative = radio_path
    if relative.startswith(BETONIARKA_LIBRARY_ROOT):
        relative = relative[len(BETONIARKA_LIBRARY_ROOT):]
    tokens, seen = [], set()
    for token in re.findall(r"\w+", relative, re.UNICODE):
        low = token.lower()
        if low == "mp3" or len(token) < 2 or low in seen:
            continue
        seen.add(low)
        tokens.append(token)
    return tokens


# ------------------------------------------------------------ radio io
def _radio_headers() -> dict:
    return {"X-Conjurer-Api-Key": CONJURER_API_KEY} if CONJURER_API_KEY else {}


def queue_via_api(radio_path: str) -> None:
    keywords = keywords_for(radio_path)
    if not keywords:
        raise RuntimeError(f"no usable keywords for {radio_path}")
    # betoniarka's wyszukaj() drops lista_slow[0] - it expects the Discord
    # command word there - so a sentinel takes that slot.
    payload = {"lista_slow": ["navidrome-bridge"] + keywords}
    response = requests.post(
        f"{BETONIARKA_URL}/request_radio_file",
        json=payload,
        headers=_radio_headers(),
        timeout=HTTP_TIMEOUT,
    )
    response.raise_for_status()


def _should_verify() -> bool:
    if VERIFY_FILE_EXISTS in ("1", "true", "yes", "on"):
        return True
    if VERIFY_FILE_EXISTS in ("0", "false", "no", "off"):
        return False
    # "auto": only meaningful when the library is actually mounted here.
    return Path(BETONIARKA_LIBRARY_ROOT).is_dir()


def queue_via_dropbox(radio_path: str) -> None:
    """Append the exact path to the spool file Liquidsoap drains.

    radio_conjurer.liq treats this file as a drop box, not as a playlist:

        text = file.lines(".../request.playlist")
        list.iter(fun(item) -> requests_queue.push.uri(item), text)
        file.remove(...)            # then recreated empty

    every 60 seconds. Consequences worth knowing:

    * lines are pushed as URIs, so this is an exact hand-off - no searching,
      no chance of the radio picking a different take of the same song;
    * pickup takes up to a minute;
    * requests_queue sits first in the fallback, so a request interrupts the
      normal rotation at the next track boundary;
    * the queue is not subject to the check_next replay guard that the regular
      playlists use, so a request plays even if the track ran recently;
    * the file is emptied under us by design - never treat it as durable state.
    """
    if _should_verify() and not Path(radio_path).exists():
        raise RuntimeError(
            f"{radio_path} not visible here - check the two library roots and the mount"
        )
    # A missing parent almost always means the radio's data dir was not
    # mounted; creating it would silently swallow requests into this
    # container's own filesystem, so fail loudly instead.
    if not REQUEST_PLAYLIST.parent.is_dir():
        raise RuntimeError(
            f"{REQUEST_PLAYLIST.parent} does not exist here - is the radio's data dir mounted?"
        )
    # One append per request: O_APPEND makes a single small write atomic, so a
    # concurrent drain can never see a half-written path.
    with REQUEST_PLAYLIST.open("a", encoding="utf-8") as handle:
        handle.write(radio_path + "\n")


def queue(radio_path: str) -> None:
    if MODE == "dropbox":
        queue_via_dropbox(radio_path)
    else:
        queue_via_api(radio_path)


# --------------------------------------------------------------- state
def load_state() -> set:
    try:
        with STATE_FILE.open(encoding="utf-8") as handle:
            return set(json.load(handle).get("queued", []))
    except (FileNotFoundError, json.JSONDecodeError, OSError):
        return set()


def save_state(queued: set) -> None:
    try:
        STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
        tmp = STATE_FILE.with_suffix(STATE_FILE.suffix + ".tmp")
        with tmp.open("w", encoding="utf-8") as handle:
            json.dump({"queued": sorted(queued)}, handle, indent=2)
        tmp.replace(STATE_FILE)
    except OSError as exc:
        LOG.warning("Cannot persist state to %s: %s", STATE_FILE, exc)


# ---------------------------------------------------------------- main
def process_once(already: set) -> set:
    """One poll cycle. Returns the updated set of handled song ids."""
    songs = fetch_starred()
    for song in songs:
        song_id = str(song.get("id", ""))
        if not song_id or song_id in already:
            continue

        navidrome_path = song.get("path", "")
        label = f"{song.get('artist', '?')} - {song.get('title', '?')}"
        if not navidrome_path:
            LOG.warning("Skipping %s (%s): Navidrome returned no path", label, song_id)
            already.add(song_id)
            continue

        radio_path = translate_path(navidrome_path)
        if DRY_RUN:
            LOG.info("[dry-run] %s\n    navidrome: %s\n    radio    : %s",
                     label, navidrome_path, radio_path)
            already.add(song_id)
            continue

        try:
            queue(radio_path)
        except (requests.exceptions.RequestException, RuntimeError, OSError) as exc:
            # Left out of `already`, so the next cycle retries it.
            LOG.error("Failed to queue %s: %s", label, exc)
            continue

        LOG.info("Queued for radio: %s -> %s", label, radio_path)
        already.add(song_id)

        if UNSTAR_AFTER_QUEUE:
            try:
                unstar(song_id)
                # The star is the trigger, not a record: once un-starred the
                # id can legitimately come back, so stop remembering it.
                already.discard(song_id)
            except (requests.exceptions.RequestException, RuntimeError) as exc:
                LOG.warning("Queued but could not un-star %s: %s", label, exc)
    return already


def main() -> int:
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
        stream=sys.stdout,
    )

    missing = [
        name for name, value in (
            ("NAVIDROME_USER", NAVIDROME_USER),
            ("NAVIDROME_PASSWORD", NAVIDROME_PASSWORD),
        ) if not value
    ]
    if missing:
        LOG.error("FATAL: missing required config: %s", ", ".join(missing))
        return 1
    if MODE not in ("api", "dropbox"):
        LOG.error(
            "FATAL: BRIDGE_MODE must be 'api' or 'dropbox' ('playlist' = 'dropbox'), got %r",
            MODE,
        )
        return 1

    if MODE == "dropbox":
        # Neither betoniarka nor the bot is involved in this mode.
        LOG.info("Navidrome %s -> drop box %s (mode=dropbox)", NAVIDROME_URL, REQUEST_PLAYLIST)
        LOG.info("Liquidsoap drains that file every 60s; path verification: %s",
                 "on" if _should_verify() else "off")
    else:
        LOG.info("Navidrome %s -> betoniarka %s (mode=api)", NAVIDROME_URL, BETONIARKA_URL)
    LOG.info("Path translation: %s -> %s", NAVIDROME_LIBRARY_ROOT, BETONIARKA_LIBRARY_ROOT)
    if DRY_RUN:
        LOG.info("DRY RUN - nothing will be queued")

    already = load_state()
    if not STATE_FILE.exists() and SKIP_EXISTING_ON_FIRST_RUN and not DRY_RUN:
        try:
            already = {str(song.get("id", "")) for song in fetch_starred()}
            save_state(already)
            LOG.info("First run: ignoring %d already-starred tracks", len(already))
        except (requests.exceptions.RequestException, RuntimeError) as exc:
            LOG.error("Could not read existing stars: %s", exc)
            return 1

    while True:
        try:
            updated = process_once(set(already))
            if updated != already:
                already = updated
                save_state(already)
        except (requests.exceptions.RequestException, RuntimeError) as exc:
            LOG.warning("Poll failed (retrying in %ss): %s", POLL_SECONDS, exc)
        time.sleep(POLL_SECONDS)


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        LOG.info("Shutdown requested")
        sys.exit(0)