mARTin-B78
c7a1e35539
Security fixes: - Block /proc /sys /dev /run /boot in /api/browse-dirs (path traversal) - Verify yt-dlp output stays inside TEMP_DIR before registration - Remove Access-Control-Allow-Origin: * from /api/proxy-audio - TTL-based temp file registry (default 2h) to prevent disk fill Performance: - Cache settings + routing rules in memory (mtime-checked); eliminates per-request disk reads on every TTS call UI: - Add container name (optional) field to Docker stack TTS/STT engine cards (Qwen3 Voice Clone, Voice Design, Custom Voice, Streaming, NVIDIA Magpie, Parakeet) — enables Stop/Start/Restart buttons on all engine cards, matching the existing Other Local TTS/STT cards Refactor — backend: - server.py: 5560 lines → 43-line entry point - core/ package: constants, registry, validation, docker_client, config, routing, audio, voice, presets, tts_helpers - routes/ package: admin, settings, library, stt, sources, docker, tts, conversation (FastAPI APIRouter modules) - Dockerfile + docker-compose.yml updated to include core/ and routes/ Refactor — frontend: - static/app.js: 8744 lines → 16 modules in static/js/ utils, voice-inspector, voice-sources, integrations, routing, settings, voice-clone, voice-library, tts-preview, benchmark, stt, init, engines, ai-backends, generation, conversation - static/loader.js updated to load modules sequentially Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
627 lines
25 KiB
Python
627 lines
25 KiB
Python
"""Voice sources scraping, proxy-audio, quick-import, yt-dlp download, ElevenLabs."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import glob
|
|
import json
|
|
import re
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from html import unescape
|
|
from pathlib import Path
|
|
from urllib.parse import quote, urljoin, urlparse
|
|
|
|
import requests
|
|
from fastapi import APIRouter, HTTPException, Request
|
|
from fastapi.responses import StreamingResponse
|
|
|
|
from core.config import _load_settings
|
|
from core.constants import _VOICES_DIR_DEFAULT, _MAX_UPLOAD_BYTES, _ALLOW_PRIVATE_DOWNLOADS
|
|
from core.registry import _registry_put, TEMP_DIR
|
|
from core.validation import _validate_http_url
|
|
from core.audio import _to_wav_24k, _duration
|
|
from core.voice import (
|
|
_AUDIO_EXTS, _active_voices_dir,
|
|
_remove_audio_variants, _export_normalized_wav, _load_meta, _save_meta,
|
|
)
|
|
|
|
router = APIRouter()
|
|
|
|
# ── Voice source helpers ──────────────────────────────────────────────────────
|
|
|
|
_VOICE_SOURCE_HEADERS = {
|
|
"User-Agent": "TTS Voice Creator voice-source-scraper/1.0 (+local)",
|
|
"Accept": "text/html,application/json,text/plain;q=0.9,*/*;q=0.8",
|
|
}
|
|
|
|
DEFAULT_VOICE_SOURCE_URLS = [
|
|
"https://aiartes.com/voiceai",
|
|
"https://sample-files.com/downloads/audio/wav/voice-sample.wav",
|
|
"https://freesound.org/people/Scott%20Simpson/",
|
|
"https://lanceblairvo.com/raw-voiceover-samples/",
|
|
"https://github.com/yaph/tts-samples/tree/main/mp3",
|
|
"https://github.com/jim-schwoebel/voice_datasets",
|
|
]
|
|
|
|
_DIRECT_AUDIO_RE = re.compile(r"\.(?:mp3|wav|ogg|flac|m4a|aac)(?:$|[?#])", re.I)
|
|
_LANG_HINTS = {
|
|
"english": "English", "german": "German", "deutsch": "German", "french": "French",
|
|
"spanish": "Spanish", "italian": "Italian", "portuguese": "Portuguese",
|
|
"dutch": "Dutch", "polish": "Polish", "russian": "Russian", "japanese": "Japanese",
|
|
"korean": "Korean", "chinese": "Chinese", "arabic": "Arabic", "swedish": "Swedish",
|
|
"turkish": "Turkish", "hindi": "Hindi",
|
|
}
|
|
_PROXY_AUDIO_DOMAINS: set[str] = {
|
|
"drive.usercontent.google.com",
|
|
"drive.google.com",
|
|
"aiartes.com",
|
|
"freesound.org",
|
|
"lanceblairvo.com",
|
|
"raw.githubusercontent.com",
|
|
"sample-files.com",
|
|
}
|
|
_GDRIVE_MIME_EXT = {
|
|
"audio/mpeg": "mp3", "audio/mp3": "mp3",
|
|
"audio/wav": "wav", "audio/x-wav": "wav",
|
|
"audio/ogg": "ogg", "audio/vorbis": "ogg",
|
|
"audio/flac": "flac", "audio/x-flac": "flac",
|
|
"audio/aac": "aac", "audio/x-aac": "aac",
|
|
"audio/x-m4a": "m4a", "audio/m4a": "m4a", "audio/mp4": "m4a",
|
|
}
|
|
|
|
|
|
def _source_import_url(value: str) -> str:
|
|
raw = str(value or "").strip()
|
|
if not re.match(r"^https?://", raw, re.I):
|
|
raise HTTPException(400, "Expected an http(s) URL")
|
|
return raw
|
|
|
|
|
|
def _source_download_limit(name: str, fallback_mb: int) -> int:
|
|
import os
|
|
value = globals().get(name)
|
|
if isinstance(value, int):
|
|
return value
|
|
return int(os.environ.get(name.lstrip("_").replace("_BYTES", "_MB"), str(fallback_mb))) * 1024 * 1024
|
|
|
|
|
|
def _source_get(url: str, timeout: int = 12) -> requests.Response:
|
|
r = requests.get(url, headers=_VOICE_SOURCE_HEADERS, timeout=timeout)
|
|
r.raise_for_status()
|
|
return r
|
|
|
|
|
|
def _plain_text(value: str) -> str:
|
|
value = re.sub(r"<[^>]+>", " ", str(value or ""))
|
|
value = unescape(value)
|
|
return re.sub(r"\s+", " ", value).strip()
|
|
|
|
|
|
def _voice_source_id(url: str) -> str:
|
|
parsed = urlparse(url)
|
|
base = (parsed.netloc + parsed.path).strip("/").lower()
|
|
base = re.sub(r"[^a-z0-9]+", "-", base).strip("-")
|
|
return base[:80] or "source"
|
|
|
|
|
|
def _source_name_from_url(url: str) -> str:
|
|
parsed = urlparse(url)
|
|
host = parsed.netloc.replace("www.", "")
|
|
tail = Path(parsed.path.rstrip("/")).stem.replace("-", " ").replace("_", " ").strip()
|
|
return f"{host} / {tail}" if tail else host or url
|
|
|
|
|
|
def _guess_language(*values: str) -> str:
|
|
text = " ".join(str(v or "") for v in values).lower()
|
|
for key, label in _LANG_HINTS.items():
|
|
if re.search(rf"\b{re.escape(key)}\b", text):
|
|
return label
|
|
code = re.search(r"(?:^|[^a-z])(en|de|fr|es|it|pt|nl|pl|ru|ja|ko|zh|ar|sv|tr|hi)(?:[^a-z]|$)", text)
|
|
return {"en": "English", "de": "German", "fr": "French", "es": "Spanish", "it": "Italian",
|
|
"pt": "Portuguese", "nl": "Dutch", "pl": "Polish", "ru": "Russian", "ja": "Japanese",
|
|
"ko": "Korean", "zh": "Chinese", "ar": "Arabic", "sv": "Swedish", "tr": "Turkish",
|
|
"hi": "Hindi"}.get(code.group(1), "Unknown") if code else "Unknown"
|
|
|
|
|
|
def _guess_gender(*values: str) -> str:
|
|
text = " ".join(str(v or "") for v in values).lower()
|
|
if re.search(r"\b(female|woman|girl|fem|_f_|-f-)\b", text):
|
|
return "Female"
|
|
if re.search(r"\b(male|man|boy|masc|_m_|-m-)\b", text):
|
|
return "Male"
|
|
return "Unknown"
|
|
|
|
|
|
def _source_item(source_id: str, source_name: str, name: str, kind: str,
|
|
page_url: str, audio_url: str = "", image_url: str = "",
|
|
category: str = "", description: str = "", file_type: str = "",
|
|
language: str = "", gender: str = "") -> dict:
|
|
if not file_type and audio_url:
|
|
match = re.search(r"\.([A-Za-z0-9]+)(?:$|[?#])", audio_url)
|
|
file_type = match.group(1).lower() if match else "audio"
|
|
language = language or _guess_language(name, kind, category, description, page_url, audio_url)
|
|
gender = gender or _guess_gender(name, kind, category, description, page_url, audio_url)
|
|
return {
|
|
"id": f"{source_id}:{uuid.uuid5(uuid.NAMESPACE_URL, page_url + audio_url + name + kind)}",
|
|
"source_id": source_id,
|
|
"source": source_name,
|
|
"name": _plain_text(name)[:160],
|
|
"kind": _plain_text(kind)[:80],
|
|
"category": _plain_text(category)[:80],
|
|
"description": _plain_text(description)[:420],
|
|
"page_url": page_url,
|
|
"audio_url": audio_url,
|
|
"image_url": image_url,
|
|
"file_type": file_type,
|
|
"language": language,
|
|
"gender": gender,
|
|
"direct_audio": bool(audio_url),
|
|
}
|
|
|
|
|
|
def _source_result(source_id: str, name: str, homepage: str, description: str, items: list[dict]) -> dict:
|
|
return {
|
|
"id": source_id,
|
|
"name": name,
|
|
"homepage": homepage,
|
|
"description": description,
|
|
"items": items,
|
|
"count": len(items),
|
|
"direct_audio": sum(1 for item in items if item.get("direct_audio")),
|
|
}
|
|
|
|
|
|
def _direct_audio_source(url: str) -> dict:
|
|
source_id = _voice_source_id(url)
|
|
name = _source_name_from_url(url)
|
|
item = _source_item(
|
|
source_id, name,
|
|
Path(urlparse(url).path).stem.replace("-", " ").replace("_", " ") or "Voice sample",
|
|
"Direct audio file", url, audio_url=url,
|
|
category="Direct audio",
|
|
description="Direct audio URL from the editable source list.",
|
|
)
|
|
return _source_result(source_id, name, url, "Single direct audio URL.", [item])
|
|
|
|
|
|
def _generic_audio_page_source(url: str) -> dict:
|
|
source_id = _voice_source_id(url)
|
|
source_name = _source_name_from_url(url)
|
|
body = _source_get(url).text
|
|
title = re.search(r"<title[^>]*>(.*?)</title>", body, re.I | re.S)
|
|
page_title = _plain_text(title.group(1)) if title else source_name
|
|
seen = set()
|
|
items = []
|
|
for match in re.finditer(r"""(?:src|href)=["']([^"']+\.(?:mp3|wav|ogg|flac|m4a|aac)(?:[^"']*)?)["']""", body, re.I):
|
|
audio_url = urljoin(url, unescape(match.group(1)))
|
|
if audio_url in seen:
|
|
continue
|
|
seen.add(audio_url)
|
|
stem = Path(urlparse(audio_url).path).stem.replace("-", " ").replace("_", " ").strip() or page_title
|
|
items.append(_source_item(
|
|
source_id, source_name, stem, "Audio sample", url,
|
|
audio_url=audio_url, category="Page audio",
|
|
description=f"Audio link found on {page_title}.",
|
|
))
|
|
if not items:
|
|
items.append(_source_item(
|
|
source_id, source_name, page_title, "Source page", url,
|
|
category="Source page",
|
|
description="No direct audio file links were found. Open the source page for previews, licensing, and download details.",
|
|
))
|
|
return _source_result(source_id, source_name, url, f"Audio links scraped from {page_title}.", items)
|
|
|
|
|
|
def _scrape_aiartes_voiceai() -> dict:
|
|
homepage = "https://aiartes.com/voiceai"
|
|
body = _source_get(homepage).text
|
|
items = []
|
|
for card in body.split('<div class="ui card card-with-info">')[1:]:
|
|
title = re.search(r'<div class="voice-title">\s*([^<]+?)\s*</div>', card, re.S)
|
|
if not title:
|
|
continue
|
|
image = re.search(r'<img[^>]+src="([^"]+)"', card, re.S)
|
|
image_url = urljoin(homepage, image.group(1)) if image else ""
|
|
pairs = re.findall(
|
|
r'<div class="voice-kind">\s*([^<]+?)\s*</div>\s*<audio[^>]*>\s*<source\s+src="([^"]+)"',
|
|
card, re.S,
|
|
)
|
|
pairs.sort(key=lambda pair: 0 if 'cloned' in pair[0].lower() else 1)
|
|
for kind, audio in pairs:
|
|
audio_url = urljoin(homepage, audio)
|
|
items.append(_source_item(
|
|
"aiartes", "Aiartes VoiceAI", title.group(1), kind, homepage,
|
|
audio_url=audio_url, image_url=image_url,
|
|
category="Voice clips",
|
|
description="Short downloadable demo clip listed on Aiartes VoiceAI.",
|
|
))
|
|
return {
|
|
"id": "aiartes",
|
|
"name": "Aiartes VoiceAI",
|
|
"homepage": homepage,
|
|
"description": "Direct MP3 voice clips scraped from the public VoiceAI gallery.",
|
|
"items": items,
|
|
"count": len(items),
|
|
"direct_audio": sum(1 for item in items if item.get("direct_audio")),
|
|
}
|
|
|
|
|
|
def _scrape_sample_files_voice_sample() -> dict:
|
|
return _direct_audio_source("https://sample-files.com/downloads/audio/wav/voice-sample.wav")
|
|
|
|
|
|
def _scrape_freesound_scott_simpson() -> dict:
|
|
return _generic_audio_page_source("https://freesound.org/people/Scott%20Simpson/")
|
|
|
|
|
|
def _scrape_lanceblair_raw_samples() -> dict:
|
|
return _generic_audio_page_source("https://lanceblairvo.com/raw-voiceover-samples/")
|
|
|
|
|
|
def _scrape_yaph_tts_samples() -> dict:
|
|
homepage = "https://github.com/yaph/tts-samples/tree/main/mp3"
|
|
api = "https://api.github.com/repos/yaph/tts-samples/git/trees/main?recursive=1"
|
|
data = _source_get(api).json()
|
|
items = []
|
|
for entry in data.get("tree", []):
|
|
path = entry.get("path", "")
|
|
if entry.get("type") != "blob" or not path.lower().startswith("mp3/") or not path.lower().endswith(".mp3"):
|
|
continue
|
|
parts = path.split("/")
|
|
language = parts[1] if len(parts) > 2 else "Unknown"
|
|
stem = Path(path).stem.replace("_", " ").replace("-", " ").strip() or Path(path).stem
|
|
raw_path = quote(path, safe="/")
|
|
raw_url = f"https://raw.githubusercontent.com/yaph/tts-samples/main/{raw_path}"
|
|
page_url = f"https://github.com/yaph/tts-samples/blob/main/{raw_path}"
|
|
items.append(_source_item(
|
|
"yaph-tts-samples", "yaph/tts-samples", stem, "MP3 sample", page_url,
|
|
audio_url=raw_url, category=language,
|
|
description=f"Language folder: {language}. Synthetic TTS sample MP3 from yaph/tts-samples.",
|
|
file_type="mp3", language=language,
|
|
))
|
|
items.sort(key=lambda x: (x.get("category", ""), x.get("name", "")))
|
|
return {
|
|
"id": "yaph-tts-samples",
|
|
"name": "yaph/tts-samples mp3",
|
|
"homepage": homepage,
|
|
"description": "GitHub-hosted TTS sample MP3 files grouped by language.",
|
|
"items": items,
|
|
"count": len(items),
|
|
"direct_audio": len(items),
|
|
}
|
|
|
|
|
|
def _scrape_jim_voice_datasets() -> dict:
|
|
homepage = "https://github.com/jim-schwoebel/voice_datasets"
|
|
raw = "https://raw.githubusercontent.com/jim-schwoebel/voice_datasets/master/README.md"
|
|
text = _source_get(raw).text
|
|
speech = text
|
|
start = text.find("### Speech datasets")
|
|
end = text.find("### Audio events", start if start >= 0 else 0)
|
|
if start >= 0:
|
|
speech = text[start:end if end >= 0 else len(text)]
|
|
items = []
|
|
for name, url, desc in re.findall(r"^\* \[([^\]]+)\]\(([^)]+)\)\s*-\s*(.+)$", speech, re.M):
|
|
items.append(_source_item(
|
|
"voice-datasets", "jim-schwoebel/voice_datasets", name, "Dataset link",
|
|
urljoin(homepage, url.strip()), category="Speech dataset", description=desc,
|
|
))
|
|
return {
|
|
"id": "voice-datasets",
|
|
"name": "jim-schwoebel/voice_datasets",
|
|
"homepage": homepage,
|
|
"description": "Curated speech and voice dataset links from the repository README.",
|
|
"items": items,
|
|
"count": len(items),
|
|
"direct_audio": 0,
|
|
}
|
|
|
|
|
|
def _scrape_google_drive_folder(url: str) -> dict:
|
|
m = re.search(r"/folders/([A-Za-z0-9_-]+)", url)
|
|
if not m:
|
|
return _generic_audio_page_source(url)
|
|
folder_id = m.group(1)
|
|
source_id = _voice_source_id(url)
|
|
source_name = "Google Drive"
|
|
embed_url = f"https://drive.google.com/embeddedfolderview?id={folder_id}#list"
|
|
try:
|
|
body = _source_get(embed_url).text
|
|
except Exception:
|
|
return _generic_audio_page_source(url)
|
|
|
|
items = []
|
|
seen: set[str] = set()
|
|
for chunk in re.split(r"(?=<div class=\"flip-entry\" id=\"entry-)", body):
|
|
eid = re.search(r'id="entry-([A-Za-z0-9_-]+)"', chunk)
|
|
title_m = re.search(r'class="flip-entry-title">(.*?)</div>', chunk)
|
|
mime_m = re.search(r"type/([a-zA-Z0-9/+\-]+)", chunk)
|
|
if not eid or not title_m:
|
|
continue
|
|
file_id = eid.group(1)
|
|
if file_id in seen:
|
|
continue
|
|
seen.add(file_id)
|
|
filename = title_m.group(1).strip()
|
|
mime = mime_m.group(1).lower() if mime_m else ""
|
|
if not mime.startswith("audio"):
|
|
continue
|
|
ext = _GDRIVE_MIME_EXT.get(mime, Path(filename).suffix.lstrip(".").lower() or "mp3")
|
|
gdrive_dl = f"https://drive.usercontent.google.com/download?id={file_id}&export=download&authuser=0"
|
|
proxy_url = f"/api/proxy-audio?url={quote(gdrive_dl)}"
|
|
stem = Path(filename).stem.replace("-", " ").replace("_", " ").strip()
|
|
lang = _guess_language(filename)
|
|
gender = _guess_gender(filename)
|
|
item = _source_item(
|
|
source_id, source_name, stem, "Google Drive audio", url,
|
|
audio_url=proxy_url, file_type=ext, category="Google Drive",
|
|
description="Audio file from Google Drive folder.",
|
|
language=lang, gender=gender,
|
|
)
|
|
item["import_url"] = gdrive_dl
|
|
items.append(item)
|
|
|
|
if not items:
|
|
items.append(_source_item(
|
|
source_id, source_name, "Google Drive folder", "Source page", url,
|
|
category="Google Drive",
|
|
description="No audio files found in this Google Drive folder. It may be private or empty.",
|
|
))
|
|
|
|
return _source_result(source_id, source_name, url, f"{len(items)} audio files from Google Drive folder.", items)
|
|
|
|
|
|
def _scrape_voice_source_url(url: str) -> dict:
|
|
lower = url.lower()
|
|
if "aiartes.com/voiceai" in lower:
|
|
return _scrape_aiartes_voiceai()
|
|
if "github.com/yaph/tts-samples" in lower:
|
|
return _scrape_yaph_tts_samples()
|
|
if "github.com/jim-schwoebel/voice_datasets" in lower:
|
|
return _scrape_jim_voice_datasets()
|
|
if "sample-files.com/downloads/audio/wav/voice-sample.wav" in lower:
|
|
return _scrape_sample_files_voice_sample()
|
|
if "freesound.org/people/scott%20simpson" in lower or "freesound.org/people/scott simpson" in lower:
|
|
return _scrape_freesound_scott_simpson()
|
|
if "lanceblairvo.com/raw-voiceover-samples" in lower:
|
|
return _scrape_lanceblair_raw_samples()
|
|
if "drive.google.com/drive/folders/" in lower or "drive.google.com/open?id=" in lower:
|
|
return _scrape_google_drive_folder(url)
|
|
if _DIRECT_AUDIO_RE.search(lower):
|
|
return _direct_audio_source(url)
|
|
return _generic_audio_page_source(url)
|
|
|
|
|
|
def _normalize_voice_source_urls(urls: list[str] | None = None) -> list[str]:
|
|
raw_urls = urls or DEFAULT_VOICE_SOURCE_URLS
|
|
normalized = []
|
|
seen = set()
|
|
for value in raw_urls:
|
|
url = str(value or "").strip()
|
|
if not url or not re.match(r"^https?://", url, re.I):
|
|
continue
|
|
if url not in seen:
|
|
seen.add(url)
|
|
normalized.append(url)
|
|
return normalized[:24]
|
|
|
|
|
|
def _voice_sources_payload(urls: list[str] | None = None) -> dict:
|
|
sources = []
|
|
errors = []
|
|
source_urls = _normalize_voice_source_urls(urls)
|
|
for url in source_urls:
|
|
try:
|
|
sources.append(_scrape_voice_source_url(url))
|
|
except Exception as e:
|
|
errors.append({"source": url, "detail": str(e)})
|
|
return {
|
|
"scraped_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
|
|
"source_urls": source_urls,
|
|
"sources": sources,
|
|
"total": sum(src.get("count", 0) for src in sources),
|
|
"direct_audio": sum(src.get("direct_audio", 0) for src in sources),
|
|
"errors": errors,
|
|
}
|
|
|
|
|
|
# ── Routes ────────────────────────────────────────────────────────────────────
|
|
|
|
@router.get("/api/voice-sources")
|
|
async def voice_sources():
|
|
return _voice_sources_payload(DEFAULT_VOICE_SOURCE_URLS)
|
|
|
|
|
|
@router.post("/api/voice-sources")
|
|
async def voice_sources_custom(request: Request):
|
|
data = await request.json()
|
|
return _voice_sources_payload(data.get("urls") if isinstance(data, dict) else None)
|
|
|
|
|
|
@router.post("/api/import-source-audio")
|
|
async def import_source_audio(request: Request):
|
|
data = await request.json()
|
|
audio_url = _source_import_url(str(data.get("audio_url") or ""))
|
|
filename = str(data.get("name") or Path(urlparse(audio_url).path).name or "source-audio").strip()
|
|
suffix = Path(urlparse(audio_url).path).suffix.lower()
|
|
if suffix not in _AUDIO_EXTS:
|
|
suffix = ".bin"
|
|
dest = TEMP_DIR / f"{uuid.uuid4().hex}{suffix}"
|
|
try:
|
|
with requests.get(audio_url, headers=_VOICE_SOURCE_HEADERS, timeout=30, stream=True) as r:
|
|
r.raise_for_status()
|
|
total = 0
|
|
with dest.open("wb") as f:
|
|
for chunk in r.iter_content(1024 * 1024):
|
|
if not chunk:
|
|
continue
|
|
total += len(chunk)
|
|
if total > _MAX_UPLOAD_BYTES:
|
|
raise HTTPException(413, "Downloaded audio is too large")
|
|
f.write(chunk)
|
|
wav = _to_wav_24k(dest)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(400, f"Source audio import failed: {e}")
|
|
fid = uuid.uuid4().hex
|
|
_registry_put(fid, wav)
|
|
return {"id": fid, "duration": _duration(wav), "filename": filename or dest.name, "audio_url": audio_url}
|
|
|
|
|
|
@router.get("/api/proxy-audio")
|
|
async def proxy_audio(url: str):
|
|
parsed = urlparse(url)
|
|
if parsed.scheme not in ("http", "https"):
|
|
raise HTTPException(400, "Only http/https URLs are supported")
|
|
domain = parsed.netloc.lower().lstrip("www.")
|
|
if not any(domain == d or domain.endswith("." + d) for d in _PROXY_AUDIO_DOMAINS):
|
|
raise HTTPException(403, f"Domain not in audio proxy allowlist: {parsed.netloc}")
|
|
try:
|
|
resp = requests.get(url, headers=_VOICE_SOURCE_HEADERS, timeout=30, stream=True)
|
|
resp.raise_for_status()
|
|
except Exception as e:
|
|
raise HTTPException(502, f"Proxy fetch failed: {e}")
|
|
content_type = resp.headers.get("content-type", "audio/mpeg")
|
|
|
|
def _stream():
|
|
for chunk in resp.iter_content(65536):
|
|
if chunk:
|
|
yield chunk
|
|
|
|
return StreamingResponse(
|
|
_stream(), media_type=content_type,
|
|
headers={"Cache-Control": "public, max-age=3600"},
|
|
)
|
|
|
|
|
|
@router.post("/api/quick-import-voice")
|
|
async def quick_import_voice(request: Request):
|
|
data = await request.json()
|
|
audio_url = _source_import_url(str(data.get("audio_url") or ""))
|
|
voice_id = re.sub(r"[^A-Za-z0-9_\-\.]", "_", str(data.get("voice_id") or "").strip())[:80]
|
|
if not voice_id:
|
|
raise HTTPException(400, "voice_id is required")
|
|
transcript = str(data.get("transcript") or "").strip()
|
|
|
|
suffix = Path(urlparse(audio_url).path).suffix.lower()
|
|
if suffix not in _AUDIO_EXTS:
|
|
suffix = ".bin"
|
|
dest = TEMP_DIR / f"{uuid.uuid4().hex}{suffix}"
|
|
try:
|
|
with requests.get(audio_url, headers=_VOICE_SOURCE_HEADERS, timeout=30, stream=True) as r:
|
|
r.raise_for_status()
|
|
total = 0
|
|
with dest.open("wb") as f:
|
|
for chunk in r.iter_content(1024 * 1024):
|
|
if not chunk:
|
|
continue
|
|
total += len(chunk)
|
|
if total > _MAX_UPLOAD_BYTES:
|
|
raise HTTPException(413, "Downloaded audio is too large")
|
|
f.write(chunk)
|
|
wav = _to_wav_24k(dest)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(400, f"Audio download failed: {e}")
|
|
|
|
settings = _load_settings()
|
|
out_dir = _active_voices_dir(settings)
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
final_id = voice_id
|
|
if (out_dir / f"{final_id}.wav").exists():
|
|
for i in range(2, 1000):
|
|
candidate = f"{voice_id}_{i}"
|
|
if not (out_dir / f"{candidate}.wav").exists():
|
|
final_id = candidate
|
|
break
|
|
|
|
wav_dest = out_dir / f"{final_id}.wav"
|
|
_remove_audio_variants(out_dir, final_id)
|
|
from core.audio import _export_normalized_wav as _enw
|
|
loudness = _enw(wav, wav_dest)
|
|
if transcript:
|
|
(out_dir / f"{final_id}.reference.txt").write_text(transcript, encoding="utf-8")
|
|
meta = _load_meta(wav_dest)
|
|
meta["enabled"] = True
|
|
meta["loudness"] = loudness
|
|
_save_meta(wav_dest, meta)
|
|
return {"voice_id": final_id, "loudness": loudness}
|
|
|
|
|
|
@router.get("/api/download-yt")
|
|
async def download_yt(url: str):
|
|
url = _validate_http_url(url, allow_private=_ALLOW_PRIVATE_DOWNLOADS)
|
|
out_path = TEMP_DIR / f"{uuid.uuid4().hex}.%(ext)s"
|
|
|
|
async def event_stream():
|
|
cmd = [
|
|
"yt-dlp", "--extract-audio", "--audio-format", "wav",
|
|
"--audio-quality", "0", "--output", str(out_path),
|
|
"--no-playlist", "--progress", "--newline", url,
|
|
]
|
|
proc = await asyncio.create_subprocess_exec(
|
|
*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT,
|
|
)
|
|
downloaded_path: Path | None = None
|
|
assert proc.stdout is not None
|
|
async for raw in proc.stdout:
|
|
line = raw.decode(errors="replace").rstrip()
|
|
if not line:
|
|
continue
|
|
if "Destination:" in line and "[ExtractAudio]" not in line:
|
|
m = re.search(r"Destination:\s+(.+)$", line)
|
|
if m:
|
|
downloaded_path = Path(m.group(1).strip())
|
|
if line.startswith("[download]") or line.startswith("[ExtractAudio]"):
|
|
pct_m = re.search(r"(\d+\.\d+)%", line)
|
|
pct = pct_m.group(1) if pct_m else None
|
|
yield f"data: {json.dumps({'msg': line, 'pct': pct})}\n\n"
|
|
await proc.wait()
|
|
if downloaded_path is None or not downloaded_path.exists():
|
|
matches = glob.glob(str(out_path).replace("%(ext)s", "*"))
|
|
if matches:
|
|
downloaded_path = Path(matches[0])
|
|
if downloaded_path is None or not downloaded_path.exists():
|
|
yield f"data: {json.dumps({'error': 'Download failed — no output file found'})}\n\n"
|
|
return
|
|
try:
|
|
downloaded_path.resolve().relative_to(TEMP_DIR.resolve())
|
|
except ValueError:
|
|
yield f"data: {json.dumps({'error': 'Download wrote outside temp directory'})}\n\n"
|
|
return
|
|
try:
|
|
wav = _to_wav_24k(downloaded_path)
|
|
except Exception as e:
|
|
yield f"data: {json.dumps({'error': f'Conversion failed: {e}'})}\n\n"
|
|
return
|
|
fid = uuid.uuid4().hex
|
|
_registry_put(fid, wav)
|
|
yield f"data: {json.dumps({'done': True, 'id': fid, 'duration': _duration(wav)})}\n\n"
|
|
|
|
return StreamingResponse(event_stream(), media_type="text/event-stream",
|
|
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})
|
|
|
|
|
|
@router.get("/api/elevenlabs/voices")
|
|
async def elevenlabs_shared_voices(request: Request):
|
|
settings = _load_settings()
|
|
api_key = (settings.get("elevenlabs_api_key") or "").strip()
|
|
_allowed = {"page_size", "page", "language", "gender", "age", "accent",
|
|
"use_case", "category", "search", "featured"}
|
|
params: dict = {k: v for k, v in request.query_params.items() if k in _allowed}
|
|
params.setdefault("page_size", "24")
|
|
headers = {"xi-api-key": api_key} if api_key else {}
|
|
try:
|
|
r = requests.get(
|
|
"https://api.elevenlabs.io/v1/shared-voices",
|
|
params=params,
|
|
headers=headers,
|
|
timeout=15,
|
|
)
|
|
return r.json()
|
|
except requests.exceptions.RequestException as exc:
|
|
raise HTTPException(502, f"ElevenLabs API error: {exc}")
|