feat(providers): add nine catalogues and the Radio Browser directory

beefy_player/providers/radiobrowser/api.py

@@ -0,0 +1,113 @@
+"""Radio Browser: the community's own station directory.
+
+``https://de1.api.radio-browser.info/json/stations/search`` is keyless, CORS-open
+and public-domain data.  It knows nothing about what a station is playing —
+there is no per-station history — so this catalogue is a *directory*: three
+hundred of the most-clicked working stations, ready to tune into.  Whatever the
+relay puts in its stream is what you hear; the app has no metadata to show.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from urllib.parse import urlencode
+
+import httpx
+
+from ... import config
+from ...models import Station, to_int
+from ..base import fetch_json
+
+API_URL = "https://de1.api.radio-browser.info/json/stations/search"
+
+#: Three hundred is a comfortable directory: enough to browse, quick to fetch.
+STATION_LIMIT = 300
+
+_PARAMS = {
+    "hidebroken": "true",
+    "order": "clickcount",
+    "reverse": "true",
+    "limit": str(STATION_LIMIT),
+}
+
+_CODECS = {"mp3": "mp3", "aac": "aac", "aac+": "aacp", "ogg": "ogg", "opus": "opus"}
+
+
+def _quality(bitrate: int) -> str:
+    if bitrate >= 192:
+        return "highest"
+    if bitrate >= 96:
+        return "high"
+    return "low" if bitrate else ""
+
+
+def station_from_entry(entry: dict[str, Any]) -> Station | None:
+    if not isinstance(entry, dict):
+        return None
+    name = str(entry.get("name") or "").strip()
+    url = str(entry.get("url_resolved") or entry.get("url") or "").strip()
+    station_id = str(entry.get("stationuuid") or "").strip()
+    if not name or not station_id or not url.startswith(("http://", "https://")):
+        return None
+
+    bitrate = to_int(entry.get("bitrate"))
+    codec = str(entry.get("codec") or "").strip().lower()
+    tags = [
+        tag.strip()
+        for tag in str(entry.get("tags") or "").split(",")
+        if tag.strip()
+    ][:3]
+    country = str(entry.get("country") or "").strip()
+    language = str(entry.get("language") or "").strip()
+    description = " · ".join(part for part in (country, language) if part)
+
+    favicon = str(entry.get("favicon") or "").strip()
+
+    return Station(
+        id=station_id,
+        title=name,
+        description=description or "internet radio",
+        genre="|".join(tags),
+        image=favicon if favicon.startswith(("http://", "https://")) else "",
+        playlists=[
+            {
+                "url": url,
+                "format": _CODECS.get(codec, codec),
+                "quality": _quality(bitrate),
+                "bitrate": str(bitrate) if bitrate else "",
+            }
+        ],
+        provider="radiobrowser",
+    )
+
+
+def parse_stations(payload: Any) -> list[Station]:
+    if not isinstance(payload, list):
+        return []
+    stations: list[Station] = []
+    seen: set[str] = set()
+    for entry in payload:
+        station = station_from_entry(entry)
+        if station is None or station.id in seen:
+            continue
+        seen.add(station.id)
+        stations.append(station)
+    stations.sort(key=lambda station: station.title.casefold())
+    return stations
+
+
+async def fetch_stations(client: httpx.AsyncClient) -> list[Station]:
+    url = f"{API_URL}?{urlencode(_PARAMS)}"
+    try:
+        payload = await fetch_json(client, url)
+    except RuntimeError as exc:
+        raise RuntimeError(f"could not fetch the Radio Browser directory: {exc}") from None
+    stations = parse_stations(payload)
+    if not stations:
+        raise RuntimeError("Radio Browser returned no stations")
+    config.write_json(config.channels_cache_path("radiobrowser"), payload)
+    return stations
+
+
+def load_cached_stations() -> list[Station]:
+    return parse_stations(config.read_json(config.channels_cache_path("radiobrowser")))