536 lines
19 KiB
Python
536 lines
19 KiB
Python
import os
|
|
import json
|
|
import uuid
|
|
import time
|
|
import asyncio
|
|
import logging
|
|
from typing import Optional, List
|
|
from pathlib import Path
|
|
from fastapi import APIRouter, File, Form, UploadFile, HTTPException, Query
|
|
from fastapi.responses import FileResponse, RedirectResponse
|
|
from pydantic import BaseModel
|
|
|
|
from core.db import db_conn
|
|
from core.config import VOICES_DIR, OUTPUTS_DIR
|
|
from core import event_bus
|
|
from core.file_cleanup import FileCleanupError, unlink_if_present
|
|
from services.ffmpeg_utils import spawn_subprocess
|
|
|
|
logger = logging.getLogger("omnivoice.gallery")
|
|
|
|
router = APIRouter()
|
|
|
|
VOICE_GALLERY_DIR = Path(os.path.join(OUTPUTS_DIR, "voice_gallery"))
|
|
VOICE_GALLERY_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Voice imports carry no project-authored taxonomy. The gallery deliberately
|
|
# ships no curated directory of named real people (celebrities, politicians,
|
|
# franchise characters): shipping such a directory would turn a neutral
|
|
# user-driven import tool into an editorial invitation to clone identifiable
|
|
# individuals (the inducement-liability line). Users paste their own URLs/files
|
|
# into a flat "My Imports" list and own the licensing call. Designed,
|
|
# real-person-free voices live in the archetype gallery (core.archetypes).
|
|
CATEGORIES: list[dict] = []
|
|
|
|
|
|
class VoiceEntry(BaseModel):
|
|
id: str
|
|
name: str
|
|
character: str
|
|
category: str
|
|
source_type: str # "youtube", "upload", "preset"
|
|
source_url: Optional[str] = None
|
|
audio_path: str
|
|
duration: float
|
|
description: Optional[str] = None
|
|
thumbnail: Optional[str] = None
|
|
tags: List[str] = []
|
|
created_at: float
|
|
|
|
|
|
def _init_gallery_db():
|
|
"""Initialize the voice gallery table."""
|
|
with db_conn() as conn:
|
|
conn.execute("""
|
|
CREATE TABLE IF NOT EXISTS voice_gallery (
|
|
id TEXT PRIMARY KEY,
|
|
name TEXT NOT NULL,
|
|
character TEXT NOT NULL,
|
|
category TEXT NOT NULL,
|
|
source_type TEXT NOT NULL,
|
|
source_url TEXT,
|
|
audio_path TEXT NOT NULL,
|
|
duration REAL NOT NULL,
|
|
description TEXT,
|
|
thumbnail TEXT,
|
|
tags TEXT,
|
|
is_favorite INTEGER NOT NULL DEFAULT 0,
|
|
created_at REAL NOT NULL
|
|
)
|
|
""")
|
|
# Migration: add is_favorite column if missing (existing DBs)
|
|
try:
|
|
conn.execute("SELECT is_favorite FROM voice_gallery LIMIT 1")
|
|
except Exception:
|
|
conn.execute("ALTER TABLE voice_gallery ADD COLUMN is_favorite INTEGER NOT NULL DEFAULT 0")
|
|
|
|
|
|
@router.get("/gallery/categories")
|
|
def list_categories():
|
|
"""List all voice gallery categories."""
|
|
return CATEGORIES
|
|
|
|
|
|
@router.get("/gallery/voices")
|
|
def list_voices(
|
|
category: Optional[str] = Query(None, description="Filter by category"),
|
|
search: Optional[str] = Query(None, description="Search by name or character"),
|
|
limit: int = Query(50, ge=1, le=200),
|
|
):
|
|
"""List voices in the gallery, optionally filtered by category or search."""
|
|
query = "SELECT * FROM voice_gallery"
|
|
params = []
|
|
conditions = []
|
|
|
|
if category:
|
|
conditions.append("category = ?")
|
|
params.append(category)
|
|
if search:
|
|
conditions.append("(name LIKE ? OR character LIKE ? OR description LIKE ?)")
|
|
params.extend([f"%{search}%", f"%{search}%", f"%{search}%"])
|
|
|
|
if conditions:
|
|
query += " WHERE " + " AND ".join(conditions)
|
|
query += " ORDER BY created_at DESC LIMIT ?"
|
|
params.append(limit)
|
|
|
|
with db_conn() as conn:
|
|
rows = conn.execute(query, params).fetchall()
|
|
|
|
results = []
|
|
for row in rows:
|
|
r = dict(row)
|
|
r["tags"] = json.loads(r.get("tags", "[]") or "[]")
|
|
results.append(r)
|
|
return results
|
|
|
|
|
|
@router.get("/gallery/voices/{voice_id}")
|
|
def get_voice(voice_id: str):
|
|
"""Get a specific voice from the gallery."""
|
|
with db_conn() as conn:
|
|
row = conn.execute(
|
|
"SELECT * FROM voice_gallery WHERE id = ?", (voice_id,)
|
|
).fetchone()
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Voice not found")
|
|
r = dict(row)
|
|
r["tags"] = json.loads(r.get("tags", "[]") or "[]")
|
|
return r
|
|
|
|
|
|
@router.delete("/gallery/voices/{voice_id}")
|
|
def delete_voice(voice_id: str):
|
|
"""Delete a voice from the gallery."""
|
|
with db_conn() as conn:
|
|
row = conn.execute(
|
|
"SELECT audio_path FROM voice_gallery WHERE id = ?", (voice_id,)
|
|
).fetchone()
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Voice not found")
|
|
|
|
audio_path = row["audio_path"]
|
|
if audio_path:
|
|
try:
|
|
unlink_if_present(audio_path)
|
|
except FileCleanupError as exc:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail="Could not delete the voice audio file. Close any app using it and retry.",
|
|
) from exc
|
|
|
|
conn.execute("DELETE FROM voice_gallery WHERE id = ?", (voice_id,))
|
|
return {"success": True}
|
|
|
|
|
|
@router.post("/gallery/search/youtube")
|
|
async def search_youtube(
|
|
query: str = Query(..., description="User-supplied search terms or video title"),
|
|
category: str = Query("import", description="Free-form tag stored with results"),
|
|
max_results: int = Query(5, ge=1, le=20),
|
|
):
|
|
"""Search a source site (via yt-dlp) for clips matching the user's query.
|
|
|
|
The query is user-supplied; the project ships no celebrity/character seed
|
|
list. Users are responsible for the licensing of whatever they import.
|
|
"""
|
|
try:
|
|
# yt-dlp is an importable module, never a PATH requirement — run it
|
|
# via the interpreter (honors the Settings → Audio tools overlay).
|
|
from services.media_tools import ytdlp_invocation
|
|
ytdlp_argv, ytdlp_env = ytdlp_invocation()
|
|
result = await spawn_subprocess(
|
|
*ytdlp_argv,
|
|
"--dump-json",
|
|
"--remote-components", "ejs:github",
|
|
f"ytsearch{max_results}:{query}",
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
env=ytdlp_env,
|
|
)
|
|
stdout, stderr = await result.communicate()
|
|
|
|
if result.returncode != 0:
|
|
logger.error(f"yt-dlp search failed: {stderr.decode()}")
|
|
raise HTTPException(
|
|
status_code=500, detail=f"YouTube search failed: {stderr.decode()}"
|
|
)
|
|
|
|
lines = stdout.decode().strip().split("\n")
|
|
results = []
|
|
for line in lines:
|
|
if not line.strip():
|
|
continue
|
|
try:
|
|
data = json.loads(line)
|
|
results.append(
|
|
{
|
|
"title": data.get("title", ""),
|
|
"video_id": data.get("id", ""),
|
|
"duration": str(data.get("duration")) if data.get("duration") is not None else None,
|
|
"thumbnail": data.get("thumbnail", None),
|
|
}
|
|
)
|
|
except json.JSONDecodeError:
|
|
logger.warning(f"Failed to parse yt-dlp JSON line: {line}")
|
|
|
|
return {"results": results, "query": query, "category": category}
|
|
except FileNotFoundError:
|
|
raise HTTPException(status_code=500, detail="yt-dlp not installed")
|
|
except Exception as e:
|
|
logger.exception("YouTube search error")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.post("/gallery/download")
|
|
async def download_youtube_clip(
|
|
video_url: str = Query(..., description="YouTube video URL"),
|
|
start_time: float = Query(0, ge=0, description="Start time in seconds"),
|
|
duration: float = Query(10, ge=1, le=30, description="Clip duration in seconds"),
|
|
character_name: str = Query(..., description="Name to label this clip"),
|
|
category: str = Query("import", description="Free-form tag stored with the clip"),
|
|
description: str = Query("", description="Optional description"),
|
|
):
|
|
"""Download a clip from YouTube for voice cloning."""
|
|
voice_id = str(uuid.uuid4())[:8]
|
|
output_path = str(VOICE_GALLERY_DIR / f"{voice_id}.wav")
|
|
temp_path = str(VOICE_GALLERY_DIR / f"{voice_id}.%(ext)s")
|
|
|
|
try:
|
|
from services.media_tools import ytdlp_invocation
|
|
ytdlp_argv, ytdlp_env = ytdlp_invocation()
|
|
cmd = [
|
|
*ytdlp_argv,
|
|
"--remote-components", "ejs:github",
|
|
"-f",
|
|
"bestaudio",
|
|
"--download-sections",
|
|
f"*{start_time:.1f}-{start_time + duration:.1f}",
|
|
"-x",
|
|
"--audio-format",
|
|
"wav",
|
|
"--audio-quality",
|
|
"0",
|
|
"-o",
|
|
temp_path,
|
|
video_url,
|
|
]
|
|
|
|
result = await spawn_subprocess(
|
|
*cmd,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
env=ytdlp_env,
|
|
)
|
|
stdout, stderr = await result.communicate()
|
|
|
|
if result.returncode != 0:
|
|
logger.error(f"yt-dlp download failed: {stderr.decode()}")
|
|
raise HTTPException(
|
|
status_code=500, detail=f"Download failed: {stderr.decode()}"
|
|
)
|
|
|
|
# Find the downloaded file (yt-dlp replaces %s with actual extension)
|
|
downloaded_files = list(VOICE_GALLERY_DIR.glob(f"{voice_id}.*"))
|
|
if not downloaded_files:
|
|
raise HTTPException(status_code=500, detail="Downloaded file not found")
|
|
|
|
actual_path = downloaded_files[0]
|
|
# Rename to output_path
|
|
final_path = Path(output_path)
|
|
actual_path.rename(final_path)
|
|
|
|
conn = db_conn()
|
|
with conn as c:
|
|
c.execute(
|
|
"""
|
|
INSERT INTO voice_gallery
|
|
(id, name, character, category, source_type, source_url, audio_path, duration, description, tags, created_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
voice_id,
|
|
character_name,
|
|
character_name,
|
|
category,
|
|
"youtube",
|
|
video_url,
|
|
output_path,
|
|
duration,
|
|
description,
|
|
json.dumps([character_name.lower(), category]),
|
|
time.time(),
|
|
),
|
|
)
|
|
|
|
return {
|
|
"success": True,
|
|
"voice_id": voice_id,
|
|
"audio_path": output_path,
|
|
"duration": duration,
|
|
}
|
|
except FileNotFoundError:
|
|
raise HTTPException(status_code=500, detail="yt-dlp not installed")
|
|
except Exception as e:
|
|
logger.exception("Download error")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.post("/gallery/upload")
|
|
async def upload_voice_clip(
|
|
name: str = Form(...),
|
|
character: str = Form(""),
|
|
category: str = Form("import"),
|
|
description: str = Form(""),
|
|
audio: UploadFile = File(...),
|
|
):
|
|
"""Upload a voice clip directly to the gallery."""
|
|
voice_id = str(uuid.uuid4())[:8]
|
|
ext = os.path.splitext(audio.filename or ".wav")[1]
|
|
audio_path = str(VOICE_GALLERY_DIR / f"{voice_id}{ext}")
|
|
|
|
with open(audio_path, "wb") as f:
|
|
f.write(await audio.read())
|
|
|
|
try:
|
|
import soundfile as sf
|
|
|
|
info = sf.info(audio_path)
|
|
duration = info.frames / info.samplerate
|
|
except Exception:
|
|
duration = 10.0
|
|
|
|
with db_conn() as conn:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO voice_gallery
|
|
(id, name, character, category, source_type, source_url, audio_path, duration, description, tags, created_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
voice_id,
|
|
name,
|
|
character,
|
|
category,
|
|
"upload",
|
|
None,
|
|
audio_path,
|
|
duration,
|
|
description,
|
|
json.dumps([character.lower(), category]),
|
|
time.time(),
|
|
),
|
|
)
|
|
|
|
return {
|
|
"id": voice_id,
|
|
"name": name,
|
|
"audio_path": audio_path,
|
|
"duration": duration,
|
|
}
|
|
|
|
|
|
@router.post("/gallery/voices/{voice_id}/save-as-profile")
|
|
async def save_voice_as_profile(
|
|
voice_id: str,
|
|
profile_name: str = Query(..., description="Name for the voice profile"),
|
|
):
|
|
"""Save a gallery voice as a voice profile for cloning."""
|
|
with db_conn() as conn:
|
|
row = conn.execute(
|
|
"SELECT * FROM voice_gallery WHERE id = ?", (voice_id,)
|
|
).fetchone()
|
|
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Voice not found")
|
|
|
|
profile_id = str(uuid.uuid4())[:8]
|
|
import shutil
|
|
|
|
ext = os.path.splitext(row["audio_path"])[1]
|
|
new_audio_path = os.path.join(VOICES_DIR, f"{profile_id}{ext}")
|
|
shutil.copy(row["audio_path"], new_audio_path)
|
|
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO voice_profiles (id, name, ref_audio_path, ref_text, instruct, language, seed, created_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
profile_id,
|
|
profile_name,
|
|
f"{profile_id}{ext}",
|
|
row["description"] or "",
|
|
row["character"] or "",
|
|
"Auto",
|
|
None,
|
|
time.time(),
|
|
),
|
|
)
|
|
event_bus.emit("profiles", {"action": "created", "id": profile_id})
|
|
|
|
return {"profile_id": profile_id, "name": profile_name}
|
|
|
|
|
|
@router.get("/gallery/voices/{voice_id}/preview")
|
|
def preview_voice(voice_id: str):
|
|
"""Get a voice clip for preview playback."""
|
|
with db_conn() as conn:
|
|
row = conn.execute(
|
|
"SELECT audio_path FROM voice_gallery WHERE id = ?", (voice_id,)
|
|
).fetchone()
|
|
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Voice not found")
|
|
|
|
audio_path = row["audio_path"]
|
|
|
|
# Debug logging
|
|
is_absolute = os.path.isabs(audio_path)
|
|
path_exists = os.path.exists(audio_path) if audio_path else False
|
|
|
|
# If absolute path, serve directly or redirect
|
|
if is_absolute and path_exists:
|
|
# Get just the relative path from outputs dir
|
|
outputs_path = str(OUTPUTS_DIR)
|
|
if audio_path.startswith(outputs_path):
|
|
# Remove outputs_dir prefix to get relative path within outputs
|
|
rel_path = os.path.relpath(audio_path, outputs_path)
|
|
# The audio_path is like: /Users/user4/.../outputs/voice_gallery/file.wav
|
|
# rel_path becomes: voice_gallery/file.wav
|
|
# We want to serve from /audio/ so: /audio/voice_gallery/file.wav
|
|
return RedirectResponse(f"/audio/{rel_path}")
|
|
return FileResponse(audio_path, media_type="audio/wav")
|
|
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="Audio file not found. It may have been deleted or moved.",
|
|
)
|
|
|
|
|
|
# ── Library management endpoints ──────────────────────────────────────────
|
|
|
|
@router.patch("/gallery/voices/{voice_id}")
|
|
def update_voice(voice_id: str, body: dict):
|
|
"""Update voice metadata — name, tags, is_favorite."""
|
|
with db_conn() as conn:
|
|
row = conn.execute("SELECT id FROM voice_gallery WHERE id = ?", (voice_id,)).fetchone()
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Voice not found")
|
|
|
|
updates = []
|
|
params = []
|
|
if "name" in body:
|
|
updates.append("name = ?")
|
|
params.append(body["name"])
|
|
if "tags" in body:
|
|
updates.append("tags = ?")
|
|
params.append(json.dumps(body["tags"]) if isinstance(body["tags"], list) else body["tags"])
|
|
if "is_favorite" in body:
|
|
updates.append("is_favorite = ?")
|
|
params.append(1 if body["is_favorite"] else 0)
|
|
if "description" in body:
|
|
updates.append("description = ?")
|
|
params.append(body["description"])
|
|
|
|
if not updates:
|
|
return {"success": True, "updated": []}
|
|
|
|
params.append(voice_id)
|
|
# `updates` holds only static, code-controlled column fragments
|
|
# ("is_favorite = ?", "description = ?"); every user value is bound via
|
|
# a `?` placeholder in `params`. No user input reaches the SQL string.
|
|
conn.execute(f"UPDATE voice_gallery SET {', '.join(updates)} WHERE id = ?", params) # nosec B608
|
|
return {"success": True, "updated": list(body.keys())}
|
|
|
|
|
|
@router.post("/gallery/voices/batch-delete")
|
|
def batch_delete_voices(body: dict):
|
|
"""Delete multiple voices by ID list."""
|
|
ids = body.get("ids", [])
|
|
if not ids:
|
|
return {"deleted": 0}
|
|
|
|
deleted = 0
|
|
failed = 0
|
|
with db_conn() as conn:
|
|
for vid in ids:
|
|
row = conn.execute("SELECT audio_path FROM voice_gallery WHERE id = ?", (vid,)).fetchone()
|
|
if row:
|
|
audio_path = row["audio_path"]
|
|
if audio_path:
|
|
try:
|
|
unlink_if_present(audio_path)
|
|
except FileCleanupError:
|
|
logger.warning("Voice audio cleanup failed for a gallery item")
|
|
failed += 1
|
|
continue
|
|
conn.execute("DELETE FROM voice_gallery WHERE id = ?", (vid,))
|
|
deleted += 1
|
|
return {"deleted": deleted, "failed": failed}
|
|
|
|
|
|
@router.post("/gallery/voices/{voice_id}/to-profile")
|
|
def voice_to_profile(voice_id: str):
|
|
"""Create a voice profile from a gallery clip."""
|
|
with db_conn() as conn:
|
|
row = conn.execute("SELECT * FROM voice_gallery WHERE id = ?", (voice_id,)).fetchone()
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Voice not found")
|
|
|
|
voice = dict(row)
|
|
audio_path = voice["audio_path"]
|
|
if not os.path.exists(audio_path):
|
|
raise HTTPException(status_code=404, detail="Audio file not found on disk")
|
|
|
|
import shutil
|
|
import uuid
|
|
|
|
profile_id = str(uuid.uuid4())[:8]
|
|
# Copy audio to voices dir
|
|
dest_filename = f"{profile_id}_gallery.wav"
|
|
dest_path = os.path.join(VOICES_DIR, dest_filename)
|
|
shutil.copy2(audio_path, dest_path)
|
|
|
|
import time
|
|
now = time.time()
|
|
conn.execute(
|
|
"""INSERT INTO voice_profiles
|
|
(id, name, ref_audio_path, ref_text, instruct, seed, is_locked, locked_audio_path, created_at, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(profile_id, voice["name"], dest_filename, "", None, None, 0, None, now, now),
|
|
)
|
|
event_bus.emit("profiles", {"action": "created", "id": profile_id})
|
|
|
|
return {"success": True, "profile_id": profile_id, "name": voice["name"]}
|