diff --git a/.env.example b/.env.example index 2868826..c1a37da 100644 --- a/.env.example +++ b/.env.example @@ -13,6 +13,15 @@ ELEVENLABS_API_KEY= ELEVENLABS_VOICE_ID=21m00Tcm4TlvDq8ikWAM # Default: Rachel ELEVENLABS_MODEL_ID=eleven_flash_v2_5 # Options: eleven_turbo_v2_5, eleven_multilingual_v2 +# 60db API Configuration (optional, for cloud-based Voice AI TTS) +# Get your API key from https://60db.ai/ +SIXTYDB_API_KEY= +SIXTYDB_VOICE_ID=fbb75ed2-975a-40c7-9e06-38e30524a9a1 # Default: Zara +SIXTYDB_MODEL_ID= # Empty = voice's default tier +SIXTYDB_SPEED=1.0 # 0.5-2.0 +SIXTYDB_STABILITY=50 # 0-100 +SIXTYDB_SIMILARITY=75 # 0-100 + # Future: Additional TTS providers # GOOGLE_TTS_API_KEY= # AZURE_TTS_API_KEY= \ No newline at end of file diff --git a/CHANGELOG.md b/CHANGELOG.md index 757a522..df9d754 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added +- 60db as an additional cloud TTS provider alongside ElevenLabs +- 60db option in the interactive installer (`ccnotify install`) +- `SixtyDBProvider` in the TTS provider factory (alias: `sixtydb`) +- `SIXTYDB_*` environment variables for runtime configuration +- Embedded 60db and ElevenLabs providers in the standalone notify hook script +- 60db section in default configuration and `.env.example` + ## [0.1.13] - 2025-08-10 ### Fixed diff --git a/README.md b/README.md index e0f7972..d611107 100644 --- a/README.md +++ b/README.md @@ -30,7 +30,7 @@ uvx ccnotify install That's it! The installer will: -- Help you choose between local (Kokoro) or cloud (ElevenLabs) TTS +- Help you choose between local (Kokoro) or cloud (ElevenLabs, 60db) TTS - Download models if needed - Install hooks into Claude Code - Create a configuration file @@ -66,7 +66,7 @@ If you use Elevenlabs as a TTS provider, [use my Affiliate link to signup](https ## Features - **Smart Filtering**: Only notifies for significant events, not routine operations -- **Dual TTS Options**: Local Kokoro models (50+ voices) or ElevenLabs cloud API +- **Dual TTS Options**: Local Kokoro models (50+ voices) or cloud APIs (ElevenLabs, 60db) - **Project Detection**: Automatically identifies your current project - **Zero Config**: Works out of the box with sensible defaults - **Privacy First**: Local processing with optional cloud TTS @@ -102,10 +102,25 @@ The installer creates a configuration file at `~/.claude/ccnotify/config.json`. } ``` +**For Cloud TTS (60db):** + +```json +{ + "tts_provider": "60db", + "api_key": "your_api_key_here", + "voice_id": "fbb75ed2-975a-40c7-9e06-38e30524a9a1", + "model_id": "", + "speed": 1.0, + "stability": 50, + "similarity": 75 +} +``` + **Voice Options:** - **Kokoro**: `af_heart` (recommended), `af_sarah`, `am_adam`, `bf_alice`, `bm_daniel` and [40+ others](https://github.com/thewh1teagle/kokoro-onnx) - **ElevenLabs**: Use voice IDs from your [ElevenLabs account](https://try.elevenlabs.io/ist8m7h95ed2) +- **60db**: Use voice IDs from your [60db account](https://60db.ai/) (`stability` and `similarity` use a 0-100 scale) **Pronunciation Customization:** @@ -159,6 +174,7 @@ Customize how project names and commands are pronounced by editing `~/.claude/cc - Claude Code CLI - **For Kokoro**: ~350MB disk space for local models - **For ElevenLabs**: API account and internet connection +- **For 60db**: API account and internet connection ## Early Version Notice diff --git a/src/ccnotify/config.py b/src/ccnotify/config.py index bd7c00a..24c4c7f 100644 --- a/src/ccnotify/config.py +++ b/src/ccnotify/config.py @@ -78,6 +78,13 @@ def get_default_config() -> Dict[str, Any]: "stability": 0.5, "similarity_boost": 0.5, }, + "sixtydb": { + "voice_id": "fbb75ed2-975a-40c7-9e06-38e30524a9a1", + "model_id": "", + "speed": 1.0, + "stability": 50, + "similarity": 75, + }, "notifications": {"enabled": True, "sound_enabled": True, "logging_enabled": False}, "replacements": {"enabled": True, "auto_add_projects": True}, } diff --git a/src/ccnotify/installer/detector.py b/src/ccnotify/installer/detector.py index 7b53061..ecb39ab 100644 --- a/src/ccnotify/installer/detector.py +++ b/src/ccnotify/installer/detector.py @@ -51,6 +51,7 @@ class ModelStatus: kokoro_downloaded: bool = False kokoro_models_count: int = 0 elevenlabs_configured: bool = False + sixtydb_configured: bool = False issues: List[str] = None def __post_init__(self): @@ -85,7 +86,9 @@ def check_existing_installation(self) -> InstallationStatus: # Check models model_status = self.get_model_status() status.models_downloaded = ( - model_status.kokoro_downloaded or model_status.elevenlabs_configured + model_status.kokoro_downloaded + or model_status.elevenlabs_configured + or model_status.sixtydb_configured ) # Collect issues @@ -156,6 +159,9 @@ def get_config_status(self) -> ConfigStatus: if status.provider == "elevenlabs": if not config.get("elevenlabs_api_key"): status.issues.append("ElevenLabs provider selected but no API key configured") + elif status.provider == "60db": + if not config.get("sixtydb_api_key"): + status.issues.append("60db provider selected but no API key configured") except json.JSONDecodeError: status.issues.append("Config file exists but contains invalid JSON") @@ -201,6 +207,20 @@ def get_model_status(self) -> ModelStatus: except Exception: status.issues.append("Could not verify ElevenLabs configuration") + elif status.provider == "60db": + # Check 60db configuration + try: + with open(self.ccnotify_dir / "config.json") as f: + config = json.load(f) + + if config.get("sixtydb_api_key"): + status.sixtydb_configured = True + else: + status.issues.append("60db provider selected but no API key configured") + + except Exception: + status.issues.append("Could not verify 60db configuration") + return status def _check_hooks_configured(self) -> bool: diff --git a/src/ccnotify/installer/flows.py b/src/ccnotify/installer/flows.py index 47f18de..eba2ef3 100644 --- a/src/ccnotify/installer/flows.py +++ b/src/ccnotify/installer/flows.py @@ -216,13 +216,14 @@ def _setup_tts_provider(self, quiet: bool = False) -> Optional[Dict[str, Any]]: table.add_row("1", "Kokoro", "Local AI models (recommended, privacy-focused)") table.add_row("2", "ElevenLabs", "Cloud-based premium quality (requires API key)") - table.add_row("3", "None", "Silent mode (no voice notifications)") + table.add_row("3", "60db", "Cloud-based Voice AI TTS (requires API key)") + table.add_row("4", "None", "Silent mode (no voice notifications)") console.print(table) console.print() while True: - choice = Prompt.ask("Select TTS provider", choices=["1", "2", "3"], default="1") + choice = Prompt.ask("Select TTS provider", choices=["1", "2", "3", "4"], default="1") if choice == "1": # Kokoro setup @@ -251,6 +252,16 @@ def _setup_tts_provider(self, quiet: bool = False) -> Optional[Dict[str, Any]]: return elevenlabs_config elif choice == "3": + # 60db setup + sixtydb_config = self._setup_sixtydb() + if not sixtydb_config: + console.print( + "[yellow]60db setup failed. Please choose another provider.[/yellow]" + ) + continue + return sixtydb_config + + elif choice == "4": # Silent mode return {"tts_provider": "none"} @@ -266,6 +277,18 @@ def _setup_elevenlabs(self) -> Optional[Dict[str, Any]]: # TODO: Validate API key return {"tts_provider": "elevenlabs", "elevenlabs_api_key": api_key} + def _setup_sixtydb(self) -> Optional[Dict[str, Any]]: + """Setup 60db TTS provider.""" + console.print("\n[bold cyan]Setting up 60db TTS...[/bold cyan]") + + api_key = Prompt.ask("Enter your 60db API key", password=True) + if not api_key: + console.print("[red]API key required for 60db[/red]") + return None + + # TODO: Validate API key + return {"tts_provider": "60db", "sixtydb_api_key": api_key} + def _install_script_and_config(self, provider_config: Dict[str, Any]) -> bool: """Install ccnotify.py script and configuration.""" try: diff --git a/src/ccnotify/notify.py b/src/ccnotify/notify.py index 2514954..d1c7acd 100644 --- a/src/ccnotify/notify.py +++ b/src/ccnotify/notify.py @@ -143,6 +143,143 @@ def generate(self, text: str, output_path: Path) -> bool: return False +class ElevenLabsTTSProvider: + """Minimal embedded ElevenLabs TTS provider""" + + def __init__(self): + self.api_key = "" + self.voice_id = "21m00Tcm4TlvDq8ikWAM" # Rachel + self.model_id = "eleven_flash_v2_5" + self.stability = 0.5 + self.similarity_boost = 0.5 + self.base_url = "https://api.elevenlabs.io/v1" + self.timeout = 30 + + def is_available(self) -> bool: + """Check if ElevenLabs is available (no network call - key presence only)""" + if not self.api_key: + logger.warning("ElevenLabs API key not configured") + return False + return True + + def get_cache_key(self, text: str) -> str: + """Generate cache key for text""" + return hashlib.md5(text.encode()).hexdigest()[:16] + + def get_file_extension(self) -> str: + """Get file extension""" + return ".mp3" + + def generate(self, text: str, output_path: Path) -> bool: + """Generate TTS audio via ElevenLabs API""" + try: + import requests + + url = f"{self.base_url}/text-to-speech/{self.voice_id}" + headers = {"xi-api-key": self.api_key, "Content-Type": "application/json"} + payload = { + "text": text, + "model_id": self.model_id, + "voice_settings": { + "stability": self.stability, + "similarity_boost": self.similarity_boost, + }, + } + + response = requests.post(url, headers=headers, json=payload, timeout=self.timeout) + + if response.status_code != 200: + logger.error(f"ElevenLabs API error {response.status_code}") + return False + + output_path.parent.mkdir(parents=True, exist_ok=True) + with open(output_path, "wb") as f: + f.write(response.content) + + return True + + except Exception as e: + logger.error(f"ElevenLabs TTS generation failed: {e}") + return False + + +class SixtyDBTTSProvider: + """Minimal embedded 60db TTS provider""" + + def __init__(self): + self.api_key = "" + self.voice_id = "fbb75ed2-975a-40c7-9e06-38e30524a9a1" # Zara + self.model_id = "" # Empty = voice's default tier + self.speed = 1.0 + self.stability = 50 # 60db uses 0-100 scale + self.similarity = 75 # 60db uses 0-100 scale + self.base_url = "https://api.60db.ai" + self.timeout = 30 + + def is_available(self) -> bool: + """Check if 60db is available (no network call - key presence only)""" + if not self.api_key: + logger.warning("60db API key not configured") + return False + return True + + def get_cache_key(self, text: str) -> str: + """Generate cache key for text""" + return hashlib.md5(text.encode()).hexdigest()[:16] + + def get_file_extension(self) -> str: + """Get file extension""" + return ".mp3" + + def generate(self, text: str, output_path: Path) -> bool: + """Generate TTS audio via 60db API""" + try: + import base64 + + import requests + + url = f"{self.base_url}/tts-synthesize" + headers = { + "Authorization": f"Bearer {self.api_key}", + "Content-Type": "application/json", + } + payload = { + "text": text, + "voice_id": self.voice_id, + "speed": self.speed, + "stability": self.stability, + "similarity": self.similarity, + } + if self.model_id: + payload["model"] = self.model_id + + response = requests.post(url, headers=headers, json=payload, timeout=self.timeout) + + if response.status_code != 200: + logger.error(f"60db API error {response.status_code}") + return False + + data = response.json() + if not data.get("success", False): + logger.error(f"60db API reported failure: {data.get('message', 'unknown')}") + return False + + audio_base64 = data.get("audio_base64", "") + if not audio_base64: + logger.error("60db API returned empty audio data") + return False + + output_path.parent.mkdir(parents=True, exist_ok=True) + with open(output_path, "wb") as f: + f.write(base64.b64decode(audio_base64)) + + return True + + except Exception as e: + logger.error(f"60db TTS generation failed: {e}") + return False + + # Configuration BASE_DIR = Path.home() / ".claude" / "ccnotify" SOUNDS_DIR = BASE_DIR / "sounds" @@ -157,10 +294,16 @@ def generate(self, text: str, output_path: Path) -> bool: # Environment variables (will be reloaded in main() if .env exists) USE_TTS = os.getenv("USE_TTS", "true").lower() == "true" # Default to true to use TTS providers USE_LOGGING = os.getenv("USE_LOGGING", "false").lower() == "true" -TTS_PROVIDER = os.getenv("TTS_PROVIDER", "kokoro") # Options: macos_say, elevenlabs, kokoro +TTS_PROVIDER = os.getenv("TTS_PROVIDER", "kokoro") # Options: macos_say, elevenlabs, 60db, kokoro ELEVENLABS_API_KEY = os.getenv("ELEVENLABS_API_KEY", "") ELEVENLABS_VOICE_ID = os.getenv("ELEVENLABS_VOICE_ID", "21m00Tcm4TlvDq8ikWAM") # Rachel ELEVENLABS_MODEL_ID = os.getenv("ELEVENLABS_MODEL_ID", "eleven_flash_v2_5") # Flash 2.5 +SIXTYDB_API_KEY = os.getenv("SIXTYDB_API_KEY", "") +SIXTYDB_VOICE_ID = os.getenv("SIXTYDB_VOICE_ID", "fbb75ed2-975a-40c7-9e06-38e30524a9a1") # Zara +SIXTYDB_MODEL_ID = os.getenv("SIXTYDB_MODEL_ID", "") # Empty = voice's default tier +SIXTYDB_SPEED = os.getenv("SIXTYDB_SPEED", "1.0") +SIXTYDB_STABILITY = os.getenv("SIXTYDB_STABILITY", "50") # 0-100 scale +SIXTYDB_SIMILARITY = os.getenv("SIXTYDB_SIMILARITY", "75") # 0-100 scale KOKORO_VOICE = os.getenv("KOKORO_VOICE", "af_heart") # KOKORO_PATH is now determined from config file KOKORO_SPEED = os.getenv("KOKORO_SPEED", "1.0") # Speed multiplier (0.5-2.0) @@ -655,6 +798,68 @@ def _init_tts_provider(self): else: logger.warning("Kokoro TTS not available - models may not be downloaded") self.tts_provider = None + + # Use embedded 60db provider + elif TTS_PROVIDER in ("60db", "sixtydb"): + sixtydb_config = config.get("sixtydb", {}) + self.tts_provider = SixtyDBTTSProvider() + self.tts_provider.api_key = ( + config.get("sixtydb_api_key") + or sixtydb_config.get("api_key") + or SIXTYDB_API_KEY + ) + self.tts_provider.voice_id = ( + sixtydb_config.get("voice_id") + or config.get("voice_id") + or SIXTYDB_VOICE_ID + ) + self.tts_provider.model_id = ( + sixtydb_config.get("model_id") or SIXTYDB_MODEL_ID + ) + self.tts_provider.speed = float(sixtydb_config.get("speed", SIXTYDB_SPEED)) + self.tts_provider.stability = float( + sixtydb_config.get("stability", SIXTYDB_STABILITY) + ) + self.tts_provider.similarity = float( + sixtydb_config.get("similarity", SIXTYDB_SIMILARITY) + ) + + if self.tts_provider.is_available(): + logger.info("Initialized embedded 60db TTS provider") + else: + logger.warning("60db TTS not available - check API key configuration") + self.tts_provider = None + + # Use embedded ElevenLabs provider + elif TTS_PROVIDER == "elevenlabs": + elevenlabs_config = config.get("elevenlabs", {}) + self.tts_provider = ElevenLabsTTSProvider() + self.tts_provider.api_key = ( + config.get("elevenlabs_api_key") + or elevenlabs_config.get("api_key") + or ELEVENLABS_API_KEY + ) + self.tts_provider.voice_id = ( + elevenlabs_config.get("voice_id") + or config.get("voice_id") + or ELEVENLABS_VOICE_ID + ) + self.tts_provider.model_id = ( + elevenlabs_config.get("model_id") or ELEVENLABS_MODEL_ID + ) + self.tts_provider.stability = float( + elevenlabs_config.get("stability", 0.5) + ) + self.tts_provider.similarity_boost = float( + elevenlabs_config.get("similarity_boost", 0.5) + ) + + if self.tts_provider.is_available(): + logger.info("Initialized embedded ElevenLabs TTS provider") + else: + logger.warning("ElevenLabs TTS not available - check API key configuration") + self.tts_provider = None + else: logger.warning(f"Unsupported TTS provider: {TTS_PROVIDER}") self.tts_provider = None diff --git a/src/ccnotify/tts/factory.py b/src/ccnotify/tts/factory.py index 3ed8018..5da60e4 100644 --- a/src/ccnotify/tts/factory.py +++ b/src/ccnotify/tts/factory.py @@ -10,6 +10,7 @@ from .base import TTSProvider, TTSProviderNotAvailable from .kokoro import KokoroProvider from .elevenlabs import ElevenLabsProvider +from .sixtydb import SixtyDBProvider logger = logging.getLogger(__name__) @@ -18,6 +19,12 @@ TTS_PROVIDERS: Dict[str, Type[TTSProvider]] = { "kokoro": KokoroProvider, "elevenlabs": ElevenLabsProvider, + "60db": SixtyDBProvider, +} + +# Aliases for provider names (normalized in get_tts_provider) +PROVIDER_ALIASES = { + "sixtydb": "60db", } @@ -44,6 +51,7 @@ def get_tts_provider( # Normalize provider name provider_name = provider_name.lower().strip() + provider_name = PROVIDER_ALIASES.get(provider_name, provider_name) # Try primary provider provider = _create_provider(provider_name, config) @@ -119,6 +127,20 @@ def _build_provider_config(provider_name: str, base_config: Dict[str, Any]) -> D } ) + elif provider_name == "60db": + config.update( + { + "api_key": os.getenv("SIXTYDB_API_KEY", base_config.get("api_key", "")), + "voice_id": os.getenv( + "SIXTYDB_VOICE_ID", base_config.get("voice_id", "fbb75ed2-975a-40c7-9e06-38e30524a9a1") + ), + "model_id": os.getenv("SIXTYDB_MODEL_ID", base_config.get("model_id", "")), + "speed": os.getenv("SIXTYDB_SPEED", base_config.get("speed", "1.0")), + "stability": os.getenv("SIXTYDB_STABILITY", base_config.get("stability", 50)), + "similarity": os.getenv("SIXTYDB_SIMILARITY", base_config.get("similarity", 75)), + } + ) + return config @@ -127,11 +149,12 @@ def _get_fallback_order(primary_provider: str) -> list: # Define fallback preferences fallback_map = { - "kokoro": ["elevenlabs"], - "elevenlabs": ["kokoro"], + "kokoro": ["elevenlabs", "60db"], + "elevenlabs": ["kokoro", "60db"], + "60db": ["elevenlabs", "kokoro"], } - return fallback_map.get(primary_provider, ["kokoro", "elevenlabs"]) + return fallback_map.get(primary_provider, ["kokoro", "elevenlabs", "60db"]) def list_available_providers() -> Dict[str, Dict[str, Any]]: @@ -170,6 +193,7 @@ def _get_provider_description(provider_name: str) -> str: descriptions = { "kokoro": "Local high-quality TTS using ONNX models", "elevenlabs": "Premium cloud-based TTS with natural voices", + "60db": "Cloud-based Voice AI TTS with natural voices", } return descriptions.get(provider_name, "Unknown provider") diff --git a/src/ccnotify/tts/sixtydb.py b/src/ccnotify/tts/sixtydb.py new file mode 100644 index 0000000..7da0fd2 --- /dev/null +++ b/src/ccnotify/tts/sixtydb.py @@ -0,0 +1,281 @@ +""" +60db TTS provider - Cloud-based Voice AI text-to-speech +""" + +import base64 +from pathlib import Path +from typing import Dict, Any +import logging + +# Import requests only when needed +try: + import requests + + REQUESTS_AVAILABLE = True +except ImportError: + REQUESTS_AVAILABLE = False + +from .base import TTSProvider, TTSProviderNotAvailable, TTSGenerationError + +logger = logging.getLogger(__name__) + +# Default 60db voice: "Zara" (60db Fast tier) +DEFAULT_VOICE_ID = "fbb75ed2-975a-40c7-9e06-38e30524a9a1" + + +class SixtyDBProvider(TTSProvider): + """60db TTS provider using cloud API""" + + def __init__(self, config: Dict[str, Any]): + super().__init__(config) + + # Configuration + self.api_key = config.get("api_key", "") + self.voice_id = config.get("voice_id", DEFAULT_VOICE_ID) + self.model_id = config.get("model_id", "") # Empty = voice's default tier + + # API settings + self.base_url = "https://api.60db.ai" + self.timeout = config.get("timeout", 30) + + # Voice settings (60db uses 0-100 scale) + self.speed = float(config.get("speed", 1.0)) + self.stability = float(config.get("stability", 50)) + self.similarity = float(config.get("similarity", 75)) + + def _get_headers(self) -> Dict[str, str]: + """Get request headers with Bearer authentication""" + return {"Authorization": f"Bearer {self.api_key}"} + + def is_available(self) -> bool: + """Check if 60db TTS is available""" + if not REQUESTS_AVAILABLE: + self.logger.warning("requests library not installed - needed for 60db") + return False + + if not self.api_key: + self.logger.warning("60db API key not configured") + return False + + try: + # Test API connectivity + response = requests.get( + f"{self.base_url}/voices", headers=self._get_headers(), timeout=5 + ) + + if response.status_code == 200: + return True + elif response.status_code == 401: + self.logger.error("60db API key invalid") + return False + else: + self.logger.warning(f"60db API returned status {response.status_code}") + return False + + except requests.RequestException as e: + self.logger.warning(f"60db API connectivity test failed: {e}") + return False + + def generate(self, text: str, output_path: Path, **kwargs) -> bool: + """ + Generate TTS audio using 60db API + + Args: + text: Text to convert to speech (max 5000 characters) + output_path: Path where audio file should be saved + **kwargs: Additional options (voice_id, model_id, speed, stability, similarity) + + Returns: + True if generation was successful, False otherwise + """ + try: + if not self.is_available(): + raise TTSProviderNotAvailable("60db TTS is not available") + + # Get generation parameters + voice_id = kwargs.get("voice_id", self.voice_id) + model_id = kwargs.get("model_id", self.model_id) + speed = float(kwargs.get("speed", self.speed)) + stability = float(kwargs.get("stability", self.stability)) + similarity = float(kwargs.get("similarity", self.similarity)) + + # Prepare request + url = f"{self.base_url}/tts-synthesize" + headers = self._get_headers() + headers["Content-Type"] = "application/json" + + payload = { + "text": text, + "voice_id": voice_id, + "speed": speed, + "stability": stability, + "similarity": similarity, + } + + if model_id: + payload["model"] = model_id + + self.logger.debug( + f"Generating TTS: '{text[:50]}...' with voice={voice_id}, model={model_id or 'default'}" + ) + + # Make API request + response = requests.post(url, headers=headers, json=payload, timeout=self.timeout) + + # Check response + if response.status_code != 200: + error_msg = f"60db API error {response.status_code}" + try: + error_detail = response.json().get("message", "") + if error_detail: + error_msg += f": {error_detail}" + except: + pass + + raise TTSGenerationError(error_msg) + + data = response.json() + + if not data.get("success", False): + raise TTSGenerationError( + f"60db API reported failure: {data.get('message', 'unknown error')}" + ) + + audio_base64 = data.get("audio_base64", "") + if not audio_base64: + raise TTSGenerationError("60db API returned empty audio data") + + # Decode and save audio data + audio_bytes = base64.b64decode(audio_base64) + + output_path.parent.mkdir(parents=True, exist_ok=True) + with open(output_path, "wb") as f: + f.write(audio_bytes) + + self.log_generation(text, output_path, True, voice_id=voice_id, model_id=model_id) + return True + + except Exception as e: + self.logger.error(f"60db TTS generation failed: {e}") + self.log_generation(text, output_path, False) + + # Clean up partial file + if output_path.exists(): + try: + output_path.unlink() + except: + pass + + return False + + def get_file_extension(self) -> str: + """Get the file extension for 60db audio files""" + return ".mp3" + + def get_available_voices(self, model: str = None) -> Dict[str, Any]: + """ + Get list of available voices from 60db API + + Args: + model: Filter by model tier ("quality" or "fast"), None for default + + Returns: + Dictionary containing voice information + """ + try: + if not self.is_available(): + return {} + + url = f"{self.base_url}/voices" + if model: + url += f"?model={model}" + + response = requests.get( + url, headers=self._get_headers(), timeout=self.timeout + ) + + if response.status_code == 200: + voices_data = response.json() + + # Organize voices by tier + voices = {} + for voice in voices_data.get("data", []): + voice_info = { + "voice_id": voice.get("voice_id"), + "name": voice.get("name"), + "model": voice.get("model", "Unknown"), + "description": voice.get("description", ""), + "labels": voice.get("labels", {}), + "categories": voice.get("categories", []), + } + + model_tier = voice_info["model"] + if model_tier not in voices: + voices[model_tier] = [] + voices[model_tier].append(voice_info) + + return voices + else: + self.logger.error(f"Failed to fetch voices: {response.status_code}") + return {} + + except Exception as e: + self.logger.error(f"Error fetching available voices: {e}") + return {} + + def get_voice_info(self, voice_id: str) -> Dict[str, Any]: + """ + Get detailed information about a specific voice + + Args: + voice_id: 60db voice ID + + Returns: + Dictionary containing voice information + """ + try: + if not self.is_available(): + return {} + + all_voices = self.get_available_voices() + for tier_voices in all_voices.values(): + for voice in tier_voices: + if voice.get("voice_id") == voice_id: + return voice + + return {} + + except Exception as e: + self.logger.error(f"Error fetching voice info for {voice_id}: {e}") + return {} + + def test_voice( + self, voice_id: str, test_text: str = "Hello, this is a test of 60db TTS" + ) -> bool: + """ + Test a specific voice by generating a sample + + Args: + voice_id: 60db voice ID to test + test_text: Text to use for testing + + Returns: + True if voice test successful, False otherwise + """ + try: + from tempfile import NamedTemporaryFile + + with NamedTemporaryFile(suffix=".mp3", delete=False) as tmp_file: + tmp_path = Path(tmp_file.name) + + success = self.generate(test_text, tmp_path, voice_id=voice_id) + + # Clean up test file + if tmp_path.exists(): + tmp_path.unlink() + + return success + + except Exception as e: + self.logger.error(f"Voice test failed for {voice_id}: {e}") + return False diff --git a/tests/test_sixtydb_provider.py b/tests/test_sixtydb_provider.py new file mode 100644 index 0000000..c229126 --- /dev/null +++ b/tests/test_sixtydb_provider.py @@ -0,0 +1,302 @@ +"""Tests for the 60db TTS provider.""" + +import base64 +import json +from pathlib import Path +from unittest.mock import patch, MagicMock + +import pytest + + +@pytest.fixture +def mock_response_ok(): + """Mock a successful 60db synthesize response.""" + audio_bytes = b"fake-mp3-audio-data" + response = MagicMock() + response.status_code = 200 + response.json.return_value = { + "success": True, + "message": "Audio generated successfully", + "audio_base64": base64.b64encode(audio_bytes).decode(), + "sample_rate": 24000, + "duration_seconds": 1.5, + "encoding": "mp3", + } + response.content = audio_bytes + return response + + +class TestSixtyDBProvider: + """Tests for the SixtyDBProvider class.""" + + def _get_provider(self): + from src.ccnotify.tts.sixtydb import SixtyDBProvider + + return SixtyDBProvider({"api_key": "test-key"}) + + def test_provider_registration(self): + """Test that 60db is registered in the factory.""" + from src.ccnotify.tts.factory import TTS_PROVIDERS + from src.ccnotify.tts.sixtydb import SixtyDBProvider + + assert TTS_PROVIDERS.get("60db") is SixtyDBProvider + + def test_alias_normalization(self): + """Test that 'sixtydb' alias resolves to '60db'.""" + from src.ccnotify.tts.factory import PROVIDER_ALIASES + + assert PROVIDER_ALIASES.get("sixtydb") == "60db" + + def test_fallback_order_includes_60db(self): + """Test that fallback orders include 60db.""" + from src.ccnotify.tts.factory import _get_fallback_order + + assert "60db" in _get_fallback_order("kokoro") + assert "60db" in _get_fallback_order("elevenlabs") + assert _get_fallback_order("60db") == ["elevenlabs", "kokoro"] + + def test_default_config(self): + """Test default configuration values.""" + provider = self._get_provider() + + assert provider.voice_id == "fbb75ed2-975a-40c7-9e06-38e30524a9a1" + assert provider.model_id == "" + assert provider.speed == 1.0 + assert provider.stability == 50 # 0-100 scale + assert provider.similarity == 75 # 0-100 scale + assert provider.base_url == "https://api.60db.ai" + + def test_get_file_extension(self): + """Test audio file extension.""" + provider = self._get_provider() + assert provider.get_file_extension() == ".mp3" + + def test_not_available_without_api_key(self): + """Test that provider is unavailable without an API key.""" + from src.ccnotify.tts.sixtydb import SixtyDBProvider + + provider = SixtyDBProvider({}) + assert provider.is_available() is False + + def test_not_available_with_invalid_key(self): + """Test that provider is unavailable when API returns 401.""" + provider = self._get_provider() + + response = MagicMock() + response.status_code = 401 + + with patch("src.ccnotify.tts.sixtydb.requests") as mock_requests: + mock_requests.get.return_value = response + assert provider.is_available() is False + + def test_available_with_valid_key(self): + """Test that provider is available when API returns 200.""" + provider = self._get_provider() + + response = MagicMock() + response.status_code = 200 + + with patch("src.ccnotify.tts.sixtydb.requests") as mock_requests: + mock_requests.get.return_value = response + assert provider.is_available() is True + + # Verify Bearer auth header is used + _, kwargs = mock_requests.get.call_args + assert kwargs["headers"]["Authorization"] == "Bearer test-key" + + def test_generate_success(self, mock_response_ok, tmp_path): + """Test successful audio generation.""" + provider = self._get_provider() + output_path = tmp_path / "test.mp3" + + with patch("src.ccnotify.tts.sixtydb.requests") as mock_requests: + mock_requests.post.return_value = mock_response_ok + + # Skip network availability check + with patch.object(provider, "is_available", return_value=True): + result = provider.generate("Hello world", output_path) + + assert result is True + assert output_path.read_bytes() == b"fake-mp3-audio-data" + + # Verify request payload + _, kwargs = mock_requests.post.call_args + assert kwargs["json"]["text"] == "Hello world" + assert kwargs["json"]["voice_id"] == "fbb75ed2-975a-40c7-9e06-38e30524a9a1" + assert "model" not in kwargs["json"] # Empty model_id = omit from payload + assert kwargs["headers"]["Authorization"] == "Bearer test-key" + + def test_generate_failure_status(self, tmp_path): + """Test generation failure on non-200 API response.""" + provider = self._get_provider() + output_path = tmp_path / "test.mp3" + + response = MagicMock() + response.status_code = 500 + response.json.return_value = {"message": "Internal error"} + + with patch("src.ccnotify.tts.sixtydb.requests") as mock_requests: + mock_requests.post.return_value = response + + with patch.object(provider, "is_available", return_value=True): + result = provider.generate("Hello world", output_path) + + assert result is False + assert not output_path.exists() + + def test_generate_failure_success_false(self, tmp_path): + """Test generation failure when API reports success=false.""" + provider = self._get_provider() + output_path = tmp_path / "test.mp3" + + response = MagicMock() + response.status_code = 200 + response.json.return_value = {"success": False, "message": "Voice not found"} + + with patch("src.ccnotify.tts.sixtydb.requests") as mock_requests: + mock_requests.post.return_value = response + + with patch.object(provider, "is_available", return_value=True): + result = provider.generate("Hello world", output_path) + + assert result is False + assert not output_path.exists() + + def test_generate_empty_audio(self, tmp_path): + """Test generation failure when API returns empty audio.""" + provider = self._get_provider() + output_path = tmp_path / "test.mp3" + + response = MagicMock() + response.status_code = 200 + response.json.return_value = {"success": True, "audio_base64": ""} + + with patch("src.ccnotify.tts.sixtydb.requests") as mock_requests: + mock_requests.post.return_value = response + + with patch.object(provider, "is_available", return_value=True): + result = provider.generate("Hello world", output_path) + + assert result is False + assert not output_path.exists() + + def test_generate_with_model_id(self, mock_response_ok, tmp_path): + """Test that model is included in payload when set.""" + provider = self._get_provider() + provider.model_id = "quality" + output_path = tmp_path / "test.mp3" + + with patch("src.ccnotify.tts.sixtydb.requests") as mock_requests: + mock_requests.post.return_value = mock_response_ok + + with patch.object(provider, "is_available", return_value=True): + result = provider.generate("Hello world", output_path) + + assert result is True + + _, kwargs = mock_requests.post.call_args + assert kwargs["json"]["model"] == "quality" + + def test_factory_config_builder(self): + """Test that _build_provider_config reads SIXTYDB_* env vars.""" + from src.ccnotify.tts.factory import _build_provider_config + + with patch("src.ccnotify.tts.factory.os.getenv") as mock_getenv: + # Env layer: SIXTYDB_API_KEY takes precedence + mock_getenv.side_effect = lambda key, default=None: { + "SIXTYDB_API_KEY": "env-key", + "SIXTYDB_VOICE_ID": "env-voice", + "SIXTYDB_MODEL_ID": "quality", + "SIXTYDB_SPEED": "1.5", + "SIXTYDB_STABILITY": "60", + "SIXTYDB_SIMILARITY": "80", + }.get(key, default) + + config = _build_provider_config("60db", {}) + + assert config["api_key"] == "env-key" + assert config["voice_id"] == "env-voice" + assert config["model_id"] == "quality" + assert config["speed"] == "1.5" + assert config["stability"] == "60" + assert config["similarity"] == "80" + + def test_get_available_voices(self): + """Test voice listing parses the 60db response shape.""" + provider = self._get_provider() + + response = MagicMock() + response.status_code = 200 + response.json.return_value = { + "success": True, + "data": [ + { + "voice_id": "fbb75ed2-975a-40c7-9e06-38e30524a9a1", + "name": "Zara", + "model": "60db Fast", + "labels": {"language": "english", "gender": "female"}, + } + ], + "meta": {"model": "fast"}, + } + + with patch("src.ccnotify.tts.sixtydb.requests") as mock_requests: + mock_requests.get.return_value = response + + with patch.object(provider, "is_available", return_value=True): + voices = provider.get_available_voices() + + assert "60db Fast" in voices + assert voices["60db Fast"][0]["name"] == "Zara" + assert voices["60db Fast"][0]["voice_id"] == "fbb75ed2-975a-40c7-9e06-38e30524a9a1" + + def test_default_config_section(self): + """Test that get_default_config includes a sixtydb section.""" + from src.ccnotify.config import get_default_config + + config = get_default_config() + assert "sixtydb" in config + assert config["sixtydb"]["voice_id"] == "fbb75ed2-975a-40c7-9e06-38e30524a9a1" + assert config["sixtydb"]["stability"] == 50 + assert config["sixtydb"]["similarity"] == 75 + + +class TestNotifyEmbeddedSixtyDB: + """Tests for the embedded provider in notify.py (standalone hook script).""" + + def test_embedded_provider_class_exists(self): + """Test that notify.py embeds a SixtyDBTTSProvider class.""" + from src.ccnotify.notify import SixtyDBTTSProvider + + provider = SixtyDBTTSProvider() + assert provider.voice_id == "fbb75ed2-975a-40c7-9e06-38e30524a9a1" + assert provider.get_file_extension() == ".mp3" + + def test_embedded_provider_not_available_without_key(self): + from src.ccnotify.notify import SixtyDBTTSProvider + + provider = SixtyDBTTSProvider() + assert provider.is_available() is False + + def test_embedded_provider_generate(self, mock_response_ok, tmp_path): + from src.ccnotify.notify import SixtyDBTTSProvider + + provider = SixtyDBTTSProvider() + provider.api_key = "test-key" + output_path = tmp_path / "test.mp3" + + with patch("src.ccnotify.notify.requests", create=True) as _: + pass # requests is imported lazily inside generate + + import types + fake_requests = types.SimpleNamespace(post=MagicMock(return_value=mock_response_ok)) + + with patch.dict("sys.modules", {"requests": fake_requests}): + result = provider.generate("Hello world", output_path) + + assert result is True + assert output_path.read_bytes() == b"fake-mp3-audio-data" + + +if __name__ == "__main__": + pytest.main([__file__, "-v"])