Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 3e55ea8a9d | |||
| e890a4aad3 |
+239
-142
@@ -15,7 +15,7 @@ import re
|
|||||||
import time
|
import time
|
||||||
from dataclasses import dataclass, field
|
from dataclasses import dataclass, field
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Any, Optional
|
from typing import Any, Optional, Union
|
||||||
|
|
||||||
import aiohttp
|
import aiohttp
|
||||||
|
|
||||||
@@ -75,6 +75,8 @@ class ModelResult:
|
|||||||
@property
|
@property
|
||||||
def filename(self) -> str:
|
def filename(self) -> str:
|
||||||
"""Get the filename from the download URL."""
|
"""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
|
return Path(self.download_url.split("/")[-1]).name
|
||||||
|
|
||||||
@property
|
@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.
|
"""Fetch the Supabase anon key from the Tone3000 website JS bundle.
|
||||||
|
|
||||||
The key is embedded in the Next.js client-side chunk.
|
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).
|
Returns None if discovery fails (silently falls back to hardcoded key).
|
||||||
"""
|
"""
|
||||||
try:
|
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()
|
html = await resp.text()
|
||||||
|
|
||||||
# Find JS chunk URLs that contain the client setup code
|
# 1. Look for embedded JSON config in the HTML itself
|
||||||
chunk_urls = re.findall(r'src="([^"]*2274[^"]*\.js[^"]*)"', html)
|
config_match = re.search(
|
||||||
if not chunk_urls:
|
r'"(?:supabaseKey|supabase_anon_key|SUPABASE_ANON_KEY'
|
||||||
# Fallback: look for any chunk with the key pattern
|
r'|supabaseAnonKey|NEXT_PUBLIC_SUPABASE_ANON_KEY)"\s*:\s*"([^"]+)"',
|
||||||
chunk_urls = re.findall(
|
html,
|
||||||
r'src="(/_[^"]*chunks/[^"]*\.js[^"]*)"', 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
|
try:
|
||||||
chunk_urls = sorted(set(chunk_urls))
|
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]:
|
# Pattern A: createBrowserClient)("url","token")
|
||||||
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(
|
match = re.search(
|
||||||
r'createBrowserClient\)\(["\']https://api\.tone3000\.com["\'],["\']([^"\']+)["\']',
|
r'createBrowserClient\)\(["\']https://api\.tone3000\.com["\'],["\']([^"\']+)["\']',
|
||||||
js,
|
js,
|
||||||
@@ -137,10 +166,41 @@ async def discover_anon_key(session: aiohttp.ClientSession) -> Optional[str]:
|
|||||||
if match:
|
if match:
|
||||||
key = match.group(1)
|
key = match.group(1)
|
||||||
if key.startswith("eyJ"):
|
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
|
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
|
return None
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.debug("Anon key discovery failed: %s", e)
|
logger.debug("Anon key discovery failed: %s", e)
|
||||||
@@ -237,25 +297,33 @@ class Tone3000Client:
|
|||||||
"""Search Tone3000 for NAM models matching the query.
|
"""Search Tone3000 for NAM models matching the query.
|
||||||
|
|
||||||
Searches model names. Results are cached for 5 minutes.
|
Searches model names. Results are cached for 5 minutes.
|
||||||
|
Returns empty list on any error (logged as warning).
|
||||||
"""
|
"""
|
||||||
cache_key = f"models:{query}:{page}"
|
try:
|
||||||
if not force_refresh:
|
cache_key = f"models:{query}:{page}"
|
||||||
cached = self._get_cached(cache_key)
|
if not force_refresh:
|
||||||
if cached is not None:
|
cached = self._get_cached(cache_key)
|
||||||
return [ModelResult(**r) for r in cached]
|
if cached is not None:
|
||||||
|
return [ModelResult(**r) for r in cached]
|
||||||
|
|
||||||
# Search models table by name similarity
|
# Search models table by name similarity
|
||||||
raw_models = await self._supabase_get("models", {
|
# Replace spaces with wildcards so multi-word queries work
|
||||||
"select": "id,name,model_url,size,pack_name,user_id,tone_id,architecture_version,created_at",
|
# (e.g. "vox ac30" → ilike.*vox*ac30* → ILIKE '%vox%ac30%')
|
||||||
"name": f"ilike.*{query}*",
|
ilike_query = query.replace(" ", "*")
|
||||||
"order": "created_at.desc",
|
raw_models = await self._supabase_get("models", {
|
||||||
"limit": str(page_size),
|
"select": "id,name,model_url,size,pack_name,user_id,tone_id,architecture_version,created_at",
|
||||||
"offset": str(page * page_size),
|
"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)
|
results = await self._enrich_models(raw_models)
|
||||||
self._set_cache(cache_key, [r.__dict__ for r in results])
|
self._set_cache(cache_key, [r.__dict__ for r in results])
|
||||||
return results
|
return results
|
||||||
|
except Exception as e:
|
||||||
|
logger.warning("search_models failed: %s", e)
|
||||||
|
return []
|
||||||
|
|
||||||
async def search_irs(
|
async def search_irs(
|
||||||
self,
|
self,
|
||||||
@@ -268,133 +336,154 @@ class Tone3000Client:
|
|||||||
|
|
||||||
IRs are stored in the tones table with gear='ir'. We search
|
IRs are stored in the tones table with gear='ir'. We search
|
||||||
by tone title and return the linked model files.
|
by tone title and return the linked model files.
|
||||||
|
Returns empty list on any error (logged as warning).
|
||||||
"""
|
"""
|
||||||
cache_key = f"irs:{query}:{page}"
|
try:
|
||||||
if not force_refresh:
|
cache_key = f"irs:{query}:{page}"
|
||||||
cached = self._get_cached(cache_key)
|
if not force_refresh:
|
||||||
if cached is not None:
|
cached = self._get_cached(cache_key)
|
||||||
return [IRResult(**r) for r in cached]
|
if cached is not None:
|
||||||
|
return [IRResult(**r) for r in cached]
|
||||||
|
|
||||||
# Search tones with gear='ir' matching the query
|
# Search tones with gear='ir' matching the query
|
||||||
raw_tones = await self._supabase_get("tones", {
|
# Replace spaces with wildcards so multi-word queries work
|
||||||
"select": "id,title,description,gear,images",
|
ilike_query = query.replace(" ", "*")
|
||||||
"title": f"ilike.*{query}*",
|
raw_tones = await self._supabase_get("tones", {
|
||||||
"gear": "eq.ir",
|
"select": "id,title,description,gear,images",
|
||||||
"limit": str(page_size),
|
"title": f"ilike.*{ilike_query}*",
|
||||||
"offset": str(page * page_size),
|
"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:
|
results: list[IRResult] = []
|
||||||
if m["id"] in seen_ids:
|
seen_ids: set[int] = set()
|
||||||
continue
|
|
||||||
seen_ids.add(m["id"])
|
|
||||||
|
|
||||||
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(
|
for m in tone_models:
|
||||||
id=m["id"],
|
if m["id"] in seen_ids:
|
||||||
name=m.get("name", ""),
|
continue
|
||||||
download_url=m.get("model_url", ""),
|
seen_ids.add(m["id"])
|
||||||
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])
|
author = await self._get_user(m.get("user_id", ""))
|
||||||
return results
|
|
||||||
|
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 ────────────────────────────────────────
|
# ── Trending / Browse ────────────────────────────────────────
|
||||||
|
|
||||||
async def get_trending_models(self, limit: int = 20) -> list[ModelResult]:
|
async def get_trending_models(self, limit: int = 20) -> list[ModelResult]:
|
||||||
"""Get trending NAM models."""
|
"""Get trending NAM models. Returns empty list on error."""
|
||||||
raw = await self._supabase_get("models", {
|
try:
|
||||||
"select": "id,name,model_url,size,pack_name,user_id,tone_id,architecture_version,created_at",
|
raw = await self._supabase_get("models", {
|
||||||
"order": "created_at.desc",
|
"select": "id,name,model_url,size,pack_name,user_id,tone_id,architecture_version,created_at",
|
||||||
"limit": str(limit),
|
"order": "created_at.desc",
|
||||||
})
|
"limit": str(limit),
|
||||||
return await self._enrich_models(raw)
|
})
|
||||||
|
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]:
|
async def get_trending_irs(self, limit: int = 20) -> list[IRResult]:
|
||||||
"""Get trending IRs."""
|
"""Get trending IRs. Returns empty list on error."""
|
||||||
raw_tones = await self._supabase_get("tones", {
|
try:
|
||||||
"select": "id,title,description,gear,images",
|
raw_tones = await self._supabase_get("tones", {
|
||||||
"gear": "eq.ir",
|
"select": "id,title,description,gear,images",
|
||||||
"order": "created_at.desc",
|
"gear": "eq.ir",
|
||||||
"limit": str(limit),
|
"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", ""))
|
results: list[IRResult] = []
|
||||||
ir = IRResult(
|
for tone in raw_tones:
|
||||||
id=m["id"],
|
tone_models = await self._supabase_get("models", {
|
||||||
name=m.get("name", ""),
|
"select": "id,name,model_url,size,pack_name,user_id,tone_id,created_at",
|
||||||
download_url=m.get("model_url", ""),
|
"tone_id": f"eq.{tone['id']}",
|
||||||
size_bytes=m.get("size"),
|
"limit": "3",
|
||||||
tone_title=tone.get("title"),
|
})
|
||||||
tone_images=tone.get("images", []),
|
for m in tone_models[:1]:
|
||||||
author_username=author.get("username") if author else None,
|
author = await self._get_user(m.get("user_id", ""))
|
||||||
author_avatar_url=author.get("avatar_url") if author else None,
|
ir = IRResult(
|
||||||
created_at=m.get("created_at"),
|
id=m["id"],
|
||||||
)
|
name=m.get("name", ""),
|
||||||
results.append(ir)
|
download_url=m.get("model_url", ""),
|
||||||
return results
|
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 ───────────────────────────────────────
|
# ── Enrichment helpers ───────────────────────────────────────
|
||||||
|
|
||||||
async def _enrich_models(
|
async def _enrich_models(
|
||||||
self, raw_models: list[dict]
|
self, raw_models: list[dict]
|
||||||
) -> list[ModelResult]:
|
) -> 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] = []
|
results: list[ModelResult] = []
|
||||||
for m in raw_models:
|
for m in raw_models:
|
||||||
tone_data = None
|
try:
|
||||||
if m.get("tone_id"):
|
tone_data = None
|
||||||
tone_data = await self._get_tone(m["tone_id"])
|
if m.get("tone_id"):
|
||||||
author = await self._get_user(m.get("user_id", ""))
|
tone_data = await self._get_tone(m["tone_id"])
|
||||||
|
author = await self._get_user(m.get("user_id", ""))
|
||||||
|
|
||||||
results.append(ModelResult(
|
results.append(ModelResult(
|
||||||
id=m["id"],
|
id=m["id"],
|
||||||
name=m.get("name", ""),
|
name=m.get("name", ""),
|
||||||
download_url=m.get("model_url", ""),
|
download_url=m.get("model_url", ""),
|
||||||
size_bytes=m.get("size"),
|
size_bytes=m.get("size"),
|
||||||
architecture=m.get("architecture_version"),
|
architecture=m.get("architecture_version"),
|
||||||
pack_name=m.get("pack_name"),
|
pack_name=m.get("pack_name"),
|
||||||
author_id=m.get("user_id"),
|
author_id=m.get("user_id"),
|
||||||
author_username=author.get("username") if author else None,
|
author_username=author.get("username") if author else None,
|
||||||
author_avatar_url=author.get("avatar_url") if author else None,
|
author_avatar_url=author.get("avatar_url") if author else None,
|
||||||
tone_id=m.get("tone_id"),
|
tone_id=m.get("tone_id"),
|
||||||
tone_title=tone_data.get("title") if tone_data else None,
|
tone_title=tone_data.get("title") if tone_data else None,
|
||||||
tone_gear=tone_data.get("gear") 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_images=tone_data.get("images", []) if tone_data else [],
|
||||||
tone_description=tone_data.get("description") if tone_data else None,
|
tone_description=tone_data.get("description") if tone_data else None,
|
||||||
created_at=m.get("created_at"),
|
created_at=m.get("created_at"),
|
||||||
))
|
))
|
||||||
|
except Exception as e:
|
||||||
|
logger.warning("Skipping model enrichment error: %s", e)
|
||||||
|
continue
|
||||||
return results
|
return results
|
||||||
|
|
||||||
async def _get_user(self, user_id: str) -> Optional[dict]:
|
async def _get_user(self, user_id: str) -> Optional[dict]:
|
||||||
@@ -515,11 +604,19 @@ class Tone3000Client:
|
|||||||
# ── Convenience helpers ───────────────────────────────────────────────
|
# ── Convenience helpers ───────────────────────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
def format_size(size_bytes: Optional[int]) -> str:
|
def format_size(size_bytes: Optional[Union[int, str]]) -> str:
|
||||||
"""Format a file size in bytes to a human-readable string."""
|
"""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:
|
if size_bytes is None:
|
||||||
return "Unknown"
|
return "Unknown"
|
||||||
mb = size_bytes / (1024 * 1024)
|
try:
|
||||||
|
nbytes = int(size_bytes)
|
||||||
|
except (ValueError, TypeError):
|
||||||
|
return "Unknown"
|
||||||
|
mb = nbytes / (1024 * 1024)
|
||||||
if mb < 1:
|
if mb < 1:
|
||||||
return f"{size_bytes / 1024:.1f} KB"
|
return f"{nbytes / 1024:.1f} KB"
|
||||||
return f"{mb:.1f} MB"
|
return f"{mb:.1f} MB"
|
||||||
+122
-52
@@ -13,6 +13,7 @@ from __future__ import annotations
|
|||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
import json
|
import json
|
||||||
|
import os
|
||||||
import logging
|
import logging
|
||||||
import time
|
import time
|
||||||
from contextlib import asynccontextmanager
|
from contextlib import asynccontextmanager
|
||||||
@@ -48,6 +49,20 @@ DEFAULT_PORT = 8080
|
|||||||
TEMPLATES_DIR = Path(__file__).resolve().parent / "templates"
|
TEMPLATES_DIR = Path(__file__).resolve().parent / "templates"
|
||||||
STATIC_DIR = Path(__file__).resolve().parent / "static"
|
STATIC_DIR = Path(__file__).resolve().parent / "static"
|
||||||
|
|
||||||
|
# React UI dist — built frontend served at /ui/
|
||||||
|
# Can be overridden via UI_DIST_DIR env var
|
||||||
|
_ui_dist_env = os.environ.get("UI_DIST_DIR")
|
||||||
|
if _ui_dist_env:
|
||||||
|
UI_DIST_DIR = Path(_ui_dist_env).resolve()
|
||||||
|
else:
|
||||||
|
# Look relative to backend package first, then common project layouts
|
||||||
|
_candidates = [
|
||||||
|
Path(__file__).resolve().parent / "ui-dist",
|
||||||
|
Path(__file__).resolve().parent.parent.parent / "frontend-react" / "dist",
|
||||||
|
Path(__file__).resolve().parent.parent.parent.parent / "pi-multifx-pedal-ui" / "dist",
|
||||||
|
]
|
||||||
|
UI_DIST_DIR = next((d for d in _candidates if d.is_dir()), None)
|
||||||
|
|
||||||
# ── Dependency container ─────────────────────────────────────────────────────
|
# ── Dependency container ─────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
@@ -145,6 +160,11 @@ class WebServer:
|
|||||||
# Mount static files
|
# Mount static files
|
||||||
app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")
|
app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")
|
||||||
|
|
||||||
|
# Mount React UI (if dist exists)
|
||||||
|
if UI_DIST_DIR and UI_DIST_DIR.is_dir():
|
||||||
|
logger.info("Mounting React UI from %s at /ui", UI_DIST_DIR)
|
||||||
|
app.mount("/ui", StaticFiles(directory=str(UI_DIST_DIR), html=True), name="ui")
|
||||||
|
|
||||||
# ── HTML pages ──────────────────────────────────────────────
|
# ── HTML pages ──────────────────────────────────────────────
|
||||||
|
|
||||||
@app.get("/", response_class=HTMLResponse)
|
@app.get("/", response_class=HTMLResponse)
|
||||||
@@ -772,62 +792,112 @@ class WebServer:
|
|||||||
@app.get("/api/models/tonedownload/search")
|
@app.get("/api/models/tonedownload/search")
|
||||||
async def tonedownload_search_models(q: str = "", page: int = 0):
|
async def tonedownload_search_models(q: str = "", page: int = 0):
|
||||||
"""Search Tone3000 for NAM models."""
|
"""Search Tone3000 for NAM models."""
|
||||||
client = await self._get_tonedownload()
|
try:
|
||||||
if not client:
|
client = await self._get_tonedownload()
|
||||||
raise HTTPException(status_code=503, detail="Tone3000 client unavailable")
|
if not client:
|
||||||
if not q.strip():
|
return JSONResponse(
|
||||||
results = await client.get_trending_models()
|
status_code=200,
|
||||||
else:
|
content={
|
||||||
results = await client.search_models(q.strip(), page=page)
|
"results": [],
|
||||||
return {
|
"status": "offline",
|
||||||
"results": [
|
"message": "Tone3000 unavailable - using local defaults",
|
||||||
{
|
},
|
||||||
"id": r.id,
|
)
|
||||||
"name": r.display_name,
|
if not q.strip():
|
||||||
"filename": r.filename,
|
results = await client.get_trending_models()
|
||||||
"size": r.size_bytes,
|
else:
|
||||||
"size_display": format_size(r.size_bytes),
|
results = await client.search_models(q.strip(), page=page)
|
||||||
"author": r.author_username or "Unknown",
|
return {
|
||||||
"author_avatar": r.author_avatar_url,
|
"results": [
|
||||||
"thumbnail": r.thumbnail_url,
|
{
|
||||||
"download_url": r.download_url,
|
"id": r.id,
|
||||||
"tone_description": r.tone_description,
|
"name": r.display_name,
|
||||||
"pack_name": r.pack_name,
|
"filename": r.filename,
|
||||||
"architecture": r.architecture,
|
"size": r.size_bytes,
|
||||||
"created_at": r.created_at,
|
"size_display": format_size(r.size_bytes),
|
||||||
}
|
"author": r.author_username or "Unknown",
|
||||||
for r in results
|
"author_avatar": r.author_avatar_url,
|
||||||
],
|
"thumbnail": r.thumbnail_url,
|
||||||
}
|
"download_url": r.download_url,
|
||||||
|
"tone_description": r.tone_description,
|
||||||
|
"pack_name": r.pack_name,
|
||||||
|
"architecture": r.architecture,
|
||||||
|
"created_at": r.created_at,
|
||||||
|
}
|
||||||
|
for r in results
|
||||||
|
],
|
||||||
|
}
|
||||||
|
except Exception as e:
|
||||||
|
logger.warning("Tone3000 model search failed: %s", e)
|
||||||
|
return JSONResponse(
|
||||||
|
status_code=200,
|
||||||
|
content={
|
||||||
|
"results": [],
|
||||||
|
"status": "offline",
|
||||||
|
"message": "Tone3000 unavailable - using local defaults",
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
@app.get("/api/irs/tonedownload/search")
|
@app.get("/api/irs/tonedownload/search")
|
||||||
async def tonedownload_search_irs(q: str = "", page: int = 0):
|
async def tonedownload_search_irs(q: str = "", page: int = 0):
|
||||||
"""Search Tone3000 for IRs."""
|
"""Search Tone3000 for IRs."""
|
||||||
client = await self._get_tonedownload()
|
try:
|
||||||
if not client:
|
client = await self._get_tonedownload()
|
||||||
raise HTTPException(status_code=503, detail="Tone3000 client unavailable")
|
if not client:
|
||||||
if not q.strip():
|
return JSONResponse(
|
||||||
results = await client.get_trending_irs()
|
status_code=200,
|
||||||
else:
|
content={
|
||||||
results = await client.search_irs(q.strip(), page=page)
|
"results": [],
|
||||||
return {
|
"status": "offline",
|
||||||
"results": [
|
"message": "Tone3000 unavailable - using local defaults",
|
||||||
{
|
},
|
||||||
"id": r.id,
|
)
|
||||||
"name": r.display_name,
|
if not q.strip():
|
||||||
"filename": r.filename,
|
results = await client.get_trending_irs()
|
||||||
"size": r.size_bytes,
|
else:
|
||||||
"size_display": format_size(r.size_bytes),
|
results = await client.search_irs(q.strip(), page=page)
|
||||||
"author": r.author_username or "Unknown",
|
return {
|
||||||
"author_avatar": r.author_avatar_url,
|
"results": [
|
||||||
"thumbnail": r.thumbnail_url,
|
{
|
||||||
"download_url": r.download_url,
|
"id": r.id,
|
||||||
"tone_description": r.tone_description,
|
"name": r.display_name,
|
||||||
"created_at": r.created_at,
|
"filename": r.filename,
|
||||||
}
|
"size": r.size_bytes,
|
||||||
for r in results
|
"size_display": format_size(r.size_bytes),
|
||||||
],
|
"author": r.author_username or "Unknown",
|
||||||
}
|
"author_avatar": r.author_avatar_url,
|
||||||
|
"thumbnail": r.thumbnail_url,
|
||||||
|
"download_url": r.download_url,
|
||||||
|
"tone_description": r.tone_description,
|
||||||
|
"created_at": r.created_at,
|
||||||
|
}
|
||||||
|
for r in results
|
||||||
|
],
|
||||||
|
}
|
||||||
|
except Exception as e:
|
||||||
|
logger.warning("Tone3000 IR search failed: %s", e)
|
||||||
|
return JSONResponse(
|
||||||
|
status_code=200,
|
||||||
|
content={
|
||||||
|
"results": [],
|
||||||
|
"status": "offline",
|
||||||
|
"message": "Tone3000 unavailable - using local defaults",
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
# ── Tonehub API alias routes ─────────────────────────────
|
||||||
|
# These are direct aliases for Tone3000 search, providing a
|
||||||
|
# simpler /api/tonehub/search path for the web UI.
|
||||||
|
|
||||||
|
@app.get("/api/tonehub/search")
|
||||||
|
async def tonehub_search_models(q: str = "", page: int = 0):
|
||||||
|
"""Alias for /api/models/tonedownload/search — search NAM models."""
|
||||||
|
return await tonedownload_search_models(q=q, page=page)
|
||||||
|
|
||||||
|
@app.get("/api/tonehub/search/irs")
|
||||||
|
async def tonehub_search_irs(q: str = "", page: int = 0):
|
||||||
|
"""Alias for /api/irs/tonedownload/search — search IRs."""
|
||||||
|
return await tonedownload_search_irs(q=q, page=page)
|
||||||
|
|
||||||
@app.post("/api/models/tonedownload/install")
|
@app.post("/api/models/tonedownload/install")
|
||||||
async def tonedownload_install_model(data: dict):
|
async def tonedownload_install_model(data: dict):
|
||||||
|
|||||||
@@ -176,6 +176,17 @@ class TestFormatSize:
|
|||||||
def test_edge_mb(self):
|
def test_edge_mb(self):
|
||||||
assert format_size(1024 * 1024) == "1.0 MB"
|
assert format_size(1024 * 1024) == "1.0 MB"
|
||||||
|
|
||||||
|
def test_string_input(self):
|
||||||
|
"""format_size should handle string inputs (Supabase can return string)."""
|
||||||
|
assert format_size("500") == "0.5 KB"
|
||||||
|
assert format_size("2097152") == "2.0 MB"
|
||||||
|
assert format_size("1024") == "1.0 KB"
|
||||||
|
|
||||||
|
def test_invalid_string(self):
|
||||||
|
"""format_size should handle unparseable string inputs."""
|
||||||
|
assert format_size("not-a-number") == "Unknown"
|
||||||
|
assert format_size("") == "Unknown"
|
||||||
|
|
||||||
|
|
||||||
# ── Client initialization ──────────────────────────────────────────────
|
# ── Client initialization ──────────────────────────────────────────────
|
||||||
|
|
||||||
@@ -389,6 +400,35 @@ class TestSearchModels:
|
|||||||
results = await client.search_models("nonexistent_xyz123", page=0)
|
results = await client.search_models("nonexistent_xyz123", page=0)
|
||||||
assert isinstance(results, list)
|
assert isinstance(results, list)
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_multi_word_query(self, client):
|
||||||
|
"""Multi-word queries should convert spaces to wildcards."""
|
||||||
|
captured_params = []
|
||||||
|
|
||||||
|
async def mock_get(path, params=None):
|
||||||
|
if path == "models" and params and "name" in params:
|
||||||
|
captured_params.append(params)
|
||||||
|
return [SAMPLE_MODEL_ROW]
|
||||||
|
if path == "users":
|
||||||
|
return [SAMPLE_USER_ROW]
|
||||||
|
if path == "tones":
|
||||||
|
return [SAMPLE_TONE_ROW]
|
||||||
|
return []
|
||||||
|
|
||||||
|
client._supabase_get = mock_get
|
||||||
|
|
||||||
|
results = await client.search_models("vox ac30", page=0)
|
||||||
|
assert len(results) == 1
|
||||||
|
assert results[0].name == "Marshall JCM800"
|
||||||
|
|
||||||
|
# Verify the Supabase param has spaces replaced with wildcards
|
||||||
|
assert len(captured_params) >= 1
|
||||||
|
name_param = captured_params[0].get("name", "")
|
||||||
|
assert "ilike.*vox*ac30*" in name_param
|
||||||
|
# Should NOT contain the raw space (which PostgREST filter parser
|
||||||
|
# can't handle in URL query values)
|
||||||
|
assert "vox ac30" not in name_param
|
||||||
|
|
||||||
|
|
||||||
class TestSearchIRs:
|
class TestSearchIRs:
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
|
|||||||
Reference in New Issue
Block a user