339 lines
11 KiB
Python
339 lines
11 KiB
Python
"""The provider seam: what a radio catalogue has to implement.
|
|
|
|
Beefy Player is an internet-radio player. Everything the UI needs from a
|
|
catalogue — the station list, the song history, the stream URLs — goes through
|
|
:class:`RadioProvider`, and the app only ever holds one of those. SomaFM
|
|
(:mod:`beefy_player.providers.somafm`) is the first implementation; adding
|
|
another station catalogue means writing another subclass and registering it in
|
|
:mod:`beefy_player.providers`, with no changes to the app or the widgets.
|
|
|
|
This module also carries the toolkit a provider needs in practice: an HTTP
|
|
client factory and playlist handling (``.pls``/``.m3u`` resolution with an
|
|
on-disk cache), which are the same for every catalogue.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import re
|
|
import time
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
from urllib.parse import quote
|
|
|
|
import httpx
|
|
|
|
from .. import config
|
|
from ..models import Broadcast, Song, Station
|
|
|
|
__all__ = [
|
|
"PLAYLIST_TTL",
|
|
"RadioProvider",
|
|
"StationPage",
|
|
"fetch_json",
|
|
"is_direct_stream",
|
|
"is_playlist",
|
|
"make_client",
|
|
"parse_pls",
|
|
"quote_station_id",
|
|
"reset_playlist_cache",
|
|
"resolve_playlist",
|
|
]
|
|
|
|
REQUEST_TIMEOUT = httpx.Timeout(15.0, connect=8.0)
|
|
PLAYLIST_TTL = 8 * 3600
|
|
|
|
_PLAYLIST_CACHE: dict[str, dict[str, Any]] | None = None
|
|
|
|
|
|
def make_client(user_agent: str = "") -> httpx.AsyncClient:
|
|
return httpx.AsyncClient(
|
|
headers={
|
|
"User-Agent": user_agent or config.USER_AGENT,
|
|
"Accept": "application/json, */*",
|
|
},
|
|
timeout=REQUEST_TIMEOUT,
|
|
follow_redirects=True,
|
|
)
|
|
|
|
|
|
def reset_playlist_cache() -> None:
|
|
"""Forget the in-memory playlist cache (the tests use this)."""
|
|
global _PLAYLIST_CACHE
|
|
_PLAYLIST_CACHE = None
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class StationPage:
|
|
"""One page of a large catalogue, in the catalogue's own order."""
|
|
|
|
stations: list[Station]
|
|
#: Offset to request next, or ``None`` when the catalogue is exhausted.
|
|
next_offset: int | None = None
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# the interface
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class RadioProvider(ABC):
|
|
"""A station catalogue plus its streams and song history.
|
|
|
|
Providers are stateless singletons registered in
|
|
:data:`beefy_player.providers.PROVIDERS`; the app keeps one for its whole
|
|
lifetime and never instantiates another.
|
|
"""
|
|
|
|
#: Stable slug, used in caches, config and ``--provider``.
|
|
id: str = ""
|
|
#: Display name, e.g. ``"SomaFM"``.
|
|
name: str = ""
|
|
#: Bare domain, for docs and the help screen.
|
|
homepage: str = ""
|
|
#: Brand shown above the station list before anything is tuned in.
|
|
brand: str = ""
|
|
#: One-liner under the brand.
|
|
tagline: str = ""
|
|
#: Where stations come from, e.g. ``"somafm.com/channels.json"``.
|
|
station_source: str = ""
|
|
#: Where the song history comes from, e.g. ``"songs/<station>.json"``.
|
|
history_source: str = ""
|
|
#: ``User-Agent`` for this catalogue's requests (SomaFM likes to know).
|
|
user_agent: str = ""
|
|
|
|
# -- header hero ----------------------------------------------------- #
|
|
|
|
#: Rows of header art drawn at the top left (rich markup is allowed).
|
|
hero: tuple[str, ...] = ()
|
|
#: Paint :attr:`hero` with the app's spectrum gradient (a wordmark) instead
|
|
#: of the colours the art carries itself (a picture).
|
|
hero_gradient: bool = False
|
|
#: Short lines printed beside the hero, top to bottom.
|
|
hero_lines: tuple[str, ...] = ()
|
|
|
|
# -- left pane ------------------------------------------------------- #
|
|
|
|
#: Show a broadcast schedule in the left pane instead of a station list.
|
|
#: Set by single-stream catalogues, where a list of one station is a waste
|
|
#: of the pane and "what is on, what is next" is the useful view.
|
|
has_schedule: bool = False
|
|
|
|
#: Look the current track up on Bandcamp and offer a link when it is found.
|
|
#: Off for catalogues whose rotation is unlikely to be there (and for
|
|
#: stations that would rather not be scraped).
|
|
bandcamp: bool = True
|
|
|
|
#: The catalogue can search its whole directory on the server. When true,
|
|
#: the app calls :meth:`search_stations` as the filter changes instead of
|
|
#: only filtering the (possibly small) list from :meth:`fetch_stations`.
|
|
server_search: bool = False
|
|
|
|
#: The catalogue is larger than one page. When true, the app fetches the
|
|
#: first page with :meth:`fetch_station_page` and asks for the next one as
|
|
#: the station list nears its end.
|
|
paginated: bool = False
|
|
|
|
#: Keep the catalogue's own ordering (popularity, schedule, ...) instead of
|
|
#: sorting the station list alphabetically.
|
|
preserve_order: bool = False
|
|
|
|
# -- catalogue ------------------------------------------------------- #
|
|
|
|
@abstractmethod
|
|
async def fetch_stations(self, client: httpx.AsyncClient) -> list[Station]:
|
|
"""Fetch the live station list and refresh the on-disk cache."""
|
|
|
|
@abstractmethod
|
|
def load_cached_stations(self) -> list[Station]:
|
|
"""The last good station list from disk, or ``[]``."""
|
|
|
|
@abstractmethod
|
|
def fallback_stations(self) -> list[Station]:
|
|
"""A bundled list, so a fresh offline start still works."""
|
|
|
|
async def search_stations(
|
|
self, client: httpx.AsyncClient, query: str
|
|
) -> list[Station]:
|
|
"""Search the catalogue for *query*.
|
|
|
|
Only called when :attr:`server_search` is true. Providers that search
|
|
locally leave this alone; the app filters the loaded list itself.
|
|
"""
|
|
return []
|
|
|
|
async def fetch_station_page(
|
|
self, client: httpx.AsyncClient, offset: int = 0
|
|
) -> StationPage:
|
|
"""One page of a paginated catalogue.
|
|
|
|
Only called when :attr:`paginated` is true. ``offset`` is the value
|
|
from the previous page's :attr:`StationPage.next_offset`.
|
|
"""
|
|
return StationPage(stations=[], next_offset=None)
|
|
|
|
async def fetch_stations_by_ids(
|
|
self, client: httpx.AsyncClient, station_ids: list[str]
|
|
) -> list[Station]:
|
|
"""Fetch specific stations by id.
|
|
|
|
Used for favourites that are not on the first page of a paginated
|
|
catalogue. Providers that return their whole catalogue at once can
|
|
leave this alone.
|
|
"""
|
|
return []
|
|
|
|
# -- metadata -------------------------------------------------------- #
|
|
|
|
@abstractmethod
|
|
async def fetch_history(
|
|
self, client: httpx.AsyncClient, station_id: str
|
|
) -> list[Song]:
|
|
"""Recently played songs, newest first (falls back to the cache)."""
|
|
|
|
@abstractmethod
|
|
def load_cached_history(self, station_id: str) -> list[Song]:
|
|
"""The last good song history from disk, or ``[]``."""
|
|
|
|
# -- schedule -------------------------------------------------------- #
|
|
|
|
async def fetch_schedule(self, client: httpx.AsyncClient) -> list[Broadcast]:
|
|
"""Upcoming broadcasts, soonest first (the live one, if any, first)."""
|
|
return []
|
|
|
|
def load_cached_schedule(self) -> list[Broadcast]:
|
|
"""The last good schedule from disk, or ``[]``."""
|
|
return []
|
|
|
|
# -- streams --------------------------------------------------------- #
|
|
|
|
def fallback_urls(self, station: Station) -> list[str]:
|
|
"""Guessed stream URLs, used when the catalogue is unreachable."""
|
|
return []
|
|
|
|
async def stream_candidates(
|
|
self, client: httpx.AsyncClient, station: Station
|
|
) -> list[str]:
|
|
"""Direct stream URLs to try, best first.
|
|
|
|
Resolves the station's own playlists (a couple of ``.pls`` files at
|
|
most) and appends :meth:`fallback_urls` in case none of them answer.
|
|
"""
|
|
urls: list[str] = []
|
|
playlists = station.sorted_playlists()
|
|
|
|
for entry in playlists[:3]:
|
|
url = entry.get("url", "")
|
|
resolved = await resolve_playlist(client, url)
|
|
for candidate in resolved:
|
|
if candidate not in urls:
|
|
urls.append(candidate)
|
|
if not resolved and url and url not in urls:
|
|
urls.append(url)
|
|
|
|
for entry in playlists[3:]:
|
|
url = entry.get("url", "")
|
|
if url and url not in urls:
|
|
urls.append(url)
|
|
|
|
for url in self.fallback_urls(station):
|
|
if url not in urls:
|
|
urls.append(url)
|
|
|
|
return urls[:10]
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# playlists
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def is_playlist(url: str) -> bool:
|
|
"""A playlist *we* resolve; HLS is handed to the player untouched."""
|
|
path = url.split("?", 1)[0].split("#", 1)[0].lower()
|
|
return path.endswith((".pls", ".m3u"))
|
|
|
|
|
|
def is_direct_stream(url: str) -> bool:
|
|
return url.startswith(("http://", "https://")) and not is_playlist(url)
|
|
|
|
|
|
def parse_pls(text: str) -> list[str]:
|
|
urls: list[str] = []
|
|
for line in text.splitlines():
|
|
line = line.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
match = re.match(r"(?i)^file\d+\s*=\s*(\S+)$", line)
|
|
candidate = match.group(1) if match else line
|
|
if candidate.startswith(("http://", "https://")) and candidate not in urls:
|
|
urls.append(candidate)
|
|
return urls
|
|
|
|
|
|
def _playlist_cache() -> dict[str, dict[str, Any]]:
|
|
global _PLAYLIST_CACHE
|
|
if _PLAYLIST_CACHE is None:
|
|
raw = config.read_json(config.PLAYLIST_CACHE)
|
|
_PLAYLIST_CACHE = raw if isinstance(raw, dict) else {}
|
|
return _PLAYLIST_CACHE
|
|
|
|
|
|
async def resolve_playlist(client: httpx.AsyncClient, url: str) -> list[str]:
|
|
"""Turn a ``.pls`` URL into direct icecast URLs (cached for a few hours)."""
|
|
if not url:
|
|
return []
|
|
if not is_playlist(url):
|
|
return [url]
|
|
|
|
cache = _playlist_cache()
|
|
now = time.time()
|
|
entry = cache.get(url)
|
|
if isinstance(entry, dict):
|
|
fetched = entry.get("ts")
|
|
urls = entry.get("urls")
|
|
if (
|
|
isinstance(fetched, (int, float))
|
|
and now - fetched < PLAYLIST_TTL
|
|
and isinstance(urls, list)
|
|
):
|
|
cached_urls = [item for item in urls if isinstance(item, str)]
|
|
if cached_urls:
|
|
return cached_urls
|
|
|
|
try:
|
|
response = await client.get(url, headers={"Accept": "*/*"})
|
|
response.raise_for_status()
|
|
text = response.text
|
|
except httpx.HTTPError:
|
|
return []
|
|
|
|
urls = parse_pls(text)
|
|
if urls:
|
|
cache[url] = {"ts": now, "urls": urls}
|
|
config.write_json(config.PLAYLIST_CACHE, cache)
|
|
return urls
|
|
|
|
|
|
def quote_station_id(station_id: str) -> str:
|
|
"""Percent-encode a station id for use in a provider URL."""
|
|
return quote(station_id, safe="")
|
|
|
|
|
|
async def fetch_json(
|
|
client: httpx.AsyncClient, url: str, *, attempts: int = 2, pause: float = 0.6
|
|
) -> Any:
|
|
"""``GET url`` and decode JSON, retrying once: flaky mirrors are normal."""
|
|
last_error: Exception | None = None
|
|
for attempt in range(attempts):
|
|
try:
|
|
response = await client.get(url)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except (httpx.HTTPError, ValueError) as exc: # includes JSON errors
|
|
last_error = exc
|
|
if attempt + 1 < attempts:
|
|
await asyncio.sleep(pause)
|
|
raise RuntimeError(f"could not fetch {url}: {last_error}")
|