114 lines
3.6 KiB
Python
114 lines
3.6 KiB
Python
"""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")))
|