197 lines
8.4 KiB
Python
197 lines
8.4 KiB
Python
"""LibriVox and Wikimedia Commons voice search (official APIs only, fixed hosts). See architecture/voice-sources.md."""
|
|
from __future__ import annotations
|
|
|
|
import html
|
|
import re
|
|
import threading
|
|
import time
|
|
from urllib.parse import urlsplit
|
|
|
|
import requests
|
|
|
|
from core.provenance import validate_provenance
|
|
|
|
LIBRIVOX_API = "https://librivox.org/api/feed/audiobooks"
|
|
COMMONS_API = "https://commons.wikimedia.org/w/api.php"
|
|
USER_AGENT = "TTSVoiceCreator/1.0 (self-hosted voice library; official-API client)" # no personal data
|
|
MAX_Q = 80
|
|
MAX_RESULTS = 10
|
|
MAX_BYTES = 8 * 1024 * 1024
|
|
TIMEOUT = 30 # the LibriVox extended feed takes ~14 s
|
|
CACHE_TTL = 600
|
|
CACHE_MAX = 200
|
|
MIN_INTERVAL = 1.0 # seconds between upstream calls per source (MediaWiki asks for serial requests)
|
|
|
|
# host suffixes allowed in result URLs (checked on every URL taken from an API response)
|
|
AUDIO_HOSTS = ("archive.org", "upload.wikimedia.org")
|
|
PAGE_HOSTS = ("librivox.org", "archive.org", "commons.wikimedia.org")
|
|
SOURCES = ("librivox", "wikimedia")
|
|
LANG_NAMES = {
|
|
"en": "English", "de": "German", "fr": "French", "es": "Spanish", "it": "Italian", "pt": "Portuguese",
|
|
"nl": "Dutch", "pl": "Polish", "ru": "Russian", "ja": "Japanese", "zh": "Chinese", "sv": "Swedish",
|
|
}
|
|
_CTRL = re.compile(r"[\x00-\x1f\x7f]")
|
|
_TAGS = re.compile(r"<[^>]*>")
|
|
|
|
_cache: dict[tuple, tuple[float, list]] = {}
|
|
_last_call: dict[str, float] = {}
|
|
_locks = {s: threading.Lock() for s in SOURCES}
|
|
_cache_lock = threading.Lock()
|
|
|
|
|
|
class SourceError(Exception):
|
|
"""Bad input (400) or upstream failure (502); message is safe to show."""
|
|
def __init__(self, msg: str, status: int = 400):
|
|
super().__init__(msg)
|
|
self.status = status
|
|
|
|
|
|
def host_ok(url: str, suffixes: tuple[str, ...]) -> bool:
|
|
try:
|
|
p = urlsplit(str(url))
|
|
h = (p.hostname or "").lower()
|
|
except ValueError:
|
|
return False
|
|
return p.scheme == "https" and not p.username and not p.password and any(h == s or h.endswith("." + s) for s in suffixes)
|
|
|
|
|
|
def clean_query(q: str) -> str:
|
|
q = _CTRL.sub(" ", str(q or "")).strip()
|
|
q = re.sub(r"\s+", " ", q)
|
|
if not q:
|
|
raise SourceError("q is required")
|
|
if len(q) > MAX_Q:
|
|
raise SourceError(f"q is too long (max {MAX_Q})")
|
|
return q
|
|
|
|
|
|
def clean_lang(lang: str) -> str:
|
|
lang = str(lang or "").strip().lower()
|
|
if lang and not re.fullmatch(r"[a-z]{2,3}", lang):
|
|
raise SourceError("lang must be a 2-3 letter code")
|
|
return lang
|
|
|
|
|
|
def _text(value, limit=200) -> str:
|
|
return _CTRL.sub(" ", html.unescape(_TAGS.sub("", str(value or "")))).strip()[:limit]
|
|
|
|
|
|
def _get_json(url: str, params: dict) -> dict:
|
|
wait = 0.0
|
|
src = "librivox" if url == LIBRIVOX_API else "wikimedia"
|
|
with _locks[src]:
|
|
wait = MIN_INTERVAL - (time.monotonic() - _last_call.get(src, 0.0))
|
|
if wait > 0:
|
|
time.sleep(wait)
|
|
_last_call[src] = time.monotonic()
|
|
try:
|
|
r = requests.get(url, params=params, headers={"User-Agent": USER_AGENT, "Accept": "application/json"},
|
|
timeout=TIMEOUT, allow_redirects=False, stream=True)
|
|
if r.status_code == 404 and src == "librivox":
|
|
return {}
|
|
r.raise_for_status()
|
|
body = b""
|
|
for chunk in r.iter_content(65536):
|
|
body += chunk
|
|
if len(body) > MAX_BYTES:
|
|
raise SourceError("Upstream response too large", 502)
|
|
import json
|
|
return json.loads(body.decode("utf-8", "replace"))
|
|
except SourceError:
|
|
raise
|
|
except Exception:
|
|
raise SourceError(f"{src} is unavailable right now", 502)
|
|
|
|
|
|
def _librivox_items(q: str, lang: str) -> list[dict]:
|
|
data = _get_json(LIBRIVOX_API, {"title": "^" + q, "format": "json", "extended": 1, "limit": 15}) # "^" = prefix match; bare title is exact
|
|
want = LANG_NAMES.get(lang, "").lower()
|
|
out = []
|
|
for b in (data.get("books") or []) if isinstance(data, dict) else []:
|
|
if not isinstance(b, dict):
|
|
continue
|
|
if want and str(b.get("language") or "").lower() != want:
|
|
continue
|
|
secs = [s for s in (b.get("sections") or []) if isinstance(s, dict) and host_ok(s.get("listen_url") or "", AUDIO_HOSTS)]
|
|
src_url = b.get("url_librivox") or ""
|
|
if not secs or not host_ok(src_url, PAGE_HOSTS):
|
|
continue
|
|
s = secs[0]
|
|
readers = ", ".join(_text(r.get("display_name"), 60) for r in (s.get("readers") or []) if isinstance(r, dict))
|
|
authors = ", ".join(_text(f"{a.get('first_name', '')} {a.get('last_name', '')}", 60) for a in (b.get("authors") or []) if isinstance(a, dict))
|
|
try:
|
|
dur = int(s.get("playtime") or 0)
|
|
except (TypeError, ValueError):
|
|
dur = 0
|
|
out.append({
|
|
"source": "librivox", "title": _text(f"{b.get('title')} - {s.get('title')}"), "reader": readers, "author": authors,
|
|
"language": _text(b.get("language"), 40), "duration": dur,
|
|
"licence": "Public domain (LibriVox, USA)", "licence_url": "https://librivox.org/pages/public-domain/",
|
|
"source_url": src_url, "audio_url": s["listen_url"], "attribution": "",
|
|
})
|
|
if len(out) >= MAX_RESULTS:
|
|
break
|
|
return out
|
|
|
|
|
|
def _wikimedia_items(q: str) -> list[dict]:
|
|
data = _get_json(COMMONS_API, {
|
|
"action": "query", "format": "json", "generator": "search", "gsrnamespace": 6, "gsrlimit": 20,
|
|
"gsrsearch": f"filetype:audio {q}", "prop": "imageinfo", "iiprop": "url|extmetadata|mediatype", "maxlag": 5,
|
|
})
|
|
pages = ((data.get("query") or {}).get("pages") or {}) if isinstance(data, dict) else {}
|
|
out = []
|
|
for p in sorted((x for x in pages.values() if isinstance(x, dict)), key=lambda x: x.get("index", 0)):
|
|
info = (p.get("imageinfo") or [{}])[0]
|
|
meta = info.get("extmetadata") or {}
|
|
mv = lambda k: (meta.get(k) or {}).get("value", "")
|
|
lic = _text(mv("LicenseShortName"), 80)
|
|
audio, page = info.get("url") or "", info.get("descriptionurl") or ""
|
|
if info.get("mediatype") != "AUDIO" or not lic or not host_ok(audio, AUDIO_HOSTS) or not host_ok(page, PAGE_HOSTS):
|
|
continue
|
|
licurl = mv("LicenseUrl") if host_ok(mv("LicenseUrl"), ("creativecommons.org", "wikimedia.org")) else ""
|
|
title = _text(re.sub(r"^File:|\.[A-Za-z0-9]+$", "", str(p.get("title") or "")))
|
|
artist = _text(mv("Artist"), 120)
|
|
out.append({
|
|
"source": "wikimedia", "title": title, "reader": artist, "author": artist, "language": "",
|
|
"duration": round(float(info.get("duration") or 0)), "licence": lic, "licence_url": licurl,
|
|
"source_url": page, "audio_url": audio,
|
|
"attribution": _text(f"{artist + ' / ' if artist else ''}{title}, {lic}, via Wikimedia Commons", 300),
|
|
})
|
|
if len(out) >= MAX_RESULTS:
|
|
break
|
|
return out
|
|
|
|
|
|
def search(source: str, q: str, lang: str = "") -> list[dict]:
|
|
if source not in SOURCES:
|
|
raise SourceError("unknown source")
|
|
q, lang = clean_query(q), clean_lang(lang)
|
|
key = (source, q.lower(), lang if source == "librivox" else "")
|
|
now = time.monotonic()
|
|
with _cache_lock:
|
|
hit = _cache.get(key)
|
|
if hit and now - hit[0] < CACHE_TTL:
|
|
return hit[1]
|
|
items = _librivox_items(q, lang) if source == "librivox" else _wikimedia_items(q)
|
|
with _cache_lock:
|
|
if len(_cache) >= CACHE_MAX:
|
|
_cache.pop(min(_cache, key=lambda k: _cache[k][0]))
|
|
_cache[key] = (now, items)
|
|
return items
|
|
|
|
|
|
def import_provenance_for(kind: str, source_url: str, licence: str, attribution: str) -> dict:
|
|
"""Provenance prefill for an import from one of these sources (raises ValueError on bad input)."""
|
|
if kind == "librivox":
|
|
return validate_provenance({
|
|
"basis": "public_domain", "source": source_url, "licence": "Public domain (LibriVox)",
|
|
"note": "LibriVox recordings are donated to the public domain (guaranteed in the USA only). "
|
|
"Check the reader's declared licence on the LibriVox page before use."})
|
|
if kind == "wikimedia":
|
|
pd = re.search(r"public domain|\bcc0\b|\bpd\b", licence, re.I)
|
|
return validate_provenance({
|
|
"basis": "public_domain" if pd else "licensed", "source": source_url, "licence": licence,
|
|
"note": ("Attribution: " + attribution)[:1000]})
|
|
raise ValueError("unknown source_kind")
|