feat(tonedownload): Tone3000 NAM + IR browser, real anon key, tests
- Tonedownload.py: Tone3000 API client with search/download/cache/enrich - Server.py: 4 endpoints (search models, search IRs, install model, install IR) - Models.html + irs.html: Browse Tone3000 tab with search, trending, offline - Models.js + irs.js: Tone3000 search UI with install tracking via localStorage - Style.css: Card layout for search results with thumbnail, meta, install btn - Tests: 47 tests covering data classes, cache, API, search, download, enrichment - Anon key: Updated from placeholder to real Supabase anon key (verified: API works)
This commit is contained in:
@@ -0,0 +1,525 @@
|
||||
"""Tone3000 API client for searching and downloading NAM models and IRs.
|
||||
|
||||
Tone3000 (successor to ToneHunt) is a Supabase-backed Next.js app that
|
||||
hosts NAM .nam model files and IR .wav files. Its Supabase public anon key
|
||||
is embedded in the client-side JavaScript — we use it here as a read-only
|
||||
API client for the Supabase REST API.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
import time
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Any, Optional
|
||||
|
||||
import aiohttp
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# ── Constants ─────────────────────────────────────────────────────────
|
||||
|
||||
SUPABASE_URL = "https://api.tone3000.com"
|
||||
SUPABASE_REST = f"{SUPABASE_URL}/rest/v1"
|
||||
SUPABASE_STORAGE = f"{SUPABASE_URL}/storage/v1/object/public"
|
||||
|
||||
# Anon key for Supabase read-only access. Extracted from the Tone3000
|
||||
# Next.js client bundle. This is a public key designed to be embedded
|
||||
# in client-side JS — it only allows read access to public tables.
|
||||
# If it ever changes, the client will dynamically re-fetch it from
|
||||
# the Tone3000 homepage JS bundles.
|
||||
SUPABASE_ANON_KEY = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZSIsInJlZiI6Imd6eWJpdW9weGtkeGJ5dG5vamRzIiwicm9sZSI6ImFub24iLCJpYXQiOjE3MzgwODIxNjUsImV4cCI6MjA1MzY1ODE2NX0.Gq66BJXjtLsqP2nAGXm9Xb9PAjoeZalWUj66K4nmVSU"
|
||||
|
||||
# Cache TTL in seconds
|
||||
CACHE_TTL = 300 # 5 minutes
|
||||
|
||||
# Default download directory
|
||||
DEFAULT_MODELS_DIR = Path.home() / ".pedal" / "models"
|
||||
DEFAULT_IRS_DIR = Path.home() / ".pedal" / "irs"
|
||||
|
||||
|
||||
# ── Data types ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@dataclass
|
||||
class ModelResult:
|
||||
"""A search result representing a NAM model on Tone3000."""
|
||||
|
||||
id: int
|
||||
name: str
|
||||
download_url: str
|
||||
size_bytes: Optional[int] = None
|
||||
architecture: Optional[str] = None
|
||||
pack_name: Optional[str] = None
|
||||
author_id: Optional[str] = None
|
||||
author_username: Optional[str] = None
|
||||
author_avatar_url: Optional[str] = None
|
||||
tone_id: Optional[int] = None
|
||||
tone_title: Optional[str] = None
|
||||
tone_gear: Optional[str] = None
|
||||
tone_images: list[str] = field(default_factory=list)
|
||||
tone_description: Optional[str] = None
|
||||
downloads: int = 0
|
||||
likes: int = 0
|
||||
created_at: Optional[str] = None
|
||||
|
||||
@property
|
||||
def is_ir(self) -> bool:
|
||||
"""Check if this model is an impulse response (.wav file)."""
|
||||
return self.tone_gear == "ir" or (self.download_url or "").lower().endswith(".wav")
|
||||
|
||||
@property
|
||||
def filename(self) -> str:
|
||||
"""Get the filename from the download URL."""
|
||||
return Path(self.download_url.split("/")[-1]).name
|
||||
|
||||
@property
|
||||
def display_name(self) -> str:
|
||||
return self.tone_title or self.name or self.filename
|
||||
|
||||
@property
|
||||
def thumbnail_url(self) -> Optional[str]:
|
||||
if self.tone_images:
|
||||
return self.tone_images[0]
|
||||
return None
|
||||
|
||||
|
||||
@dataclass
|
||||
class IRResult(ModelResult):
|
||||
"""A search result representing an IR file."""
|
||||
|
||||
sample_rate: Optional[int] = None
|
||||
num_taps: Optional[int] = None
|
||||
length_ms: Optional[float] = None
|
||||
|
||||
@property
|
||||
def is_ir(self) -> bool:
|
||||
return True
|
||||
|
||||
|
||||
# ── Anon key discovery ────────────────────────────────────────────────
|
||||
|
||||
|
||||
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.
|
||||
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:
|
||||
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
|
||||
)
|
||||
# Sort by name to get a deterministic order
|
||||
chunk_urls = sorted(set(chunk_urls))
|
||||
|
||||
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")
|
||||
match = re.search(
|
||||
r'createBrowserClient\)\(["\']https://api\.tone3000\.com["\'],["\']([^"\']+)["\']',
|
||||
js,
|
||||
)
|
||||
if match:
|
||||
key = match.group(1)
|
||||
if key.startswith("eyJ"):
|
||||
logger.info("Discovered Supabase anon key from JS bundle")
|
||||
return key
|
||||
|
||||
logger.warning("Could not discover anon key from Tone3000 JS")
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.debug("Anon key discovery failed: %s", e)
|
||||
return None
|
||||
|
||||
|
||||
# ── API Client ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class Tone3000Client:
|
||||
"""Client for the Tone3000 Supabase REST API.
|
||||
|
||||
Uses the public anon key embedded in the Tone3000 web client for
|
||||
read-only access to the models, tones, and users tables.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
anon_key: str = SUPABASE_ANON_KEY,
|
||||
models_dir: Path = DEFAULT_MODELS_DIR,
|
||||
irs_dir: Path = DEFAULT_IRS_DIR,
|
||||
session: Optional[aiohttp.ClientSession] = None,
|
||||
) -> None:
|
||||
self._anon_key = anon_key
|
||||
self._models_dir = models_dir
|
||||
self._irs_dir = irs_dir
|
||||
self._session = session
|
||||
self._cache: dict[str, tuple[float, list[dict]]] = {}
|
||||
self._headers = {
|
||||
"apikey": anon_key,
|
||||
"Authorization": f"Bearer {anon_key}",
|
||||
"Accept": "application/json",
|
||||
}
|
||||
self._named_user_cache: dict[str, Optional[dict]] = {}
|
||||
self._named_tone_cache: dict[int, Optional[dict]] = {}
|
||||
|
||||
# ── Session management ───────────────────────────────────────
|
||||
|
||||
async def _get_session(self) -> aiohttp.ClientSession:
|
||||
if self._session is None or self._session.closed:
|
||||
self._session = aiohttp.ClientSession(
|
||||
headers=self._headers,
|
||||
timeout=aiohttp.ClientTimeout(total=30),
|
||||
)
|
||||
return self._session
|
||||
|
||||
async def close(self) -> None:
|
||||
if self._session and not self._session.closed:
|
||||
await self._session.close()
|
||||
|
||||
# ── Caching ──────────────────────────────────────────────────
|
||||
|
||||
def _get_cached(self, key: str) -> Optional[list[dict]]:
|
||||
entry = self._cache.get(key)
|
||||
if entry and (time.monotonic() - entry[0]) < CACHE_TTL:
|
||||
return entry[1]
|
||||
return None
|
||||
|
||||
def _set_cache(self, key: str, data: list[dict]) -> None:
|
||||
self._cache[key] = (time.monotonic(), data)
|
||||
|
||||
# ── Core API calls ───────────────────────────────────────────
|
||||
|
||||
async def _supabase_get(
|
||||
self, path: str, params: Optional[dict] = None
|
||||
) -> list[dict]:
|
||||
"""Make a GET request to the Supabase REST API."""
|
||||
session = await self._get_session()
|
||||
url = f"{SUPABASE_REST}/{path}"
|
||||
try:
|
||||
async with session.get(url, params=params) as resp:
|
||||
if resp.status in (200, 206):
|
||||
data = await resp.json()
|
||||
return data if isinstance(data, list) else [data]
|
||||
else:
|
||||
text = await resp.text()
|
||||
logger.warning(
|
||||
"Supabase API error %d: %s", resp.status, text[:200]
|
||||
)
|
||||
return []
|
||||
except (aiohttp.ClientError, asyncio.TimeoutError) as e:
|
||||
logger.warning("Supabase request failed: %s", e)
|
||||
return []
|
||||
|
||||
# ── Search ───────────────────────────────────────────────────
|
||||
|
||||
async def search_models(
|
||||
self,
|
||||
query: str,
|
||||
page: int = 0,
|
||||
page_size: int = 20,
|
||||
force_refresh: bool = False,
|
||||
) -> list[ModelResult]:
|
||||
"""Search Tone3000 for NAM models matching the query.
|
||||
|
||||
Searches model names. Results are cached for 5 minutes.
|
||||
"""
|
||||
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),
|
||||
})
|
||||
|
||||
results = await self._enrich_models(raw_models)
|
||||
self._set_cache(cache_key, [r.__dict__ for r in results])
|
||||
return results
|
||||
|
||||
async def search_irs(
|
||||
self,
|
||||
query: str,
|
||||
page: int = 0,
|
||||
page_size: int = 20,
|
||||
force_refresh: bool = False,
|
||||
) -> list[IRResult]:
|
||||
"""Search Tone3000 for IRs (impulse responses).
|
||||
|
||||
IRs are stored in the tones table with gear='ir'. We search
|
||||
by tone title and return the linked model files.
|
||||
"""
|
||||
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",
|
||||
})
|
||||
|
||||
for m in tone_models:
|
||||
if m["id"] in seen_ids:
|
||||
continue
|
||||
seen_ids.add(m["id"])
|
||||
|
||||
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
|
||||
|
||||
# ── 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)
|
||||
|
||||
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",
|
||||
})
|
||||
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
|
||||
|
||||
# ── Enrichment helpers ───────────────────────────────────────
|
||||
|
||||
async def _enrich_models(
|
||||
self, raw_models: list[dict]
|
||||
) -> list[ModelResult]:
|
||||
"""Enrich raw model data with user/tone info."""
|
||||
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", ""))
|
||||
|
||||
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"),
|
||||
))
|
||||
return results
|
||||
|
||||
async def _get_user(self, user_id: str) -> Optional[dict]:
|
||||
if not user_id:
|
||||
return None
|
||||
if user_id in self._named_user_cache:
|
||||
return self._named_user_cache[user_id]
|
||||
data = await self._supabase_get("users", {
|
||||
"select": "id,username,avatar_url,display_name",
|
||||
"id": f"eq.{user_id}",
|
||||
"limit": "1",
|
||||
})
|
||||
result = data[0] if data else None
|
||||
self._named_user_cache[user_id] = result
|
||||
return result
|
||||
|
||||
async def _get_tone(self, tone_id: int) -> Optional[dict]:
|
||||
if tone_id in self._named_tone_cache:
|
||||
return self._named_tone_cache[tone_id]
|
||||
data = await self._supabase_get("tones", {
|
||||
"select": "id,title,description,gear,images",
|
||||
"id": f"eq.{tone_id}",
|
||||
"limit": "1",
|
||||
})
|
||||
result = data[0] if data else None
|
||||
self._named_tone_cache[tone_id] = result
|
||||
return result
|
||||
|
||||
# ── Download ─────────────────────────────────────────────────
|
||||
|
||||
async def download_model(
|
||||
self,
|
||||
model: ModelResult,
|
||||
progress_callback: Optional[callable] = None,
|
||||
) -> Optional[Path]:
|
||||
"""Download a NAM model to the pedal's models directory.
|
||||
|
||||
Returns the local path to the downloaded file, or None on failure.
|
||||
"""
|
||||
dest_dir = self._models_dir
|
||||
dest_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
filename = model.filename
|
||||
if not filename.endswith(".nam"):
|
||||
filename += ".nam"
|
||||
dest_path = dest_dir / filename
|
||||
|
||||
return await self._download_file(model.download_url, dest_path, progress_callback)
|
||||
|
||||
async def download_ir(
|
||||
self,
|
||||
ir: IRResult,
|
||||
progress_callback: Optional[callable] = None,
|
||||
) -> Optional[Path]:
|
||||
"""Download an IR file to the pedal's IRs directory."""
|
||||
dest_dir = self._irs_dir
|
||||
dest_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
filename = ir.filename
|
||||
if not filename.endswith(".wav"):
|
||||
filename += ".wav"
|
||||
dest_path = dest_dir / filename
|
||||
|
||||
return await self._download_file(ir.download_url, dest_path, progress_callback)
|
||||
|
||||
async def _download_file(
|
||||
self,
|
||||
url: str,
|
||||
dest_path: Path,
|
||||
progress_callback: Optional[callable] = None,
|
||||
) -> Optional[Path]:
|
||||
"""Download a file with optional progress reporting."""
|
||||
if not url:
|
||||
logger.error("No download URL available")
|
||||
return None
|
||||
|
||||
session = await self._get_session()
|
||||
try:
|
||||
async with session.get(
|
||||
url, timeout=aiohttp.ClientTimeout(total=120)
|
||||
) as resp:
|
||||
if resp.status not in (200, 302, 307):
|
||||
logger.warning("Download failed: HTTP %d", resp.status)
|
||||
return None
|
||||
|
||||
total = int(resp.headers.get("Content-Length", 0)) or 0
|
||||
downloaded = 0
|
||||
|
||||
with open(dest_path, "wb") as f:
|
||||
async for chunk in resp.content.iter_chunked(8192):
|
||||
f.write(chunk)
|
||||
downloaded += len(chunk)
|
||||
if progress_callback and total:
|
||||
progress_callback(downloaded / total)
|
||||
|
||||
logger.info(
|
||||
"Downloaded %s (%d bytes)", dest_path.name, downloaded
|
||||
)
|
||||
return dest_path
|
||||
|
||||
except (aiohttp.ClientError, asyncio.TimeoutError, OSError) as e:
|
||||
logger.error("Download failed: %s", e)
|
||||
if dest_path.exists():
|
||||
dest_path.unlink()
|
||||
return None
|
||||
|
||||
# ── Connectivity check ───────────────────────────────────────
|
||||
|
||||
async def ping(self) -> bool:
|
||||
"""Check if the Tone3000 API is reachable."""
|
||||
try:
|
||||
data = await self._supabase_get("models", {"limit": "1"})
|
||||
return len(data) > 0
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
# ── Convenience helpers ───────────────────────────────────────────────
|
||||
|
||||
|
||||
def format_size(size_bytes: Optional[int]) -> str:
|
||||
"""Format a file size in bytes to a human-readable string."""
|
||||
if size_bytes is None:
|
||||
return "Unknown"
|
||||
mb = size_bytes / (1024 * 1024)
|
||||
if mb < 1:
|
||||
return f"{size_bytes / 1024:.1f} KB"
|
||||
return f"{mb:.1f} MB"
|
||||
Reference in New Issue
Block a user