beefy-player/beefy_player/bandcamp.py

"""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)