bee_server.py

#!/usr/bin/env python3
"""
bee_server.py — Premium TTS sidecar for The Bee Bot
Port 7874 | FastAPI + uvicorn

Voice engine priority:
  1. Microsoft Edge Neural TTS (edge-tts) — premium quality, ~200 ms latency,
     free, no API key, uses the same Azure Cognitive Services backend as
     Microsoft Edge browser and Cortana.  Requires internet.
  2. Piper TTS (local neural) — offline fallback.  Default voice downloads ~78 MB
     on first use.  Good quality, always available once downloaded.

Endpoints:
  GET  /health                      -> {"status":"ok","primary_engine":"...","edge_available":bool,...}
  GET  /diagnostics                 -> full diagnostic detail for the panel
  POST /speak                       -> body {"text":"..."} -> audio/mpeg (edge) or audio/wav (piper)
  GET  /voices                      -> catalog of available Piper voices with download/active status
  POST /voices/download             -> body {"voice_id":"..."} -> start background download
  GET  /voices/progress/{voice_id}  -> {"voice_id":"...", "progress":0-100, "status":"idle|downloading|done|error"}
  POST /voices/activate             -> body {"voice_id":"..."} -> set active Piper voice
  GET  /voices/edge                 -> list of available Edge neural voices
  POST /voices/set_edge_voice       -> body {"voice":"en-US-JennyNeural"} -> switch Edge voice
"""

# ── stdout/stderr fix for Windows (avoids charmap crash on special chars) ─────
import sys
try:
    if sys.platform == "win32":
        sys.stdout.reconfigure(encoding="utf-8", errors="replace")
        sys.stderr.reconfigure(encoding="utf-8", errors="replace")
except AttributeError:
    pass  # Python < 3.7 doesn't have reconfigure

import asyncio
import json
import os
import platform
import secrets
import shutil
import struct
import subprocess
import threading
import urllib.request
import zipfile
from pathlib import Path
from typing import Optional

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import Response
from pydantic import BaseModel
import uvicorn

# ── Constants ─────────────────────────────────────────────────────────────────

PORT    = 7874
# Writable Bee voice state intentionally lives outside Tauri app data at
# ~/.more_ai/bee. App data is reserved for app DBs/logs; executable scripts load
# from bundled resources in release or the source checkout in dev/no-bundle.
BEE_DIR = Path.home() / ".more_ai" / "bee"
BEE_SCRIPT_SOURCE = os.environ.get("MORE_AI_BEE_TTS_SCRIPT_SOURCE", "unknown")
BEE_SCRIPT_PATH = os.environ.get("MORE_AI_BEE_TTS_SCRIPT_PATH", str(Path(__file__).resolve()))
BEE_PYTHON_PATH = os.environ.get("MORE_AI_BEE_TTS_PYTHON_PATH", sys.executable)
BEE_LOG_DIR = os.environ.get("MORE_AI_BEE_TTS_LOG_DIR", "")
EDGE_TTS_DISABLED = os.environ.get("MORE_AI_BEE_TTS_DISABLE_EDGE", "").strip().lower() in {"1", "true", "yes", "on"}

# ── Microsoft Edge Neural TTS ─────────────────────────────────────────────────

EDGE_RATE   = "+0%"
EDGE_PITCH  = "+0Hz"
EDGE_VOLUME = "+0%"

# Edge voice catalog — all free, no API key required.
EDGE_VOICE_CATALOG = [
    {"id": "en-US-GuyNeural",    "name": "Guy",    "lang": "en-US", "gender": "Male",   "style": "Natural, neutral (default)"},
    {"id": "en-US-JennyNeural",  "name": "Jenny",  "lang": "en-US", "gender": "Female", "style": "Warm, conversational"},
    {"id": "en-US-AriaNeural",   "name": "Aria",   "lang": "en-US", "gender": "Female", "style": "Expressive, clear"},
    {"id": "en-US-DavisNeural",  "name": "Davis",  "lang": "en-US", "gender": "Male",   "style": "Deep, confident"},
    {"id": "en-US-EmmaNeural",   "name": "Emma",   "lang": "en-US", "gender": "Female", "style": "Very natural, smooth"},
    {"id": "en-US-NancyNeural",  "name": "Nancy",  "lang": "en-US", "gender": "Female", "style": "Professional"},
    {"id": "en-GB-LibbyNeural",  "name": "Libby",  "lang": "en-GB", "gender": "Female", "style": "British, friendly"},
    {"id": "en-GB-RyanNeural",   "name": "Ryan",   "lang": "en-GB", "gender": "Male",   "style": "British, natural"},
    {"id": "en-AU-NatashaNeural","name": "Natasha","lang": "en-AU", "gender": "Female", "style": "Australian, warm"},
    {"id": "en-AU-WilliamNeural","name": "William","lang": "en-AU", "gender": "Male",   "style": "Australian, confident"},
]

_EDGE_VOICE_PREFS_FILE = BEE_DIR / "edge_voice.txt"

def _load_edge_voice() -> str:
    try:
        if _EDGE_VOICE_PREFS_FILE.exists():
            v = _EDGE_VOICE_PREFS_FILE.read_text(encoding="utf-8").strip()
            if v:
                return v
    except Exception:
        pass
    return "en-US-GuyNeural"

def _save_edge_voice(voice: str) -> None:
    BEE_DIR.mkdir(parents=True, exist_ok=True)
    _EDGE_VOICE_PREFS_FILE.write_text(voice, encoding="utf-8")

# Active edge voice — mutable global, persisted to file
EDGE_VOICE = _load_edge_voice()

_edge_available: Optional[bool] = None   # None = not yet checked

# ── Piper TTS voice catalog ───────────────────────────────────────────────────

_HF_BASE = "https://huggingface.co/rhasspy/piper-voices/resolve/main"

# Each entry: voice_id → metadata.
# The URLs follow the pattern: {HF_BASE}/{lang_folder}/{locale_folder}/{speaker}/{quality}/{voice_id}.onnx
PIPER_VOICE_CATALOG = {
    "en_US-lessac-medium": {
        "name": "Lessac",
        "lang": "en-US",
        "gender": "Female",
        "quality": "medium",
        "size_mb": 63,
        "description": "Clear, neutral American English (default offline voice)",
        "onnx_url": f"{_HF_BASE}/en/en_US/lessac/medium/en_US-lessac-medium.onnx",
        "json_url":  f"{_HF_BASE}/en/en_US/lessac/medium/en_US-lessac-medium.onnx.json",
    },
    "en_US-ryan-medium": {
        "name": "Ryan",
        "lang": "en-US",
        "gender": "Male",
        "quality": "medium",
        "size_mb": 63,
        "description": "Natural, professional American male",
        "onnx_url": f"{_HF_BASE}/en/en_US/ryan/medium/en_US-ryan-medium.onnx",
        "json_url":  f"{_HF_BASE}/en/en_US/ryan/medium/en_US-ryan-medium.onnx.json",
    },
    "en_US-amy-medium": {
        "name": "Amy",
        "lang": "en-US",
        "gender": "Female",
        "quality": "medium",
        "size_mb": 63,
        "description": "Clear, friendly American female",
        "onnx_url": f"{_HF_BASE}/en/en_US/amy/medium/en_US-amy-medium.onnx",
        "json_url":  f"{_HF_BASE}/en/en_US/amy/medium/en_US-amy-medium.onnx.json",
    },
    "en_US-joe-medium": {
        "name": "Joe",
        "lang": "en-US",
        "gender": "Male",
        "quality": "medium",
        "size_mb": 36,
        "description": "Casual, conversational American male",
        "onnx_url": f"{_HF_BASE}/en/en_US/joe/medium/en_US-joe-medium.onnx",
        "json_url":  f"{_HF_BASE}/en/en_US/joe/medium/en_US-joe-medium.onnx.json",
    },
    "en_GB-alan-medium": {
        "name": "Alan",
        "lang": "en-GB",
        "gender": "Male",
        "quality": "medium",
        "size_mb": 55,
        "description": "Warm British English male",
        "onnx_url": f"{_HF_BASE}/en/en_GB/alan/medium/en_GB-alan-medium.onnx",
        "json_url":  f"{_HF_BASE}/en/en_GB/alan/medium/en_GB-alan-medium.onnx.json",
    },
    "en_GB-alba-medium": {
        "name": "Alba",
        "lang": "en-GB",
        "gender": "Female",
        "quality": "medium",
        "size_mb": 66,
        "description": "Expressive British English female",
        "onnx_url": f"{_HF_BASE}/en/en_GB/alba/medium/en_GB-alba-medium.onnx",
        "json_url":  f"{_HF_BASE}/en/en_GB/alba/medium/en_GB-alba-medium.onnx.json",
    },
}

_PIPER_ACTIVE_FILE = BEE_DIR / "active_piper_voice.txt"
_DEFAULT_PIPER_VOICE = "en_US-lessac-medium"

def _load_active_piper_voice() -> str:
    try:
        if _PIPER_ACTIVE_FILE.exists():
            v = _PIPER_ACTIVE_FILE.read_text(encoding="utf-8").strip()
            if v in PIPER_VOICE_CATALOG:
                return v
    except Exception:
        pass
    return _DEFAULT_PIPER_VOICE

def _save_active_piper_voice(voice_id: str) -> None:
    BEE_DIR.mkdir(parents=True, exist_ok=True)
    _PIPER_ACTIVE_FILE.write_text(voice_id, encoding="utf-8")

# Mutable globals for active Piper voice
_active_piper_voice: str = _load_active_piper_voice()

# Download state per voice_id: 0-100 = progress, -1 = error, 101 = done
_download_progress: dict[str, int] = {}
_download_errors: dict[str, str] = {}
_download_lock = threading.Lock()

# ── Runtime diagnostics (updated on every /speak call) ─────────────────────
import time as _time
_last_engine_used: str = "none"
_last_error: str = ""
_last_fallback_reason: str = ""
_last_latency_ms: float = 0.0
_total_speak_calls: int = 0
_edge_success_count: int = 0
_piper_success_count: int = 0
_fallback_count: int = 0

def _classify_storage_error(exc: Exception) -> str:
    msg = str(exc).lower()
    if "access is denied" in msg or "permission denied" in msg or isinstance(exc, PermissionError):
        return "permission_denied"
    if "read-only" in msg or "readonly" in msg:
        return "readonly"
    if "no space" in msg or "disk full" in msg:
        return "disk_full"
    return type(exc).__name__

def _probe_bee_storage() -> dict:
    """Probe Bee TTS writable storage separately from SQLite database health."""
    probe_path = BEE_DIR / ".bee_tts_write_probe.tmp"
    try:
        BEE_DIR.mkdir(parents=True, exist_ok=True)
        probe_path.write_text("ok", encoding="utf-8")
        probe_path.unlink(missing_ok=True)
        return {
            "ok": True,
            "path": str(BEE_DIR),
            "detail": "Bee TTS storage is writable.",
            "likely_cause": None,
        }
    except Exception as exc:
        return {
            "ok": False,
            "path": str(BEE_DIR),
            "detail": f"{type(exc).__name__}: {exc}",
            "likely_cause": _classify_storage_error(exc),
        }


# ── Edge TTS helpers ──────────────────────────────────────────────────────────

def _piper_status_detail() -> dict:
    voice_downloaded = _voice_downloaded(_active_piper_voice)
    model_path = _voice_model_path(_active_piper_voice)
    cfg_path = _voice_cfg_path(_active_piper_voice)
    binary_ready = PIPER_EXE.exists()
    repair_status = "ready" if binary_ready and voice_downloaded else "needs_download"
    if binary_ready and not voice_downloaded:
        repair_message = f"Download or repair Piper voice '{_active_piper_voice}' for offline DugBee speech."
    elif not binary_ready:
        repair_message = f"Download or repair Piper binary at {PIPER_EXE} for offline DugBee speech."
    else:
        repair_message = "Piper offline fallback assets are present."
    return {
        "piper_binary_path": str(PIPER_EXE),
        "piper_binary_ready": binary_ready,
        "piper_active_voice_downloaded": voice_downloaded,
        "piper_active_voice_model_path": str(model_path),
        "piper_active_voice_config_path": str(cfg_path),
        "piper_repair_status": repair_status,
        "piper_repair_message": repair_message,
    }


def _ensure_edge_tts() -> bool:
    """Import edge_tts from the bundled runtime.  Caches result globally."""
    global _edge_available, _last_error, _last_fallback_reason
    if EDGE_TTS_DISABLED:
        _edge_available = False
        _last_error = ""
        _last_fallback_reason = "edge-tts disabled by MORE_AI_BEE_TTS_DISABLE_EDGE; Piper offline fallback will be used"
        print(f"[Bee TTS] {_last_fallback_reason}", flush=True)
        return False
    if _edge_available is not None:
        return _edge_available

    try:
        import edge_tts  # noqa: F401
        _edge_available = True
        print("[Bee TTS] edge-tts ready — Microsoft Neural voice active", flush=True)
    except ImportError as e:
        _edge_available = False
        _last_error = f"edge-tts import failed: {e}"
        _last_fallback_reason = "edge-tts missing from bundled Python runtime; runtime pip install is disabled"
        print(f"[Bee TTS] {_last_fallback_reason}. Rebuild python_env.zip with edge-tts.", flush=True)

    return bool(_edge_available)


def _fix_pronunciation(text: str) -> str:
    """Apply phonetic corrections before sending to the neural voice engine.

    Rules:
      • "Trier"   → "Treer"    rhymes with rear/deer/steer (TREER, one syllable)
      • "Ollama"  → "Oh-lama"  prevent "olla-mah" mispronunciation
      • "Tauri"   → "Tawree"   the Rust desktop framework (TORE-ee)
      • "LLM"     → "L L M"    spell out the acronym
      • "LLMs"    → "L L M s"
      • "P2P"     → "peer to peer"
      • "WebRTC"  → "Web R T C"
      • "DMARC"   → "dee mark"
      • "ECDSA"   → "E C D S A"
      • "IDE"     → "I D E"    prevent "idle" mispronunciation
      • "API"     → "A P I"    ensure letters not word
      • "APIs"    → "A P I s"
      • "SSML"    → "S S M L"
      • "RTMP"    → "R T M P"
      • "UUID"    → "U U I D"
      • "SQLite"  → "sequel lite"
      • "SQL"     → "sequel"
      • "SFX"     → "S F X"
      • "MSI"     → "M S I"
      • "Qwen"    → "Chwen"    Chinese AI model, ch sound
      • "uvicorn" → "you-vi-corn"
      • "FastAPI" → "Fast A P I"
      • "Nostr"   → "No-str"
      • "PyTorch" → "Pie torch"
    """
    import re

    # Brand name — "Trier" rhymes with rear/deer/steer (TREER, one syllable)
    text = re.sub(r'Trier\s+OS', 'Treer OS', text, flags=re.IGNORECASE)
    text = re.sub(r'\bTrier\b',  'Treer',    text, flags=re.IGNORECASE)

    # Compound tech names — do these before single-letter acronym expansion
    text = re.sub(r'\bFastAPI\b',  'Fast A P I',     text, flags=re.IGNORECASE)
    text = re.sub(r'\bWebRTC\b',   'Web R T C',      text, flags=re.IGNORECASE)
    text = re.sub(r'\bSQLite\b',   'sequel lite',    text, flags=re.IGNORECASE)
    text = re.sub(r'\bPyTorch\b',  'Pie torch',      text, flags=re.IGNORECASE)
    text = re.sub(r'\buvicorn\b',  'you-vi-corn',    text, flags=re.IGNORECASE)
    text = re.sub(r'\bNostr\b',    'No-str',         text, flags=re.IGNORECASE)

    # Framework/tool names
    text = re.sub(r'\bOllama\b',   'Oh-lama',        text, flags=re.IGNORECASE)
    text = re.sub(r'\bTauri\b',    'Tawree',         text, flags=re.IGNORECASE)
    text = re.sub(r'\bQwen\b',     'Chwen',          text, flags=re.IGNORECASE)

    # Acronyms — order matters: longest/most-specific first
    text = re.sub(r'\bLLMs\b',     'L L M s',        text)
    text = re.sub(r'\bLLM\b',      'L L M',          text)
    text = re.sub(r'\bP2P\b',      'peer to peer',   text, flags=re.IGNORECASE)
    text = re.sub(r'\bDMARC\b',    'dee mark',       text, flags=re.IGNORECASE)
    text = re.sub(r'\bECDSA\b',    'E C D S A',      text, flags=re.IGNORECASE)
    text = re.sub(r'\bSSML\b',     'S S M L',        text, flags=re.IGNORECASE)
    text = re.sub(r'\bRTMP\b',     'R T M P',        text, flags=re.IGNORECASE)
    text = re.sub(r'\bUUID\b',     'U U I D',        text, flags=re.IGNORECASE)
    text = re.sub(r'\bSQL\b',      'sequel',         text, flags=re.IGNORECASE)
    text = re.sub(r'\bSFX\b',      'S F X',          text, flags=re.IGNORECASE)
    text = re.sub(r'\bMSI\b',      'M S I',          text, flags=re.IGNORECASE)
    text = re.sub(r'\bIDEs?\b',    lambda m: 'I D E s' if m.group().endswith('s') else 'I D E', text)
    text = re.sub(r'\bAPIs\b',     'A P I s',        text, flags=re.IGNORECASE)
    text = re.sub(r'\bAPI\b',      'A P I',          text, flags=re.IGNORECASE)

    # ── Weather: wind direction arrows → compass words ───────────────────────
    # wttr.in returns Unicode arrows like ↘8mph which TTS reads as "arrow 8 M P H"
    text = text.replace('↖', 'northwest at ')
    text = text.replace('↑', 'north at ')
    text = text.replace('↗', 'northeast at ')
    text = text.replace('→', 'east at ')
    text = text.replace('↘', 'southeast at ')
    text = text.replace('↓', 'south at ')
    text = text.replace('↙', 'southwest at ')
    text = text.replace('←', 'west at ')

    # ── Weather: speed abbreviations ─────────────────────────────────────────
    text = re.sub(r'\bmph\b',  'miles per hour',       text, flags=re.IGNORECASE)
    text = re.sub(r'\bkm/h\b', 'kilometers per hour',  text, flags=re.IGNORECASE)
    text = re.sub(r'\bkph\b',  'kilometers per hour',  text, flags=re.IGNORECASE)

    # ── Weather: temperature symbols ─────────────────────────────────────────
    text = text.replace('°F', ' degrees Fahrenheit')
    text = text.replace('°C', ' degrees Celsius')

    # Collapse any double-spaces introduced above
    text = re.sub(r'  +', ' ', text)

    return text


async def _synthesize_edge(text: str) -> bytes:
    """Call Microsoft Edge Neural TTS.  Returns raw MP3 bytes.
    NOTE: pronunciation fixes are applied before this call at the /speak endpoint."""
    import edge_tts
    communicate = edge_tts.Communicate(
        text=text,
        voice=EDGE_VOICE,
        rate=EDGE_RATE,
        pitch=EDGE_PITCH,
        volume=EDGE_VOLUME,
    )
    chunks: list[bytes] = []
    async for chunk in communicate.stream():
        if chunk["type"] == "audio":
            chunks.append(chunk["data"])
    data = b"".join(chunks)
    if not data:
        raise RuntimeError("edge-tts returned empty audio")
    return data


# ── Piper TTS helpers ─────────────────────────────────────────────────────────

_PIPER_RELEASE = "2023.11.14-2"
if platform.system() == "Windows":
    _PIPER_URL = (
        f"https://github.com/rhasspy/piper/releases/download/{_PIPER_RELEASE}"
        "/piper_windows_amd64.zip"
    )
else:
    _PIPER_URL = (
        f"https://github.com/rhasspy/piper/releases/download/{_PIPER_RELEASE}"
        "/piper_linux_x86_64.tar.gz"
    )

_VOICE_DIR = BEE_DIR / "voices"
PIPER_EXE  = BEE_DIR / ("piper.exe" if platform.system() == "Windows" else "piper")

_piper_ready: bool = False   # True once the active voice is downloaded and Piper binary is present


def _voice_model_path(voice_id: str) -> Path:
    return _VOICE_DIR / f"{voice_id}.onnx"

def _voice_cfg_path(voice_id: str) -> Path:
    return _VOICE_DIR / f"{voice_id}.onnx.json"

def _voice_downloaded(voice_id: str) -> bool:
    return _voice_model_path(voice_id).exists() and _voice_cfg_path(voice_id).exists()


def _download_file(url: str, dest: Path, label: str,
                   progress_callback=None, total_bytes: int = 0) -> None:
    """Download url → dest with optional progress reporting."""
    print(f"[Bee TTS] Downloading {label}...", flush=True)
    dest.parent.mkdir(parents=True, exist_ok=True)
    tmp = dest.with_suffix(dest.suffix + ".tmp")
    try:
        req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
        with urllib.request.urlopen(req, timeout=120) as resp:
            file_size = int(resp.headers.get("Content-Length", total_bytes) or 0)
            downloaded = 0
            chunk_size = 65536
            with open(tmp, "wb") as f:
                while True:
                    chunk = resp.read(chunk_size)
                    if not chunk:
                        break
                    f.write(chunk)
                    downloaded += len(chunk)
                    if progress_callback and file_size > 0:
                        progress_callback(int(downloaded * 100 / file_size))
        tmp.replace(dest)
        print(f"[Bee TTS] {label} saved ({downloaded // 1024} KB)", flush=True)
    except Exception as e:
        tmp.unlink(missing_ok=True)
        raise RuntimeError(f"Failed to download {label}: {e}") from e


def _download_legacy(url: str, dest: Path, label: str) -> None:
    """Simple download without progress (used for Piper binary)."""
    _download_file(url, dest, label)


def _ensure_piper() -> None:
    if PIPER_EXE.exists():
        return
    BEE_DIR.mkdir(parents=True, exist_ok=True)
    if platform.system() == "Windows":
        arc = BEE_DIR / "piper_windows.zip"
        _download_legacy(_PIPER_URL, arc, "Piper TTS (Windows)")
        print("[Bee TTS] Extracting Piper...", flush=True)
        xdir = BEE_DIR / "_piper_extract"
        xdir.mkdir(exist_ok=True)
        with zipfile.ZipFile(str(arc), "r") as zf:
            zf.extractall(str(xdir))
        sub = xdir / "piper"
        if not sub.exists():
            sub = xdir
        for item in sub.iterdir():
            dst = BEE_DIR / item.name
            if dst.exists():
                shutil.rmtree(str(dst)) if dst.is_dir() else dst.unlink()
            shutil.move(str(item), str(dst))
        shutil.rmtree(str(xdir), ignore_errors=True)
        arc.unlink(missing_ok=True)
    else:
        arc = BEE_DIR / "piper_linux.tar.gz"
        _download_legacy(_PIPER_URL, arc, "Piper TTS (Linux)")
        print("[Bee TTS] Extracting Piper...", flush=True)
        xdir = BEE_DIR / "_piper_extract"
        xdir.mkdir(exist_ok=True)
        import tarfile
        with tarfile.open(str(arc), "r:gz") as tf:
            tf.extractall(str(xdir))
        sub = xdir / "piper"
        if not sub.exists():
            sub = xdir
        for item in sub.iterdir():
            dst = BEE_DIR / item.name
            if dst.exists():
                shutil.rmtree(str(dst)) if dst.is_dir() else dst.unlink()
            shutil.move(str(item), str(dst))
        shutil.rmtree(str(xdir), ignore_errors=True)
        arc.unlink(missing_ok=True)
        if PIPER_EXE.exists():
            PIPER_EXE.chmod(PIPER_EXE.stat().st_mode | 0o111)
    if not PIPER_EXE.exists():
        raise RuntimeError(f"Piper binary not found at {PIPER_EXE} after extraction")
    print("[Bee TTS] Piper binary ready", flush=True)


def _ensure_voice_files(voice_id: str, progress_callback=None) -> None:
    """Download the .onnx and .onnx.json files for a voice if not already present."""
    if voice_id not in PIPER_VOICE_CATALOG:
        raise ValueError(f"Unknown voice_id: {voice_id}")
    meta = PIPER_VOICE_CATALOG[voice_id]
    _VOICE_DIR.mkdir(parents=True, exist_ok=True)
    model_path = _voice_model_path(voice_id)
    cfg_path   = _voice_cfg_path(voice_id)
    size_bytes = meta["size_mb"] * 1024 * 1024

    if not model_path.exists():
        _download_file(
            meta["onnx_url"], model_path, f"{voice_id}.onnx",
            progress_callback=progress_callback,
            total_bytes=size_bytes,
        )
    if not cfg_path.exists():
        _download_file(meta["json_url"], cfg_path, f"{voice_id}.onnx.json")


def _pcm_to_wav(pcm: bytes, sr: int = 22050) -> bytes:
    n = len(pcm)
    hdr = struct.pack(
        "<4sI4s4sIHHIIHH4sI",
        b"RIFF", 36 + n, b"WAVE", b"fmt ", 16,
        1, 1, sr, sr * 2, 2, 16, b"data", n,
    )
    return hdr + pcm


def _setup_piper_background() -> None:
    """Download Piper binary + active voice model in a daemon thread."""
    global _piper_ready
    try:
        _ensure_piper()
        _ensure_voice_files(_active_piper_voice)
        _piper_ready = True
        print("[Bee TTS] Piper offline fallback ready", flush=True)
    except Exception as e:
        print(f"[Bee TTS] Piper setup failed (offline TTS unavailable): {e}", flush=True)


def _download_voice_background(voice_id: str) -> None:
    """Background thread: download a Piper voice, tracking progress."""
    global _piper_ready
    with _download_lock:
        if _download_progress.get(voice_id) == 101:
            return  # already done
        _download_progress[voice_id] = 0

    def on_progress(pct: int) -> None:
        _download_progress[voice_id] = min(pct, 99)

    try:
        _ensure_piper()
        _ensure_voice_files(voice_id, progress_callback=on_progress)
        _download_progress[voice_id] = 101  # done
        _download_errors.pop(voice_id, None)
        # If this is the active voice, mark piper ready
        if voice_id == _active_piper_voice:
            _piper_ready = True
        print(f"[Bee TTS] Voice '{voice_id}' downloaded successfully", flush=True)
    except Exception as e:
        _download_progress[voice_id] = -1
        _download_errors[voice_id] = str(e)
        print(f"[Bee TTS] Voice '{voice_id}' download failed: {e}", flush=True)


async def _synthesize_piper(text: str) -> bytes:
    """Synthesize speech using the currently active Piper voice."""
    voice_id = _active_piper_voice
    model_path = _voice_model_path(voice_id)
    if not (PIPER_EXE.exists() and model_path.exists()):
        raise RuntimeError(f"Piper not ready (voice: {voice_id})")
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(
        None,
        lambda: subprocess.run(
            [str(PIPER_EXE), "--model", str(model_path), "--output_raw"],
            input=text.encode("utf-8"),
            capture_output=True,
            timeout=30,
        ),
    )
    if result.returncode != 0:
        raise RuntimeError(result.stderr.decode("utf-8", errors="replace"))
    return _pcm_to_wav(result.stdout)


# ── FastAPI app ───────────────────────────────────────────────────────────────

app = FastAPI(title="Bee TTS Server", version="3.0.0")

# CAT-5-002: Token auth middleware
import os as _os, sys as _sys
_SESSION_TOKEN = ""
for _i, _a in enumerate(_sys.argv):
    if _a == "--token" and _i + 1 < len(_sys.argv):
        _SESSION_TOKEN = _sys.argv[_i + 1]
        break

try:
    from starlette.middleware.base import BaseHTTPMiddleware
    from starlette.responses import Response as _SR
    class _TokenAuth(BaseHTTPMiddleware):
        async def dispatch(self, request, call_next):
            if request.method == "OPTIONS":
                return await call_next(request)
            provided = request.headers.get("x-more-ai-token", "")
            if not _SESSION_TOKEN:
                return _SR("Bee TTS auth token is not configured", status_code=503)
            if not provided:
                return _SR("Missing Bee TTS auth token", status_code=401)
            if not secrets.compare_digest(provided, _SESSION_TOKEN):
                return _SR("Invalid or stale Bee TTS auth token", status_code=401)
            return await call_next(request)
    app.add_middleware(_TokenAuth)
except Exception:
    raise RuntimeError("Bee TTS auth middleware is required")

# CORS — required so the Tauri WebView (tauri://localhost in production,
# http://localhost:1420 in dev) can make fetch() calls to this sidecar.
# Without this every /speak, /diagnostics, /voices call is silently blocked
# by the browser and the client sees "Connection refused".
app.add_middleware(
    CORSMiddleware,
    allow_origins=["tauri://localhost", "http://tauri.localhost", "http://localhost:1420", "http://127.0.0.1:1420"],       # localhost sidecar — no external traffic
    allow_methods=["*"],
    allow_headers=["*"],
    expose_headers=["X-More-AI-TTS-Engine"],
)


@app.get("/health")
async def health():
    preferred = f"edge-tts ({EDGE_VOICE})" if _edge_available else (
        f"piper ({_active_piper_voice})" if _piper_ready else "none"
    )
    storage_probe = _probe_bee_storage()
    return {
        "status": "ok",
        "primary_engine": preferred,
        "script_source": BEE_SCRIPT_SOURCE,
        "script_path": BEE_SCRIPT_PATH,
        "python_path": BEE_PYTHON_PATH,
        "log_dir": BEE_LOG_DIR,
        "tts_storage_root": str(BEE_DIR),
        "tts_storage_probe": storage_probe,
        "piper_voice_dir": str(_VOICE_DIR),
        **_piper_status_detail(),
        "edge_available": bool(_edge_available),
        "edge_tts_disabled": EDGE_TTS_DISABLED,
        "edge_voice": EDGE_VOICE,
        "piper_ready": _piper_ready,
        "piper_active_voice": _active_piper_voice,
        # Runtime diagnostics
        "last_engine_used": _last_engine_used,
        "last_error": _last_error,
        "last_fallback_reason": _last_fallback_reason,
        "last_latency_ms": round(_last_latency_ms, 1),
        "total_speak_calls": _total_speak_calls,
        "edge_success_count": _edge_success_count,
        "piper_success_count": _piper_success_count,
        "web_fallback_count": _fallback_count,
    }


@app.get("/diagnostics")
async def diagnostics():
    """Detailed diagnostics endpoint — used by the DugBee Voice Diagnostics panel."""
    preferred = f"edge-tts ({EDGE_VOICE})" if _edge_available else (
        f"piper ({_active_piper_voice})" if _piper_ready else "none"
    )
    storage_probe = _probe_bee_storage()
    return {
        "server": "online",
        "preferred_engine": preferred,
        "script_source": BEE_SCRIPT_SOURCE,
        "script_path": BEE_SCRIPT_PATH,
        "python_path": BEE_PYTHON_PATH,
        "log_dir": BEE_LOG_DIR,
        "tts_storage_root": str(BEE_DIR),
        "tts_storage_probe": storage_probe,
        "piper_voice_dir": str(_VOICE_DIR),
        **_piper_status_detail(),
        "edge_voice_prefs_file": str(_EDGE_VOICE_PREFS_FILE),
        "piper_active_voice_file": str(_PIPER_ACTIVE_FILE),
        "edge_tts_available": bool(_edge_available),
        "edge_tts_disabled": EDGE_TTS_DISABLED,
        "edge_voice": EDGE_VOICE,
        "edge_rate": EDGE_RATE,
        "piper_ready": _piper_ready,
        "piper_active_voice": _active_piper_voice,
        "last_engine_used": _last_engine_used,
        "last_error": _last_error,
        "last_fallback_reason": _last_fallback_reason,
        "last_latency_ms": round(_last_latency_ms, 1),
        "total_speak_calls": _total_speak_calls,
        "edge_success_count": _edge_success_count,
        "piper_success_count": _piper_success_count,
        "web_fallback_count": _fallback_count,
    }


@app.get("/voices")
async def list_voices():
    """Return all available Piper voices with download and active status."""
    voices = []
    for voice_id, meta in PIPER_VOICE_CATALOG.items():
        downloaded = _voice_downloaded(voice_id)
        prog = _download_progress.get(voice_id)
        if prog == 101:
            dl_status = "done"
        elif prog == -1:
            dl_status = "error"
        elif prog is not None:
            dl_status = "downloading"
        else:
            dl_status = "idle"
        voices.append({
            "voice_id": voice_id,
            "name": meta["name"],
            "lang": meta["lang"],
            "gender": meta["gender"],
            "quality": meta["quality"],
            "size_mb": meta["size_mb"],
            "description": meta["description"],
            "downloaded": downloaded,
            "active": voice_id == _active_piper_voice,
            "download_status": dl_status,
            "download_progress": prog if (prog is not None and 0 <= prog <= 100) else None,
            "download_error": _download_errors.get(voice_id),
        })
    return {"voices": voices, "piper_binary_ready": PIPER_EXE.exists()}


class DownloadRequest(BaseModel):
    voice_id: str

@app.post("/voices/download")
async def download_voice(req: DownloadRequest):
    """Start downloading a Piper voice in the background."""
    voice_id = req.voice_id
    if voice_id not in PIPER_VOICE_CATALOG:
        raise HTTPException(400, f"Unknown voice_id: {voice_id}")
    if _voice_downloaded(voice_id):
        return {"status": "already_downloaded", "voice_id": voice_id}
    prog = _download_progress.get(voice_id)
    if prog is not None and 0 <= prog < 101:
        return {"status": "already_downloading", "voice_id": voice_id, "progress": prog}
    # Start background download
    t = threading.Thread(target=_download_voice_background, args=(voice_id,), daemon=True)
    t.start()
    return {"status": "started", "voice_id": voice_id}


@app.get("/voices/progress/{voice_id}")
async def voice_progress(voice_id: str):
    """Poll download progress for a voice."""
    if voice_id not in PIPER_VOICE_CATALOG:
        raise HTTPException(400, f"Unknown voice_id: {voice_id}")
    if _voice_downloaded(voice_id):
        return {"voice_id": voice_id, "progress": 100, "status": "done"}
    prog = _download_progress.get(voice_id)
    if prog is None:
        return {"voice_id": voice_id, "progress": 0, "status": "idle"}
    if prog == 101:
        return {"voice_id": voice_id, "progress": 100, "status": "done"}
    if prog == -1:
        return {"voice_id": voice_id, "progress": 0, "status": "error",
                "error": _download_errors.get(voice_id, "unknown")}
    return {"voice_id": voice_id, "progress": prog, "status": "downloading"}


class ActivateRequest(BaseModel):
    voice_id: str

@app.post("/voices/activate")
async def activate_voice(req: ActivateRequest):
    """Set a Piper voice as the active offline fallback voice."""
    global _active_piper_voice, _piper_ready
    voice_id = req.voice_id
    if voice_id not in PIPER_VOICE_CATALOG:
        raise HTTPException(400, f"Unknown voice_id: {voice_id}")
    if not _voice_downloaded(voice_id):
        raise HTTPException(400, f"Voice '{voice_id}' is not downloaded yet")
    _active_piper_voice = voice_id
    _piper_ready = PIPER_EXE.exists() and _voice_downloaded(voice_id)
    _save_active_piper_voice(voice_id)
    return {"status": "activated", "voice_id": voice_id, "piper_ready": _piper_ready}


@app.get("/voices/edge")
async def list_edge_voices():
    """Return the Edge neural voice catalog."""
    return {
        "voices": [
            {**v, "active": v["id"] == EDGE_VOICE}
            for v in EDGE_VOICE_CATALOG
        ],
        "current_voice": EDGE_VOICE,
        "edge_available": bool(_edge_available),
    }


class SetEdgeVoiceRequest(BaseModel):
    voice: str

@app.post("/voices/set_edge_voice")
async def set_edge_voice(req: SetEdgeVoiceRequest):
    """Switch the active Microsoft Edge Neural voice."""
    global EDGE_VOICE
    valid_ids = {v["id"] for v in EDGE_VOICE_CATALOG}
    if req.voice not in valid_ids:
        raise HTTPException(400, f"Unknown Edge voice: {req.voice}")
    EDGE_VOICE = req.voice
    _save_edge_voice(req.voice)
    return {"status": "ok", "voice": EDGE_VOICE}


class SpeakRequest(BaseModel):
    text: str


@app.post("/speak")
async def speak(req: SpeakRequest):
    global _last_engine_used, _last_error, _last_fallback_reason
    global _last_latency_ms, _total_speak_calls
    global _edge_success_count, _piper_success_count, _fallback_count

    text = req.text.strip()
    if not text:
        raise HTTPException(400, "text is empty")

    # Apply phonetic corrections once here — covers ALL engines (Edge, Piper, future)
    text = _fix_pronunciation(text)

    _total_speak_calls += 1
    _t0 = _time.monotonic()

    # ── 1. Microsoft Edge Neural TTS (premium, ~200 ms, internet required) ──
    if _edge_available:
        try:
            mp3 = await asyncio.wait_for(_synthesize_edge(text), timeout=10.0)
            _last_latency_ms = (_time.monotonic() - _t0) * 1000
            _last_engine_used = f"edge-tts ({EDGE_VOICE})"
            _last_error = ""
            _last_fallback_reason = ""
            _edge_success_count += 1
            return Response(
                content=mp3,
                media_type="audio/mpeg",
                headers={"X-More-AI-TTS-Engine": _last_engine_used},
            )
        except asyncio.TimeoutError:
            _last_error = "edge-tts timeout (>10 s)"
            _last_fallback_reason = "edge-tts timed out — no internet or Azure blocked"
            print(f"[Bee TTS] {_last_error}, falling back to Piper", flush=True)
        except Exception as e:
            _last_error = f"edge-tts: {e}"
            _last_fallback_reason = f"edge-tts exception: {type(e).__name__}"
            print(f"[Bee TTS] edge-tts error ({e}), falling back to Piper", flush=True)

    # ── 2. Piper TTS (local neural, offline capable) ──
    try:
        wav = await _synthesize_piper(text)
        _last_latency_ms = (_time.monotonic() - _t0) * 1000
        _last_engine_used = f"piper ({_active_piper_voice})"
        _piper_success_count += 1
        return Response(
            content=wav,
            media_type="audio/wav",
            headers={"X-More-AI-TTS-Engine": _last_engine_used},
        )
    except Exception as e:
        _last_error = f"piper: {e}"
        _last_fallback_reason = f"piper exception: {type(e).__name__}"
        _fallback_count += 1
        raise HTTPException(503, f"All TTS engines unavailable: {e}")


# ── Entrypoint ────────────────────────────────────────────────────────────────

if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser(description="Bee TTS Server")
    parser.add_argument("--port", type=int, default=PORT,
                        help="TCP port to listen on (default: 7874)")
    parser.add_argument("--token", default="", help=argparse.SUPPRESS)
    args = parser.parse_args()
    actual_port = args.port if args.port > 0 else PORT

    print(f"[Bee TTS] Starting on port {actual_port}...", flush=True)
    print(f"[Bee TTS] script_source={BEE_SCRIPT_SOURCE}", flush=True)
    print(f"[Bee TTS] script_path={BEE_SCRIPT_PATH}", flush=True)
    print(f"[Bee TTS] python_path={BEE_PYTHON_PATH}", flush=True)
    print(f"[Bee TTS] log_dir={BEE_LOG_DIR or 'not configured'}", flush=True)

    # Verify bundled edge-tts (primary engine). Runtime pip install is disabled.
    _ensure_edge_tts()

    # Piper downloads happen in background so the server is immediately ready
    # for edge-tts requests without waiting for model downloads
    threading.Thread(target=_setup_piper_background, daemon=True).start()

    uvicorn.run(app, host="127.0.0.1", port=actual_port, log_level="warning")