"""The public SomaFM endpoints, and nothing but. * ``https://somafm.com/channels.json`` -> every station + listener counts * ``https://somafm.com/.pls`` -> tiny PLS playlists (resolved to ice URLs) * ``https://somafm.com/songs/.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. 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 re import time from collections.abc import Iterable from typing import Any import httpx 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" # --------------------------------------------------------------------------- # # 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, provider="somafm", ) 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]: 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") 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(station_id)) 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) # --------------------------------------------------------------------------- # # streams # --------------------------------------------------------------------------- # def synthetic_mirrors(station_id: str) -> 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/{station_id}{suffix}") return urls