463 lines
14 KiB
Python
463 lines
14 KiB
Python
"""Async client for the public SomaFM endpoints.
|
||
|
||
The app talks to three things:
|
||
|
||
* ``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.
|
||
"""
|
||
|
||
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
|
||
|
||
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 = ""
|
||
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)
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# channels
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def station_from_payload(item: dict[str, Any]) -> Station | None:
|
||
if not isinstance(item, dict):
|
||
return None
|
||
channel_id = item.get("id")
|
||
if not isinstance(channel_id, str) or not channel_id.strip():
|
||
return None
|
||
channel_id = channel_id.strip()
|
||
|
||
playlists: list[dict[str, str]] = []
|
||
raw_playlists = item.get("playlists")
|
||
if isinstance(raw_playlists, list):
|
||
for entry in raw_playlists:
|
||
if not isinstance(entry, dict):
|
||
continue
|
||
url = entry.get("url")
|
||
if not isinstance(url, str) or not url.startswith(("http://", "https://")):
|
||
continue
|
||
playlists.append(
|
||
{
|
||
"url": url,
|
||
"format": str(entry.get("format") or "").strip().lower(),
|
||
"quality": str(entry.get("quality") or "").strip().lower(),
|
||
}
|
||
)
|
||
|
||
station = Station(
|
||
id=channel_id,
|
||
title=str(item.get("title") or channel_id).strip(),
|
||
description=str(item.get("description") or "").strip(),
|
||
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")),
|
||
playlists=playlists,
|
||
)
|
||
last_playing = item.get("lastPlaying")
|
||
if isinstance(last_playing, str):
|
||
station.last_playing = last_playing.strip()
|
||
station.now_playing = parse_last_playing(station.last_playing)
|
||
if not station.image:
|
||
station.image = f"https://somafm.com/logos/256/{channel_id}256.png"
|
||
return station
|
||
|
||
|
||
def parse_channels(payload: Any) -> list[Station]:
|
||
if isinstance(payload, dict):
|
||
items = payload.get("channels")
|
||
elif isinstance(payload, list):
|
||
items = payload
|
||
else:
|
||
items = None
|
||
if not isinstance(items, list):
|
||
return []
|
||
stations: list[Station] = []
|
||
seen: set[str] = set()
|
||
for item in items:
|
||
station = station_from_payload(item)
|
||
if station is None or station.id in seen:
|
||
continue
|
||
seen.add(station.id)
|
||
stations.append(station)
|
||
return stations
|
||
|
||
|
||
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}")
|
||
stations = parse_channels(payload)
|
||
if not stations:
|
||
raise RuntimeError("SomaFM returned an empty channel list")
|
||
config.write_json(config.CHANNELS_CACHE, payload)
|
||
return stations
|
||
|
||
|
||
def load_cached_channels() -> list[Station]:
|
||
return parse_channels(config.read_json(config.CHANNELS_CACHE))
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# song history
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def parse_songs(payload: Any) -> list[Song]:
|
||
if isinstance(payload, dict):
|
||
items = payload.get("songs")
|
||
elif isinstance(payload, list):
|
||
items = payload
|
||
else:
|
||
items = None
|
||
if not isinstance(items, list):
|
||
return []
|
||
songs: list[Song] = []
|
||
for entry in items:
|
||
if not isinstance(entry, dict):
|
||
continue
|
||
song = Song(
|
||
title=str(entry.get("title") or "").strip(),
|
||
artist=str(entry.get("artist") or "").strip(),
|
||
album=str(entry.get("album") or "").strip(),
|
||
art_url=str(
|
||
entry.get("albumArt") or entry.get("albumart") or entry.get("art") or ""
|
||
).strip(),
|
||
played_at=parse_date(entry.get("date")),
|
||
)
|
||
if song.title or song.artist:
|
||
songs.append(song)
|
||
# Newest first; the upstream feed is already sorted but let's be sure.
|
||
if all(song.played_at for song in songs):
|
||
songs.sort(key=lambda song: song.played_at, reverse=True) # type: ignore[arg-type,return-value]
|
||
return songs
|
||
|
||
|
||
def _songs_cache_path(station_id: str):
|
||
safe = re.sub(r"[^A-Za-z0-9_.-]", "_", station_id)
|
||
return config.SONGS_CACHE_DIR / f"{safe}.json"
|
||
|
||
|
||
def load_cached_songs(station_id: str) -> list[Song]:
|
||
raw = config.read_json(_songs_cache_path(station_id))
|
||
if isinstance(raw, dict):
|
||
return parse_songs(raw.get("songs"))
|
||
return []
|
||
|
||
|
||
def _songs_to_cache(songs: Iterable[Song]) -> list[dict[str, Any]]:
|
||
return [
|
||
{
|
||
"title": song.title,
|
||
"artist": song.artist,
|
||
"album": song.album,
|
||
"albumArt": song.art_url,
|
||
"date": song.played_at.isoformat() if song.played_at else "",
|
||
}
|
||
for song in songs
|
||
]
|
||
|
||
|
||
async def fetch_history(client: httpx.AsyncClient, station_id: str) -> list[Song]:
|
||
url = SONGS_URL.format(station=quote(station_id, safe=""))
|
||
try:
|
||
response = await client.get(url)
|
||
response.raise_for_status()
|
||
payload = response.json()
|
||
except (httpx.HTTPError, ValueError):
|
||
return load_cached_songs(station_id)
|
||
|
||
songs = parse_songs(payload)
|
||
if songs:
|
||
config.write_json(
|
||
_songs_cache_path(station_id),
|
||
{"fetched": time.time(), "songs": _songs_to_cache(songs)},
|
||
)
|
||
return songs
|
||
return load_cached_songs(station_id)
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# playlist resolution
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
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]:
|
||
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
|
||
|
||
|
||
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]
|