feat(bandcamp): simulate typed autocomplete searches

somafm_tui/bandcamp.py

@@ -1,8 +1,11 @@
 """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
+(``/api/bcsearch_public_api/1/autocomplete_elastic``).  The endpoint is meant
-of scraping the search page.  It first searches for the artist and track,
+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.
@@ -10,9 +13,9 @@ result set — is treated as "no link" rather than as an error.
 
 from __future__ import annotations
 
+import difflib
 import html
 import json
-import re
 import unicodedata
 from dataclasses import dataclass
 from html.parser import HTMLParser
@@ -51,6 +54,27 @@ 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:
@@ -226,15 +250,43 @@ def parse_album_tracks(markup: str, album_url: str) -> list[SearchResult]:
 
 
 def build_query(*values: str) -> str:
-    terms: list[str] = []
+    """Join values into the lower-case query string sent while typing."""
-    seen: set[str] = set()
+    terms = [_clean_query_text(value) for value in values]
-    for value in values:
+    return " ".join(term for term in terms if term)
-        text = _clean_text(value)
+
-        folded = text.casefold()
+
-        if text and folded not in seen:
+MIN_TYPING_GROUP = 3
-            seen.add(folded)
+MAX_TYPING_GROUP = 5
-            terms.append(text)
+
-    return " ".join(terms)
+
+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:
@@ -242,8 +294,10 @@ def _normalise(value: str) -> str:
     ascii_text = "".join(
         character for character in decomposed if not unicodedata.combining(character)
     )
-    ascii_text = ascii_text.replace("&", " and ")
+    ascii_text = ascii_text.replace("&", " and ").casefold()
-    return " ".join(re.findall(r"[a-z0-9]+", ascii_text.casefold()))
+    return " ".join(
+        "".join(character if character.isalnum() else " " for character in ascii_text).split()
+    )
 
 
 def _tokens(value: str) -> set[str]:
@@ -251,16 +305,27 @@ def _tokens(value: str) -> set[str]:
 
 
 def _matches(expected: str, actual: str) -> bool:
-    expected_tokens = _tokens(expected)
+    """Case- and whitespace-insensitive fuzzy comparison of two labels."""
-    if not expected_tokens:
+    expected_key = _normalise(expected)
+    if not expected_key:
         return True
-    actual_tokens = _tokens(actual)
+    actual_key = _normalise(actual)
-    if not actual_tokens:
+    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)
-    return overlap / len(expected_tokens) >= 0.6
+    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(
@@ -336,13 +401,28 @@ async def _album_tracks(
     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.
 
-    Search for the artist + track first.  If that finds nothing, search for
+    Type the artist + track query into the autocomplete endpoint first.  If
-    the artist + album, open the album page and look for the track there.
+    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)
@@ -351,8 +431,11 @@ async def find_track(
         return None
 
     track_query = build_query(artist, title)
-    url = pick_track(
+    url = await _search_while_typing(
-        await _search(client, track_query, "t"), artist, "", title
+        client,
+        track_query,
+        "t",
+        lambda results: pick_track(results, artist, "", title),
     )
     if url:
         return url
@@ -361,7 +444,12 @@ async def find_track(
         return None
 
     album_query = build_query(artist, album)
-    album_url = pick_album(await _search(client, album_query, "a"), 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
 

tests/test_bandcamp.py

@@ -151,26 +151,60 @@ def test_pick_track_and_album_ignore_mismatches() -> None:
     assert bandcamp.pick_album(results, "Someone Else", "Gravity") is None
 
 
+def test_build_query_normalises_for_autocomplete() -> None:
+    assert bandcamp.build_query(
+        " Diskette Park ", "Dreams — [Extended Mix] (Live)!"
+    ) == "diskette park dreams extended mix live!"
+    assert bandcamp.build_query("Don't Stop", "AC/DC") == "dont stop acdc"
+
+
+def test_build_prefixes_grows_in_small_typing_groups() -> None:
+    assert bandcamp.build_prefixes("diskette park dreams") == [
+        "dis",
+        "disket",
+        "diskette ",
+        "diskette par",
+        "diskette park d",
+        "diskette park dreams",
+    ]
+    assert bandcamp.build_prefixes("abcde") == ["abcde"]
+    assert bandcamp.build_prefixes("") == []
+
+
+def test_matches_is_case_and_whitespace_insensitive() -> None:
+    assert bandcamp._matches("Jens   Buchert", "jens buchert")
+    assert bandcamp._matches("Rauschebart", "RAUSCHEBART (Live)")
+    assert not bandcamp._matches("Dreams", "Something Else")
+
+
 @pytest.mark.asyncio
 async def test_find_track_uses_track_search_first() -> None:
     requests: list[httpx.Request] = []
+    match_text = "jens buchert ra"
 
     def handler(request: httpx.Request) -> httpx.Response:
         requests.append(request)
         assert request.method == "POST"
-        assert json.loads(request.content) == {
+        body = json.loads(request.content)
-            "search_text": "Jens Buchert Rauschebart",
+        assert body["search_filter"] == "t"
-            "search_filter": "t",
+        assert body["full_page"] is True
-            "full_page": True,
+        assert body["fan_id"] is None
-            "fan_id": None,
+        payload = TRACK_PAYLOAD if body["search_text"] == match_text else {
+            "auto": {"results": []}
         }
-        return httpx.Response(200, json=TRACK_PAYLOAD)
+        return httpx.Response(200, json=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
+    assert [json.loads(request.content)["search_text"] for request in requests] == [
+        "jen",
+        "jens b",
+        "jens buch",
+        "jens buchert",
+        "jens buchert ra",
+    ]
 
 
 @pytest.mark.asyncio
@@ -184,7 +218,12 @@ async def test_find_track_falls_back_to_album_then_track_on_page() -> None:
             if body["search_filter"] == "t":
                 return httpx.Response(200, json={"auto": {"results": []}})
             if body["search_filter"] == "a":
-                return httpx.Response(200, json=ALBUM_PAYLOAD)
+                payload = (
+                    ALBUM_PAYLOAD
+                    if body["search_text"] == "jens buchert"
+                    else {"auto": {"results": []}}
+                )
+                return httpx.Response(200, json=payload)
         if request.url.path == "/album/gravity":
             return httpx.Response(200, text=ALBUM_PAGE_HTML)
         return httpx.Response(404)
@@ -195,15 +234,30 @@ async def test_find_track_falls_back_to_album_then_track_on_page() -> None:
         )
 
     assert url == "https://jensbuchert.bandcamp.com/track/rauschebart"
-    assert len(requests) == 3
+    search_bodies = [
-    search_bodies = [json.loads(request.content) for request in requests[:2]]
+        json.loads(request.content)
-    assert [body["search_text"] for body in search_bodies] == [
+        for request in requests
-        "Jens Buchert Rauschebart",
+        if request.method == "POST"
-        "Jens Buchert Gravity",
     ]
-    assert [body["search_filter"] for body in search_bodies] == ["t", "a"]
+    assert [body["search_text"] for body in search_bodies] == [
-    assert requests[2].method == "GET"
+        "jen",
-    assert requests[2].url.path == "/album/gravity"
+        "jens b",
+        "jens buch",
+        "jens buchert",
+        "jens buchert ra",
+        "jens buchert rausc",
+        "jens buchert rauscheb",
+        "jens buchert rauschebart",
+        "jen",
+        "jens b",
+        "jens buch",
+        "jens buchert",
+    ]
+    assert [body["search_filter"] for body in search_bodies] == ["t"] * 8 + [
+        "a"
+    ] * 4
+    assert requests[-1].method == "GET"
+    assert requests[-1].url.path == "/album/gravity"
 
 
 @pytest.mark.asyncio
@@ -216,4 +270,6 @@ async def test_find_track_returns_none_for_challenge_or_no_results() -> None:
 
     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
+    assert len(requests) == len(bandcamp.build_prefixes("a c")) + len(
+        bandcamp.build_prefixes("a b")
+    )