refactor!: rename somafm-tui to beefy-player
The app is now a general internet-radio player rather than a SomaFM client, so the package, console script, MPRIS identity, mpv namespaces, macOS tap names, XDG directories and docs all move to beefy-player. Environment variables are now BEEFY_* (the old SOMAFM_* spellings are still honoured) and an existing ~/.config/somafm-tui/config.json is migrated on first run, so favourites and volume survive the rename. The visible interface is untouched: same layout, same logo, same keys.
This commit is contained in:
beefcarver committed
1 parent
4bed90b489
commit
e30ca09627
44 files changed
+329
-192
No files matched your search
@@ -0,0 +1,456 @@
|
||||
"""Look up the now-playing track on Bandcamp.
|
||||
|
||||
Bandcamp's own search box talks to an unauthenticated JSON endpoint
|
||||
(``/api/bcsearch_public_api/1/autocomplete_elastic``). The endpoint is meant
|
||||
to be driven by a person typing, so this module mimics that: the query is
|
||||
normalised, then sent as a series of growing chunks (three to five characters
|
||||
at a time). After every chunk the returned results are matched fuzzily
|
||||
against the artist and track. It first searches for the artist and track,
|
||||
then falls back to the artist and album and reads the track link from that
|
||||
album page. Anything unexpected — a bot challenge, a layout change, an empty
|
||||
result set — is treated as "no link" rather than as an error.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import difflib
|
||||
import html
|
||||
import json
|
||||
import unicodedata
|
||||
from dataclasses import dataclass
|
||||
from html.parser import HTMLParser
|
||||
from urllib.parse import urljoin, urlparse, urlunparse
|
||||
|
||||
import httpx
|
||||
|
||||
SEARCH_API_URL = (
|
||||
"https://bandcamp.com/api/bcsearch_public_api/1/autocomplete_elastic"
|
||||
)
|
||||
BROWSER_UA = (
|
||||
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
|
||||
"(KHTML, like Gecko) Chrome/124.0 Safari/537.36"
|
||||
)
|
||||
SEARCH_HEADERS = {
|
||||
"User-Agent": BROWSER_UA,
|
||||
"Accept": "application/json, text/plain, */*",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
PAGE_HEADERS = {
|
||||
"User-Agent": BROWSER_UA,
|
||||
"Accept": "text/html,application/xhtml+xml",
|
||||
}
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class SearchResult:
|
||||
title: str = ""
|
||||
artist: str = ""
|
||||
album: str = ""
|
||||
url: str = ""
|
||||
item_type: str = ""
|
||||
|
||||
|
||||
def _clean_text(value: str) -> str:
|
||||
return " ".join(html.unescape(value or "").split())
|
||||
|
||||
|
||||
def _clean_query_text(value: str) -> str:
|
||||
"""Normalise text for the autocomplete endpoint.
|
||||
|
||||
Em dashes, brackets, parentheses and every other punctuation mark except
|
||||
comma, period, exclamation mark and question mark are removed. The
|
||||
result is lower-cased and whitespace is collapsed. Em dashes become a
|
||||
space so that ``Artist—Track`` does not turn into one word.
|
||||
"""
|
||||
decomposed = unicodedata.normalize("NFKD", html.unescape(value or ""))
|
||||
text = "".join(
|
||||
character for character in decomposed if not unicodedata.combining(character)
|
||||
).casefold()
|
||||
cleaned: list[str] = []
|
||||
for character in text:
|
||||
if character.isalnum() or character in ",.!?":
|
||||
cleaned.append(character)
|
||||
elif character.isspace() or character in "—–―":
|
||||
cleaned.append(" ")
|
||||
return " ".join("".join(cleaned).split())
|
||||
|
||||
|
||||
def _clean_url(value: str, base: str = "https://bandcamp.com/") -> str:
|
||||
text = html.unescape(value or "").strip()
|
||||
if not text:
|
||||
return ""
|
||||
absolute = urljoin(base, text)
|
||||
if any(character.isspace() for character in absolute):
|
||||
return ""
|
||||
parsed = urlparse(absolute)
|
||||
host = (parsed.hostname or "").casefold()
|
||||
if parsed.scheme not in ("http", "https") or not host:
|
||||
return ""
|
||||
if parsed.username or parsed.password:
|
||||
return ""
|
||||
path = parsed.path.rstrip("/")
|
||||
if "/track/" not in path and "/album/" not in path:
|
||||
return ""
|
||||
# Search results can live on a Bandcamp Pro custom domain (for example
|
||||
# music.example.com), so keep the host instead of requiring bandcamp.com.
|
||||
# Drop search tracking parameters and fragments.
|
||||
return urlunparse(("https", host, path, "", "", ""))
|
||||
|
||||
|
||||
def parse_search_payload(payload: object) -> list[SearchResult]:
|
||||
"""Convert an autocomplete_elastic response into search results."""
|
||||
auto = payload.get("auto") if isinstance(payload, dict) else None
|
||||
entries = auto.get("results") if isinstance(auto, dict) else None
|
||||
if not isinstance(entries, list):
|
||||
return []
|
||||
|
||||
results: list[SearchResult] = []
|
||||
for entry in entries:
|
||||
if not isinstance(entry, dict):
|
||||
continue
|
||||
kind = str(entry.get("type") or "").casefold()
|
||||
if kind not in ("a", "t"):
|
||||
continue
|
||||
title = _clean_text(str(entry.get("name") or ""))
|
||||
url = _clean_url(
|
||||
str(entry.get("item_url_path") or entry.get("item_url_root") or "")
|
||||
)
|
||||
if not title or not url:
|
||||
continue
|
||||
results.append(
|
||||
SearchResult(
|
||||
title=title,
|
||||
artist=_clean_text(str(entry.get("band_name") or "")),
|
||||
album=_clean_text(str(entry.get("album_name") or "")),
|
||||
url=url,
|
||||
item_type="album" if kind == "a" else "track",
|
||||
)
|
||||
)
|
||||
return results
|
||||
|
||||
|
||||
class _AlbumParser(HTMLParser):
|
||||
"""Collect album-page track links from tralbum/JSON-LD/anchors."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
super().__init__(convert_charrefs=True)
|
||||
self.tralbum = ""
|
||||
self.json_ld: list[str] = []
|
||||
self.anchors: list[tuple[str, str]] = []
|
||||
self._script_active = False
|
||||
self._script_buffer: list[str] = []
|
||||
self._anchor_href = ""
|
||||
self._anchor_buffer: list[str] = []
|
||||
|
||||
def handle_starttag(self, tag: str, attrs) -> None:
|
||||
attributes = {key: value or "" for key, value in attrs}
|
||||
raw = attributes.get("data-tralbum", "")
|
||||
if raw and not self.tralbum:
|
||||
self.tralbum = raw
|
||||
if tag == "script" and "ld+json" in attributes.get("type", "").casefold():
|
||||
self._script_active = True
|
||||
self._script_buffer = []
|
||||
if tag == "a" and attributes.get("href"):
|
||||
self._anchor_href = attributes["href"].strip()
|
||||
self._anchor_buffer = []
|
||||
|
||||
def handle_data(self, data: str) -> None:
|
||||
if self._script_active:
|
||||
self._script_buffer.append(data)
|
||||
if self._anchor_href:
|
||||
self._anchor_buffer.append(data)
|
||||
|
||||
def handle_endtag(self, tag: str) -> None:
|
||||
if tag == "script" and self._script_active:
|
||||
self.json_ld.append("".join(self._script_buffer))
|
||||
self._script_active = False
|
||||
self._script_buffer = []
|
||||
if tag == "a" and self._anchor_href:
|
||||
self.anchors.append(
|
||||
(self._anchor_href, _clean_text("".join(self._anchor_buffer)))
|
||||
)
|
||||
self._anchor_href = ""
|
||||
self._anchor_buffer = []
|
||||
|
||||
|
||||
def _json_tracks(payload: object) -> list[SearchResult]:
|
||||
"""Find MusicRecording-ish objects in an arbitrary JSON-LD document."""
|
||||
found: list[SearchResult] = []
|
||||
if isinstance(payload, list):
|
||||
for item in payload:
|
||||
found.extend(_json_tracks(item))
|
||||
return found
|
||||
if not isinstance(payload, dict):
|
||||
return found
|
||||
|
||||
name = payload.get("name")
|
||||
url = payload.get("url") or payload.get("@id")
|
||||
if isinstance(name, str) and isinstance(url, str) and "/track/" in url:
|
||||
found.append(SearchResult(title=_clean_text(name), url=url))
|
||||
|
||||
for value in payload.values():
|
||||
if isinstance(value, (dict, list)):
|
||||
found.extend(_json_tracks(value))
|
||||
return found
|
||||
|
||||
|
||||
def parse_album_tracks(markup: str, album_url: str) -> list[SearchResult]:
|
||||
"""Extract track links from an album page."""
|
||||
parser = _AlbumParser()
|
||||
try:
|
||||
parser.feed(markup or "")
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
results: list[SearchResult] = []
|
||||
seen: set[str] = set()
|
||||
|
||||
def add(title: str, url: str) -> None:
|
||||
clean = _clean_url(url, album_url)
|
||||
if not clean or "/track/" not in clean or clean in seen:
|
||||
return
|
||||
seen.add(clean)
|
||||
results.append(SearchResult(title=_clean_text(title), url=clean))
|
||||
|
||||
if parser.tralbum:
|
||||
try:
|
||||
data = json.loads(html.unescape(parser.tralbum))
|
||||
except (TypeError, ValueError):
|
||||
data = None
|
||||
trackinfo = data.get("trackinfo") if isinstance(data, dict) else None
|
||||
if isinstance(trackinfo, list):
|
||||
for entry in trackinfo:
|
||||
if not isinstance(entry, dict):
|
||||
continue
|
||||
add(
|
||||
str(entry.get("title") or ""),
|
||||
str(
|
||||
entry.get("title_link")
|
||||
or entry.get("titleLink")
|
||||
or entry.get("url")
|
||||
or ""
|
||||
),
|
||||
)
|
||||
|
||||
if not results:
|
||||
for raw in parser.json_ld:
|
||||
try:
|
||||
payload = json.loads(html.unescape(raw))
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
for item in _json_tracks(payload):
|
||||
add(item.title, item.url)
|
||||
|
||||
if not results:
|
||||
for href, text in parser.anchors:
|
||||
if "/track/" in href:
|
||||
add(text, href)
|
||||
|
||||
return results
|
||||
|
||||
|
||||
def build_query(*values: str) -> str:
|
||||
"""Join values into the lower-case query string sent while typing."""
|
||||
terms = [_clean_query_text(value) for value in values]
|
||||
return " ".join(term for term in terms if term)
|
||||
|
||||
|
||||
MIN_TYPING_GROUP = 3
|
||||
MAX_TYPING_GROUP = 5
|
||||
|
||||
|
||||
def build_prefixes(query: str) -> list[str]:
|
||||
"""Return the growing strings produced by "typing" *query*.
|
||||
|
||||
Each group is between :data:`MIN_TYPING_GROUP` and
|
||||
:data:`MAX_TYPING_GROUP` characters long, and every prefix contains all
|
||||
of the text typed so far. For example ``diskette park dreams`` produces
|
||||
``dis``, ``disket``, ``diskette ``, ``diskette par``, …
|
||||
"""
|
||||
text = query.strip()
|
||||
if not text:
|
||||
return []
|
||||
|
||||
groups: list[str] = []
|
||||
index = 0
|
||||
while index < len(text):
|
||||
remaining = len(text) - index
|
||||
if remaining <= MAX_TYPING_GROUP:
|
||||
groups.append(text[index:])
|
||||
break
|
||||
groups.append(text[index : index + MIN_TYPING_GROUP])
|
||||
index += MIN_TYPING_GROUP
|
||||
|
||||
prefixes: list[str] = []
|
||||
typed = ""
|
||||
for group in groups:
|
||||
typed += group
|
||||
prefixes.append(typed)
|
||||
return prefixes
|
||||
|
||||
|
||||
def _normalise(value: str) -> str:
|
||||
decomposed = unicodedata.normalize("NFKD", value or "")
|
||||
ascii_text = "".join(
|
||||
character for character in decomposed if not unicodedata.combining(character)
|
||||
)
|
||||
ascii_text = ascii_text.replace("&", " and ").casefold()
|
||||
return " ".join(
|
||||
"".join(character if character.isalnum() else " " for character in ascii_text).split()
|
||||
)
|
||||
|
||||
|
||||
def _tokens(value: str) -> set[str]:
|
||||
return set(_normalise(value).split())
|
||||
|
||||
|
||||
def _matches(expected: str, actual: str) -> bool:
|
||||
"""Case- and whitespace-insensitive fuzzy comparison of two labels."""
|
||||
expected_key = _normalise(expected)
|
||||
if not expected_key:
|
||||
return True
|
||||
actual_key = _normalise(actual)
|
||||
if not actual_key:
|
||||
return False
|
||||
if expected_key == actual_key:
|
||||
return True
|
||||
|
||||
expected_tokens = set(expected_key.split())
|
||||
actual_tokens = set(actual_key.split())
|
||||
if expected_tokens <= actual_tokens or actual_tokens <= expected_tokens:
|
||||
return True
|
||||
overlap = len(expected_tokens & actual_tokens)
|
||||
if overlap / len(expected_tokens) >= 0.6:
|
||||
return True
|
||||
|
||||
# Tolerate small spelling differences and typos once the word-level
|
||||
# comparison has failed.
|
||||
return difflib.SequenceMatcher(None, expected_key, actual_key).ratio() >= 0.85
|
||||
|
||||
|
||||
def pick_track(
|
||||
results: list[SearchResult], artist: str, album: str, title: str
|
||||
) -> str | None:
|
||||
"""Return the best matching Bandcamp *track* URL, if there is one."""
|
||||
for result in results:
|
||||
if "/track/" not in result.url:
|
||||
continue
|
||||
if result.item_type and "track" not in result.item_type:
|
||||
continue
|
||||
if not _matches(title, result.title):
|
||||
continue
|
||||
# Track results normally carry the artist; if a layout change drops
|
||||
# it, the search query still keeps the title match reasonably
|
||||
# specific.
|
||||
if artist and result.artist and not _matches(artist, result.artist):
|
||||
continue
|
||||
if album and result.album and not _matches(album, result.album):
|
||||
continue
|
||||
return result.url
|
||||
return None
|
||||
|
||||
|
||||
def pick_album(
|
||||
results: list[SearchResult], artist: str, album: str
|
||||
) -> str | None:
|
||||
"""Return the best matching Bandcamp album URL, if there is one."""
|
||||
for result in results:
|
||||
if "/album/" not in result.url:
|
||||
continue
|
||||
if result.item_type and "album" not in result.item_type:
|
||||
continue
|
||||
if not _matches(album, result.title):
|
||||
continue
|
||||
if artist and result.artist and not _matches(artist, result.artist):
|
||||
continue
|
||||
return result.url
|
||||
return None
|
||||
|
||||
|
||||
async def _search(
|
||||
client: httpx.AsyncClient, query: str, search_filter: str
|
||||
) -> list[SearchResult]:
|
||||
if not query:
|
||||
return []
|
||||
try:
|
||||
response = await client.post(
|
||||
SEARCH_API_URL,
|
||||
json={
|
||||
"search_text": query,
|
||||
"search_filter": search_filter,
|
||||
"full_page": True,
|
||||
"fan_id": None,
|
||||
},
|
||||
headers=SEARCH_HEADERS,
|
||||
)
|
||||
response.raise_for_status()
|
||||
payload = response.json()
|
||||
except (httpx.HTTPError, ValueError):
|
||||
return []
|
||||
return parse_search_payload(payload)
|
||||
|
||||
|
||||
async def _album_tracks(
|
||||
client: httpx.AsyncClient, album_url: str
|
||||
) -> list[SearchResult]:
|
||||
try:
|
||||
response = await client.get(album_url, headers=PAGE_HEADERS)
|
||||
response.raise_for_status()
|
||||
except httpx.HTTPError:
|
||||
return []
|
||||
return parse_album_tracks(response.text, album_url)
|
||||
|
||||
|
||||
async def _search_while_typing(
|
||||
client: httpx.AsyncClient,
|
||||
query: str,
|
||||
search_filter: str,
|
||||
pick,
|
||||
) -> str | None:
|
||||
"""Send growing prefixes of *query* and return the first selected URL."""
|
||||
for prefix in build_prefixes(query):
|
||||
url = pick(await _search(client, prefix, search_filter))
|
||||
if url:
|
||||
return url
|
||||
return None
|
||||
|
||||
|
||||
async def find_track(
|
||||
client: httpx.AsyncClient, artist: str, album: str, title: str
|
||||
) -> str | None:
|
||||
"""Find the now-playing track on Bandcamp.
|
||||
|
||||
Type the artist + track query into the autocomplete endpoint first. If
|
||||
that finds nothing, do the same for the artist + album, open the album
|
||||
page and look for the track there.
|
||||
"""
|
||||
artist = _clean_text(artist)
|
||||
album = _clean_text(album)
|
||||
title = _clean_text(title)
|
||||
if not title:
|
||||
return None
|
||||
|
||||
track_query = build_query(artist, title)
|
||||
url = await _search_while_typing(
|
||||
client,
|
||||
track_query,
|
||||
"t",
|
||||
lambda results: pick_track(results, artist, "", title),
|
||||
)
|
||||
if url:
|
||||
return url
|
||||
|
||||
if not album:
|
||||
return None
|
||||
|
||||
album_query = build_query(artist, album)
|
||||
album_url = await _search_while_typing(
|
||||
client,
|
||||
album_query,
|
||||
"a",
|
||||
lambda results: pick_album(results, artist, album),
|
||||
)
|
||||
if not album_url:
|
||||
return None
|
||||
|
||||
return pick_track(await _album_tracks(client, album_url), artist, "", title)
|
||||
Reference in new issue
Block a user