routes/airadio_bridge.py
Rama main · réplica de lectura del repositorio original.
"""
routes/airadio_bridge.py — AI-Radio Bridge Blueprint
Bridges OpenVoiceUI to the per-VPS AI-Radio service (Next.js, on the
shared `jambot-shared` Docker network at http://ai-radio:3000).
All endpoints are prefixed `/api/airadio/`. Frontend voice-tag parser
and canvas pages call these routes; the Flask app then proxies to the
scoped AI-Radio "agent API" at `${AIRADIO_URL}/api/agent/*`, signing
every outbound request with `X-JamBot-Agent-Key` + `X-JamBot-User`.
Spec: /mnt/system/base/airadio/docs/PLAN/18-OVU-INTEGRATION.md
Endpoints registered:
POST /api/airadio/push-song
POST /api/airadio/push-playlist
POST /api/airadio/set-image
GET /api/airadio/resolve
POST /api/airadio/play
GET /api/airadio/inbox
POST /api/airadio/send-to-friend
POST /api/airadio/vote
POST /api/airadio/friend-request
POST /api/airadio/friend-accept
POST /api/airadio/friend-decline
GET /api/airadio/library
"""
import base64
import hashlib
import json
import logging
import mimetypes
import os
import re
from pathlib import Path
import requests as http_requests
from flask import Blueprint, jsonify, request
from services.paths import (
GENERATED_MUSIC_DIR,
KNOWN_FACES_DIR,
MUSIC_DIR,
RUNTIME_DIR,
UPLOADS_DIR,
)
from routes.music import (
load_generated_music_metadata,
load_music_metadata,
)
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Configuration (env at module load)
# ---------------------------------------------------------------------------
# Default to the public internet URL so standalone OVU installs (no JamBot
# provisioner) can still reach the platform's public catalog without any
# additional config. JamBot clients set AIRADIO_URL=http://ai-radio:3000
# in their compose .env to short-circuit over the shared Docker network.
AIRADIO_URL = os.environ.get("AIRADIO_URL", "https://ai-radio.jam-bot.com").rstrip("/")
# When set, the bridge will answer read-only requests via AI-Radio's
# /public/* endpoints even when no API key is configured. True by default —
# keyless browse and play is the whole point of the public API. Writes
# (push, vote, friend, save, inbox) always require a key regardless.
AIRADIO_PUBLIC_FALLBACK = os.environ.get("AIRADIO_PUBLIC_FALLBACK", "1") not in ("0", "false", "False", "")
# Per-user key (aia_sk_*) takes precedence over the platform-shared agent key
# when set. Generated by the user at ai-radio.jam-bot.com/settings.
AIRADIO_USER_KEY = os.environ.get("AIRADIO_USER_KEY", "")
AIRADIO_AGENT_KEY = os.environ.get("AIRADIO_AGENT_KEY", "")
AIRADIO_JAMBOT_USER = os.environ.get("AIRADIO_JAMBOT_USER", "")
AIRADIO_TIMEOUT = int(os.environ.get("AIRADIO_TIMEOUT", "30"))
# Persistent override for AIRADIO_USER_KEY. Lives in the runtime dir which is
# bind-mounted from the host (/mnt/clients/<user>/openvoiceui/runtime-config),
# so the value survives container restarts and re-images. Written by
# /api/airadio/set-user-key, read at request time to avoid module-reload churn.
_USER_KEY_PERSIST_PATH = RUNTIME_DIR / "airadio-user-key"
def _read_persisted_user_key() -> str:
"""Return the persisted aia_sk_* key, or empty string if none."""
try:
if _USER_KEY_PERSIST_PATH.is_file():
val = _USER_KEY_PERSIST_PATH.read_text().strip()
if val.startswith("aia_sk_"):
return val
except OSError:
pass
return ""
def _effective_user_key() -> str:
"""Persisted key wins; env var is the fallback."""
return _read_persisted_user_key() or AIRADIO_USER_KEY
# ---------------------------------------------------------------------------
# Whitelisted local directories for set-image (security: NO arbitrary paths)
# ---------------------------------------------------------------------------
# Paths the agent is allowed to upload from. Anything outside these is rejected.
# Mirrors the rule in PLAN/18-OVU-INTEGRATION.md §3.3.
_FACES_DIR = RUNTIME_DIR / "faces"
_TEMP_DIR = RUNTIME_DIR / "1temp"
_IMAGE_WHITELIST_DIRS = [
UPLOADS_DIR.resolve(),
_FACES_DIR.resolve(),
KNOWN_FACES_DIR.resolve(),
GENERATED_MUSIC_DIR.resolve(),
_TEMP_DIR.resolve(),
]
# URL-prefix → on-disk dir map for paths like "/uploads/foo.png"
_URL_PREFIX_MAP = {
"/uploads/": UPLOADS_DIR,
"/faces/": _FACES_DIR,
"/known_faces/": KNOWN_FACES_DIR,
"/generated_music/": GENERATED_MUSIC_DIR,
"/1temp/": _TEMP_DIR,
}
# ---------------------------------------------------------------------------
# Blueprint
# ---------------------------------------------------------------------------
airadio_bp = Blueprint("airadio", __name__)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _bridge_has_auth() -> bool:
"""True iff we have a key to authenticate writes (push/vote/friend/etc)."""
return bool(AIRADIO_URL) and bool(_effective_user_key() or AIRADIO_AGENT_KEY)
def _bridge_has_backend() -> bool:
"""True iff we know where AI-Radio lives — enough for public-only reads."""
return bool(AIRADIO_URL)
# Back-compat shim — old code calls _bridge_configured(). Reads can proceed in
# public mode; writes still need auth. Caller context determines which matters.
def _bridge_configured() -> bool:
return _bridge_has_auth()
def _bridge_not_configured_response(for_write: bool = False):
"""503 when env vars are missing. `for_write=True` clarifies the message."""
missing = []
if not AIRADIO_URL:
missing.append("AIRADIO_URL")
if for_write and not _effective_user_key() and not AIRADIO_AGENT_KEY:
missing.append(
"AIRADIO_USER_KEY (generate at ai-radio.jam-bot.com/settings) "
"or legacy AIRADIO_AGENT_KEY + AIRADIO_JAMBOT_USER"
)
msg = (
"AI-Radio write requires an account — the public browse/play surface "
"works without one, but push/vote/friend/save need an aia_sk_* key. "
f"Missing env: {', '.join(missing) or 'unknown'}"
) if for_write else (
"AI-Radio bridge is not configured. "
f"Missing env: {', '.join(missing) or 'unknown'}"
)
return (
jsonify({
"ok": False,
"error": {"code": "BRIDGE_NOT_CONFIGURED", "message": msg},
}),
503,
)
# Paths that are available via /public/* without auth. Each entry maps the
# authenticated /api/agent/<...> path to its public equivalent. If a path is
# NOT in this table, falling back to public mode isn't possible and the
# endpoint will return 503 when no auth is present.
def _public_equivalent(agent_path: str) -> str | None:
"""Return the /public/* path for a given /api/agent/* path, or None."""
if not agent_path.startswith("/api/agent/"):
return None
suffix = agent_path[len("/api/agent/"):]
# Direct 1:1 mappings
MAP = {
"catalog": "/public/catalog",
"interpret": "/public/interpret",
}
if suffix in MAP:
return MAP[suffix]
# Prefix-based mappings
if suffix.startswith("catalog/"):
# /api/agent/catalog/<id> → /public/song/<id>
return f"/public/song/{suffix[len('catalog/'):]}"
if suffix.startswith("queue"):
# All /api/agent/queue/* have identical shape at /public/queue/*
return f"/public/{suffix}"
if suffix.startswith("u/"):
# /api/agent/u/<user> → /public/u/<user>
# /api/agent/u/<user>/songs → /public/u/<user>/songs
# /api/agent/u/<user>/playlists → /public/u/<user>/playlists
return f"/public/{suffix}"
if suffix.startswith("playlist/"):
# /api/agent/playlist/<id> (GET) → /public/playlist/<id>
return f"/public/{suffix}"
if suffix.startswith("song/") and suffix.endswith("/stream-url"):
# No signed-url needed publicly — caller should use /public/stream/<id>.mp3 directly
song_id = suffix[len("song/"):-len("/stream-url")]
return f"/public/stream/{song_id}.mp3"
return None
def _agent_headers() -> dict:
"""
Auth headers attached to every outbound AI-Radio call.
Preferred path: Authorization: Bearer aia_sk_* (per-user key from the
AI-Radio settings page). Identity is embedded in the key.
Legacy fallback: X-JamBot-Agent-Key + X-JamBot-User (platform-shared
secret + subdomain-as-user). Retained so existing clients keep working
during migration.
"""
eff_key = _effective_user_key()
if eff_key:
return {"Authorization": f"Bearer {eff_key}"}
return {
"X-JamBot-Agent-Key": AIRADIO_AGENT_KEY,
"X-JamBot-User": AIRADIO_JAMBOT_USER,
}
def _airadio_call(method, path, json=None, files=None, params=None, data=None):
"""
Wrapper around `requests` that targets AI-Radio with the auth headers.
If the bridge has no auth key but the requested path has a public
equivalent AND the method is GET, transparently rewrite to /public/* and
drop the auth headers. This is what makes keyless OVU installs able to
browse + play without any API key at all. `mode: "public"` is tagged on
the response so the agent can surface "to upload your own songs, connect
an account" without guessing.
Returns a dict shaped:
{"ok": bool, "data": <json or None>, "error": <dict or None>,
"status": int, "mode": "auth"|"public"}
"""
mode = "auth"
outbound_path = path
outbound_headers = _agent_headers()
# Public fallback: only if there's no auth to send and the endpoint has a
# public equivalent. `_public_equivalent()` ONLY returns mappings for
# read-safe / inference paths (catalog, song detail, queue, u/<handle>,
# playlist, interpret, stream). Anything that mutates server state (push,
# vote, friend, save, set-image, send-to-friend) is never mapped, so
# writes cannot silently fall through to an unauthenticated call.
if AIRADIO_PUBLIC_FALLBACK and not _bridge_has_auth():
pub = _public_equivalent(path)
if pub:
outbound_path = pub
outbound_headers = {}
mode = "public"
url = f"{AIRADIO_URL}{outbound_path}"
try:
print(f"📻 AI-Radio [{mode}] {method} {outbound_path}")
resp = http_requests.request(
method,
url,
headers=outbound_headers,
json=json,
files=files,
params=params,
data=data,
timeout=AIRADIO_TIMEOUT,
)
except http_requests.Timeout as exc:
logger.warning("AI-Radio timeout: %s %s — %s", method, path, exc)
return {
"ok": False,
"data": None,
"status": 504,
"mode": mode,
"error": {
"code": "BRIDGE_TIMEOUT",
"message": f"AI-Radio timed out after {AIRADIO_TIMEOUT}s",
},
}
except http_requests.RequestException as exc:
logger.warning("AI-Radio request failed: %s %s — %s", method, path, exc)
return {
"ok": False,
"data": None,
"status": 502,
"mode": mode,
"error": {
"code": "BRIDGE_UNREACHABLE",
"message": f"Could not reach AI-Radio: {exc}",
},
}
# Try to parse a JSON body — AI-Radio always returns JSON per its contract.
try:
body = resp.json()
except ValueError:
body = None
if 200 <= resp.status_code < 300:
# AI-Radio's own contract is { ok, data?, error? }. Pass it through if
# present; otherwise treat the parsed body as `data`.
if isinstance(body, dict) and "ok" in body:
return {
"ok": bool(body.get("ok")),
"data": body.get("data"),
"error": body.get("error"),
"status": resp.status_code,
"mode": mode,
}
return {
"ok": True,
"data": body,
"error": None,
"status": resp.status_code,
"mode": mode,
}
# Non-2xx — propagate AI-Radio's error if it gave us one.
if isinstance(body, dict) and body.get("error"):
return {
"ok": False,
"data": body.get("data"),
"error": body.get("error"),
"status": resp.status_code,
"mode": mode,
}
return {
"ok": False,
"data": None,
"status": resp.status_code,
"mode": mode,
"error": {
"code": "AIRADIO_HTTP_ERROR",
"message": f"AI-Radio returned HTTP {resp.status_code}",
"body": (resp.text or "")[:400],
},
}
def _bridge_response(result):
"""Convert an `_airadio_call` result into a Flask (json, status) tuple."""
payload = {
"ok": result["ok"],
"data": result.get("data"),
"error": result.get("error"),
"mode": result.get("mode", "auth"),
}
# Preserve AI-Radio's status code for non-success cases; otherwise 200.
status = 200 if result["ok"] else (result.get("status") or 502)
return jsonify(payload), status
_AUDIO_EXTS = (".mp3", ".wav", ".m4a", ".ogg", ".webm", ".flac")
def _norm_title(s: str) -> str:
"""Lowercase, collapse non-alphanumerics to single hyphens. Used to match a
spoken/slugified title (`ship-that-shit`, `Ship That Shit`) to a file."""
return re.sub(r"[^a-z0-9]+", "-", (s or "").lower()).strip("-")
def _in_base(candidate: Path, base_dir: Path) -> bool:
"""True if candidate resolves inside base_dir (path-traversal guard)."""
try:
candidate.resolve().relative_to(base_dir.resolve())
return True
except ValueError:
return False
def _resolve_local_song(filename: str, playlist: str):
"""
Look up a local audio track in either the library or the generated
playlist. Accepts an exact filename, a stem without extension, OR the
song's title (fuzzy) — matching what the skill/voice-tag promises. Returns
(Path, metadata_dict) or (None, None) if not found / playlist invalid /
path-traversal attempt.
"""
if not filename:
return None, None
if playlist == "generated":
base_dir = GENERATED_MUSIC_DIR
meta = load_generated_music_metadata()
elif playlist == "library":
base_dir = MUSIC_DIR
meta = load_music_metadata()
else:
return None, None
# Strict containment — reject `..`, absolute paths, anything not in base.
safe_name = Path(filename).name
if safe_name != filename or not base_dir.is_dir():
return None, None
# 1. Exact filename as given (already has a real extension).
cand = base_dir / safe_name
if _in_base(cand, base_dir) and cand.is_file():
return cand, meta.get(safe_name, {})
# 2. Stem given without extension → try the known audio extensions.
stem = Path(safe_name).suffix.lower() in _AUDIO_EXTS and Path(safe_name).stem or safe_name
for ext in _AUDIO_EXTS:
cand = base_dir / (stem + ext)
if _in_base(cand, base_dir) and cand.is_file():
return cand, meta.get(cand.name, {})
# 3. Fuzzy: normalize the query and match against each file's stem AND its
# metadata title. Exact-normalized hit wins; otherwise first containment.
want = _norm_title(stem)
if not want:
return None, None
fallback = None
for f in sorted(base_dir.iterdir()):
if not f.is_file() or f.suffix.lower() not in _AUDIO_EXTS:
continue
m = meta.get(f.name, {})
norms = {_norm_title(f.stem)}
if m.get("title"):
norms.add(_norm_title(m["title"]))
if want in norms:
return f, m
if fallback is None and any(want in n or n in want for n in norms if n):
fallback = (f, m)
return fallback if fallback else (None, None)
def _resolve_image_path(raw_path: str):
"""
Map a user-supplied image path (URL-style like `/uploads/foo.png` or a
bare filename) to an absolute Path inside one of the whitelisted dirs.
Returns Path on success, None on rejection. Strict — no `..`, no absolute
paths outside the whitelist.
"""
if not raw_path or not isinstance(raw_path, str):
return None
# Reject anything containing `..` outright before anything else.
if ".." in raw_path:
return None
# Reject absolute paths that aren't already inside a whitelisted dir.
candidate = None
if raw_path.startswith("/"):
# First, try matching a known URL prefix.
for prefix, base in _URL_PREFIX_MAP.items():
if raw_path.startswith(prefix):
relative = raw_path[len(prefix):]
if "/" in relative or "\\" in relative:
# No nested subdirs allowed (matches existing /uploads style).
return None
candidate = (base / relative).resolve()
break
if candidate is None:
# An absolute path that doesn't begin with one of our prefixes is
# never accepted — that would let the agent reach into the host FS.
return None
else:
# Bare filename — assume uploads/ by convention.
if "/" in raw_path or "\\" in raw_path:
return None
candidate = (UPLOADS_DIR / raw_path).resolve()
if not candidate.is_file():
return None
# Final containment check against the whitelist.
for allowed in _IMAGE_WHITELIST_DIRS:
try:
candidate.relative_to(allowed)
return candidate
except ValueError:
continue
return None
# Keys AI-Radio's push-song `meta` schema actually accepts (see
# /mnt/system/base/airadio/app/api/agent/push-song/route.ts). Everything else
# the bridge tracks internally (source, playlist, playlistId, ...) is dropped
# here so it never reaches — nor is silently mangled by — the server validator.
_AIRADIO_META_KEYS = (
"title", "artist", "genre", "duration",
"isPublic", "sunoId", "sourceHash", "cover",
)
def _encode_cover_data_url(cover_path: str) -> str | None:
"""Read a local image file and return a `data:` URL, or None if unreadable.
AI-Radio's push-song decodes a `meta.cover` data URL into a song_cover
image. Used so an agent can attach artwork in the same call that uploads
the audio."""
try:
p = Path(cover_path)
if not p.is_file():
return None
raw = p.read_bytes()
if not raw:
return None
mime, _ = mimetypes.guess_type(str(p))
if not mime or not mime.startswith("image/"):
mime = "image/png"
return f"data:{mime};base64," + base64.b64encode(raw).decode("ascii")
except OSError as exc:
logger.warning("AI-Radio cover read failed for %s: %s", cover_path, exc)
return None
_COVER_EXTS = (".png", ".jpg", ".jpeg", ".webp")
def _find_sibling_cover(song_path: Path) -> str | None:
"""Return artwork that lives next to the song, or None.
Looks for `<stem>.<img>` beside the audio file, then a `covers/<stem>.<img>`
subdir. This is what makes covers AUTOMATIC: save artwork as `<song>.png`
alongside `<song>.mp3` and every push attaches it with no explicit
cover_path and no change to the voice-tag."""
for ext in _COVER_EXTS:
cand = song_path.with_suffix(ext)
if cand.is_file():
return str(cand)
covers_dir = song_path.parent / "covers"
if covers_dir.is_dir():
for ext in _COVER_EXTS:
cand = covers_dir / (song_path.stem + ext)
if cand.is_file():
return str(cand)
return None
def _build_song_meta(fields: dict, audio_bytes: bytes, cover_path: str | None) -> dict:
"""Translate the bridge's flat field dict into AI-Radio's `meta` JSON shape.
AI-Radio's /api/agent/push-song expects multipart `audio` (File) + `meta`
(JSON string). `duration` MUST be an int; an optional cover travels as a
`data:` URL inside meta; absent a sunoId we hash the bytes into a
sourceHash so re-pushes dedupe instead of piling up duplicate rows."""
meta: dict = {}
for k in _AIRADIO_META_KEYS:
if fields.get(k) not in (None, ""):
meta[k] = fields[k]
# server schema is z.number().int(); callers hand us a string.
if "duration" in meta:
try:
meta["duration"] = int(float(meta["duration"]))
except (TypeError, ValueError):
del meta["duration"]
if not meta.get("sunoId") and not meta.get("sourceHash"):
meta["sourceHash"] = hashlib.sha256(audio_bytes).hexdigest()
if cover_path and not meta.get("cover"):
cov = _encode_cover_data_url(cover_path)
if cov:
meta["cover"] = cov
return meta
def _stream_song(local_path: Path, fields: dict, cover_path: str | None = None):
"""
POST a song file to AI-Radio's /api/agent/push-song as multipart/form-data.
AI-Radio expects field `audio` (the File) + field `meta` (a JSON string) —
NOT `file` + flat form fields. `fields` is the bridge's internal flat
metadata dict; `_build_song_meta` reshapes it to the contract the deployed
server validates, so every caller (push-song, push-playlist) hits it
correctly. AI-Radio dedupes via sunoId/sourceHash.
"""
raw = local_path.read_bytes()
mime, _ = mimetypes.guess_type(str(local_path))
if not mime:
mime = "audio/mpeg"
# No explicit cover? Auto-attach artwork sitting next to the song.
if not cover_path:
cover_path = _find_sibling_cover(local_path)
meta = _build_song_meta(fields, raw, cover_path)
files = {"audio": (local_path.name, raw, mime)}
return _airadio_call(
"POST",
"/api/agent/push-song",
files=files,
data={"meta": json.dumps(meta)},
)
def _stream_image(local_path: Path, fields: dict):
"""POST an image file to AI-Radio's /api/agent/set-image.
The server expects multipart `image` (the File) + `meta` (a JSON string of
`{target, targetId?}`) — NOT `file` + flat fields. `fields` already carries
the right keys, so we just JSON-encode it under `meta`."""
mime, _ = mimetypes.guess_type(str(local_path))
if not mime:
mime = "application/octet-stream"
with open(local_path, "rb") as fh:
files = {"image": (local_path.name, fh, mime)}
return _airadio_call(
"POST",
"/api/agent/set-image",
files=files,
data={"meta": json.dumps(fields or {})},
)
# ---------------------------------------------------------------------------
# Endpoints
# ---------------------------------------------------------------------------
@airadio_bp.route("/api/airadio/push-song", methods=["POST"])
def push_song():
"""
Push a single song from MUSIC_DIR (library) or GENERATED_MUSIC_DIR
(generated) to AI-Radio.
Body: { "filename": "...", "playlist": "library"|"generated",
"title": optional, "cover_path": optional local image path }
"""
if not _bridge_has_auth():
return _bridge_not_configured_response(for_write=True)
body = request.get_json(silent=True) or {}
filename = body.get("filename")
playlist = body.get("playlist", "library")
title = body.get("title")
cover_path = body.get("cover_path")
local_path, meta = _resolve_local_song(filename, playlist)
if local_path is None:
# Auto-detect: if not found in the stated dir, try the other one.
# The JS bridge never sends a playlist param, so it always defaults to
# "library" — but most agent-pushed songs live in "generated".
alt_playlist = "generated" if playlist == "library" else "library"
local_path, meta = _resolve_local_song(filename, alt_playlist)
if local_path is not None:
playlist = alt_playlist
else:
return jsonify({
"ok": False,
"error": {
"code": "NOT_FOUND",
"message": f"Song '{filename}' not found in library or generated playlist",
},
}), 404
fields = {
"title": title or (meta or {}).get("title") or local_path.stem,
"source": "ovu-library" if playlist == "library" else "ovu-generated",
"playlist": playlist,
}
# Carry through any useful metadata so AI-Radio doesn't lose it.
for src_key, dst_key in (
("artist", "artist"),
("genre", "genre"),
("description", "description"),
("duration_seconds", "duration"),
("suno_id", "sunoId"),
):
if meta and meta.get(src_key) is not None:
fields[dst_key] = str(meta[src_key])
print(f"📻 push-song: {local_path.name} → AI-Radio (playlist={playlist})")
result = _stream_song(local_path, fields, cover_path=cover_path)
if not result["ok"]:
return _bridge_response(result)
# Normalize success shape: { ok, songId, url } as the spec requires,
# while still passing through anything else AI-Radio gave us.
data = result.get("data") or {}
out = {
"ok": True,
"songId": data.get("songId") or data.get("id"),
"url": data.get("url"),
"data": data,
}
return jsonify(out), 200
@airadio_bp.route("/api/airadio/push-playlist", methods=["POST"])
def push_playlist():
"""
Push a whole playlist to AI-Radio (inline mode — one song per request).
Body: { "name": "...", "playlist": "library"|"generated",
"description": optional }
"""
if not _bridge_has_auth():
return _bridge_not_configured_response(for_write=True)
body = request.get_json(silent=True) or {}
name = body.get("name")
playlist = body.get("playlist", "library")
description = body.get("description", "")
if not name:
return jsonify({
"ok": False,
"error": {"code": "VALIDATION_FAILED", "message": "Missing 'name' field"},
}), 400
if playlist == "generated":
meta = load_generated_music_metadata()
elif playlist == "library":
meta = load_music_metadata()
else:
return jsonify({
"ok": False,
"error": {"code": "VALIDATION_FAILED", "message": f"Unknown playlist '{playlist}'"},
}), 400
# The metadata files are flat {filename: {...}}. There's no concept of a
# named in-file playlist yet — the social spec says we should look for
# songs whose metadata mentions the playlist name. For MVP, we treat the
# playlist arg as "all songs in this directory" if `name` matches a tag,
# otherwise we send all songs in the dir. AI-Radio dedupes by sourceHash.
matching_files = []
name_lower = (name or "").strip().lower()
for fname, entry in (meta or {}).items():
playlist_tags = entry.get("playlists") or []
if isinstance(playlist_tags, str):
playlist_tags = [playlist_tags]
tag_match = any(
isinstance(t, str) and t.lower() == name_lower for t in playlist_tags
)
if tag_match:
matching_files.append(fname)
if not matching_files:
# Fallback: the whole directory.
base_dir = GENERATED_MUSIC_DIR if playlist == "generated" else MUSIC_DIR
for f in base_dir.iterdir():
if f.is_file() and f.suffix.lower() in {".mp3", ".wav", ".ogg", ".m4a", ".webm"}:
matching_files.append(f.name)
print(f"📻 push-playlist: '{name}' ({playlist}) — {len(matching_files)} song(s)")
# 1. Create the playlist on AI-Radio first.
create_payload = {
"name": name,
"description": description,
"source": "ovu-library" if playlist == "library" else "ovu-generated",
}
create_result = _airadio_call("POST", "/api/agent/push-playlist", json=create_payload)
if not create_result["ok"]:
return _bridge_response(create_result)
playlist_data = create_result.get("data") or {}
playlist_id = playlist_data.get("id") or playlist_data.get("playlistId")
# 2. Push each song inline. Track per-song results so the caller sees
# which ones failed without aborting the whole batch.
pushed = []
failed = []
for fname in matching_files:
local_path, song_meta = _resolve_local_song(fname, playlist)
if local_path is None:
failed.append({"filename": fname, "error": "NOT_FOUND"})
continue
fields = {
"title": (song_meta or {}).get("title") or local_path.stem,
"source": "ovu-library" if playlist == "library" else "ovu-generated",
"playlist": playlist,
"playlistId": str(playlist_id) if playlist_id else "",
"playlistName": name,
}
for src_key, dst_key in (
("artist", "artist"),
("genre", "genre"),
("description", "description"),
("duration_seconds", "duration"),
("suno_id", "sunoId"),
):
if song_meta and song_meta.get(src_key) is not None:
fields[dst_key] = str(song_meta[src_key])
song_result = _stream_song(local_path, fields)
if song_result["ok"]:
sd = song_result.get("data") or {}
pushed.append({
"filename": fname,
"songId": sd.get("songId") or sd.get("id"),
"url": sd.get("url"),
})
else:
failed.append({
"filename": fname,
"error": (song_result.get("error") or {}).get("code", "PUSH_FAILED"),
"message": (song_result.get("error") or {}).get("message"),
})
return jsonify({
"ok": True,
"playlist": playlist_data,
"pushed_count": len(pushed),
"failed_count": len(failed),
"pushed": pushed,
"failed": failed,
}), 200
@airadio_bp.route("/api/airadio/set-image", methods=["POST"])
def set_image():
"""
Update an avatar / banner / song cover / playlist cover on AI-Radio.
Body: { "target": "avatar"|"banner"|"song_cover"|"playlist_cover",
"target_id": optional, "path": "/uploads/..." }
"""
if not _bridge_has_auth():
return _bridge_not_configured_response(for_write=True)
body = request.get_json(silent=True) or {}
target = body.get("target")
target_id = body.get("target_id", "")
raw_path = body.get("path", "")
if target not in ("avatar", "banner", "song_cover", "playlist_cover"):
return jsonify({
"ok": False,
"error": {
"code": "VALIDATION_FAILED",
"message": "target must be one of avatar, banner, song_cover, playlist_cover",
},
}), 400
local_path = _resolve_image_path(raw_path)
if local_path is None:
return jsonify({
"ok": False,
"error": {
"code": "VALIDATION_FAILED",
"message": (
"Image path must point to an existing file under one of the "
"whitelisted dirs (uploads, faces, known_faces, generated_music, 1temp). "
"No '..' or absolute paths outside the whitelist."
),
},
}), 400
if target in ("song_cover", "playlist_cover") and not target_id:
return jsonify({
"ok": False,
"error": {
"code": "VALIDATION_FAILED",
"message": f"target_id is required when target='{target}'",
},
}), 400
fields = {"target": target}
if target_id:
fields["targetId"] = str(target_id)
print(f"📻 set-image: target={target} id={target_id or '-'} file={local_path.name}")
result = _stream_image(local_path, fields)
return _bridge_response(result)
@airadio_bp.route("/api/airadio/resolve", methods=["GET"])
def resolve():
"""
Fuzzy-resolve a song or playlist query → { id, url, title, artist }.
Query params: type=song|playlist, query=<text>
"""
if not _bridge_configured():
return _bridge_not_configured_response()
rtype = request.args.get("type", "song")
query = request.args.get("query", "")
if rtype not in ("song", "playlist"):
return jsonify({
"ok": False,
"error": {"code": "VALIDATION_FAILED", "message": "type must be 'song' or 'playlist'"},
}), 400
if not query:
return jsonify({
"ok": False,
"error": {"code": "VALIDATION_FAILED", "message": "Missing 'query' param"},
}), 400
print(f"📻 resolve: type={rtype} query='{query}'")
result = _airadio_call(
"GET",
"/api/agent/resolve",
params={"type": rtype, "query": query},
)
return _bridge_response(result)
@airadio_bp.route("/api/airadio/play", methods=["POST"])
def play():
"""
Resolve a song / playlist if needed, then return its URL+metadata.
The frontend music player handles actual playback — this endpoint is a
helper for the voice tag parser that needs `{ url, title, ... }`.
Body: { "type": "song"|"playlist", "id": "..." OR "query": "..." }
"""
if not _bridge_configured():
return _bridge_not_configured_response()
body = request.get_json(silent=True) or {}
rtype = body.get("type", "song")
rid = body.get("id")
query = body.get("query")
if rtype not in ("song", "playlist"):
return jsonify({
"ok": False,
"error": {"code": "VALIDATION_FAILED", "message": "type must be 'song' or 'playlist'"},
}), 400
if not rid and not query:
return jsonify({
"ok": False,
"error": {"code": "VALIDATION_FAILED", "message": "Provide either 'id' or 'query'"},
}), 400
# Resolve a query → id first if necessary.
if not rid:
print(f"📻 play: resolving '{query}' (type={rtype})")
resolved = _airadio_call(
"GET",
"/api/agent/resolve",
params={"type": rtype, "query": query},
)
if not resolved["ok"]:
return _bridge_response(resolved)
rdata = resolved.get("data") or {}
rid = rdata.get("id")
if not rid:
return jsonify({
"ok": False,
"error": {"code": "NOT_FOUND", "message": f"No {rtype} matching '{query}'"},
}), 404
print(f"📻 play: type={rtype} id={rid}")
# Ask AI-Radio for the canonical URL for that ID.
result = _airadio_call(
"GET",
"/api/agent/resolve",
params={"type": rtype, "id": rid},
)
if not result["ok"]:
return _bridge_response(result)
data = result.get("data") or {}
return jsonify({
"ok": True,
"type": rtype,
"id": data.get("id") or rid,
"url": data.get("url"),
"title": data.get("title"),
"artist": data.get("artist"),
"data": data,
}), 200
@airadio_bp.route("/api/airadio/inbox", methods=["GET"])
def inbox():
"""Songs friends have sent to the current user (unread + recent)."""
if not _bridge_has_auth():
return _bridge_not_configured_response(for_write=True)
print("📻 inbox: fetching")
result = _airadio_call("GET", "/api/agent/inbox")
return _bridge_response(result)
@airadio_bp.route("/api/airadio/send-to-friend", methods=["POST"])
def send_to_friend():
"""
Send a song to a friend.
Body: { "song_id": optional, "song_title": optional,
"receiver_handle": "@example", "note": optional }
"""
if not _bridge_has_auth():
return _bridge_not_configured_response(for_write=True)
body = request.get_json(silent=True) or {}
song_id = body.get("song_id")
song_title = body.get("song_title")
receiver = body.get("receiver_handle")
note = body.get("note", "")
if not receiver:
return jsonify({
"ok": False,
"error": {"code": "VALIDATION_FAILED", "message": "Missing 'receiver_handle'"},
}), 400
if not song_id and not song_title:
return jsonify({
"ok": False,
"error": {"code": "VALIDATION_FAILED", "message": "Provide 'song_id' or 'song_title'"},
}), 400
if not song_id:
print(f"📻 send-to-friend: resolving song '{song_title}'")
resolved = _airadio_call(
"GET",
"/api/agent/resolve",
params={"type": "song", "query": song_title},
)
if not resolved["ok"]:
return _bridge_response(resolved)
song_id = (resolved.get("data") or {}).get("id")
if not song_id:
return jsonify({
"ok": False,
"error": {"code": "NOT_FOUND", "message": f"No song matching '{song_title}'"},
}), 404
print(f"📻 send-to-friend: song={song_id} receiver={receiver}")
result = _airadio_call(
"POST",
"/api/agent/send-to-friend",
json={"songId": song_id, "receiverHandle": receiver, "note": note},
)
return _bridge_response(result)
@airadio_bp.route("/api/airadio/vote", methods=["POST"])
def vote():
"""
Vote on a song.
Body: { "song_id": optional, "song_title": optional, "value": 1|-1|0 }
"""
if not _bridge_has_auth():
return _bridge_not_configured_response(for_write=True)
body = request.get_json(silent=True) or {}
song_id = body.get("song_id")
song_title = body.get("song_title")
value = body.get("value")
if value not in (1, -1, 0):
return jsonify({
"ok": False,
"error": {"code": "VALIDATION_FAILED", "message": "'value' must be 1, -1, or 0"},
}), 400
if not song_id and not song_title:
return jsonify({
"ok": False,
"error": {"code": "VALIDATION_FAILED", "message": "Provide 'song_id' or 'song_title'"},
}), 400
if not song_id:
print(f"📻 vote: resolving song '{song_title}'")
resolved = _airadio_call(
"GET",
"/api/agent/resolve",
params={"type": "song", "query": song_title},
)
if not resolved["ok"]:
return _bridge_response(resolved)
song_id = (resolved.get("data") or {}).get("id")
if not song_id:
return jsonify({
"ok": False,
"error": {"code": "NOT_FOUND", "message": f"No song matching '{song_title}'"},
}), 404
print(f"📻 vote: song={song_id} value={value}")
result = _airadio_call("POST", "/api/agent/vote", json={"songId": song_id, "value": value})
return _bridge_response(result)
# ---------------------------------------------------------------------------
# Friend management — pure pass-through
# ---------------------------------------------------------------------------
def _friend_passthrough(action_path: str):
if not _bridge_has_auth():
return _bridge_not_configured_response(for_write=True)
body = request.get_json(silent=True) or {}
handle = body.get("handle")
if not handle:
return jsonify({
"ok": False,
"error": {"code": "VALIDATION_FAILED", "message": "Missing 'handle'"},
}), 400
print(f"📻 {action_path}: handle={handle}")
result = _airadio_call("POST", action_path, json={"handle": handle})
return _bridge_response(result)
@airadio_bp.route("/api/airadio/friend-request", methods=["POST"])
def friend_request():
return _friend_passthrough("/api/agent/friend-request")
@airadio_bp.route("/api/airadio/friend-accept", methods=["POST"])
def friend_accept():
return _friend_passthrough("/api/agent/friend-accept")
@airadio_bp.route("/api/airadio/friend-decline", methods=["POST"])
def friend_decline():
return _friend_passthrough("/api/agent/friend-decline")
# ---------------------------------------------------------------------------
# Library
# ---------------------------------------------------------------------------
@airadio_bp.route("/api/airadio/library", methods=["GET"])
def library():
"""Current user's library on AI-Radio (own songs + playlists)."""
if not _bridge_has_auth():
return _bridge_not_configured_response(for_write=True)
print("📻 library: fetching")
result = _airadio_call("GET", "/api/agent/library")
return _bridge_response(result)
# ---------------------------------------------------------------------------
# Full catalog API (doc 22 — additive, cross-user)
# ---------------------------------------------------------------------------
# Query params we forward to /api/agent/catalog. Anything else is dropped.
_CATALOG_PASSTHROUGH_PARAMS = ("q", "genre", "artist", "sort", "cursor", "limit")
@airadio_bp.route("/api/airadio/catalog", methods=["GET"])
def catalog():
"""
Platform-wide public song catalog search.
Query params:
q, genre, artist, sort=new|trending|top_week|top_all|plays,
cursor, limit (<=100, default 24).
Returns {items: [...], nextCursor}. Each item has a signed `streamUrl`
valid for 15 minutes that can be fed directly to the OVU music player
— no Clerk session needed.
"""
if not _bridge_has_backend():
return _bridge_not_configured_response()
params = {
k: request.args.get(k)
for k in _CATALOG_PASSTHROUGH_PARAMS
if request.args.get(k) is not None
}
print(f"📻 catalog: {params}")
result = _airadio_call("GET", "/api/agent/catalog", params=params or None)
return _bridge_response(result)
@airadio_bp.route("/api/airadio/catalog/<song_id>", methods=["GET"])
def catalog_detail(song_id):
"""Single song detail (catalog shape + viewer flags: liked, voted, saved)."""
if not _bridge_has_backend():
return _bridge_not_configured_response()
if not song_id:
return jsonify({
"ok": False,
"error": {"code": "VALIDATION_FAILED", "message": "Missing song id"},
}), 400
print(f"📻 catalog/{song_id}")
result = _airadio_call("GET", f"/api/agent/catalog/{song_id}")
return _bridge_response(result)
@airadio_bp.route("/api/airadio/u/<username>/songs", methods=["GET"])
def user_songs(username):
"""Paginated public songs by @<username> (all songs if it's the caller)."""
if not _bridge_has_backend():
return _bridge_not_configured_response()
if not username:
return jsonify({
"ok": False,
"error": {"code": "VALIDATION_FAILED", "message": "Missing username"},
}), 400
handle = username.lstrip("@")
params = {}
for k in ("cursor", "limit"):
v = request.args.get(k)
if v is not None:
params[k] = v
print(f"📻 u/{handle}/songs: {params}")
result = _airadio_call(
"GET",
f"/api/agent/u/{handle}/songs",
params=params or None,
)
return _bridge_response(result)
@airadio_bp.route("/api/airadio/song/<song_id>/stream-url", methods=["POST"])
def song_stream_url(song_id):
"""Issue a signed 15-min stream URL for the given song."""
if not _bridge_has_backend():
return _bridge_not_configured_response()
if not song_id:
return jsonify({
"ok": False,
"error": {"code": "VALIDATION_FAILED", "message": "Missing song id"},
}), 400
print(f"📻 stream-url: song={song_id}")
result = _airadio_call("POST", f"/api/agent/song/{song_id}/stream-url")
return _bridge_response(result)
@airadio_bp.route("/api/airadio/library/save", methods=["POST"])
def library_save():
"""
Save another user's song to the caller's library.
Body: { "song_id": "..." } (accepts "songId" too for symmetry)
"""
if not _bridge_has_auth():
return _bridge_not_configured_response(for_write=True)
body = request.get_json(silent=True) or {}
song_id = body.get("song_id") or body.get("songId")
if not song_id:
return jsonify({
"ok": False,
"error": {"code": "VALIDATION_FAILED", "message": "Missing 'song_id'"},
}), 400
print(f"📻 library/save: song={song_id}")
result = _airadio_call(
"POST",
"/api/agent/library/save",
json={"songId": song_id},
)
return _bridge_response(result)
@airadio_bp.route("/api/airadio/playlist", methods=["POST"])
def playlist_create():
"""
Create a new AI-Radio playlist owned by the caller.
Body: { "name": "...", "description"?: "...", "is_public"?: bool }
"""
if not _bridge_has_auth():
return _bridge_not_configured_response(for_write=True)
body = request.get_json(silent=True) or {}
name = body.get("name")
if not name:
return jsonify({
"ok": False,
"error": {"code": "VALIDATION_FAILED", "message": "Missing 'name'"},
}), 400
payload = {"name": name}
if body.get("description") is not None:
payload["description"] = body.get("description")
# Accept both is_public and isPublic from the caller.
if "is_public" in body:
payload["isPublic"] = bool(body.get("is_public"))
elif "isPublic" in body:
payload["isPublic"] = bool(body.get("isPublic"))
print(f"📻 playlist/create: name={name}")
result = _airadio_call("POST", "/api/agent/playlist", json=payload)
return _bridge_response(result)
@airadio_bp.route("/api/airadio/playlist/<playlist_id>", methods=["GET"])
def playlist_read(playlist_id):
"""Read a playlist + ordered songs (each with a signed streamUrl)."""
if not _bridge_has_backend():
return _bridge_not_configured_response()
if not playlist_id:
return jsonify({
"ok": False,
"error": {"code": "VALIDATION_FAILED", "message": "Missing playlist id"},
}), 400
print(f"📻 playlist/{playlist_id}")
result = _airadio_call("GET", f"/api/agent/playlist/{playlist_id}")
return _bridge_response(result)
# ---------------------------------------------------------------------------
# Queue / Radio — voice-shaped shortcuts (doc 22+). Every endpoint returns
# {items, reason, nextCursor}. OVU can feed `items` straight into the music
# player. Each item has a 15-min signed streamUrl.
# ---------------------------------------------------------------------------
# Query params forwarded to every queue endpoint.
_QUEUE_PASSTHROUGH_PARAMS = (
"mood",
"genre",
"freshness",
"audience",
"similarTo",
"artist",
"excludeHeard",
"window",
"q",
"limit",
"cursor",
# bpmMin/bpmMax are rejected by the upstream today — forward so the
# caller sees the VALIDATION_FAILED from AI-Radio unchanged.
"bpmMin",
"bpmMax",
)
def _queue_params():
return {
k: request.args.get(k)
for k in _QUEUE_PASSTHROUGH_PARAMS
if request.args.get(k) is not None
}
def _queue_proxy(path):
if not _bridge_has_backend():
return _bridge_not_configured_response()
params = _queue_params()
print(f"📻 {path}: {params}")
result = _airadio_call("GET", path, params=params or None)
return _bridge_response(result)
@airadio_bp.route("/api/airadio/queue", methods=["GET"])
def queue_generic():
"""Generic queue resolver — all QueueParams are URL params."""
return _queue_proxy("/api/agent/queue")
@airadio_bp.route("/api/airadio/queue/new", methods=["GET"])
def queue_new():
return _queue_proxy("/api/agent/queue/new")
@airadio_bp.route("/api/airadio/queue/trending", methods=["GET"])
def queue_trending():
return _queue_proxy("/api/agent/queue/trending")
@airadio_bp.route("/api/airadio/queue/top", methods=["GET"])
def queue_top():
return _queue_proxy("/api/agent/queue/top")
@airadio_bp.route("/api/airadio/queue/random", methods=["GET"])
def queue_random():
return _queue_proxy("/api/agent/queue/random")
@airadio_bp.route("/api/airadio/queue/friends", methods=["GET"])
def queue_friends():
return _queue_proxy("/api/agent/queue/friends")
@airadio_bp.route("/api/airadio/queue/following", methods=["GET"])
def queue_following():
return _queue_proxy("/api/agent/queue/following")
@airadio_bp.route("/api/airadio/queue/me", methods=["GET"])
def queue_me():
return _queue_proxy("/api/agent/queue/me")
@airadio_bp.route("/api/airadio/queue/liked", methods=["GET"])
def queue_liked():
return _queue_proxy("/api/agent/queue/liked")
@airadio_bp.route("/api/airadio/queue/unheard", methods=["GET"])
def queue_unheard():
return _queue_proxy("/api/agent/queue/unheard")
@airadio_bp.route("/api/airadio/queue/genre/<slug>", methods=["GET"])
def queue_genre(slug):
if not slug:
return jsonify({
"ok": False,
"error": {"code": "VALIDATION_FAILED", "message": "Missing genre slug"},
}), 400
return _queue_proxy(f"/api/agent/queue/genre/{slug}")
@airadio_bp.route("/api/airadio/queue/mood/<mood>", methods=["GET"])
def queue_mood(mood):
if not mood:
return jsonify({
"ok": False,
"error": {"code": "VALIDATION_FAILED", "message": "Missing mood"},
}), 400
return _queue_proxy(f"/api/agent/queue/mood/{mood}")
@airadio_bp.route("/api/airadio/queue/artist/<username>", methods=["GET"])
def queue_artist(username):
# Accept "@example" as well as "example"
handle = (username or "").lstrip("@")
if not handle:
return jsonify({
"ok": False,
"error": {"code": "VALIDATION_FAILED", "message": "Missing username"},
}), 400
return _queue_proxy(f"/api/agent/queue/artist/{handle}")
@airadio_bp.route("/api/airadio/queue/similar/<song_id>", methods=["GET"])
def queue_similar(song_id):
if not song_id:
return jsonify({
"ok": False,
"error": {"code": "VALIDATION_FAILED", "message": "Missing song id"},
}), 400
return _queue_proxy(f"/api/agent/queue/similar/{song_id}")
@airadio_bp.route("/api/airadio/interpret", methods=["POST"])
def interpret():
"""Natural-language phrase → QueueParams + items."""
if not _bridge_has_backend():
return _bridge_not_configured_response()
body = request.get_json(silent=True) or {}
query = body.get("query")
if not query:
return jsonify({
"ok": False,
"error": {"code": "VALIDATION_FAILED", "message": "Missing 'query'"},
}), 400
print(f"📻 interpret: '{query}'")
result = _airadio_call("POST", "/api/agent/interpret", json={"query": query})
return _bridge_response(result)
# ---------------------------------------------------------------------------
# Library search (dedup) — "is this song already on my AI-Radio?"
# ---------------------------------------------------------------------------
def _title_tokens(s: str):
import re as _re
return [t for t in _re.split(r"[^a-z0-9]+", (s or "").lower()) if t]
@airadio_bp.route("/api/airadio/library/search", methods=["GET"])
def library_search():
"""
Search ONLY the current user's AI-Radio library by title.
Use this before pushing a song to avoid re-uploading a track the user
already owns on AI-Radio. The upstream dedupes by `sourceHash`, but the
agent should still know BEFORE the push whether the song exists so it
can tell the user instead of silently no-op'ing.
Query params:
q — required, title text (fuzzy-matched)
limit — optional, default 10, max 50
Returns:
{
ok: true,
query: "<q>",
exists: bool, // any fuzzy match?
exact: bool, // true iff at least one match is an exact title
matches: [
{ id, title, artist, coverUrl, plays, createdAt, ... }
]
}
"""
if not _bridge_has_auth():
return _bridge_not_configured_response(for_write=True)
q = (request.args.get("q") or "").strip()
if not q:
return jsonify({
"ok": False,
"error": {"code": "VALIDATION_FAILED", "message": "Missing 'q'"},
}), 400
try:
limit = max(1, min(50, int(request.args.get("limit", "10"))))
except (TypeError, ValueError):
limit = 10
print(f"📻 library/search: q='{q}' limit={limit}")
# Pull the caller's library — the upstream scopes this to the authenticated
# user automatically (Bearer aia_sk_* or platform key + X-JamBot-User).
lib = _airadio_call("GET", "/api/agent/library")
if not lib["ok"]:
return _bridge_response(lib)
data = lib.get("data") or {}
# The library payload shape is { songs: [...], playlists: [...] } — be
# defensive, the API version still moves.
songs = data.get("songs") or data.get("items") or []
if not isinstance(songs, list):
songs = []
q_lower = q.lower().strip()
q_tokens = set(_title_tokens(q))
matches = []
exact = False
for s in songs:
if not isinstance(s, dict):
continue
title = (s.get("title") or "").strip()
if not title:
continue
t_lower = title.lower()
if t_lower == q_lower:
exact = True
matches.append((0, s))
continue
if q_lower and q_lower in t_lower:
matches.append((1, s))
continue
t_tokens = set(_title_tokens(title))
if q_tokens and q_tokens.issubset(t_tokens):
matches.append((2, s))
continue
# Token overlap fallback: all-but-one token match
if q_tokens and len(q_tokens) > 1:
overlap = q_tokens & t_tokens
if len(overlap) >= max(1, len(q_tokens) - 1):
matches.append((3, s))
matches.sort(key=lambda pair: pair[0])
slimmed = []
for _, s in matches[:limit]:
slimmed.append({
"id": s.get("id") or s.get("songId"),
"title": s.get("title"),
"artist": s.get("artist") or s.get("artistHandle") or s.get("owner"),
"coverUrl": s.get("coverUrl") or s.get("cover"),
"plays": s.get("plays"),
"createdAt": s.get("createdAt") or s.get("created_at"),
"isPublic": s.get("isPublic"),
})
return jsonify({
"ok": True,
"query": q,
"exists": len(matches) > 0,
"exact": exact,
"count": len(matches),
"matches": slimmed,
}), 200
# ---------------------------------------------------------------------------
# Per-user API key — write aia_sk_* to the workspace .env so subsequent
# requests authenticate as that specific AI-Radio user.
# ---------------------------------------------------------------------------
@airadio_bp.route("/api/airadio/set-user-key", methods=["POST"])
def set_user_key():
"""
Persist a per-user AI-Radio API key for this workspace.
Body: { "key": "aia_sk_..." }
The key is written to RUNTIME_DIR/airadio-user-key, which is bind-mounted
from the host so it survives container restarts and image upgrades.
Subsequent /api/airadio/* calls pick the new key up immediately
(no restart required) — `_effective_user_key()` reads the file at
request time.
"""
body = request.get_json(silent=True) or {}
key = (body.get("key") or "").strip()
if not key:
return jsonify({
"ok": False,
"error": {"code": "VALIDATION_FAILED", "message": "Missing 'key'"},
}), 400
if not key.startswith("aia_sk_") or len(key) < 20:
return jsonify({
"ok": False,
"error": {
"code": "VALIDATION_FAILED",
"message": "Key must look like 'aia_sk_...' (minimum 20 chars)",
},
}), 400
# Probe the key before persisting so a typo doesn't replace a working key.
try:
probe = http_requests.get(
f"{AIRADIO_URL}/api/agent/library",
headers={"Authorization": f"Bearer {key}"},
timeout=AIRADIO_TIMEOUT,
)
except http_requests.RequestException as exc:
return jsonify({
"ok": False,
"error": {"code": "BRIDGE_UNREACHABLE", "message": str(exc)},
}), 502
if probe.status_code in (401, 403):
return jsonify({
"ok": False,
"error": {
"code": "AUTH_API_KEY_INVALID",
"message": "AI-Radio rejected the key. Check that it was copied correctly and not rotated.",
},
}), 401
if probe.status_code >= 500:
return jsonify({
"ok": False,
"error": {
"code": "AIRADIO_HTTP_ERROR",
"message": f"AI-Radio returned HTTP {probe.status_code} during key probe",
},
}), 502
try:
_USER_KEY_PERSIST_PATH.parent.mkdir(parents=True, exist_ok=True)
_USER_KEY_PERSIST_PATH.write_text(key + "\n")
# Best-effort tighten — runtime dir lives on a 777 mount; the key file
# itself we keep readable only by the running user when possible.
try:
os.chmod(_USER_KEY_PERSIST_PATH, 0o600)
except OSError:
pass
except OSError as exc:
return jsonify({
"ok": False,
"error": {"code": "WORKSPACE_KEY_WRITE_FAILED", "message": str(exc)},
}), 500
print(f"📻 set-user-key: persisted AIRADIO_USER_KEY → {_USER_KEY_PERSIST_PATH}")
return jsonify({
"ok": True,
"message": "AI-Radio user key saved and active for this workspace.",
"path": str(_USER_KEY_PERSIST_PATH),
"restartRequired": False,
}), 200
# ---------------------------------------------------------------------------
# Reply to a received song (wraps send-to-friend with a reply-of pointer)
# ---------------------------------------------------------------------------
@airadio_bp.route("/api/airadio/reply-to-send", methods=["POST"])
def reply_to_send():
"""
Reply to a song a friend sent you by sending another song back.
Body: { "send_id": "<inbox entry id>",
"song_id": optional,
"song_title": optional,
"note": optional }
"""
if not _bridge_has_auth():
return _bridge_not_configured_response(for_write=True)
body = request.get_json(silent=True) or {}
send_id = body.get("send_id") or body.get("sendId")
song_id = body.get("song_id") or body.get("songId")
song_title = body.get("song_title") or body.get("songTitle")
note = body.get("note", "")
if not send_id:
return jsonify({
"ok": False,
"error": {"code": "VALIDATION_FAILED", "message": "Missing 'send_id'"},
}), 400
if not song_id and not song_title:
return jsonify({
"ok": False,
"error": {"code": "VALIDATION_FAILED", "message": "Provide 'song_id' or 'song_title'"},
}), 400
if not song_id:
resolved = _airadio_call(
"GET",
"/api/agent/resolve",
params={"type": "song", "query": song_title},
)
if not resolved["ok"]:
return _bridge_response(resolved)
song_id = (resolved.get("data") or {}).get("id")
if not song_id:
return jsonify({
"ok": False,
"error": {"code": "NOT_FOUND", "message": f"No song matching '{song_title}'"},
}), 404
print(f"📻 reply-to-send: in-reply-to={send_id} song={song_id}")
result = _airadio_call(
"POST",
"/api/agent/reply-to-send",
json={"sendId": send_id, "songId": song_id, "note": note},
)
return _bridge_response(result)