|
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", |
|
"X-Title": "URL to Podcast Generator", |
|
"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 |
|
|
|
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") |
|
|
|
|
|
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 |
|
) 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 |
|
|