369 lines
12 KiB
Python
369 lines
12 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``); this uses that instead
|
|
of scraping the search page. 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 html
|
|
import json
|
|
import re
|
|
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_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:
|
|
terms: list[str] = []
|
|
seen: set[str] = set()
|
|
for value in values:
|
|
text = _clean_text(value)
|
|
folded = text.casefold()
|
|
if text and folded not in seen:
|
|
seen.add(folded)
|
|
terms.append(text)
|
|
return " ".join(terms)
|
|
|
|
|
|
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 ")
|
|
return " ".join(re.findall(r"[a-z0-9]+", ascii_text.casefold()))
|
|
|
|
|
|
def _tokens(value: str) -> set[str]:
|
|
return set(_normalise(value).split())
|
|
|
|
|
|
def _matches(expected: str, actual: str) -> bool:
|
|
expected_tokens = _tokens(expected)
|
|
if not expected_tokens:
|
|
return True
|
|
actual_tokens = _tokens(actual)
|
|
if not actual_tokens:
|
|
return False
|
|
if expected_tokens <= actual_tokens or actual_tokens <= expected_tokens:
|
|
return True
|
|
overlap = len(expected_tokens & actual_tokens)
|
|
return overlap / len(expected_tokens) >= 0.6
|
|
|
|
|
|
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 find_track(
|
|
client: httpx.AsyncClient, artist: str, album: str, title: str
|
|
) -> str | None:
|
|
"""Find the now-playing track on Bandcamp.
|
|
|
|
Search for the artist + track first. If that finds nothing, search 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 = pick_track(
|
|
await _search(client, track_query, "t"), artist, "", title
|
|
)
|
|
if url:
|
|
return url
|
|
|
|
if not album:
|
|
return None
|
|
|
|
album_query = build_query(artist, album)
|
|
album_url = pick_album(await _search(client, album_query, "a"), artist, album)
|
|
if not album_url:
|
|
return None
|
|
|
|
return pick_track(await _album_tracks(client, album_url), artist, "", title)
|