Add ttstoy cog
This commit is contained in:
@@ -0,0 +1,901 @@
|
||||
"""
|
||||
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://127.0.0.1:8099")
|
||||
|
||||
|
||||
def get_dectalk_url() -> str:
|
||||
cfg = get_global_config()
|
||||
return cfg.get("dectalk_api_url", "http://127.0.0.1:3001")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 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 (no VC playback). Chatterbox only."""
|
||||
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)
|
||||
|
||||
if mode != "chatterbox":
|
||||
return jsonify({"error": "Generate Only is available for Chatterbox mode only. Switch to Chatterbox or use Speak in VC."}), 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,
|
||||
"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 — 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)
|
||||
Reference in New Issue
Block a user