462 lines
14 KiB
Python
462 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]
|