beefy-player/beefy_player/config.py

"""Paths, persistent settings and tiny cache helpers.

Everything lives in the usual XDG places:

    ~/.config/beefy-player/config.json          favourites, volume, station
    ~/.cache/beefy-player/channels.json         last good station list (SomaFM)
    ~/.cache/beefy-player/channels-<id>.json    ... for every other catalogue
    ~/.cache/beefy-player/schedule-<id>.json    last good broadcast schedule
    ~/.cache/beefy-player/playlists.json        resolved .pls -> stream URLs
    ~/.cache/beefy-player/songs/<id>.json       last good song history
    ~/.cache/beefy-player/artwork/<sha1>        cached album/channel artwork

Settings from the pre-rename ``somafm-tui`` directory are copied over once, so
upgrading does not lose favourites or volume.
"""

from __future__ import annotations

import contextlib
import json
import os
import re
import shutil
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any

from . import __version__

#: Directory name for config, cache and runtime files.
APP_NAME = "beefy-player"

#: Name the app used before it became Beefy Player; only read, once, to migrate.
LEGACY_APP_NAME = "somafm-tui"

#: What we send as ``User-Agent`` to catalogues, icecast mirrors and mpv.
USER_AGENT = f"{APP_NAME}/{__version__} (terminal radio)"

#: Environment variables are namespaced ``BEEFY_*``; the old ``SOMAFM_*`` names
#: are still honoured so existing shell profiles keep working.
ENV_PREFIX = "BEEFY_"
LEGACY_ENV_PREFIX = "SOMAFM_"


def _xdg(env_var: str, fallback: str) -> Path:
    value = os.environ.get(env_var)
    if value:
        return Path(value).expanduser()
    return Path.home() / fallback


def env_var(suffix: str) -> str:
    """``BEEFY_<suffix>``, falling back to the legacy ``SOMAFM_<suffix>``."""
    value = os.environ.get(ENV_PREFIX + suffix)
    if value is None:
        value = os.environ.get(LEGACY_ENV_PREFIX + suffix)
    return (value or "").strip()


CONFIG_DIR = _xdg("XDG_CONFIG_HOME", ".config") / APP_NAME
CACHE_DIR = _xdg("XDG_CACHE_HOME", ".cache") / APP_NAME
RUNTIME_DIR = Path(os.environ.get("XDG_RUNTIME_DIR") or "/tmp") / APP_NAME

CONFIG_PATH = CONFIG_DIR / "config.json"
CHANNELS_CACHE = CACHE_DIR / "channels.json"
PLAYLIST_CACHE = CACHE_DIR / "playlists.json"
SONGS_CACHE_DIR = CACHE_DIR / "songs"
ARTWORK_CACHE_DIR = CACHE_DIR / "artwork"

#: Pre-rename location, read once by :meth:`Config.load`.
LEGACY_CONFIG_PATH = _xdg("XDG_CONFIG_HOME", ".config") / LEGACY_APP_NAME / "config.json"

VISUALIZERS = ("both", "bars", "wave", "waterfall", "radial", "braille")


def channels_cache_path(provider_id: str) -> Path:
    """Where a catalogue caches its last good station list.

    SomaFM keeps the historical ``channels.json``; every other provider gets
    its own file, so two catalogues can never overwrite each other's cache.
    """
    slug = re.sub(r"[^a-z0-9_-]", "_", provider_id.strip().lower())
    if not slug or slug == "somafm":
        return CHANNELS_CACHE
    return CACHE_DIR / f"channels-{slug}.json"


def schedule_cache_path(provider_id: str) -> Path:
    """Where a catalogue caches its last good broadcast schedule."""
    slug = re.sub(r"[^a-z0-9_-]", "_", provider_id.strip().lower()) or "provider"
    return CACHE_DIR / f"schedule-{slug}.json"


def ensure_dirs() -> None:
    for directory in (CONFIG_DIR, CACHE_DIR, SONGS_CACHE_DIR):
        with contextlib.suppress(OSError):
            directory.mkdir(parents=True, exist_ok=True)


def atomic_write_text(path: Path, text: str) -> None:
    """Write ``text`` to ``path`` without ever leaving a half-written file."""
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_name(f".{path.name}.tmp")
    tmp.write_text(text, encoding="utf-8")
    os.replace(tmp, path)


def read_json(path: Path) -> Any:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return None


def write_json(path: Path, payload: Any) -> None:
    with contextlib.suppress(OSError):
        atomic_write_text(path, json.dumps(payload, indent=2, ensure_ascii=False))


def migrate_legacy_state() -> None:
    """Copy a pre-rename config file into place, once.

    Favourites, volume and the last station are the only state worth keeping;
    caches are rebuilt on the next run.
    """
    try:
        if CONFIG_PATH.exists() or not LEGACY_CONFIG_PATH.is_file():
            return
        CONFIG_DIR.mkdir(parents=True, exist_ok=True)
        shutil.copyfile(LEGACY_CONFIG_PATH, CONFIG_PATH)
    except OSError:
        pass


@dataclass(slots=True)
class Config:
    favorites: list[str] = field(default_factory=list)
    last_station: str = "groovesalad"
    #: Which station each catalogue was last tuned to, keyed by provider slug,
    #: so switching back lands where you left off.
    last_stations: dict[str, str] = field(default_factory=dict)
    volume: int = 70
    visualizer: str = "both"
    autoplay: bool = False
    #: Which catalogue to use by default (see :mod:`beefy_player.providers`).
    provider: str = "somafm"
    #: Publish what is playing to the desktop media session (MPRIS on Linux,
    #: macOS Now Playing via mpv) and honour the media keys.  The JSON key is
    #: spelled ``mpris`` for backwards compatibility.
    mpris: bool = True

    @classmethod
    def load(cls) -> Config:
        migrate_legacy_state()
        raw = read_json(CONFIG_PATH)
        config = cls()
        if not isinstance(raw, dict):
            return config

        favorites = raw.get("favorites")
        if isinstance(favorites, list):
            config.favorites = [str(item) for item in favorites if isinstance(item, str)]

        last_station = raw.get("last_station")
        if isinstance(last_station, str) and last_station:
            config.last_station = last_station

        last_stations = raw.get("last_stations")
        if isinstance(last_stations, dict):
            config.last_stations = {
                str(key): str(value)
                for key, value in last_stations.items()
                if isinstance(value, str) and value
            }

        volume = raw.get("volume")
        if isinstance(volume, (int, float)):
            config.volume = max(0, min(100, int(volume)))

        visualizer = raw.get("visualizer")
        if visualizer in VISUALIZERS:
            config.visualizer = visualizer

        provider = raw.get("provider")
        if isinstance(provider, str) and provider.strip():
            config.provider = provider.strip().lower()

        if isinstance(raw.get("autoplay"), bool):
            config.autoplay = raw["autoplay"]
        if isinstance(raw.get("mpris"), bool):
            config.mpris = raw["mpris"]
        return config

    def save(self) -> None:
        try:
            ensure_dirs()
            atomic_write_text(
                CONFIG_PATH, json.dumps(asdict(self), indent=2, ensure_ascii=False)
            )
        except OSError:
            pass

    def last_station_for(self, provider_id: str) -> str:
        """The station to re-tune for a catalogue (old configs have one only)."""
        return self.last_stations.get(provider_id) or self.last_station

    def is_favorite(self, station_id: str) -> bool:
        return station_id in self.favorites

    def toggle_favorite(self, station_id: str) -> bool:
        if station_id in self.favorites:
            self.favorites.remove(station_id)
            return False
        self.favorites.append(station_id)
        return True