""" Genesis Talking Website Widget - Voice Handler Telnyx integration for voice conversations (TTS/STT) """ import os import base64 import io import tempfile from typing import Dict, Any, Optional, Tuple import telnyx import httpx from datetime import datetime class VoiceHandler: """Handler for Telnyx voice operations (TTS, STT, WebRTC).""" def __init__(self): self.api_key = os.getenv("TELNYX_API_KEY", "KEY019BE7A3A2D749FCA8681CFF8448A7F0_vTMM1n77CtQxLDT2ra3P1z") self.voice = os.getenv("TELNYX_VOICE", "eucalyptus") # Australian female self.model = os.getenv("TELNYX_MODEL", "google/gemini-2.5-flash") telnyx.api_key = self.api_key async def process_voice_input(self, audio_base64: str) -> str: """ Convert audio to text using Telnyx STT. Args: audio_base64: Base64 encoded audio (WAV/MP3) Returns: Transcribed text """ try: # Decode base64 audio audio_bytes = base64.b64decode(audio_base64) # For MVP: Use a simple STT service # Telnyx doesn't have built-in STT, so we'd use: # - OpenAI Whisper API # - Google Cloud Speech-to-Text # - AssemblyAI # For now, placeholder for STT integration # TODO: Integrate actual STT service transcription = "[STT_PLACEHOLDER] User speech would be transcribed here" return transcription except Exception as e: print(f"Voice input processing error: {e}") return "" async def generate_voice_response( self, text: str, business_id: str, voice_name: Optional[str] = None ) -> str: """ Convert text to speech using Telnyx TTS. Args: text: Text to convert to speech business_id: Business identifier voice_name: Optional custom voice (defaults to eucalyptus) Returns: Base64 encoded audio (WAV) """ try: # Use Telnyx Text-to-Speech API # Endpoint: POST https://api.telnyx.com/v2/text-to-speech voice = voice_name or self.voice url = "https://api.telnyx.com/v2/text-to-speech" headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json" } payload = { "voice": f"Telnyx.NaturalHD.{voice}", "text": text, "output_format": "wav" } async with httpx.AsyncClient() as client: response = await client.post(url, json=payload, headers=headers, timeout=30.0) if response.status_code == 200: # Telnyx returns audio file directly audio_bytes = response.content audio_base64 = base64.b64encode(audio_bytes).decode('utf-8') return audio_base64 else: print(f"TTS API error: {response.status_code} - {response.text}") # Fallback: Return empty (frontend will show text) return "" except Exception as e: print(f"Voice generation error: {e}") return "" async def generate_ai_response( self, visitor_message: str, business_context: Dict[str, Any], session_context: Dict[str, Any] ) -> Tuple[str, Dict[str, Any]]: """ Generate AI response using Gemini or other LLM. Args: visitor_message: What the visitor said/typed business_context: Business knowledge base session_context: Current session state Returns: (response_text, updated_session_context) """ try: # For MVP, use a simple rule-based system # In production, integrate with: # - Google Gemini 2.5 Flash (Kinan has credits) # - OpenRouter Claude/GPT # - Genesis local LLM business_name = business_context.get("name", "our business") agent_name = business_context.get("agent_name", "Sarah") # Simple intent detection message_lower = visitor_message.lower() if any(word in message_lower for word in ["hours", "open", "close"]): response = f"We're open Monday to Friday, 9 AM to 5 PM. How else can I help you?" elif any(word in message_lower for word in ["price", "cost", "how much"]): response = f"I'd be happy to discuss pricing with you. Can I get your name and phone number so one of our team can follow up?" session_context["intent"] = "pricing_inquiry" session_context["lead_capture_ready"] = True elif any(word in message_lower for word in ["book", "appointment", "schedule"]): response = f"I can help you book an appointment! What's your name and phone number?" session_context["intent"] = "booking" session_context["lead_capture_ready"] = True elif any(word in message_lower for word in ["hi", "hello", "hey"]): response = f"Hi there! I'm {agent_name}, the virtual assistant for {business_name}. How can I help you today?" else: response = f"Thanks for your message. Let me connect you with the right person. Can I get your name and phone number?" session_context["intent"] = "general_inquiry" session_context["lead_capture_ready"] = True # Update session context session_context["last_message"] = visitor_message session_context["last_response"] = response session_context["message_count"] = session_context.get("message_count", 0) + 1 return response, session_context except Exception as e: print(f"AI response generation error: {e}") return "I'm having trouble processing that. Can I get your phone number so someone can call you back?", session_context async def detect_lead_info( self, message: str, session_context: Dict[str, Any] ) -> Optional[Dict[str, str]]: """ Extract lead information from visitor message. Args: message: Visitor's message session_context: Current session state Returns: Dict with name, phone, email if detected """ import re lead_info = {} # Simple regex for phone extraction (Australian format) phone_patterns = [ r'\b0[2-478]\d{8}\b', # Australian landline/mobile r'\b\+61\s?[2-478]\s?\d{4}\s?\d{4}\b', # International format r'\b\d{4}\s?\d{3}\s?\d{3}\b' # Common formatting ] for pattern in phone_patterns: match = re.search(pattern, message) if match: lead_info["phone"] = match.group(0) break # Email extraction email_match = re.search(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', message) if email_match: lead_info["email"] = email_match.group(0) # Name extraction (basic - looks for "my name is X" or "I'm X") name_patterns = [ r"(?:my name is|i'm|i am|this is)\s+([A-Z][a-z]+(?:\s+[A-Z][a-z]+)?)", r"^([A-Z][a-z]+(?:\s+[A-Z][a-z]+)?)(?:\s+here|\s+calling)" ] for pattern in name_patterns: match = re.search(pattern, message, re.IGNORECASE) if match: lead_info["name"] = match.group(1).strip() break return lead_info if lead_info else None def test_connection(self) -> bool: """Test Telnyx API connectivity.""" try: # Test by listing available voices url = "https://api.telnyx.com/v2/text-to-speech/voices" headers = {"Authorization": f"Bearer {self.api_key}"} import requests response = requests.get(url, headers=headers, timeout=10) return response.status_code == 200 except Exception as e: print(f"Telnyx connection failed: {e}") return False # Global voice handler instance voice_handler = VoiceHandler() if __name__ == "__main__": import asyncio print("Genesis Widget API - Voice Handler Test") print("=" * 60) # Test connection print("Testing Telnyx API connection...") if voice_handler.test_connection(): print("✅ Connected to Telnyx") else: print("❌ Connection failed") exit(1) # Test TTS async def test_tts(): print("\nTesting Text-to-Speech...") text = "Hi! This is Sarah from ReceptionistAI. This is a test of the Telnyx voice system." audio_base64 = await voice_handler.generate_voice_response(text, "test_business") if audio_base64: print(f"✅ Generated {len(audio_base64)} bytes of audio (base64)") else: print("⚠️ TTS returned empty (check API key/quota)") asyncio.run(test_tts()) print("\n🎙️ Voice handler ready")