feat(providers): add a pluggable station-catalogue seam
The app now holds one RadioProvider and asks it for stations, song history and stream URLs instead of importing SomaFM's HTTP client directly. - models.py: provider-neutral Song/Station plus the shared parsers every catalogue needs (Station.provider/.uid keep ids unique once there is more than one). - providers/base.py: the RadioProvider interface and the toolkit it needs in practice - httpx client factory, .pls/.m3u resolution with an 8h disk cache, and a default stream_candidates() strategy. - providers/somafm/: everything SomaFM-specific (channels.json, songs/<id>.json, PLS layout, guessed icecast mirrors) behind SomaFMProvider. - providers/__init__.py: registry, get_provider() and --provider, remembered in the config file. Widgets render branding, splash copy and the help credits from the active provider, so a second catalogue needs no UI changes; with SomaFM selected the screen is byte-identical to before.
This commit is contained in:
beefcarver committed
1 parent
e30ca09627
commit
c4cc91ee1a
13 files changed
+708
-372
No files matched your search
@@ -1,197 +1,30 @@
|
||||
"""Async client for the public SomaFM endpoints.
|
||||
|
||||
The app talks to three things:
|
||||
"""The public SomaFM endpoints, and nothing but.
|
||||
|
||||
* ``https://somafm.com/channels.json`` -> every station + listener counts
|
||||
* ``https://somafm.com/<station>.pls`` -> tiny PLS playlists (resolved to ice URLs)
|
||||
* ``https://somafm.com/songs/<station>.json`` -> the "last played" song history
|
||||
|
||||
All parsing is deliberately forgiving: SomaFM has changed field names before
|
||||
and we would rather show a slightly degraded station than crash.
|
||||
and we would rather show a slightly degraded station than crash. Nothing
|
||||
outside :mod:`beefy_player.providers.somafm` should import this module — the
|
||||
app talks to :class:`beefy_player.providers.somafm.SomaFMProvider` instead.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import re
|
||||
import time
|
||||
from collections.abc import Iterable
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
from urllib.parse import quote
|
||||
|
||||
import httpx
|
||||
|
||||
from . import config
|
||||
from ... import config
|
||||
from ...models import Song, Station, parse_date, parse_last_playing, to_int
|
||||
from ..base import fetch_json, quote_station_id
|
||||
|
||||
CHANNELS_URL = "https://somafm.com/channels.json"
|
||||
SONGS_URL = "https://somafm.com/songs/{station}.json"
|
||||
USER_AGENT = "somafm-tui/0.1 (terminal radio; +https://somafm.com)"
|
||||
REQUEST_TIMEOUT = httpx.Timeout(15.0, connect=8.0)
|
||||
PLAYLIST_TTL = 8 * 3600
|
||||
|
||||
_QUALITY_RANK = {"highest": 0, "high": 1, "low": 2}
|
||||
_FORMAT_RANK = {"mp3": 0, "aac": 1, "aacp": 2}
|
||||
|
||||
_PLAYLIST_CACHE: dict[str, dict[str, Any]] | None = None
|
||||
|
||||
|
||||
def make_client() -> httpx.AsyncClient:
|
||||
return httpx.AsyncClient(
|
||||
headers={"User-Agent": USER_AGENT, "Accept": "application/json, */*"},
|
||||
timeout=REQUEST_TIMEOUT,
|
||||
follow_redirects=True,
|
||||
)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# models
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class Song:
|
||||
title: str = ""
|
||||
artist: str = ""
|
||||
album: str = ""
|
||||
art_url: str = ""
|
||||
bandcamp_url: str = ""
|
||||
played_at: datetime | None = None
|
||||
|
||||
@property
|
||||
def display_title(self) -> str:
|
||||
return self.title.strip() or "Unknown track"
|
||||
|
||||
@property
|
||||
def display_artist(self) -> str:
|
||||
return self.artist.strip() or "Unknown artist"
|
||||
|
||||
def key(self) -> str:
|
||||
return f"{self.artist}\x1f{self.title}\x1f{self.album}"
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class Station:
|
||||
id: str
|
||||
title: str
|
||||
description: str = ""
|
||||
genre: str = ""
|
||||
dj: str = ""
|
||||
image: str = ""
|
||||
listeners: int = 0
|
||||
last_playing: str = ""
|
||||
playlists: list[dict[str, str]] = field(default_factory=list)
|
||||
now_playing: Song | None = None
|
||||
history: list[Song] = field(default_factory=list)
|
||||
favorite: bool = False
|
||||
|
||||
@property
|
||||
def genre_label(self) -> str:
|
||||
return self.genre.replace("|", " · ") or "eclectic"
|
||||
|
||||
def sorted_playlists(self) -> list[dict[str, str]]:
|
||||
return sorted(
|
||||
self.playlists,
|
||||
key=lambda pl: (
|
||||
_QUALITY_RANK.get(pl.get("quality", ""), 3),
|
||||
_FORMAT_RANK.get(pl.get("format", ""), 3),
|
||||
),
|
||||
)
|
||||
|
||||
@property
|
||||
def format_label(self) -> str:
|
||||
playlists = self.sorted_playlists()
|
||||
if not playlists:
|
||||
return "stream"
|
||||
first = playlists[0]
|
||||
fmt = {"aacp": "aac+", "aac": "aac", "mp3": "mp3"}.get(
|
||||
first.get("format", ""), first.get("format") or "stream"
|
||||
)
|
||||
return f"{fmt} · {_guess_bitrate(first.get('url', ''))}k"
|
||||
|
||||
def synthetic_mirrors(self) -> list[str]:
|
||||
"""Direct icecast URLs, handy when the PLS files are unreachable."""
|
||||
urls: list[str] = []
|
||||
for suffix in ("-128-mp3", "-64-aac"):
|
||||
for host in ("ice1", "ice2", "ice3", "ice4"):
|
||||
urls.append(f"https://{host}.somafm.com/{self.id}{suffix}")
|
||||
return urls
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# small helpers
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def _guess_bitrate(url: str) -> int:
|
||||
match = re.search(
|
||||
r"(256|192|130|128|96|64|48|32)(?=\.pls|\.m3u|-mp3|-aac)", url
|
||||
)
|
||||
if not match:
|
||||
return 128
|
||||
value = int(match.group(1))
|
||||
return 128 if value == 130 else value
|
||||
|
||||
|
||||
def _to_int(value: Any, default: int = 0) -> int:
|
||||
if isinstance(value, bool):
|
||||
return default
|
||||
if isinstance(value, int):
|
||||
return value
|
||||
if isinstance(value, float):
|
||||
return int(value)
|
||||
if isinstance(value, str):
|
||||
digits = re.sub(r"[^0-9]", "", value)
|
||||
if digits:
|
||||
try:
|
||||
return int(digits)
|
||||
except ValueError:
|
||||
return default
|
||||
return default
|
||||
|
||||
|
||||
def parse_date(value: Any) -> datetime | None:
|
||||
if value is None or value == "":
|
||||
return None
|
||||
if isinstance(value, (int, float)):
|
||||
try:
|
||||
return datetime.fromtimestamp(float(value), tz=timezone.utc)
|
||||
except (OverflowError, OSError, ValueError):
|
||||
return None
|
||||
if isinstance(value, str):
|
||||
text = value.strip()
|
||||
if text.isdigit():
|
||||
try:
|
||||
return datetime.fromtimestamp(int(text), tz=timezone.utc)
|
||||
except (OverflowError, OSError, ValueError):
|
||||
return None
|
||||
try:
|
||||
parsed = datetime.fromisoformat(text.replace("Z", "+00:00"))
|
||||
except ValueError:
|
||||
parsed = None
|
||||
for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y/%m/%d %H:%M:%S"):
|
||||
try:
|
||||
parsed = datetime.strptime(text, fmt)
|
||||
break
|
||||
except ValueError:
|
||||
continue
|
||||
if parsed is not None:
|
||||
return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
|
||||
return None
|
||||
|
||||
|
||||
def parse_last_playing(text: str | None) -> Song | None:
|
||||
if not isinstance(text, str):
|
||||
return None
|
||||
cleaned = text.strip()
|
||||
if not cleaned:
|
||||
return None
|
||||
for separator in (" - ", " – ", " — ", " -- "):
|
||||
if separator in cleaned:
|
||||
artist, title = cleaned.split(separator, 1)
|
||||
return Song(title=title.strip(), artist=artist.strip())
|
||||
return Song(title=cleaned)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
@@ -231,8 +64,9 @@ def station_from_payload(item: dict[str, Any]) -> Station | None:
|
||||
genre=str(item.get("genre") or "").strip(),
|
||||
dj=str(item.get("dj") or "").strip(),
|
||||
image=str(item.get("image") or item.get("largeimage") or "").strip(),
|
||||
listeners=_to_int(item.get("listeners")),
|
||||
listeners=to_int(item.get("listeners")),
|
||||
playlists=playlists,
|
||||
provider="somafm",
|
||||
)
|
||||
last_playing = item.get("lastPlaying")
|
||||
if isinstance(last_playing, str):
|
||||
@@ -264,20 +98,10 @@ def parse_channels(payload: Any) -> list[Station]:
|
||||
|
||||
|
||||
async def fetch_channels(client: httpx.AsyncClient) -> list[Station]:
|
||||
payload: Any = None
|
||||
last_error: Exception | None = None
|
||||
for attempt in range(2):
|
||||
try:
|
||||
response = await client.get(CHANNELS_URL)
|
||||
response.raise_for_status()
|
||||
payload = response.json()
|
||||
break
|
||||
except (httpx.HTTPError, ValueError) as exc: # includes JSON errors
|
||||
last_error = exc
|
||||
if attempt == 0:
|
||||
await asyncio.sleep(0.6)
|
||||
if payload is None:
|
||||
raise RuntimeError(f"could not fetch SomaFM channels: {last_error}")
|
||||
try:
|
||||
payload = await fetch_json(client, CHANNELS_URL)
|
||||
except RuntimeError as exc:
|
||||
raise RuntimeError(f"could not fetch SomaFM channels: {exc}") from None
|
||||
stations = parse_channels(payload)
|
||||
if not stations:
|
||||
raise RuntimeError("SomaFM returned an empty channel list")
|
||||
@@ -350,7 +174,7 @@ def _songs_to_cache(songs: Iterable[Song]) -> list[dict[str, Any]]:
|
||||
|
||||
|
||||
async def fetch_history(client: httpx.AsyncClient, station_id: str) -> list[Song]:
|
||||
url = SONGS_URL.format(station=quote(station_id, safe=""))
|
||||
url = SONGS_URL.format(station=quote_station_id(station_id))
|
||||
try:
|
||||
response = await client.get(url)
|
||||
response.raise_for_status()
|
||||
@@ -369,95 +193,14 @@ async def fetch_history(client: httpx.AsyncClient, station_id: str) -> list[Song
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# playlist resolution
|
||||
# streams
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def is_playlist(url: str) -> bool:
|
||||
path = url.split("?", 1)[0].split("#", 1)[0].lower()
|
||||
return path.endswith((".pls", ".m3u", ".m3u8"))
|
||||
|
||||
|
||||
def is_direct_stream(url: str) -> bool:
|
||||
return url.startswith(("http://", "https://")) and not is_playlist(url)
|
||||
|
||||
|
||||
def parse_pls(text: str) -> list[str]:
|
||||
def synthetic_mirrors(station_id: str) -> list[str]:
|
||||
"""Direct icecast URLs, handy when the PLS files are unreachable."""
|
||||
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)
|
||||
for suffix in ("-128-mp3", "-64-aac"):
|
||||
for host in ("ice1", "ice2", "ice3", "ice4"):
|
||||
urls.append(f"https://{host}.somafm.com/{station_id}{suffix}")
|
||||
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
|
||||
|
||||
|
||||
async def stream_candidates(client: httpx.AsyncClient, station: Station) -> list[str]:
|
||||
"""Direct URLs first, then unresolved playlists, then guessed mirrors."""
|
||||
playlists = station.sorted_playlists()
|
||||
urls: list[str] = []
|
||||
|
||||
for entry in playlists[:3]:
|
||||
resolved = await resolve_playlist(client, entry["url"])
|
||||
for url in resolved:
|
||||
if url not in urls:
|
||||
urls.append(url)
|
||||
if not resolved and entry["url"] not in urls:
|
||||
urls.append(entry["url"])
|
||||
|
||||
for entry in playlists[3:]:
|
||||
if entry["url"] not in urls:
|
||||
urls.append(entry["url"])
|
||||
|
||||
for url in station.synthetic_mirrors():
|
||||
if url not in urls:
|
||||
urls.append(url)
|
||||
|
||||
return urls[:10]
|
||||
Reference in new issue
Block a user