feat(bandcamp): add Bandcamp track lookup

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)

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 = """
+<html><body>
+  <div data-tralbum='{"trackinfo": [
+    {"title": "Rauschebart", "title_link": "/track/rauschebart?from=album", "track_id": 1},
+    {"title": "Other Track", "title_link": "/track/other-track", "track_id": 2}
+  ]}'></div>
+</body></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 = """
+    <script type="application/ld+json">
+    {"@type": "MusicAlbum", "track": [
+      {"@type": "MusicRecording", "name": "Rauschebart",
+       "url": "https://jensbuchert.bandcamp.com/track/rauschebart"}
+    ]}
+    </script>
+    """
+    results = bandcamp.parse_album_tracks(
+        json_ld, "https://jensbuchert.bandcamp.com/album/gravity"
+    )
+    assert results[0].url == "https://jensbuchert.bandcamp.com/track/rauschebart"
+
+    anchors = """
+    <html><body>
+      <a href="/track/rauschebart">Rauschebart</a>
+      <a href="/album/gravity">Gravity</a>
+    </body></html>
+    """
+    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="<html><title>Client Challenge</title></html>")
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(challenge)) as client:
+        assert await bandcamp.find_track(client, "A", "B", "C") is None
+    assert len(requests) == 2