"""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"]*>(.*?)", 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('
')[1:]: title = re.search(r'
\s*([^<]+?)\s*
', card, re.S) if not title: continue image = re.search(r']+src="([^"]+)"', card, re.S) image_url = urljoin(homepage, image.group(1)) if image else "" pairs = re.findall( r'
\s*([^<]+?)\s*
\s*]*>\s* 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"(?=
(.*?)
', 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']+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" 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']*>\s*(?:<[^>]+>\s*)*(?:]+>[^<]*Prev[^<]*|«|«|‹)', 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'', '\n', body, flags=re.I) body = re.sub(r'', '\n', body, flags=re.I) body = re.sub(r'', '\n', body, flags=re.I) body = re.sub(r']+href="https?://www\.(?:definitions|abbreviations)\.com/[^>]+>(.*?)', 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 
 block."""
    m = re.search(r"]*>([\s\S]+?)
", html, re.I) if not m: return "" raw = m.group(1) # Preserve text inside and (character names / stage dirs), strip others raw = re.sub(r"(.*?)", r"\1", raw, flags=re.DOTALL | re.I) raw = re.sub(r"(.*?)", 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 
 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"]*>([\s\S]+?)|]*>([\s\S]+?)", 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']*>([^<]+)', 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