211 lines
6.6 KiB
Python
211 lines
6.6 KiB
Python
"""Provider-neutral data types.
|
||
|
||
Every radio catalogue — SomaFM today, anything else later — is reduced to
|
||
:class:`Station` and :class:`Song` before the UI sees it, so the widgets and
|
||
the app never have to know which catalogue a station came from. The parsers
|
||
at the bottom are the small, boring helpers (numbers, dates, "Artist - Title"
|
||
strings) that every provider ends up needing.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime, timezone
|
||
from typing import Any
|
||
|
||
_QUALITY_RANK = {"highest": 0, "high": 1, "low": 2}
|
||
_FORMAT_RANK = {"mp3": 0, "aac": 1, "aacp": 2}
|
||
_FORMAT_LABEL = {"aacp": "aac+", "aac": "aac", "mp3": "mp3"}
|
||
|
||
|
||
@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 Broadcast:
|
||
"""One entry in a catalogue's schedule: a show, a DJ slot, a stream."""
|
||
|
||
name: str
|
||
dj: str = ""
|
||
#: Where the broadcast comes from ("Poznan, Poland"), when the feed says.
|
||
location: str = ""
|
||
start: datetime | None = None
|
||
end: datetime | None = None
|
||
#: True for the broadcast that is on air right now.
|
||
live: bool = False
|
||
|
||
@property
|
||
def display_name(self) -> str:
|
||
return self.name.strip() or self.dj.strip() or "untitled broadcast"
|
||
|
||
def key(self) -> str:
|
||
return f"{self.name}\x1f{self.start.isoformat() if self.start else ''}"
|
||
|
||
|
||
@dataclass(slots=True)
|
||
class Station:
|
||
id: str
|
||
title: str
|
||
description: str = ""
|
||
genre: str = ""
|
||
dj: str = ""
|
||
#: Who is presenting right now when that is more than a DJ name —
|
||
#: ``"auto dj · library rotation"`` for a station between live shows.
|
||
presenter: 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
|
||
#: Id of the provider that produced this station (see
|
||
#: :mod:`beefy_player.providers`). Station ids are only unique within a
|
||
#: catalogue, so anything persisted across providers should use
|
||
#: :attr:`uid` instead.
|
||
provider: str = ""
|
||
|
||
@property
|
||
def uid(self) -> str:
|
||
"""Provider-qualified id, stable once more than one catalogue exists."""
|
||
return f"{self.provider}:{self.id}" if self.provider else self.id
|
||
|
||
@property
|
||
def genre_label(self) -> str:
|
||
return self.genre.replace("|", " · ") or "eclectic"
|
||
|
||
@property
|
||
def presenter_label(self) -> str:
|
||
"""What the ON AIR metadata line says about who is on the desk."""
|
||
if self.presenter.strip():
|
||
return self.presenter.strip()
|
||
if self.dj.strip():
|
||
return f"dj {self.dj.strip()}"
|
||
return ""
|
||
|
||
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 = _FORMAT_LABEL.get(
|
||
first.get("format", ""), first.get("format") or "stream"
|
||
)
|
||
return f"{fmt} · {playlist_bitrate(first)}k"
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# small, provider-agnostic parsers
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def playlist_bitrate(playlist: dict[str, str]) -> int:
|
||
"""The bitrate a catalogue declared, else the best guess from the URL."""
|
||
declared = to_int(playlist.get("bitrate"))
|
||
return declared or guess_bitrate(playlist.get("url", ""))
|
||
|
||
|
||
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:
|
||
"""Forgiving ``int`` — catalogues like to send ``"1,873"`` as text."""
|
||
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:
|
||
"""Unix timestamp, ISO 8601 or one of the usual ``%Y-%m-%d`` shapes."""
|
||
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
|
||
|
||
|
||
_SEPARATORS = (" - ", " – ", " — ", " -- ")
|
||
|
||
|
||
def parse_last_playing(text: str | None) -> Song | None:
|
||
"""Split the ``"Artist - Title"`` blurb most catalogues send."""
|
||
if not isinstance(text, str):
|
||
return None
|
||
cleaned = text.strip()
|
||
if not cleaned:
|
||
return None
|
||
for separator in _SEPARATORS:
|
||
if separator in cleaned:
|
||
artist, title = cleaned.split(separator, 1)
|
||
return Song(title=title.strip(), artist=artist.strip())
|
||
return Song(title=cleaned)
|