1062 lines
39 KiB
Python
1062 lines
39 KiB
Python
"""
|
|
TtsToy Web UI — Flask app
|
|
Reads RedBot config directly from disk. No RedBot dependency at runtime.
|
|
Run: python app.py (or gunicorn app:app)
|
|
"""
|
|
import os
|
|
import json
|
|
import logging
|
|
import secrets
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
from functools import wraps
|
|
|
|
import requests
|
|
from flask import (
|
|
Flask, render_template, request, redirect, url_for,
|
|
session, jsonify, send_file, abort, flash
|
|
)
|
|
|
|
log = logging.getLogger("ttstoy.webui")
|
|
|
|
app = Flask(__name__)
|
|
|
|
# ── Load stack config for shared session ──
|
|
_STACK_FILE = Path(__file__).parent / "stack.json"
|
|
def _load_stack():
|
|
try:
|
|
with open(_STACK_FILE) as f:
|
|
return json.load(f)
|
|
except Exception:
|
|
return {}
|
|
|
|
_stack = _load_stack()
|
|
app.secret_key = _stack.get("session_secret", os.environ.get("WEBUI_SECRET", secrets.token_hex(32)))
|
|
app.config["SESSION_COOKIE_DOMAIN"] = _stack.get("cookie_domain", None)
|
|
app.config["SESSION_COOKIE_NAME"] = "scrapyard_session"
|
|
app.config["SESSION_COOKIE_SAMESITE"] = "Lax"
|
|
app.config["SESSION_COOKIE_SECURE"] = True
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Config — edit these or set via environment variables
|
|
# ---------------------------------------------------------------------------
|
|
REDBOT_DATA_PATH = Path(os.environ.get(
|
|
"REDBOT_DATA_PATH",
|
|
Path.home() / ".local/share/Red-DiscordBot/data"
|
|
))
|
|
BOT_INSTANCE = os.environ.get("BOT_INSTANCE", "redbot")
|
|
WEBUI_PORT = int(os.environ.get("WEBUI_PORT", 8098))
|
|
# Hardcoded public URL used for login links (Discord DM).
|
|
# This web UI is typically reached via Cloudflared at this hostname.
|
|
WEBUI_URL = "https://ttstoy.kingstons-scrapyard.net"
|
|
|
|
# Path where RedBot stores cog config
|
|
COG_CONFIG_PATH = REDBOT_DATA_PATH / BOT_INSTANCE / "cogs" / "TtsToy"
|
|
# Path where RedBot stores per-user config
|
|
CORE_CONFIG_PATH = REDBOT_DATA_PATH / BOT_INSTANCE / "core"
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Shared state init — runs at import time so gunicorn workers pick it up
|
|
# ---------------------------------------------------------------------------
|
|
_internal_secret = os.environ.get("TTSTOY_INTERNAL_SECRET") or secrets.token_hex(16)
|
|
app.config["INTERNAL_SECRET"] = _internal_secret
|
|
SHARED_STATE_FILE = Path(os.environ.get("TTSTOY_SHARED_STATE", "/tmp/ttstoy_webui_state.json"))
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers — RedBot config reading (defined early; needed by _init_shared_state)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _read_json(path: Path) -> dict:
|
|
try:
|
|
with open(path) as f:
|
|
return json.load(f)
|
|
except FileNotFoundError:
|
|
log.debug(f"Config file not found: {path}")
|
|
return {}
|
|
except Exception as e:
|
|
log.warning(f"Failed to read {path}: {e}")
|
|
return {}
|
|
|
|
|
|
def _init_shared_state():
|
|
try:
|
|
state = _read_json(SHARED_STATE_FILE) if SHARED_STATE_FILE.exists() else {}
|
|
state["internal_secret"] = _internal_secret
|
|
state["webui_url"] = WEBUI_URL
|
|
SHARED_STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(SHARED_STATE_FILE, "w") as f:
|
|
json.dump(state, f)
|
|
except Exception as e:
|
|
log.warning(f"Could not init shared state: {e}")
|
|
# Also ensure shared_auth.json exists with the secret
|
|
try:
|
|
auth = _read_auth_file()
|
|
auth["secret"] = _internal_secret
|
|
_write_auth_file(auth)
|
|
except Exception:
|
|
pass
|
|
|
|
_init_shared_state()
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ---------------------------------------------------------------------------
|
|
# File-based login token store: ttstoy/shared_auth.json
|
|
# All apps in the stack read/write this file via stack.json paths.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _get_auth_file() -> Path:
|
|
"""Resolve shared_auth.json location from stack.json, fallback to relative."""
|
|
ttstoy_path = _stack.get("ttstoy", "")
|
|
if ttstoy_path:
|
|
return Path(ttstoy_path) / "shared_auth.json"
|
|
return Path(__file__).parent.parent / "shared_auth.json"
|
|
|
|
def _read_auth_file() -> dict:
|
|
p = _get_auth_file()
|
|
try:
|
|
with open(p) as f:
|
|
return json.load(f)
|
|
except (FileNotFoundError, json.JSONDecodeError):
|
|
return {"tokens": {}, "secret": _internal_secret}
|
|
|
|
def _write_auth_file(data: dict):
|
|
p = _get_auth_file()
|
|
p.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(p, "w") as f:
|
|
json.dump(data, f, indent=2)
|
|
|
|
_token_lock = threading.Lock()
|
|
TOKEN_TTL = 300 # 5 minutes to use the token
|
|
|
|
# In-memory TTS output store: {job_id: {status, path, text, user, engine}}
|
|
_tts_jobs: dict = {}
|
|
|
|
|
|
# RedBot stores TtsToy config under the cog's numeric identifier (0x0A0A0A0A)
|
|
COG_IDENTIFIER = "168430090"
|
|
|
|
|
|
def get_global_config() -> dict:
|
|
"""Read TtsToy global config from RedBot data dir."""
|
|
p = COG_CONFIG_PATH / "settings.json"
|
|
data = _read_json(p)
|
|
if not data:
|
|
return {}
|
|
cfg = data.get(COG_IDENTIFIER, {}).get("GLOBAL", {})
|
|
log.debug(f"Global config: mode={cfg.get('tts_mode')} chatterbox={cfg.get('chatterbox_api_url')}")
|
|
return cfg
|
|
|
|
|
|
def get_user_config(user_id: str) -> dict:
|
|
"""Read TtsToy per-user config."""
|
|
p = COG_CONFIG_PATH / "settings.json"
|
|
data = _read_json(p)
|
|
if not data:
|
|
return {}
|
|
cfg = data.get(COG_IDENTIFIER, {}).get("USER", {}).get(str(user_id), {})
|
|
log.debug(f"User config for {user_id}: voice={cfg.get('minimax_voice')} voices={cfg.get('chatterbox_voices')}")
|
|
return cfg
|
|
|
|
|
|
def get_shared_state() -> dict:
|
|
return _read_json(SHARED_STATE_FILE)
|
|
|
|
|
|
def get_chatterbox_url() -> str:
|
|
cfg = get_global_config()
|
|
return cfg.get("chatterbox_api_url", "http://192.168.0.200:8099")
|
|
|
|
|
|
def get_dectalk_url() -> str:
|
|
cfg = get_global_config()
|
|
return cfg.get("dectalk_api_url", "http://127.0.0.1:33001")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Auth helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def create_login_token(user_id: str, discord_name: str, avatar_url: str = "", is_owner: bool = False, guild_id: str = "", guild_name: str = "") -> str:
|
|
token = secrets.token_urlsafe(24)
|
|
with _token_lock:
|
|
auth = _read_auth_file()
|
|
auth.setdefault("tokens", {})[token] = {
|
|
"user_id": user_id,
|
|
"discord_name": discord_name,
|
|
"avatar_url": avatar_url,
|
|
"is_owner": is_owner,
|
|
"guild_id": guild_id,
|
|
"guild_name": guild_name,
|
|
"expires": time.time() + TOKEN_TTL,
|
|
}
|
|
auth["secret"] = _internal_secret
|
|
_write_auth_file(auth)
|
|
log.debug(f"Login token created for user {user_id} ({discord_name}) guild={guild_id}, expires in {TOKEN_TTL}s")
|
|
return token
|
|
|
|
|
|
def consume_login_token(token: str) -> dict | None:
|
|
"""Validate a token. Does NOT delete it — the session cookie handles auth
|
|
across all apps via the shared cookie domain."""
|
|
with _token_lock:
|
|
auth = _read_auth_file()
|
|
tokens = auth.get("tokens", {})
|
|
entry = tokens.get(token)
|
|
if not entry:
|
|
log.debug("consume_login_token: token not found")
|
|
return None
|
|
if time.time() > entry["expires"]:
|
|
log.debug(f"consume_login_token: token expired for user {entry['user_id']}")
|
|
del tokens[token]
|
|
_write_auth_file(auth)
|
|
return None
|
|
log.debug(f"consume_login_token: success for user {entry['user_id']} ({entry['discord_name']})")
|
|
return entry
|
|
|
|
|
|
def login_required(f):
|
|
@wraps(f)
|
|
def decorated(*args, **kwargs):
|
|
if "user_id" not in session:
|
|
return redirect(url_for("login_page"))
|
|
return f(*args, **kwargs)
|
|
return decorated
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Routes — Auth
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@app.route("/")
|
|
def index():
|
|
if "user_id" in session:
|
|
return redirect(url_for("dashboard"))
|
|
return redirect(url_for("login_page"))
|
|
|
|
|
|
@app.route("/login")
|
|
def login_page():
|
|
return render_template("login.html", webui_url=WEBUI_URL)
|
|
|
|
|
|
@app.route("/login", methods=["POST"])
|
|
def login_submit():
|
|
token = request.form.get("token", "").strip()
|
|
log.debug(f"Login attempt with token length={len(token)}")
|
|
entry = consume_login_token(token)
|
|
if not entry:
|
|
log.warning("Login failed: invalid or expired token")
|
|
flash("Invalid or expired token. Run [p]ttstoy login in Discord to get a new one.", "error")
|
|
return redirect(url_for("login_page"))
|
|
session["user_id"] = entry["user_id"]
|
|
session["discord_name"] = entry["discord_name"]
|
|
session["avatar_url"] = entry["avatar_url"]
|
|
session["is_owner"] = entry.get("is_owner", False)
|
|
session["guild_id"] = entry.get("guild_id", "")
|
|
session["guild_name"] = entry.get("guild_name", "")
|
|
# Persist user→guild mapping server-side so it survives session overwrites
|
|
try:
|
|
state = _read_json(SHARED_STATE_FILE) if SHARED_STATE_FILE.exists() else {}
|
|
ugm = state.setdefault("user_guild_map", {})
|
|
if entry.get("guild_id"):
|
|
ugm[entry["user_id"]] = entry["guild_id"]
|
|
with open(SHARED_STATE_FILE, "w") as f:
|
|
json.dump(state, f)
|
|
except Exception:
|
|
pass
|
|
log.info(f"User logged in: {entry['discord_name']} ({entry['user_id']}) owner={session['is_owner']} guild={session['guild_id']}")
|
|
return redirect(url_for("dashboard"))
|
|
|
|
|
|
@app.route("/logout")
|
|
def logout():
|
|
session.clear()
|
|
return redirect(url_for("login_page"))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Routes — Dashboard
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@app.route("/dashboard")
|
|
@login_required
|
|
def dashboard():
|
|
user_id = session["user_id"]
|
|
gcfg = get_global_config()
|
|
ucfg = get_user_config(user_id)
|
|
global_mode = gcfg.get("tts_mode", "minimax")
|
|
# Per-user mode override stored in shared state
|
|
state = get_shared_state()
|
|
user_mode_overrides = state.get("user_mode_overrides", {})
|
|
mode = user_mode_overrides.get(str(user_id), global_mode)
|
|
user_voice = ucfg.get("minimax_voice")
|
|
# Build display name for active voice
|
|
if user_voice and "_" in user_voice:
|
|
active_voice_display = user_voice.split("_", 1)[1].rsplit(".", 1)[0]
|
|
else:
|
|
active_voice_display = user_voice or "Global default"
|
|
return render_template("dashboard.html",
|
|
mode=mode,
|
|
global_mode=global_mode,
|
|
user_voice=user_voice,
|
|
active_voice_display=active_voice_display,
|
|
gcfg=gcfg,
|
|
ucfg=ucfg,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Routes — TTS Submit
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@app.route("/api/tts", methods=["POST"])
|
|
@login_required
|
|
def api_tts():
|
|
"""Queue a TTS request. If generate_only is true, generate directly without VC."""
|
|
data = request.get_json(force=True)
|
|
text = (data.get("text") or "").strip()
|
|
if not text:
|
|
return jsonify({"error": "No text provided"}), 400
|
|
|
|
# If generate_only, redirect to the direct generation endpoint
|
|
if data.get("generate_only"):
|
|
return api_tts_generate()
|
|
|
|
user_id = session["user_id"]
|
|
discord_name = session.get("discord_name", "Unknown")
|
|
gcfg = get_global_config()
|
|
|
|
# Allow per-request mode/voice override from the web UI
|
|
state = get_shared_state()
|
|
user_mode_overrides = state.get("user_mode_overrides", {})
|
|
global_mode = gcfg.get("tts_mode", "minimax")
|
|
mode = data.get("mode") or user_mode_overrides.get(str(user_id), global_mode)
|
|
voice = data.get("voice") or None # None = use user's saved voice
|
|
|
|
job_id = str(uuid.uuid4())
|
|
log.info(f"TTS job {job_id}: user={discord_name} ({user_id}) mode={mode} voice={voice} text={text[:80]!r}")
|
|
_tts_jobs[job_id] = {"status": "pending", "text": text, "user": discord_name, "engine": mode}
|
|
|
|
_write_bot_command(user_id, "speak_in_vc", {
|
|
"job_id": job_id,
|
|
"text": text,
|
|
"user": discord_name,
|
|
"mode_override": mode,
|
|
"voice_override": voice,
|
|
"guild_id": _resolve_guild_id(),
|
|
})
|
|
|
|
return jsonify({"job_id": job_id})
|
|
|
|
|
|
@app.route("/api/tts/<job_id>/status")
|
|
@login_required
|
|
def tts_job_status(job_id):
|
|
job = _tts_jobs.get(job_id)
|
|
if not job:
|
|
return jsonify({"status": "not_found"}), 404
|
|
return jsonify({k: v for k, v in job.items() if k != "path"})
|
|
|
|
|
|
@app.route("/api/tts/<job_id>/audio")
|
|
@login_required
|
|
def tts_job_audio(job_id):
|
|
job = _tts_jobs.get(job_id)
|
|
if not job or job.get("status") != "done":
|
|
abort(404)
|
|
return send_file(job["path"], mimetype="audio/mpeg", as_attachment=False)
|
|
|
|
|
|
@app.route("/api/tts/<job_id>/update", methods=["POST"])
|
|
def tts_job_update(job_id):
|
|
"""Called by the bot to update a job's status (done/error) and provide the audio path."""
|
|
secret = request.headers.get("X-Internal-Secret", "")
|
|
if secret != app.config.get("INTERNAL_SECRET", ""):
|
|
abort(403)
|
|
data = request.get_json(force=True)
|
|
if job_id in _tts_jobs:
|
|
_tts_jobs[job_id].update(data)
|
|
log.debug(f"Job {job_id} updated by bot: {data}")
|
|
return jsonify({"ok": True})
|
|
|
|
|
|
@app.route("/api/tts/generate", methods=["POST"])
|
|
@login_required
|
|
def api_tts_generate():
|
|
"""Generate TTS via the bot or directly via TTS servers."""
|
|
data = request.get_json(force=True)
|
|
text = (data.get("text") or "").strip()
|
|
if not text:
|
|
return jsonify({"error": "No text provided"}), 400
|
|
|
|
user_id = session["user_id"]
|
|
discord_name = session.get("discord_name", "Unknown")
|
|
gcfg = get_global_config()
|
|
state = get_shared_state()
|
|
user_mode_overrides = state.get("user_mode_overrides", {})
|
|
global_mode = gcfg.get("tts_mode", "minimax")
|
|
mode = data.get("mode") or user_mode_overrides.get(str(user_id), global_mode)
|
|
|
|
# Direct server modes — generate audio without the bot
|
|
if mode == "morshu":
|
|
try:
|
|
r = requests.get("http://127.0.0.1:33002/say", params={"text": text}, timeout=60)
|
|
r.raise_for_status()
|
|
job_id = str(uuid.uuid4())
|
|
out_path = tempfile.mktemp(suffix=".wav")
|
|
with open(out_path, "wb") as f:
|
|
f.write(r.content)
|
|
_tts_jobs[job_id] = {"status": "done", "path": out_path, "text": text, "user": discord_name, "engine": "morshu"}
|
|
return jsonify({"job_id": job_id})
|
|
except Exception as e:
|
|
return jsonify({"error": f"Morshu server error: {e}"}), 500
|
|
|
|
if mode == "dectalk":
|
|
try:
|
|
voice = data.get("voice") or ""
|
|
dectalk_text = f"{voice}{text}" if voice else text
|
|
dectalk_url = gcfg.get("dectalk_api_url", "http://127.0.0.1:33001")
|
|
r = requests.get(f"{dectalk_url}/say", params={"text": dectalk_text}, timeout=30)
|
|
r.raise_for_status()
|
|
job_id = str(uuid.uuid4())
|
|
out_path = tempfile.mktemp(suffix=".wav")
|
|
with open(out_path, "wb") as f:
|
|
f.write(r.content)
|
|
_tts_jobs[job_id] = {"status": "done", "path": out_path, "text": text, "user": discord_name, "engine": "dectalk"}
|
|
return jsonify({"job_id": job_id})
|
|
except Exception as e:
|
|
return jsonify({"error": f"DECTalk server error: {e}"}), 500
|
|
|
|
if mode == "vox":
|
|
try:
|
|
pack = gcfg.get("vox_pack", "vox")
|
|
r = requests.get("http://127.0.0.1:33003/say", params={"text": text, "pack": pack}, timeout=60)
|
|
r.raise_for_status()
|
|
job_id = str(uuid.uuid4())
|
|
out_path = tempfile.mktemp(suffix=".wav")
|
|
with open(out_path, "wb") as f:
|
|
f.write(r.content)
|
|
_tts_jobs[job_id] = {"status": "done", "path": out_path, "text": text, "user": discord_name, "engine": "vox"}
|
|
return jsonify({"job_id": job_id})
|
|
except Exception as e:
|
|
return jsonify({"error": f"VOX server error: {e}"}), 500
|
|
|
|
if mode != "chatterbox":
|
|
return jsonify({"error": f"Generate Only is not available for '{mode}' mode via web UI."}), 400
|
|
|
|
voice = data.get("voice") or None
|
|
job_id = str(uuid.uuid4())
|
|
_tts_jobs[job_id] = {"status": "pending", "text": text, "user": discord_name, "engine": mode}
|
|
|
|
_write_bot_command(user_id, "speak_in_vc", {
|
|
"job_id": job_id,
|
|
"text": text,
|
|
"user": discord_name,
|
|
"mode_override": mode,
|
|
"voice_override": voice,
|
|
"skip_vc": True,
|
|
"skip_post": not data.get("post_to_discord", True),
|
|
"guild_id": _resolve_guild_id(),
|
|
})
|
|
|
|
return jsonify({"job_id": job_id})
|
|
|
|
|
|
def _notify_bot_post(job_id, path, text, user, engine):
|
|
"""Write a pending post request to the shared state file for the bot to pick up."""
|
|
try:
|
|
state = _read_json(SHARED_STATE_FILE) if SHARED_STATE_FILE.exists() else {}
|
|
pending = state.get("pending_posts", [])
|
|
pending.append({"job_id": job_id, "path": path, "text": text, "user": user, "engine": engine})
|
|
state["pending_posts"] = pending
|
|
with open(SHARED_STATE_FILE, "w") as f:
|
|
json.dump(state, f)
|
|
log.debug(f"Queued Discord post for job {job_id} (user={user} engine={engine})")
|
|
except Exception as e:
|
|
log.error(f"Failed to queue Discord post for job {job_id}: {e}")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Routes — Settings API
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@app.route("/api/settings", methods=["GET"])
|
|
@login_required
|
|
def api_get_settings():
|
|
user_id = session["user_id"]
|
|
gcfg = get_global_config()
|
|
ucfg = get_user_config(user_id)
|
|
from ttstoy_constants import MINIMAX_VOICES
|
|
|
|
# Merge in any locally-cached param changes (written immediately on save,
|
|
# before the bot has had a chance to process the command and update RedBot config)
|
|
state = get_shared_state()
|
|
param_cache = state.get("chatterbox_params_cache", {}).get(str(user_id), {})
|
|
|
|
# Cache keys match the params keys: temperature, exaggeration, volume_db, speed
|
|
# RedBot config keys: chatterbox_temperature_offsets, etc.
|
|
def _merged(redbot_key, cache_key):
|
|
from_disk = ucfg.get(redbot_key, {})
|
|
from_cache = param_cache.get(cache_key, {})
|
|
return {**from_disk, **from_cache}
|
|
|
|
return jsonify({
|
|
"mode": gcfg.get("tts_mode", "minimax"),
|
|
"user_voice": ucfg.get("minimax_voice"),
|
|
"chatterbox_voices": ucfg.get("chatterbox_voices", []),
|
|
"chatterbox_temperature_offsets": _merged("chatterbox_temperature_offsets", "temperature"),
|
|
"chatterbox_exaggeration_offsets": _merged("chatterbox_exaggeration_offsets", "exaggeration"),
|
|
"chatterbox_volume_offsets": _merged("chatterbox_volume_offsets", "volume_db"),
|
|
"chatterbox_speed_offsets": _merged("chatterbox_speed_offsets", "speed"),
|
|
"global_voice": gcfg.get("minimax_voice"),
|
|
"vox_pack": gcfg.get("vox_pack", "vox"),
|
|
"sfx_volume": gcfg.get("sfx_volume", 100),
|
|
"minimax_voices": MINIMAX_VOICES,
|
|
})
|
|
|
|
|
|
@app.route("/api/settings/voice", methods=["POST"])
|
|
@login_required
|
|
def api_set_voice():
|
|
"""Tell the bot to update the user's voice via shared state."""
|
|
data = request.get_json(force=True)
|
|
voice = data.get("voice", "").strip()
|
|
if not voice:
|
|
return jsonify({"error": "No voice provided"}), 400
|
|
_write_bot_command(session["user_id"], "set_voice", {"voice": voice})
|
|
return jsonify({"ok": True})
|
|
|
|
|
|
@app.route("/api/settings/mode", methods=["POST"])
|
|
@login_required
|
|
def api_set_mode():
|
|
data = request.get_json(force=True)
|
|
mode = data.get("mode", "").strip()
|
|
if mode not in ("minimax", "chatterbox", "dectalk", "morshu", "vox"):
|
|
return jsonify({"error": "Invalid mode"}), 400
|
|
# Store per-user mode override in shared state (doesn't change global bot mode)
|
|
state = _read_json(SHARED_STATE_FILE) if SHARED_STATE_FILE.exists() else {}
|
|
overrides = state.get("user_mode_overrides", {})
|
|
overrides[str(session["user_id"])] = mode
|
|
state["user_mode_overrides"] = overrides
|
|
with open(SHARED_STATE_FILE, "w") as f:
|
|
json.dump(state, f)
|
|
log.info(f"User {session['user_id']} set mode override → {mode}")
|
|
return jsonify({"ok": True})
|
|
|
|
|
|
@app.route("/api/settings/chatterbox_params", methods=["POST"])
|
|
@login_required
|
|
def api_set_chatterbox_params():
|
|
data = request.get_json(force=True)
|
|
voice = data.get("voice", "").strip()
|
|
raw = data.get("params") or {}
|
|
params = {k: raw[k] for k in ("temperature", "exaggeration", "volume_db", "speed") if raw.get(k) is not None}
|
|
|
|
# Write to shared state immediately so /api/settings reflects the change right away,
|
|
# without waiting for the bot to process the command and write back to RedBot config.
|
|
try:
|
|
state = _read_json(SHARED_STATE_FILE) if SHARED_STATE_FILE.exists() else {}
|
|
user_id = str(session["user_id"])
|
|
cache = state.setdefault("chatterbox_params_cache", {}).setdefault(user_id, {})
|
|
for k, v in params.items():
|
|
# params keys: temperature, exaggeration, volume_db, speed
|
|
# store flat: {voice: value} under each key
|
|
cache.setdefault(k, {})[voice] = v
|
|
with open(SHARED_STATE_FILE, "w") as f:
|
|
json.dump(state, f)
|
|
except Exception as e:
|
|
log.warning(f"Could not cache chatterbox params: {e}")
|
|
|
|
_write_bot_command(session["user_id"], "set_chatterbox_params", {"voice": voice, "params": params})
|
|
return jsonify({"ok": True})
|
|
|
|
|
|
def _write_bot_command(user_id, cmd, payload):
|
|
"""Queue a command for the bot cog to execute."""
|
|
try:
|
|
state = _read_json(SHARED_STATE_FILE) if SHARED_STATE_FILE.exists() else {}
|
|
cmds = state.get("pending_commands", [])
|
|
cmds.append({"user_id": str(user_id), "cmd": cmd, "payload": payload, "ts": time.time()})
|
|
state["pending_commands"] = cmds
|
|
with open(SHARED_STATE_FILE, "w") as f:
|
|
json.dump(state, f)
|
|
log.debug(f"Queued bot command: user={user_id} cmd={cmd} payload={payload}")
|
|
except Exception as e:
|
|
log.error(f"Failed to write bot command {cmd} for user {user_id}: {e}")
|
|
|
|
|
|
def _resolve_guild_id() -> str:
|
|
"""Get the guild_id for the current user: session first, then server-side map."""
|
|
gid = session.get("guild_id", "")
|
|
if gid:
|
|
return gid
|
|
try:
|
|
state = _read_json(SHARED_STATE_FILE) if SHARED_STATE_FILE.exists() else {}
|
|
return state.get("user_guild_map", {}).get(session.get("user_id", ""), "")
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Routes — Minecraft Mod API
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@app.route("/api/mc/login", methods=["POST"])
|
|
def api_mc_login():
|
|
"""Minecraft mod login endpoint. Accepts JSON {"token": "..."} and returns user info + personal voices."""
|
|
data = request.get_json(force=True) if request.is_json else {}
|
|
token = data.get("token", "").strip()
|
|
if not token:
|
|
return jsonify({"error": "Missing token"}), 400
|
|
|
|
entry = consume_login_token(token)
|
|
if not entry:
|
|
return jsonify({"error": "Invalid or expired token. Run [p]ttstoy login in Discord."}), 401
|
|
|
|
user_id = entry["user_id"]
|
|
ucfg = get_user_config(user_id)
|
|
voices = ucfg.get("chatterbox_voices", [])
|
|
# Only return voices that belong to this user (prefixed with their user_id)
|
|
result = []
|
|
for v in voices:
|
|
if v.startswith(f"{user_id}_"):
|
|
display = v.split("_", 1)[1].rsplit(".", 1)[0]
|
|
result.append({"filename": v, "display": display})
|
|
|
|
return jsonify({
|
|
"user_id": user_id,
|
|
"discord_name": entry.get("discord_name", "Unknown"),
|
|
"voices": result,
|
|
})
|
|
|
|
|
|
@app.route("/api/mc/voices", methods=["GET"])
|
|
def api_mc_voices():
|
|
"""List predefined voices from the Chatterbox server (excludes user-uploaded voices)."""
|
|
try:
|
|
chatterbox_url = get_chatterbox_url()
|
|
r = requests.get(f"{chatterbox_url}/get_predefined_voices", timeout=10)
|
|
r.raise_for_status()
|
|
voices = r.json()
|
|
# Filter out user-uploaded voices (format: userid_name.wav where userid is numeric)
|
|
filtered = [v for v in voices if not v.get("filename", "").split("_")[0].isdigit()]
|
|
return jsonify(filtered)
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 502
|
|
|
|
|
|
@app.route("/api/mc/tts", methods=["POST"])
|
|
def api_mc_tts():
|
|
"""Generate TTS audio. Returns WAV bytes.
|
|
JSON body: {"text": "...", "mode": "chatterbox|dectalk|morshu", "voice": "Emily.wav"}
|
|
"""
|
|
data = request.get_json(force=True) if request.is_json else {}
|
|
text = (data.get("text") or "").strip()
|
|
if not text:
|
|
return jsonify({"error": "Missing text"}), 400
|
|
|
|
mode = data.get("mode", "chatterbox")
|
|
voice = data.get("voice", "Emily.wav")
|
|
|
|
try:
|
|
if mode == "chatterbox":
|
|
chatterbox_url = get_chatterbox_url()
|
|
payload = {
|
|
"text": text,
|
|
"voice_mode": "predefined",
|
|
"predefined_voice_id": voice,
|
|
"output_format": "wav",
|
|
"split_text": False,
|
|
}
|
|
r = requests.post(f"{chatterbox_url}/tts", json=payload, timeout=120)
|
|
r.raise_for_status()
|
|
return r.content, 200, {"Content-Type": "audio/wav"}
|
|
|
|
elif mode == "clone":
|
|
chatterbox_url = get_chatterbox_url()
|
|
payload = {
|
|
"text": text,
|
|
"voice_mode": "clone",
|
|
"reference_audio_filename": voice,
|
|
"output_format": "wav",
|
|
"split_text": False,
|
|
}
|
|
r = requests.post(f"{chatterbox_url}/tts", json=payload, timeout=120)
|
|
r.raise_for_status()
|
|
return r.content, 200, {"Content-Type": "audio/wav"}
|
|
|
|
elif mode == "dectalk":
|
|
dectalk_url = get_dectalk_url()
|
|
r = requests.get(f"{dectalk_url}/say", params={"text": text}, timeout=30)
|
|
r.raise_for_status()
|
|
return r.content, 200, {"Content-Type": "audio/wav"}
|
|
|
|
elif mode == "morshu":
|
|
r = requests.get("http://127.0.0.1:33002/say", params={"text": text}, timeout=60)
|
|
r.raise_for_status()
|
|
return r.content, 200, {"Content-Type": "audio/wav"}
|
|
|
|
elif mode == "vox":
|
|
gcfg = get_global_config()
|
|
pack = gcfg.get("vox_pack", "vox")
|
|
r = requests.get("http://127.0.0.1:33003/say", params={"text": text, "pack": pack}, timeout=60)
|
|
r.raise_for_status()
|
|
return r.content, 200, {"Content-Type": "audio/wav"}
|
|
|
|
else:
|
|
return jsonify({"error": f"Unknown mode: {mode}"}), 400
|
|
|
|
except requests.exceptions.ConnectionError:
|
|
return jsonify({"error": f"{mode} server not reachable"}), 502
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Routes — Chatterbox voice management
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@app.route("/api/chatterbox/voices")
|
|
@login_required
|
|
def api_chatterbox_voices():
|
|
user_id = session["user_id"]
|
|
ucfg = get_user_config(user_id)
|
|
voices = ucfg.get("chatterbox_voices", [])
|
|
result = []
|
|
for v in voices:
|
|
parts = v.split("_", 1)
|
|
display = parts[1].rsplit(".", 1)[0] if len(parts) > 1 else v.rsplit(".", 1)[0]
|
|
result.append({"filename": v, "display": display})
|
|
return jsonify(result)
|
|
|
|
|
|
@app.route("/api/chatterbox/voices/upload", methods=["POST"])
|
|
@login_required
|
|
def api_chatterbox_upload():
|
|
if "file" not in request.files:
|
|
log.warning("Voice upload: no file in request")
|
|
return jsonify({"error": "No file"}), 400
|
|
f = request.files["file"]
|
|
name = request.form.get("name", "").strip()
|
|
if not name:
|
|
name = f.filename.rsplit(".", 1)[0]
|
|
import re
|
|
safe_name = re.sub(r'[^a-zA-Z0-9_]', '', name) or "voice"
|
|
ext = ".wav" if f.filename.lower().endswith(".wav") else ".mp3"
|
|
filename = f"{session['user_id']}_{safe_name}{ext}"
|
|
cb_url = get_chatterbox_url()
|
|
log.debug(f"Voice upload: filename={filename} cb_url={cb_url}")
|
|
audio_bytes = f.read()
|
|
log.debug(f"Voice upload: raw audio size={len(audio_bytes)} bytes")
|
|
try:
|
|
from pydub import AudioSegment
|
|
import io as _io
|
|
seg = AudioSegment.from_file(_io.BytesIO(audio_bytes))
|
|
log.debug(f"Voice upload: clip duration={len(seg)}ms")
|
|
if len(seg) < 6000:
|
|
loops = (6000 // len(seg)) + 1
|
|
seg = seg * loops
|
|
log.debug(f"Voice upload: looped clip to {len(seg)}ms")
|
|
buf = _io.BytesIO()
|
|
seg.export(buf, format="wav")
|
|
audio_bytes = buf.getvalue()
|
|
filename = f"{session['user_id']}_{safe_name}.wav"
|
|
log.debug(f"Voice upload: processed wav size={len(audio_bytes)} bytes")
|
|
except Exception as e:
|
|
log.warning(f"Voice upload: pydub processing failed ({e}), uploading raw")
|
|
try:
|
|
r = requests.post(f"{cb_url.rstrip('/')}/upload_predefined_voice",
|
|
files={"files": (filename, audio_bytes, "audio/wav")}, timeout=30)
|
|
log.debug(f"Voice upload: Chatterbox response {r.status_code}")
|
|
r.raise_for_status()
|
|
except Exception as ex:
|
|
log.error(f"Voice upload failed: {ex}")
|
|
return jsonify({"error": str(ex)}), 500
|
|
_write_bot_command(session["user_id"], "add_voice", {"filename": filename})
|
|
log.info(f"Voice uploaded: {filename} for user {session['user_id']}")
|
|
return jsonify({"ok": True, "filename": filename})
|
|
|
|
|
|
@app.route("/api/chatterbox/voices/<filename>", methods=["DELETE"])
|
|
@login_required
|
|
def api_chatterbox_delete_voice(filename):
|
|
user_id = session["user_id"]
|
|
if not filename.startswith(f"{user_id}_"):
|
|
return jsonify({"error": "Not your voice"}), 403
|
|
cb_url = get_chatterbox_url()
|
|
try:
|
|
r = requests.delete(f"{cb_url.rstrip('/')}/delete_predefined_voice/{filename}", timeout=10)
|
|
if r.status_code not in (200, 204, 404):
|
|
return jsonify({"error": f"Server returned {r.status_code}"}), 500
|
|
except Exception as ex:
|
|
return jsonify({"error": str(ex)}), 500
|
|
_write_bot_command(user_id, "remove_voice", {"filename": filename})
|
|
return jsonify({"ok": True})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Routes — Pages
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@app.route("/api/sfx_emojis")
|
|
@login_required
|
|
def api_sfx_emojis():
|
|
"""Return the emoji→sfx folder mapping for the SFX picker."""
|
|
mapping = _load_emoji_map()
|
|
return jsonify([{"emoji": e, "name": n} for e, n in mapping.items()])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SFX helpers
|
|
# ---------------------------------------------------------------------------
|
|
def _sfx_root() -> Path:
|
|
return Path(__file__).resolve().parent.parent / "sfx"
|
|
|
|
|
|
def _emoji_map_path() -> Path:
|
|
return _sfx_root() / "emoji_map.json"
|
|
|
|
|
|
def _load_emoji_map() -> dict:
|
|
"""Load emoji→folder mapping from the shared JSON file."""
|
|
p = _emoji_map_path()
|
|
if p.exists():
|
|
try:
|
|
with open(p) as f:
|
|
return json.load(f)
|
|
except Exception:
|
|
pass
|
|
# Fallback: try importing the live dict from the cog
|
|
try:
|
|
import sys
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
from ttstoy import EMOJI_SFX_FOLDERS
|
|
return dict(EMOJI_SFX_FOLDERS)
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
def _save_emoji_map(mapping: dict):
|
|
"""Save emoji→folder mapping to the shared JSON file and reload in the cog."""
|
|
p = _emoji_map_path()
|
|
p.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(p, "w") as f:
|
|
json.dump(mapping, f, ensure_ascii=False, indent=2)
|
|
# Try to reload the live dict in the cog (same process won't help for
|
|
# the bot, but the bot's state processor or next read will pick it up)
|
|
try:
|
|
import sys
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
from ttstoy import _reload_emoji_sfx_map
|
|
_reload_emoji_sfx_map()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _owner_required(f):
|
|
@wraps(f)
|
|
def decorated(*args, **kwargs):
|
|
if not session.get("is_owner"):
|
|
return jsonify({"error": "Owner only"}), 403
|
|
return f(*args, **kwargs)
|
|
return decorated
|
|
|
|
|
|
@app.route("/api/sfx", methods=["GET"])
|
|
@login_required
|
|
@_owner_required
|
|
def api_sfx_list():
|
|
"""List all SFX entries: one emoji + one file per folder."""
|
|
mapping = _load_emoji_map()
|
|
|
|
root = _sfx_root()
|
|
# Reverse map: folder → emoji
|
|
folder_emoji = {}
|
|
for emoji, folder in mapping.items():
|
|
folder_emoji[folder] = emoji # last one wins (1:1 model)
|
|
|
|
result = []
|
|
if root.exists():
|
|
for d in sorted(root.iterdir()):
|
|
if d.is_dir():
|
|
files = sorted(f.name for f in d.iterdir() if f.is_file() and f.suffix in (".mp3", ".wav", ".ogg"))
|
|
result.append({
|
|
"name": d.name,
|
|
"emoji": folder_emoji.get(d.name, ""),
|
|
"file": files[0] if files else "",
|
|
})
|
|
return jsonify(result)
|
|
|
|
|
|
@app.route("/api/sfx/<folder_name>/upload", methods=["POST"])
|
|
@login_required
|
|
@_owner_required
|
|
def api_sfx_upload(folder_name):
|
|
"""Upload an audio file to an SFX folder. Creates the folder if needed."""
|
|
import re
|
|
if not re.match(r'^[a-zA-Z0-9_-]+$', folder_name):
|
|
return jsonify({"error": "Invalid folder name"}), 400
|
|
if "file" not in request.files:
|
|
return jsonify({"error": "No file"}), 400
|
|
f = request.files["file"]
|
|
if not f.filename:
|
|
return jsonify({"error": "No filename"}), 400
|
|
ext = Path(f.filename).suffix.lower()
|
|
if ext not in (".mp3", ".wav", ".ogg"):
|
|
return jsonify({"error": "Only .mp3, .wav, .ogg allowed"}), 400
|
|
|
|
safe_name = re.sub(r'[^a-zA-Z0-9_.-]', '_', f.filename)
|
|
folder = _sfx_root() / folder_name
|
|
folder.mkdir(parents=True, exist_ok=True)
|
|
dest = folder / safe_name
|
|
f.save(str(dest))
|
|
log.info(f"SFX uploaded: {dest}")
|
|
return jsonify({"ok": True, "path": str(dest.relative_to(_sfx_root()))})
|
|
|
|
|
|
@app.route("/api/sfx/<folder_name>", methods=["DELETE"])
|
|
@login_required
|
|
@_owner_required
|
|
def api_sfx_delete_folder(folder_name):
|
|
"""Delete an SFX and its emoji mapping."""
|
|
import shutil
|
|
folder = _sfx_root() / folder_name
|
|
if not folder.exists():
|
|
return jsonify({"error": "Folder not found"}), 404
|
|
shutil.rmtree(folder)
|
|
log.info(f"SFX folder deleted: {folder}")
|
|
# Also remove any emoji mappings pointing to this folder
|
|
mapping = _load_emoji_map()
|
|
changed = {e: f for e, f in mapping.items() if f != folder_name}
|
|
if len(changed) != len(mapping):
|
|
_save_emoji_map(changed)
|
|
return jsonify({"ok": True})
|
|
|
|
|
|
@app.route("/api/sfx/emoji_map", methods=["POST"])
|
|
@login_required
|
|
@_owner_required
|
|
def api_emoji_map_set():
|
|
"""Set or clear the emoji for an SFX folder. Enforces 1 emoji per folder."""
|
|
data = request.get_json(force=True)
|
|
emoji = (data.get("emoji") or "").strip()
|
|
folder = (data.get("folder") or "").strip()
|
|
name = (data.get("name") or folder or "").strip()
|
|
|
|
mapping = _load_emoji_map()
|
|
|
|
# Remove any existing emoji that pointed to this folder (1:1)
|
|
if name:
|
|
mapping = {e: f for e, f in mapping.items() if f != name}
|
|
|
|
if emoji and folder:
|
|
# Also remove this emoji from any other folder
|
|
mapping = {e: f for e, f in mapping.items() if e != emoji}
|
|
mapping[emoji] = folder
|
|
# If emoji is empty, we already removed the folder's old mapping above
|
|
|
|
_save_emoji_map(mapping)
|
|
log.info(f"Emoji map updated: {emoji!r} → {folder!r}")
|
|
return jsonify({"ok": True})
|
|
|
|
|
|
@app.route("/sfx")
|
|
@login_required
|
|
def page_sfx():
|
|
if not session.get("is_owner"):
|
|
return redirect(url_for("dashboard"))
|
|
return render_template("sfx.html")
|
|
|
|
|
|
@app.route("/chatterbox")
|
|
@login_required
|
|
def page_chatterbox():
|
|
return render_template("chatterbox.html")
|
|
|
|
|
|
@app.route("/minimax")
|
|
@login_required
|
|
def page_minimax():
|
|
gcfg = get_global_config()
|
|
from ttstoy_constants import MINIMAX_VOICES
|
|
return render_template("minimax.html", voices=MINIMAX_VOICES, gcfg=gcfg)
|
|
|
|
|
|
@app.route("/dectalk")
|
|
@login_required
|
|
def page_dectalk():
|
|
gcfg = get_global_config()
|
|
return render_template("dectalk.html", gcfg=gcfg)
|
|
|
|
|
|
@app.route("/vox")
|
|
@login_required
|
|
def page_vox():
|
|
gcfg = get_global_config()
|
|
try:
|
|
import sys
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
from vox_engine import get_available_words, get_available_packs
|
|
pack = gcfg.get("vox_pack", "vox")
|
|
words = get_available_words(pack)
|
|
packs = get_available_packs()
|
|
except Exception:
|
|
words, packs, pack = [], ["vox"], "vox"
|
|
return render_template("vox.html", words=words, packs=packs, current_pack=pack)
|
|
|
|
|
|
@app.route("/profile")
|
|
@login_required
|
|
def page_profile():
|
|
user_id = session["user_id"]
|
|
gcfg = get_global_config()
|
|
ucfg = get_user_config(user_id)
|
|
state = get_shared_state()
|
|
user_info = state.get("users", {}).get(str(user_id), {})
|
|
return render_template("profile.html", gcfg=gcfg, ucfg=ucfg, user_info=user_info)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Token endpoint — called by the bot cog to register a login token
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@app.route("/api/register_token", methods=["POST"])
|
|
def api_register_token():
|
|
"""Called by the bot (locally) to register a login token for a user."""
|
|
secret = request.headers.get("X-Internal-Secret", "")
|
|
if secret != app.config.get("INTERNAL_SECRET", ""):
|
|
log.warning("api_register_token: rejected request with wrong internal secret")
|
|
abort(403)
|
|
data = request.get_json(force=True)
|
|
log.debug(f"api_register_token: creating token for user {data.get('user_id')} ({data.get('discord_name')})")
|
|
token = create_login_token(
|
|
str(data["user_id"]),
|
|
data.get("discord_name", "Unknown"),
|
|
data.get("avatar_url", ""),
|
|
data.get("is_owner", False),
|
|
data.get("guild_id", ""),
|
|
data.get("guild_name", ""),
|
|
)
|
|
return jsonify({"token": token})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Main
|
|
# ---------------------------------------------------------------------------
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
import logging as _logging
|
|
_logging.basicConfig(
|
|
stream=sys.stdout,
|
|
level=_logging.DEBUG,
|
|
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
|
|
)
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
log.info(f"TtsToy WebUI starting — port={WEBUI_PORT} url={WEBUI_URL}")
|
|
app.run(host="0.0.0.0", port=WEBUI_PORT, debug=False)
|