backend/app/utils/sample_generator.py

"""Bootstrap sample audio for the demo's Sample Library.

Strategy (per slot):
  1. If file already present and non-empty, leave it alone.
  2. For "real" slots: download a small public-domain LibriSpeech clip from
     HuggingFace's CDN (stable URLs used in their docs for years). Trim,
     resample to 16 kHz mono, save as PCM_16 WAV.
  3. If download fails, fall back to the high-quality source-filter
     synthesizer in ``speech_synth.py`` — much better than the previous
     formant-resonator approach.
  4. For "fake" slots: always synthesise (these are *meant* to be TTS-like).

Result: the Sample Library has actually-real speech for the real slots when
the user has internet, gracefully degrading to plausible synthesis offline.
"""
from __future__ import annotations

import io
from pathlib import Path
from typing import Optional

import numpy as np
import soundfile as sf

from app.features.audio_preprocessor import preprocess_audio
from app.logging_setup import get_logger
from app.utils.speech_synth import (
    synth_real_speech,
    synth_fake_neural_tts,
    synth_fake_tts_commercial,
    synth_fake_voice_clone,
)

logger = get_logger(__name__)

SR = 16000

# Stable, long-lived public URLs for real LibriSpeech samples. These are the
# clips HuggingFace ships in their wav2vec2 docs and have been stable for
# years. Both are CC-BY 4.0 (LibriSpeech license).
_REAL_DOWNLOAD_CANDIDATES: list[tuple[str, str]] = [
    ("real_news_excerpt.wav",   "https://cdn-media.huggingface.co/speech_samples/sample1.flac"),
    ("real_conversation.wav",   "https://cdn-media.huggingface.co/speech_samples/sample2.flac"),
    # Third real slot uses the same source but the second half — we trim
    # different segments client-side so we get audibly distinct clips.
    ("real_lecture.wav",        "https://cdn-media.huggingface.co/speech_samples/sample1.flac"),
]


def _try_download_speech(url: str, max_seconds: float, segment: str) -> Optional[np.ndarray]:
    """Download → decode → trim → resample to 16 kHz mono. Returns None on failure.

    ``segment`` selects which slice to keep:
        'head'   -> first max_seconds
        'mid'    -> middle max_seconds
        'tail'   -> last max_seconds
    """
    try:
        import httpx

        with httpx.Client(timeout=15.0, follow_redirects=True) as client:
            r = client.get(url)
            if r.status_code != 200 or not r.content:
                logger.warning("download %s -> HTTP %s", url, r.status_code)
                return None
            data = r.content
    except Exception as exc:  # noqa: BLE001
        logger.warning("download %s failed: %s", url, exc)
        return None

    try:
        loaded = preprocess_audio(data, target_sr=SR, max_seconds=60.0)
    except Exception as exc:  # noqa: BLE001
        logger.warning("decode of %s failed: %s", url, exc)
        return None

    wav = loaded.waveform.squeeze().detach().cpu().numpy().astype(np.float32)
    target_len = int(max_seconds * SR)
    if wav.size <= target_len:
        return wav

    if segment == "head":
        return wav[:target_len]
    if segment == "tail":
        return wav[-target_len:]
    # 'mid' — pick a deterministic offset
    start = (wav.size - target_len) // 2
    return wav[start : start + target_len]


# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------

def ensure_sample_audios(target_dir: Path) -> int:
    """Create any missing sample WAV files. Returns number of new files written."""
    target_dir.mkdir(parents=True, exist_ok=True)
    rng = np.random.default_rng(42)
    created = 0

    # --- REAL slots: try download first --------------------------------- #
    real_segments = ["head", "head", "tail"]
    for (filename, url), seg in zip(_REAL_DOWNLOAD_CANDIDATES, real_segments):
        path = target_dir / filename
        if path.exists() and path.stat().st_size > 4096:
            continue
        wav = _try_download_speech(url, max_seconds=4.5, segment=seg)
        if wav is None or wav.size < SR:
            logger.info("Falling back to synthesised real speech for %s", filename)
            wav = synth_real_speech(seconds=4.5, rng=rng)
        else:
            logger.info("Downloaded real speech for %s (%.1fs)", filename, wav.size / SR)
        sf.write(str(path), wav, SR, subtype="PCM_16")
        created += 1

    # --- FAKE slots: always synthesise ---------------------------------- #
    fake_specs = [
        ("fake_tts_commercial.wav", lambda r: synth_fake_tts_commercial(4.5, r)),
        ("fake_voice_clone.wav",    lambda r: synth_fake_voice_clone(4.5, r)),
        ("fake_neural_tts.wav",     lambda r: synth_fake_neural_tts(4.5, r)),
    ]
    for filename, gen in fake_specs:
        path = target_dir / filename
        if path.exists() and path.stat().st_size > 4096:
            continue
        wav = gen(rng)
        sf.write(str(path), wav, SR, subtype="PCM_16")
        logger.info("Generated fake sample: %s", filename)
        created += 1

    if created:
        logger.info("Sample audio bootstrap complete: %d new files in %s", created, target_dir)
    return created