beefcarver
e30ca09627
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.
457 lines
15 KiB
Python
457 lines
15 KiB
Python
"""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)
|