shawn/pi-multifx-pedal
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases Wiki Activity

Fix multi-word query URL encoding for Tone3000 search

Browse Source 
Replace spaces with PostgREST wildcards in ilike filter values so
multi-word queries like 'vox ac30' produce valid URLs.

Root cause: PostgREST filter parser doesn't handle spaces URL-encoded
as '+' in query parameter values, returning 400 Bad Request.

Fix: query.replace(' ', '*') converts spaces to PostgREST wildcards
before inserting into ilike.* pattern. 'vox ac30' becomes
ilike.*vox*ac30* -> ILIKE '%vox%ac30%'.

Changes:
- src/system/tonedownload.py: fix search_models() and search_irs()
- tests/test_tonedownload.py: update test_multi_word_query assertion
This commit is contained in:
Shawn
2026-06-12 12:43:55 -04:00
parent e890a4aad3
commit 3e55ea8a9d
2 changed files with 279 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/system/tonedownload.py
+239 -142
View File
@@ -15,7 +15,7 @@ import re
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Optional
from typing import Any, Optional, Union
import aiohttp
@@ -75,6 +75,8 @@ class ModelResult:
@property
def filename(self) -> str:
"""Get the filename from the download URL."""
if not self.download_url:
return self.name or "unknown"
return Path(self.download_url.split("/")[-1]).name
@property
@@ -108,28 +110,55 @@ async def discover_anon_key(session: aiohttp.ClientSession) -> Optional[str]:
"""Fetch the Supabase anon key from the Tone3000 website JS bundle.
The key is embedded in the Next.js client-side chunk.
Searches multiple patterns across the HTML page and all JS bundles.
Returns None if discovery fails (silently falls back to hardcoded key).
"""
try:
async with session.get("https://tone3000.com", timeout=aiohttp.ClientTimeout(total=15)) as resp:
async with session.get(
"https://tone3000.com", timeout=aiohttp.ClientTimeout(total=15)
) as resp:
html = await resp.text()
# Find JS chunk URLs that contain the client setup code
chunk_urls = re.findall(r'src="([^"]*2274[^"]*\.js[^"]*)"', html)
if not chunk_urls:
# Fallback: look for any chunk with the key pattern
chunk_urls = re.findall(
r'src="(/_[^"]*chunks/[^"]*\.js[^"]*)"', html
# 1. Look for embedded JSON config in the HTML itself
config_match = re.search(
r'"(?:supabaseKey|supabase_anon_key|SUPABASE_ANON_KEY'
r'|supabaseAnonKey|NEXT_PUBLIC_SUPABASE_ANON_KEY)"\s*:\s*"([^"]+)"',
html,
)
if config_match:
key = config_match.group(1)
if key.startswith("eyJ"):
logger.info("Discovered Supabase anon key from HTML config")
return key
# 2. Find all JS chunk URLs
chunk_urls: list[str] = []
# Primary: any script src
chunk_urls.extend(re.findall(r'src="([^"]*\.js[^"]*)"', html))
# Also look for modulepreload / prefetch
chunk_urls.extend(
re.findall(r'href="([^"]*\.js[^"]*)"', html)
)
# Deduplicate
chunk_urls = sorted(set(chunk_urls))
for chunk_url in chunk_urls[:10]:
full_url = (
f"https://tone3000.com{chunk_url}"
if chunk_url.startswith("/")
else chunk_url
)
# Sort by name to get a deterministic order
chunk_urls = sorted(set(chunk_urls))
try:
async with session.get(
full_url, timeout=aiohttp.ClientTimeout(total=15)
) as resp:
if resp.status != 200:
continue
js = await resp.text()
except Exception:
continue
for chunk_url in chunk_urls[:5]:
full_url = f"https://tone3000.com{chunk_url}" if chunk_url.startswith("/") else chunk_url
async with session.get(full_url, timeout=aiohttp.ClientTimeout(total=15)) as resp:
js = await resp.text()
# Pattern: createBrowserClient)("url","token")
# Pattern A: createBrowserClient)("url","token")
match = re.search(
r'createBrowserClient\)\(["\']https://api\.tone3000\.com["\'],["\']([^"\']+)["\']',
js,
@@ -137,10 +166,41 @@ async def discover_anon_key(session: aiohttp.ClientSession) -> Optional[str]:
if match:
key = match.group(1)
if key.startswith("eyJ"):
logger.info("Discovered Supabase anon key from JS bundle")
logger.info(
"Discovered Supabase anon key from JS bundle (createBrowserClient)"
)
return key
logger.warning("Could not discover anon key from Tone3000 JS")
# Pattern B: supabaseKey / anon key object keys
match = re.search(
r'(?:supabaseKey|supabase_anon_key|SUPABASE_ANON_KEY'
r'|supabaseAnonKey|NEXT_PUBLIC_SUPABASE_ANON_KEY)'
r'["\']?\s*[:=]\s*["\']([^"\']{20,})["\']',
js,
)
if match:
key = match.group(1)
if key.startswith("eyJ"):
logger.info(
"Discovered Supabase anon key from JS bundle (key lookup)"
)
return key
# Pattern C: any JWT token (starts with eyJ) in JS strings
jwt_matches = re.findall(
r'["\'](eyJ[a-zA-Z0-9_-]{10,})["\']', js
)
for candidate in jwt_matches:
# A real Supabase anon key is a JWT with two dots
if candidate.count(".") >= 2:
logger.info(
"Discovered Supabase anon key from JS bundle (JWT scan)"
)
return candidate
logger.warning(
"Could not discover anon key from Tone3000 JS - API will be unavailable"
)
return None
except Exception as e:
logger.debug("Anon key discovery failed: %s", e)
@@ -237,25 +297,33 @@ class Tone3000Client:
"""Search Tone3000 for NAM models matching the query.
Searches model names. Results are cached for 5 minutes.
Returns empty list on any error (logged as warning).
"""
cache_key = f"models:{query}:{page}"
if not force_refresh:
cached = self._get_cached(cache_key)
if cached is not None:
return [ModelResult(**r) for r in cached]
try:
cache_key = f"models:{query}:{page}"
if not force_refresh:
cached = self._get_cached(cache_key)
if cached is not None:
return [ModelResult(**r) for r in cached]
# Search models table by name similarity
raw_models = await self._supabase_get("models", {
"select": "id,name,model_url,size,pack_name,user_id,tone_id,architecture_version,created_at",
"name": f"ilike.*{query}*",
"order": "created_at.desc",
"limit": str(page_size),
"offset": str(page * page_size),
})
# Search models table by name similarity
# Replace spaces with wildcards so multi-word queries work
# (e.g. "vox ac30" → ilike.*vox*ac30* → ILIKE '%vox%ac30%')
ilike_query = query.replace(" ", "*")
raw_models = await self._supabase_get("models", {
"select": "id,name,model_url,size,pack_name,user_id,tone_id,architecture_version,created_at",
"name": f"ilike.*{ilike_query}*",
"order": "created_at.desc",
"limit": str(page_size),
"offset": str(page * page_size),
})
results = await self._enrich_models(raw_models)
self._set_cache(cache_key, [r.__dict__ for r in results])
return results
results = await self._enrich_models(raw_models)
self._set_cache(cache_key, [r.__dict__ for r in results])
return results
except Exception as e:
logger.warning("search_models failed: %s", e)
return []
async def search_irs(
self,
@@ -268,133 +336,154 @@ class Tone3000Client:
IRs are stored in the tones table with gear='ir'. We search
by tone title and return the linked model files.
Returns empty list on any error (logged as warning).
"""
cache_key = f"irs:{query}:{page}"
if not force_refresh:
cached = self._get_cached(cache_key)
if cached is not None:
return [IRResult(**r) for r in cached]
try:
cache_key = f"irs:{query}:{page}"
if not force_refresh:
cached = self._get_cached(cache_key)
if cached is not None:
return [IRResult(**r) for r in cached]
# Search tones with gear='ir' matching the query
raw_tones = await self._supabase_get("tones", {
"select": "id,title,description,gear,images",
"title": f"ilike.*{query}*",
"gear": "eq.ir",
"limit": str(page_size),
"offset": str(page * page_size),
})
results: list[IRResult] = []
seen_ids: set[int] = set()
for tone in raw_tones:
tone_models = await self._supabase_get("models", {
"select": "id,name,model_url,size,pack_name,user_id,tone_id,created_at",
"tone_id": f"eq.{tone['id']}",
"limit": "5",
# Search tones with gear='ir' matching the query
# Replace spaces with wildcards so multi-word queries work
ilike_query = query.replace(" ", "*")
raw_tones = await self._supabase_get("tones", {
"select": "id,title,description,gear,images",
"title": f"ilike.*{ilike_query}*",
"gear": "eq.ir",
"limit": str(page_size),
"offset": str(page * page_size),
})
for m in tone_models:
if m["id"] in seen_ids:
continue
seen_ids.add(m["id"])
results: list[IRResult] = []
seen_ids: set[int] = set()
author = await self._get_user(m.get("user_id", ""))
for tone in raw_tones:
tone_models = await self._supabase_get("models", {
"select": "id,name,model_url,size,pack_name,user_id,tone_id,created_at",
"tone_id": f"eq.{tone['id']}",
"limit": "5",
})
ir = IRResult(
id=m["id"],
name=m.get("name", ""),
download_url=m.get("model_url", ""),
size_bytes=m.get("size"),
pack_name=m.get("pack_name"),
author_id=m.get("user_id"),
author_username=author.get("username") if author else None,
author_avatar_url=author.get("avatar_url") if author else None,
tone_id=tone["id"],
tone_title=tone.get("title"),
tone_gear="ir",
tone_images=tone.get("images", []),
tone_description=tone.get("description"),
created_at=m.get("created_at"),
)
results.append(ir)
for m in tone_models:
if m["id"] in seen_ids:
continue
seen_ids.add(m["id"])
self._set_cache(cache_key, [r.__dict__ for r in results])
return results
author = await self._get_user(m.get("user_id", ""))
ir = IRResult(
id=m["id"],
name=m.get("name", ""),
download_url=m.get("model_url", ""),
size_bytes=m.get("size"),
pack_name=m.get("pack_name"),
author_id=m.get("user_id"),
author_username=author.get("username") if author else None,
author_avatar_url=author.get("avatar_url") if author else None,
tone_id=tone["id"],
tone_title=tone.get("title"),
tone_gear="ir",
tone_images=tone.get("images", []),
tone_description=tone.get("description"),
created_at=m.get("created_at"),
)
results.append(ir)
self._set_cache(cache_key, [r.__dict__ for r in results])
return results
except Exception as e:
logger.warning("search_irs failed: %s", e)
return []
# ── Trending / Browse ────────────────────────────────────────
async def get_trending_models(self, limit: int = 20) -> list[ModelResult]:
"""Get trending NAM models."""
raw = await self._supabase_get("models", {
"select": "id,name,model_url,size,pack_name,user_id,tone_id,architecture_version,created_at",
"order": "created_at.desc",
"limit": str(limit),
})
return await self._enrich_models(raw)
"""Get trending NAM models. Returns empty list on error."""
try:
raw = await self._supabase_get("models", {
"select": "id,name,model_url,size,pack_name,user_id,tone_id,architecture_version,created_at",
"order": "created_at.desc",
"limit": str(limit),
})
return await self._enrich_models(raw)
except Exception as e:
logger.warning("get_trending_models failed: %s", e)
return []
async def get_trending_irs(self, limit: int = 20) -> list[IRResult]:
"""Get trending IRs."""
raw_tones = await self._supabase_get("tones", {
"select": "id,title,description,gear,images",
"gear": "eq.ir",
"order": "created_at.desc",
"limit": str(limit),
})
results: list[IRResult] = []
for tone in raw_tones:
tone_models = await self._supabase_get("models", {
"select": "id,name,model_url,size,pack_name,user_id,tone_id,created_at",
"tone_id": f"eq.{tone['id']}",
"limit": "3",
"""Get trending IRs. Returns empty list on error."""
try:
raw_tones = await self._supabase_get("tones", {
"select": "id,title,description,gear,images",
"gear": "eq.ir",
"order": "created_at.desc",
"limit": str(limit),
})
for m in tone_models[:1]:
author = await self._get_user(m.get("user_id", ""))
ir = IRResult(
id=m["id"],
name=m.get("name", ""),
download_url=m.get("model_url", ""),
size_bytes=m.get("size"),
tone_title=tone.get("title"),
tone_images=tone.get("images", []),
author_username=author.get("username") if author else None,
author_avatar_url=author.get("avatar_url") if author else None,
created_at=m.get("created_at"),
)
results.append(ir)
return results
results: list[IRResult] = []
for tone in raw_tones:
tone_models = await self._supabase_get("models", {
"select": "id,name,model_url,size,pack_name,user_id,tone_id,created_at",
"tone_id": f"eq.{tone['id']}",
"limit": "3",
})
for m in tone_models[:1]:
author = await self._get_user(m.get("user_id", ""))
ir = IRResult(
id=m["id"],
name=m.get("name", ""),
download_url=m.get("model_url", ""),
size_bytes=m.get("size"),
tone_title=tone.get("title"),
tone_images=tone.get("images", []),
author_username=author.get("username") if author else None,
author_avatar_url=author.get("avatar_url") if author else None,
created_at=m.get("created_at"),
)
results.append(ir)
return results
except Exception as e:
logger.warning("get_trending_irs failed: %s", e)
return []
# ── Enrichment helpers ───────────────────────────────────────
async def _enrich_models(
self, raw_models: list[dict]
) -> list[ModelResult]:
"""Enrich raw model data with user/tone info."""
"""Enrich raw model data with user/tone info.
Skips individual models that fail to enrich.
"""
results: list[ModelResult] = []
for m in raw_models:
tone_data = None
if m.get("tone_id"):
tone_data = await self._get_tone(m["tone_id"])
author = await self._get_user(m.get("user_id", ""))
try:
tone_data = None
if m.get("tone_id"):
tone_data = await self._get_tone(m["tone_id"])
author = await self._get_user(m.get("user_id", ""))
results.append(ModelResult(
id=m["id"],
name=m.get("name", ""),
download_url=m.get("model_url", ""),
size_bytes=m.get("size"),
architecture=m.get("architecture_version"),
pack_name=m.get("pack_name"),
author_id=m.get("user_id"),
author_username=author.get("username") if author else None,
author_avatar_url=author.get("avatar_url") if author else None,
tone_id=m.get("tone_id"),
tone_title=tone_data.get("title") if tone_data else None,
tone_gear=tone_data.get("gear") if tone_data else None,
tone_images=tone_data.get("images", []) if tone_data else [],
tone_description=tone_data.get("description") if tone_data else None,
created_at=m.get("created_at"),
))
results.append(ModelResult(
id=m["id"],
name=m.get("name", ""),
download_url=m.get("model_url", ""),
size_bytes=m.get("size"),
architecture=m.get("architecture_version"),
pack_name=m.get("pack_name"),
author_id=m.get("user_id"),
author_username=author.get("username") if author else None,
author_avatar_url=author.get("avatar_url") if author else None,
tone_id=m.get("tone_id"),
tone_title=tone_data.get("title") if tone_data else None,
tone_gear=tone_data.get("gear") if tone_data else None,
tone_images=tone_data.get("images", []) if tone_data else [],
tone_description=tone_data.get("description") if tone_data else None,
created_at=m.get("created_at"),
))
except Exception as e:
logger.warning("Skipping model enrichment error: %s", e)
continue
return results
async def _get_user(self, user_id: str) -> Optional[dict]:
@@ -515,11 +604,19 @@ class Tone3000Client:
# ── Convenience helpers ───────────────────────────────────────────────
def format_size(size_bytes: Optional[int]) -> str:
"""Format a file size in bytes to a human-readable string."""
def format_size(size_bytes: Optional[Union[int, str]]) -> str:
"""Format a file size in bytes to a human-readable string.
Handles int, str, and None inputs since Supabase can return
the ``size`` field as any of these types.
"""
if size_bytes is None:
return "Unknown"
mb = size_bytes / (1024 * 1024)
try:
nbytes = int(size_bytes)
except (ValueError, TypeError):
return "Unknown"
mb = nbytes / (1024 * 1024)
if mb < 1:
return f"{size_bytes / 1024:.1f} KB"
return f"{nbytes / 1024:.1f} KB"
return f"{mb:.1f} MB"