diff --git a/somafm_tui/bandcamp.py b/somafm_tui/bandcamp.py new file mode 100644 index 0000000..8ac5c64 --- /dev/null +++ b/somafm_tui/bandcamp.py @@ -0,0 +1,368 @@ +"""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) diff --git a/tests/test_bandcamp.py b/tests/test_bandcamp.py new file mode 100644 index 0000000..f8458ea --- /dev/null +++ b/tests/test_bandcamp.py @@ -0,0 +1,219 @@ +from __future__ import annotations + +import json + +import httpx +import pytest + +from somafm_tui import bandcamp + +TRACK_PAYLOAD = { + "auto": { + "results": [ + { + "type": "t", + "name": "Rauschebart", + "band_name": "Jens Buchert", + "album_name": "Gravity", + "item_url_root": "https://jensbuchert.bandcamp.com", + "item_url_path": ( + "https://jensbuchert.bandcamp.com/track/rauschebart?from=search" + ), + } + ] + } +} + +ALBUM_PAYLOAD = { + "auto": { + "results": [ + { + "type": "a", + "name": "Gravity", + "band_name": "Jens Buchert", + "item_url_root": "https://jensbuchert.bandcamp.com", + "item_url_path": ( + "https://jensbuchert.bandcamp.com/album/gravity?from=search" + ), + } + ] + } +} + +ALBUM_PAGE_HTML = """ +
+ + +""" + + +def test_parse_search_payload() -> None: + results = bandcamp.parse_search_payload(TRACK_PAYLOAD) + assert len(results) == 1 + result = results[0] + assert result.title == "Rauschebart" + assert result.artist == "Jens Buchert" + assert result.album == "Gravity" + assert result.item_type == "track" + # Tracking parameters are removed. + assert result.url == "https://jensbuchert.bandcamp.com/track/rauschebart" + + +def test_parse_search_payload_accepts_custom_domains_and_skips_other_types() -> None: + payload = { + "auto": { + "results": [ + { + "type": "t", + "name": "Rauschebart", + "band_name": "Jens Buchert", + "album_name": "Gravity", + "item_url_path": "https://music.example.com/track/rauschebart", + }, + { + "type": "b", + "name": "Jens Buchert", + "item_url_root": "https://jensbuchert.bandcamp.com", + }, + {"type": "t", "name": "", "item_url_path": "https://x.test/track/x"}, + ] + } + } + results = bandcamp.parse_search_payload(payload) + assert [result.url for result in results] == [ + "https://music.example.com/track/rauschebart" + ] + + +def test_parse_album_tracks_from_tralbum() -> None: + results = bandcamp.parse_album_tracks( + ALBUM_PAGE_HTML, "https://jensbuchert.bandcamp.com/album/gravity" + ) + assert [entry.title for entry in results] == ["Rauschebart", "Other Track"] + assert results[0].url == "https://jensbuchert.bandcamp.com/track/rauschebart" + + +def test_parse_album_tracks_from_json_ld_and_anchor_fallback() -> None: + json_ld = """ + + """ + results = bandcamp.parse_album_tracks( + json_ld, "https://jensbuchert.bandcamp.com/album/gravity" + ) + assert results[0].url == "https://jensbuchert.bandcamp.com/track/rauschebart" + + anchors = """ + + Rauschebart + Gravity + + """ + results = bandcamp.parse_album_tracks( + anchors, "https://jensbuchert.bandcamp.com/album/gravity" + ) + assert [entry.title for entry in results] == ["Rauschebart"] + + +def test_pick_track_and_album_ignore_mismatches() -> None: + results = bandcamp.parse_search_payload( + { + "auto": { + "results": [ + { + "type": "a", + "name": "Gravity", + "band_name": "Jens Buchert", + "item_url_path": "https://x.bandcamp.com/album/gravity", + }, + { + "type": "t", + "name": "Rauschebart", + "band_name": "Someone Else", + "album_name": "Other", + "item_url_path": "https://other.bandcamp.com/track/rauschebart", + }, + ] + } + } + ) + assert bandcamp.pick_track(results, "Jens Buchert", "Gravity", "Rauschebart") is None + assert bandcamp.pick_album(results, "Jens Buchert", "Gravity") == ( + "https://x.bandcamp.com/album/gravity" + ) + assert bandcamp.pick_album(results, "Someone Else", "Gravity") is None + + +@pytest.mark.asyncio +async def test_find_track_uses_track_search_first() -> None: + requests: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + assert request.method == "POST" + assert json.loads(request.content) == { + "search_text": "Jens Buchert Rauschebart", + "search_filter": "t", + "full_page": True, + "fan_id": None, + } + return httpx.Response(200, json=TRACK_PAYLOAD) + + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + url = await bandcamp.find_track(client, "Jens Buchert", "Gravity", "Rauschebart") + + assert url == "https://jensbuchert.bandcamp.com/track/rauschebart" + assert len(requests) == 1 + + +@pytest.mark.asyncio +async def test_find_track_falls_back_to_album_then_track_on_page() -> None: + requests: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + if request.url.path == "/api/bcsearch_public_api/1/autocomplete_elastic": + body = json.loads(request.content) + if body["search_filter"] == "t": + return httpx.Response(200, json={"auto": {"results": []}}) + if body["search_filter"] == "a": + return httpx.Response(200, json=ALBUM_PAYLOAD) + if request.url.path == "/album/gravity": + return httpx.Response(200, text=ALBUM_PAGE_HTML) + return httpx.Response(404) + + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + url = await bandcamp.find_track( + client, "Jens Buchert", "Gravity", "Rauschebart" + ) + + assert url == "https://jensbuchert.bandcamp.com/track/rauschebart" + assert len(requests) == 3 + search_bodies = [json.loads(request.content) for request in requests[:2]] + assert [body["search_text"] for body in search_bodies] == [ + "Jens Buchert Rauschebart", + "Jens Buchert Gravity", + ] + assert [body["search_filter"] for body in search_bodies] == ["t", "a"] + assert requests[2].method == "GET" + assert requests[2].url.path == "/album/gravity" + + +@pytest.mark.asyncio +async def test_find_track_returns_none_for_challenge_or_no_results() -> None: + requests: list[httpx.Request] = [] + + def challenge(request: httpx.Request) -> httpx.Response: + requests.append(request) + return httpx.Response(200, text="