1002 lines
41 KiB
Python
1002 lines
41 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, CONFIG_DIR
|
|
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,
|
|
)
|
|
from core.voice_index import upsert_voice_in_index
|
|
|
|
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",
|
|
"platform.r2.fish.audio",
|
|
"fish.audio",
|
|
}
|
|
_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)
|
|
await asyncio.to_thread(upsert_voice_in_index, settings, wav_dest)
|
|
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}")
|
|
|
|
|
|
# ── Fish.audio public voice library (api.fish.audio/model) ────────────────────
|
|
|
|
_FISHAUDIO_API = "https://api.fish.audio/model"
|
|
|
|
|
|
def _fishaudio_voice(item: dict) -> dict:
|
|
"""Normalize a fish.audio model record into a voice card with a clonable MP3 sample."""
|
|
samples = item.get("samples") or []
|
|
sample = samples[0] if samples else {}
|
|
langs = item.get("languages") or []
|
|
tags = item.get("tags") or []
|
|
gender = next((t for t in tags if t.lower() in ("male", "female", "neutral")), "")
|
|
age = next((t for t in tags if t.lower() in ("young", "middle-aged", "middle_aged", "old")), "")
|
|
author = item.get("author")
|
|
cover = item.get("cover_image")
|
|
return {
|
|
"id": item.get("_id", ""),
|
|
"title": _plain_text(item.get("title", ""))[:120],
|
|
"description": _plain_text(item.get("description", ""))[:400],
|
|
"tags": tags[:12],
|
|
"language": (langs[0] if langs else "").upper(),
|
|
"gender": gender.capitalize(),
|
|
"age": age.replace("_", "-").replace("middle-aged", "Middle-aged").capitalize() if age else "",
|
|
"plays": item.get("task_count", 0),
|
|
"likes": item.get("like_count", 0),
|
|
"author": author.get("nickname", "") if isinstance(author, dict) else (author or ""),
|
|
"image": f"https://public-platform.r2.fish.audio/{cover}" if cover else "",
|
|
"sample_audio": sample.get("audio", ""), # direct .mp3 URL (public, no auth)
|
|
"sample_text": _plain_text(sample.get("text", ""))[:600],
|
|
"default_text": _plain_text(item.get("default_text", ""))[:600],
|
|
}
|
|
|
|
|
|
@router.get("/api/fishaudio/voices")
|
|
async def fishaudio_voices(request: Request):
|
|
"""Browse the public fish.audio voice library. Returns clonable MP3 samples + transcripts."""
|
|
qp = request.query_params
|
|
# fish.audio only accepts these sort keys — clamp so a stale value never 422s.
|
|
sort_by = qp.get("sort_by", "score")
|
|
if sort_by not in ("score", "task_count", "created_at"):
|
|
sort_by = "score"
|
|
params: dict = {
|
|
"page_size": qp.get("page_size", "24"),
|
|
"page_number": qp.get("page", "1"),
|
|
"sort_by": sort_by,
|
|
}
|
|
if qp.get("search"):
|
|
params["title"] = qp["search"]
|
|
if qp.get("language"):
|
|
params["language"] = qp["language"]
|
|
# gender / age / any number of use-case & quality tags all map to fish.audio's
|
|
# repeatable `tag` filter
|
|
tags = [t for t in (list(qp.getlist("tag")) + [qp.get("gender"), qp.get("age")]) if t]
|
|
if tags:
|
|
params["tag"] = tags
|
|
import hashlib
|
|
cache_dir = CONFIG_DIR / "fishaudio_cache"
|
|
cache_key = hashlib.md5(json.dumps(params, sort_keys=True, default=str).encode()).hexdigest()
|
|
cache_file = cache_dir / f"{cache_key}.json"
|
|
try:
|
|
r = await asyncio.to_thread(
|
|
lambda: requests.get(_FISHAUDIO_API, params=params, headers={"Accept": "application/json"}, timeout=15)
|
|
)
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
except requests.exceptions.RequestException as exc:
|
|
# Offline fallback — serve a previously-cached copy of this query if we have one.
|
|
if cache_file.is_file():
|
|
cached = json.loads(cache_file.read_text(encoding="utf-8"))
|
|
cached["offline"] = True
|
|
return cached
|
|
raise HTTPException(502, f"fish.audio API error: {exc}")
|
|
items = [_fishaudio_voice(it) for it in data.get("items", []) if it.get("type") == "tts"]
|
|
result = {
|
|
"items": items,
|
|
"total": data.get("total", len(items)),
|
|
"page": int(params["page_number"]),
|
|
"page_size": int(params["page_size"]),
|
|
}
|
|
# Cache the normalized response so the browser keeps working if fish.audio goes offline.
|
|
try:
|
|
cache_dir.mkdir(parents=True, exist_ok=True)
|
|
cache_file.write_text(json.dumps(result), encoding="utf-8")
|
|
except Exception:
|
|
pass
|
|
return result
|
|
|
|
|
|
# ── Web script fetch (IMSDb + generic screenplay URLs) ────────────────────────
|
|
|
|
_SCRAPE_HEADERS = {
|
|
"User-Agent": (
|
|
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
|
|
"(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
|
|
),
|
|
"Accept": "text/html,application/xhtml+xml;q=0.9,*/*;q=0.8",
|
|
"Accept-Language": "en-US,en;q=0.9",
|
|
}
|
|
|
|
|
|
def _imsdb_title_from_url(raw_url: str) -> str:
|
|
"""Extract the movie title from any IMSDb URL variant (script page or detail page)."""
|
|
from urllib.parse import unquote as _unq
|
|
path = _unq(urlparse(raw_url).path)
|
|
# Direct script URL: /scripts/Some-Title.html → "Some Title"
|
|
m = re.match(r"^/scripts/(.+?)\.html?$", path, re.I)
|
|
if m:
|
|
return m.group(1).replace("-", " ").strip()
|
|
# Movie Scripts detail page: /Movie Scripts/Some Title[-Script].html
|
|
m2 = re.match(r"^/Movie[\s_]Scripts/(.+?)(?:[\s-]Script)?\.html?$", path, re.I)
|
|
if m2:
|
|
return m2.group(1).strip().rstrip("-").strip()
|
|
return ""
|
|
|
|
|
|
def _imsdb_read_link(detail_html: str) -> str:
|
|
"""Pull the real '/scripts/Title.html' target out of a detail page's 'Read … Script' anchor.
|
|
|
|
IMSDb occasionally emits a broken href like '/scripts/.' (its template chokes on
|
|
titles containing '?'); requiring a '.html' suffix rejects those so we fall back to slugs.
|
|
"""
|
|
m = re.search(r'<a[^>]+href="(/scripts/[^"]+\.html)"[^>]*>\s*Read', detail_html, re.I)
|
|
return m.group(1).strip() if m else ""
|
|
|
|
|
|
def _imsdb_script_candidates(title: str) -> list[str]:
|
|
"""Slug guesses for /scripts/{slug}.html — IMSDb's slugs are inconsistent so we try a few."""
|
|
base = re.sub(r"\s+", "-", title.strip())
|
|
variants = [
|
|
base, # spaces → dashes, punctuation kept (e.g. "1492:-Conquest…")
|
|
re.sub(r"[?!*]", "", base), # drop chars IMSDb strips (e.g. "Who-Framed-Roger-Rabbit")
|
|
re.sub(r"[?!*:,'\"]", "", base), # drop most punctuation (e.g. "Star-Wars-A-New-Hope")
|
|
]
|
|
out, seen = [], set()
|
|
for slug in variants:
|
|
if slug and slug not in seen:
|
|
seen.add(slug)
|
|
out.append(f"https://www.imsdb.com/scripts/{slug}.html")
|
|
return out
|
|
|
|
|
|
def _looks_like_script(html: str) -> bool:
|
|
return bool(re.search(r"<pre", html, re.I)) and len(html) > 2500
|
|
|
|
|
|
def _fetch_imsdb_script(raw_url: str, headers: dict) -> tuple[requests.Response, str]:
|
|
"""Resolve any IMSDb URL to its real screenplay page and fetch it.
|
|
|
|
Strategy mirrors the site's own navigation: all-scripts → detail page → 'Read Script' link.
|
|
Falls back to slug guesses when the detail link is missing or broken. Returns (response, title).
|
|
"""
|
|
title = _imsdb_title_from_url(raw_url)
|
|
candidates: list[str] = []
|
|
|
|
# 1. If we were handed a direct /scripts/ URL, try it as-is first.
|
|
if re.match(r"^/scripts/", urlparse(raw_url).path, re.I):
|
|
candidates.append(f"https://www.imsdb.com{urlparse(raw_url).path}")
|
|
|
|
# 2. Resolve via the detail page's authoritative "Read Script" link.
|
|
if title:
|
|
detail_url = f"https://www.imsdb.com/Movie Scripts/{title} Script.html"
|
|
try:
|
|
dresp = requests.get(detail_url, headers=headers, timeout=25)
|
|
if dresp.ok:
|
|
link = _imsdb_read_link(dresp.text)
|
|
if link:
|
|
candidates.append(urljoin("https://www.imsdb.com/", link))
|
|
except requests.exceptions.RequestException:
|
|
pass
|
|
# 3. Slug guesses as a fallback for broken/missing detail links.
|
|
candidates.extend(_imsdb_script_candidates(title))
|
|
|
|
if not candidates:
|
|
candidates.append(raw_url)
|
|
|
|
last: requests.Response | None = None
|
|
for url in dict.fromkeys(candidates): # dedupe, preserve order
|
|
try:
|
|
resp = requests.get(url, headers=headers, timeout=25, allow_redirects=True)
|
|
except requests.exceptions.RequestException:
|
|
continue
|
|
last = resp
|
|
if resp.ok and _looks_like_script(resp.text):
|
|
return resp, title
|
|
if last is None:
|
|
raise HTTPException(502, "Could not reach IMSDb for that script")
|
|
return last, title
|
|
|
|
|
|
|
|
def _extract_scripts_com_text(html: str) -> str:
|
|
"""Extract the visible screenplay section from a Scripts.com page."""
|
|
body = html
|
|
start = re.search(r'<hr[^>]*>\s*(?:<[^>]+>\s*)*(?:<a[^>]+>[^<]*Prev[^<]*</a>|«|«|‹)', body, re.I)
|
|
if start:
|
|
body = body[start.end():]
|
|
stop = re.search(r'Rate this script:|Submitted by|#### Discuss this', body, re.I)
|
|
if stop:
|
|
body = body[:stop.start()]
|
|
body = re.sub(r'<br\s*/?>', '\n', body, flags=re.I)
|
|
body = re.sub(r'</p\s*>', '\n', body, flags=re.I)
|
|
body = re.sub(r'</div\s*>', '\n', body, flags=re.I)
|
|
body = re.sub(r'<a[^>]+href="https?://www\.(?:definitions|abbreviations)\.com/[^>]+>(.*?)</a>', r'\1', body, flags=re.I | re.S)
|
|
body = re.sub(r'<[^>]+>', ' ', body)
|
|
text = unescape(body)
|
|
text = text.replace('\xa0', ' ')
|
|
lines = []
|
|
for raw in text.splitlines():
|
|
line = re.sub(r'\s+', ' ', raw).strip()
|
|
if not line:
|
|
if lines and lines[-1] != '':
|
|
lines.append('')
|
|
continue
|
|
if (line in {'|', '»', 'Prev', 'Next', '« Prev | Next »'}
|
|
or line.startswith('Next »')
|
|
or re.fullmatch(r'[|«»\s]*(?:Prev)?\s*\|?\s*Next\s*»?', line, re.I)):
|
|
continue
|
|
lines.append(line)
|
|
text = '\n'.join(lines).strip()
|
|
text = re.sub(r'\n{3,}', '\n\n', text)
|
|
return text
|
|
|
|
def _extract_screenplay_text(html: str) -> str:
|
|
"""Extract and clean screenplay text from a page's <pre> block."""
|
|
m = re.search(r"<pre[^>]*>([\s\S]+?)</pre>", html, re.I)
|
|
if not m:
|
|
return ""
|
|
raw = m.group(1)
|
|
# Preserve text inside <b> and <i> (character names / stage dirs), strip others
|
|
raw = re.sub(r"<b>(.*?)</b>", r"\1", raw, flags=re.DOTALL | re.I)
|
|
raw = re.sub(r"<i>(.*?)</i>", r"\1", raw, flags=re.DOTALL | re.I)
|
|
raw = re.sub(r"<[^>]+>", "", raw)
|
|
text = unescape(raw).replace("\r\n", "\n").replace("\r", "\n").strip()
|
|
# Collapse runs of 4+ blank lines to 2
|
|
text = re.sub(r"\n{4,}", "\n\n\n", text)
|
|
return text
|
|
|
|
|
|
@router.post("/api/fetch-web-script")
|
|
async def fetch_web_script(request: Request):
|
|
"""Fetch a screenplay from a web URL (IMSDb and plain text/HTML supported)."""
|
|
data = await request.json()
|
|
raw_url: str = (data.get("url") or "").strip()
|
|
if not raw_url:
|
|
raise HTTPException(400, "No URL provided")
|
|
|
|
# Validate URL
|
|
parsed = urlparse(raw_url)
|
|
if parsed.scheme not in ("http", "https") or not parsed.netloc:
|
|
raise HTTPException(400, "Invalid URL — must start with http:// or https://")
|
|
|
|
host = parsed.netloc.lower().replace("www.", "")
|
|
title = ""
|
|
|
|
if "imsdb.com" in host:
|
|
headers = {**_SCRAPE_HEADERS, "Referer": "https://www.imsdb.com/"}
|
|
try:
|
|
resp, title = await asyncio.to_thread(_fetch_imsdb_script, raw_url, headers)
|
|
resp.raise_for_status()
|
|
except HTTPException:
|
|
raise
|
|
except requests.exceptions.RequestException as e:
|
|
raise HTTPException(502, f"Fetch failed: {e}")
|
|
fetch_url = resp.url
|
|
else:
|
|
fetch_url = raw_url
|
|
headers = _SCRAPE_HEADERS
|
|
try:
|
|
resp = await asyncio.to_thread(
|
|
lambda: requests.get(fetch_url, headers=headers, timeout=25, allow_redirects=True)
|
|
)
|
|
resp.raise_for_status()
|
|
except requests.exceptions.RequestException as e:
|
|
raise HTTPException(502, f"Fetch failed: {e}")
|
|
|
|
content_type = resp.headers.get("content-type", "").lower()
|
|
if "text/plain" in content_type:
|
|
text = resp.text.strip()
|
|
elif "scripts.com" in host:
|
|
text = _extract_scripts_com_text(resp.text)
|
|
if not text:
|
|
raise HTTPException(422, "Could not find screenplay text on that Scripts.com page. Try a concrete /script/... page URL.")
|
|
title = title or _plain_title_from_html(resp.text)
|
|
else:
|
|
# HTML — try to extract <pre> screenplay block
|
|
text = _extract_screenplay_text(resp.text)
|
|
if not text:
|
|
raise HTTPException(422, "Could not find screenplay text on that page. Try pasting the script manually.")
|
|
|
|
if len(text) < 200:
|
|
raise HTTPException(422, "The page loaded but the script appears empty or too short.")
|
|
# (continues below)
|
|
return _web_script_result(text, title, fetch_url, parsed)
|
|
|
|
|
|
def _plain_title_from_html(html: str) -> str:
|
|
m = re.search(r"<h1[^>]*>([\s\S]+?)</h1>|<title[^>]*>([\s\S]+?)</title>", html, re.I)
|
|
if not m:
|
|
return ""
|
|
return re.sub(r"\s+", " ", unescape(re.sub(r"<[^>]+>", " ", m.group(1) or m.group(2) or ""))).strip()
|
|
|
|
def _web_script_result(text, title, fetch_url, parsed):
|
|
if not title:
|
|
title = parsed.path.split("/")[-1].replace(".html", "").replace("-", " ").replace("_", " ").strip()
|
|
return {"text": text, "title": title, "url": fetch_url, "chars": len(text)}
|
|
|
|
|
|
# ── IMSDb library browser ─────────────────────────────────────────────────────
|
|
|
|
_IMSDB_LIST_CACHE: dict = {"items": None, "ts": 0.0}
|
|
_POSTER_CACHE: dict = {} # title → poster url (or "" if none found)
|
|
|
|
|
|
@router.get("/api/imsdb/list")
|
|
async def imsdb_list():
|
|
"""Return the full IMSDb script catalogue: [{title, fetch_url}]. Cached 6 h in-memory."""
|
|
import time as _t
|
|
now = _t.time()
|
|
if _IMSDB_LIST_CACHE["items"] and (now - _IMSDB_LIST_CACHE["ts"] < 6 * 3600):
|
|
return {"items": _IMSDB_LIST_CACHE["items"], "cached": True}
|
|
try:
|
|
resp = await asyncio.to_thread(
|
|
lambda: requests.get("https://imsdb.com/all-scripts.html", headers=_SCRAPE_HEADERS, timeout=25)
|
|
)
|
|
resp.raise_for_status()
|
|
except requests.exceptions.RequestException as e:
|
|
raise HTTPException(502, f"Could not reach IMSDb: {e}")
|
|
|
|
items = []
|
|
seen = set()
|
|
for href, title in re.findall(r'<a href="(/Movie Scripts/[^"]+)"[^>]*>([^<]+)</a>', resp.text):
|
|
title = unescape(title).strip()
|
|
if not title or title in seen:
|
|
continue
|
|
seen.add(title)
|
|
# Hand back the detail-page URL; the fetch route follows its real "Read Script"
|
|
# link (IMSDb's per-script slugs are inconsistent, so a guessed slug often 404s).
|
|
items.append({"title": title, "fetch_url": f"https://www.imsdb.com{href}"})
|
|
items.sort(key=lambda x: x["title"].lower())
|
|
_IMSDB_LIST_CACHE["items"] = items
|
|
_IMSDB_LIST_CACHE["ts"] = now
|
|
return {"items": items, "cached": False}
|
|
|
|
|
|
@router.get("/api/movie-poster")
|
|
async def movie_poster(title: str = ""):
|
|
"""Look up a movie poster + year via OMDb. Cached per title in-memory."""
|
|
title = (title or "").strip()
|
|
if not title:
|
|
return {"poster": "", "year": ""}
|
|
if title in _POSTER_CACHE:
|
|
return _POSTER_CACHE[title]
|
|
from urllib.parse import quote as _q
|
|
# Strip trailing year/qualifiers that hurt matching
|
|
clean = re.sub(r"\s*\(.*?\)\s*$", "", title).strip()
|
|
result = {"poster": "", "year": "", "title": title}
|
|
try:
|
|
url = f"http://www.omdbapi.com/?t={_q(clean)}&type=movie&apikey=trilogy"
|
|
r = await asyncio.to_thread(lambda: requests.get(url, timeout=8))
|
|
if r.ok:
|
|
d = r.json()
|
|
poster = d.get("Poster", "")
|
|
if poster and poster != "N/A":
|
|
result["poster"] = poster
|
|
result["year"] = d.get("Year", "") or ""
|
|
except Exception:
|
|
pass
|
|
_POSTER_CACHE[title] = result
|
|
return result
|