from functools import lru_cache
from typing import List, Tuple, Optional
import aiohttp
import elevenlabs
import time
from contextlib import asynccontextmanager
from logger import setup_logger, log_execution_time, log_async_execution_time
from models import OpenRouterModel
logger = setup_logger("api_clients")
class OpenRouterClient:
"""Handles OpenRouter API interactions with comprehensive logging and error tracking"""
def __init__(self, api_key: str):
logger.info("Initializing OpenRouter client")
self.api_key = api_key
self.base_url = "https://openrouter.ai/api/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"HTTP-Referer": "https://localhost:7860", # Required by OpenRouter
"X-Title": "URL to Podcast Generator", # Required by OpenRouter
"Content-Type": "application/json"
}
logger.debug("OpenRouter client initialized successfully")
@property
def api_key(self):
return self._api_key
@api_key.setter
def api_key(self, value: str):
if not value or len(value) < 32:
logger.error("Invalid API key format")
raise ValueError("Invalid OpenRouter API key")
self._api_key = value
# Update headers when API key changes
self.headers = {
"Authorization": f"Bearer {value}",
"HTTP-Referer": "https://localhost:7860",
"X-Title": "URL to Podcast Generator",
"Content-Type": "application/json",
}
logger.info("OpenRouter API key updated successfully")
@asynccontextmanager
async def get_session(self):
logger.debug("Creating new aiohttp session")
async with aiohttp.ClientSession(headers=self.headers) as session:
yield session
@lru_cache(maxsize=1)
async def get_models(self) -> List[Tuple[str, str]]:
"""
Fetch available models from OpenRouter API using pydantic models
Returns:
List of tuples containing (model_id, model_id) where both values are the same
"""
logger.info("Fetching available models from OpenRouter")
async with self.get_session() as session:
async with session.get(f"{self.base_url}/models") as response:
response.raise_for_status()
data = await response.json()
models = [OpenRouterModel(**model) for model in data["data"]]
logger.info(f"Successfully fetched {len(models)} models")
return [(model.name, model.id) for model in models]
@log_async_execution_time(logger)
async def generate_script(self, content: str, prompt: str, model_id: str) -> str:
"""
Generate a podcast script with detailed progress tracking and validation
Performance metrics and content analysis are logged at each step.
"""
logger.info(f"Starting script generation with model: {model_id}")
logger.debug(f"Input metrics - Content: {len(content)} chars, Prompt: {len(prompt)} chars")
# Validate inputs
if not content or len(content) < 100:
logger.error("Content too short for meaningful script generation")
raise ValueError("Insufficient content for script generation")
if not prompt or len(prompt) < 10:
logger.error("Prompt too short or missing")
raise ValueError("Please provide a more detailed prompt")
system_prompt = """DO NOT WRITE ASIDES OR ACTION DESCRIPTIONS, YOU WRITE DIALOG ONLY!!. You are an expert podcast dialog writer with these specific requirements:
1. Start the content immediately - no introductions, timestamps, or meta-commentary
2. Write in a natural, conversational tone suitable for speaking
3. Structure the podcast dialog with clear paragraphs and natural pauses
4. Use informal language while maintaining professionalism
5. Focus on narrative flow and engaging delivery
6. Keep technical terms simple and explained
7. Include vocal variety cues through punctuation
8. Write as if speaking directly to the listener
9. Use storytelling techniques to maintain interest
10. Do not add muscial queues or sound effects
11. Add host and show intros, outros, and transitions as needed
"""
user_prompt = f"""Write podcast dialog for a single person based on the following content. Make it engaging and easy to follow.
Context: {prompt if prompt else 'Create an informative and engaging podcast episode'}
Content:
{content}
Format the dialog in a clear, readable way with appropriate spacing. Do not add asides or action descriptions. Only add spoken dialog."""
try:
request_data = {
"model": model_id,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
"temperature": 0.7,
"max_tokens": 2000
}
async with self.get_session() as session:
async with session.post(
f"{self.base_url}/chat/completions",
json=request_data
) as response:
if response.status != 200:
error_text = await response.text()
logger.error(f"OpenRouter API error: {error_text}")
raise ValueError(f"API request failed: {error_text}")
data = await response.json()
return data['choices'][0]['message']['content']
except Exception as e:
logger.error(f"Script generation failed", exc_info=True)
raise
class ElevenLabsClient:
def __init__(self, api_key: str):
self.api_key = api_key
elevenlabs.set_api_key(api_key)
def get_voices(self) -> List[Tuple[str, str]]:
"""
Synchronously get available voices from ElevenLabs
Returns:
List of tuples containing (voice_id, display_name)
where display_name shows the name and description but not the ID
"""
try:
voices = elevenlabs.voices()
return [(
f"{voice.name} ({voice.labels.get('accent', 'No accent')})" +
(f" - {voice.description[:50]}..." if voice.description else ""),
voice.voice_id # Value (hidden from user)
) for voice in voices]
except Exception as e:
logger.error("Failed to fetch voices from ElevenLabs", exc_info=True)
raise
def generate_audio(self, text: str, voice_id: str):
"""Generate audio synchronously"""
logger.info(f"Starting audio generation with voice: {voice_id}")
logger.debug(f"Input text length: {len(text)} chars")
if len(text) > 5000:
logger.warning(f"Long text detected ({len(text)} chars), may impact performance")
try:
start_time = time.time()
audio = elevenlabs.generate(
text=text,
voice=voice_id,
model="eleven_monolingual_v1"
)
duration = time.time() - start_time
audio_size = len(audio)
logger.info(f"Audio generated: {audio_size} bytes in {duration:.2f} seconds")
logger.debug(f"Audio generation rate: {len(text)/duration:.2f} chars/second")
return audio
except Exception as e:
logger.error("Audio generation failed", exc_info=True)
raise