speech_engine.py

"""
Speech Engine — Uses edge-tts (Microsoft Edge's neural TTS) to render
speech with voice and prosody control.

edge-tts provides free access to Microsoft's neural TTS voices with support for:
    - 50+ neural voices
    - Rate control (speaking speed)
    - Pitch control (in Hz offset)
    - Volume control
"""

import asyncio
import re
import tempfile

import edge_tts


def _sanitize_pitch(pitch_str: str) -> str:
    """
    Convert any pitch format to Hz format required by edge-tts.
    edge-tts requires pitch in '+NHz' or '-NHz' format.
    """
    if not pitch_str or pitch_str in ("+0%", "0%", "0", "+0", "+0Hz"):
        return "+0Hz"

    # Already in Hz format
    match_hz = re.match(r"([+-]?\d+)Hz$", pitch_str)
    if match_hz:
        val = int(match_hz.group(1))
        sign = "+" if val >= 0 else ""
        return f"{sign}{val}Hz"

    # Percentage format → convert to Hz (roughly 1% ≈ 1Hz)
    match_pct = re.match(r"([+-]?)(\d+)%?$", pitch_str)
    if match_pct:
        sign = match_pct.group(1) or "+"
        value = int(match_pct.group(2))
        return f"{sign}{value}Hz"

    return "+0Hz"


def _sanitize_rate(rate_str: str) -> str:
    """Ensure rate is in valid format for edge-tts (e.g., '+20%')."""
    if not rate_str or rate_str in ("0", "+0", "normal"):
        return "+0%"

    match = re.match(r"([+-]?)(\d+)%?$", rate_str)
    if match:
        sign = match.group(1) or "+"
        value = int(match.group(2))
        # Clamp to reasonable range
        value = min(100, max(0, value))
        return f"{sign}{value}%"

    return "+0%"


def _sanitize_volume(volume_str: str) -> str:
    """Ensure volume is in valid format for edge-tts (e.g., '+20%')."""
    if not volume_str or volume_str in ("0", "+0", "default", "medium"):
        return "+0%"

    # Already percentage
    match = re.match(r"([+-]?)(\d+)%?$", volume_str)
    if match:
        sign = match.group(1) or "+"
        value = int(match.group(2))
        value = min(50, max(0, value))
        return f"{sign}{value}%"

    return "+0%"


async def _synthesize_async(
    text: str,
    voice: str = "en-US-AriaNeural",
    rate: str = "+0%",
    pitch: str = "+0Hz",
    volume: str = "+0%",
    output_path: str | None = None,
) -> tuple[bytes, str]:
    """Internal async synthesis function."""
    if output_path is None:
        output_path = tempfile.mktemp(suffix=".mp3")

    # Sanitize all parameters
    pitch = _sanitize_pitch(pitch)
    rate = _sanitize_rate(rate)
    volume = _sanitize_volume(volume)

    communicate = edge_tts.Communicate(
        text=text,
        voice=voice,
        rate=rate,
        pitch=pitch,
        volume=volume,
    )

    await communicate.save(output_path)

    with open(output_path, "rb") as f:
        audio_bytes = f.read()

    return audio_bytes, output_path


def synthesize(
    text: str,
    voice: str = "en-US-AriaNeural",
    rate: str = "+0%",
    pitch: str = "+0Hz",
    volume: str = "+0%",
    output_path: str | None = None,
) -> tuple[bytes, str]:
    """
    Synthesize speech from text with voice parameters.

    Args:
        text: Content text to speak
        voice: Microsoft neural voice name
        rate: Speaking rate (e.g., "+20%", "-10%")
        pitch: Pitch adjustment in Hz (e.g., "+10Hz", "-5Hz")
        volume: Volume adjustment (e.g., "+20%", "-10%")
        output_path: Save path (None = temp file)

    Returns:
        (audio_bytes, output_path)
    """
    return asyncio.run(_synthesize_async(
        text=text,
        voice=voice,
        rate=rate,
        pitch=pitch,
        volume=volume,
        output_path=output_path,
    ))


async def _list_voices_async() -> list[dict]:
    """List all available voices."""
    voices = await edge_tts.list_voices()
    return voices


def list_voices(language: str = "en") -> list[dict]:
    """List available voices, optionally filtered by language."""
    voices = asyncio.run(_list_voices_async())
    if language:
        voices = [v for v in voices if v["Locale"].startswith(language)]
    return voices