interface.py

import asyncio
import os
import time
from dataclasses import dataclass
from typing import List, Optional, AsyncGenerator
import gradio as gr
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from rich.console import Console
from rich.panel import Panel
from rich.text import Text
from logger import setup_logger, log_execution_time, log_async_execution_time

from browser_use import Agent, Browser
from browser_use.browser.browser import BrowserContext
from api_clients import OpenRouterClient, ElevenLabsClient

load_dotenv()

console = Console()
logger = setup_logger("interface")

@dataclass
class ActionResult:
	is_done: bool
	extracted_content: Optional[str]
	error: Optional[str]
	include_in_memory: bool


@dataclass
class AgentHistoryList:
	all_results: List[ActionResult]
	all_model_outputs: List[dict]


def parse_agent_history(history_str: str) -> None:
	# Split the content into sections based on ActionResult entries
	sections = history_str.split('ActionResult(')

	for i, section in enumerate(sections[1:], 1):  # Skip first empty section
		# Extract relevant information
		content = ''
		if 'extracted_content=' in section:
			content = section.split('extracted_content=')[1].split(',')[0].strip("'")

		if content:
			header = Text(f'Step {i}', style='bold blue')
			panel = Panel(content, title=header, border_style='blue')
			console.print(panel)
			console.print()


async def run_browser_task(
	task: str,
	api_key: str,
	provider: str = 'openai',
	model: str = 'gpt-4-vision',
	headless: bool = True,
) -> str:
	if not api_key.strip():
		return 'Please provide an API key'

	if provider == 'openai':
		os.environ['OPENAI_API_KEY'] = api_key
		llm = ChatOpenAI(model=model)
	elif provider == 'anthropic':
		os.environ['ANTHROPIC_API_KEY'] = api_key
		llm = ChatAnthropic(model=model)
	else:  # google
		os.environ['GOOGLE_API_KEY'] = api_key
		llm = ChatGoogleGenerativeAI(model=model)

	try:
		agent = Agent(
			task=task,
			llm=llm,
			browser=Browser(BrowserContext(headless=True))
		)
		result = await agent.run()
		#  TODO: The result cloud be parsed better
		return result
	except Exception as e:
		return f'Error: {str(e)}'


@log_async_execution_time(logger)
async def scrape_content(url: str) -> str:
    """
    Scrape and summarize content from the given URL using browser automation
    
    This function performs the following steps:
    1. Validates the input URL
    2. Initializes the browser agent
    3. Extracts and summarizes the content
    
    Args:
        url: Target URL to scrape
        
    Returns:
        Summarized content suitable for podcast generation
        
    Raises:
        ValueError: If URL is invalid or content extraction fails
    """
    logger.info(f"Starting content scrape for URL: {url}")
    
    # Input validation
    if not url.startswith(('http://', 'https://')):
        logger.error(f"Invalid URL format: {url}")
        raise ValueError("URL must start with http:// or https://")
    
    try:
        logger.debug("Initializing LLM and browser agent")
        llm = ChatOpenAI(model="gpt-4")
        agent = Agent(
            task=f"Visit this URL: {url} and extract the main content. Summarize it in a clear and concise way.",
            llm=llm,
            browser=Browser(BrowserContext(headless=True))
        )
        
        logger.info("Executing content extraction")
        result = await agent.run()
        
        logger.debug(f"Content extraction successful. Length: {len(result)} chars")
        logger.debug(f"Content preview: {result[:200]}...")
        
        return result
    except Exception as e:
        logger.error(f"Content extraction failed for {url}", exc_info=True)
        raise

@log_async_execution_time(logger)
async def create_podcast(
    url: str,
    prompt: str,
    elevenlabs_key: str,
    voice_id: str,
    openrouter_key: str,
    model_id: str,
) -> AsyncGenerator[tuple[Optional[str], str], None]:
    """
    Create a podcast through a multi-step process:
    1. Content extraction from URL
    2. Script generation using AI
    3. Voice synthesis
    
    Progress updates are yielded at each step for UI feedback.
    """
    logger.info(f"Starting podcast creation for URL: {url}")
    logger.debug(f"Parameters - Voice: {voice_id}, Model: {model_id}")
    logger.debug(f"Prompt length: {len(prompt)} chars")
    
    try:
        # Initialize clients with validation
        logger.debug("Initializing API clients")
        openrouter = OpenRouterClient(openrouter_key)
        elevenlabs = ElevenLabsClient(elevenlabs_key)
        
        # Phase 1: Content scraping
        logger.info("Phase 1/3: Content scraping")
        yield None, "Scraping website content..."
        content = await scrape_content(url)
        logger.debug(f"Scraped content length: {len(content)} chars")
        
        # Phase 2: Script generation
        logger.info("Phase 2/3: Script generation")
        yield None, "Generating podcast script..."
        script = await openrouter.generate_script(content, prompt, model_id)
        logger.debug(f"Generated script length: {len(script)} chars")
        
        # Phase 3: Audio synthesis
        logger.info("Phase 3/3: Audio generation")
        yield None, "Converting to audio..."
        audio = elevenlabs.generate_audio(script, voice_id)
        logger.debug(f"Generated audio size: {len(audio)} bytes")
        
        # Save output
        audio_path = f"podcast_{int(time.time())}.mp3"
        logger.debug(f"Saving audio to: {audio_path}")
        with open(audio_path, "wb") as f:
            f.write(audio)
        
        logger.info("Podcast creation completed successfully")
        yield audio_path, "Podcast created successfully!"
        
    except Exception as e:
        logger.error("Podcast creation failed", exc_info=True)
        yield None, f"Error: {str(e)}"

def create_ui():
    logger.info("Initializing Gradio interface")
    
    # Default choices for dropdowns
    default_voices = [("", "Enter API key to load voices")]
    default_models = [("", "Enter API key to load models")]
    
    with gr.Blocks(title='PodcastCreator', theme=gr.themes.Soft()) as interface:
        with gr.Row():
            with gr.Column(scale=2):
                url_input = gr.Textbox(label='Source URL', placeholder='Enter the URL...')
                prompt = gr.Textbox(label='Podcast Topic', lines=3)
                
                with gr.Row():
                    with gr.Column():
                        elevenlabs_key = gr.Textbox(
                            label='ElevenLabs API Key',
                            type='password',
                            placeholder='Enter key...'
                        )
                        voice = gr.Dropdown(
                            label='Voice',
                            choices=default_voices,
                            value=None,
                            allow_custom_value=True
                        )
                    
                    with gr.Column():
                        openrouter_key = gr.Textbox(
                            label='OpenRouter API Key',
                            type='password',
                            placeholder='Enter key...'
                        )
                        model = gr.Dropdown(
                            label='AI Model',
                            choices=default_models,
                            value=None,
                            allow_custom_value=True
                        )
                
                submit_btn = gr.Button('Create Podcast', variant='primary')

            with gr.Column(scale=1):
                audio_output = gr.Audio(label="Generated Podcast")
                status = gr.Textbox(label='Status', interactive=False)

        # Event handlers
        def update_voices(key):
            if not key:
                return gr.Dropdown(choices=default_voices, value=default_voices[0][0])
            try:
                client = ElevenLabsClient(key)
                voices = client.get_voices()
                return gr.Dropdown(choices=voices, value=voices[0][0] if voices else None)
            except Exception as e:
                logger.error(f"Failed to load voices: {e}")
                return gr.Dropdown(choices=[(None, f"Error: {str(e)}")], value=None)

        async def update_models(key):
            if not key:
                return gr.Dropdown(choices=default_models, value=default_models[0][0])
            try:
                client = OpenRouterClient(key)
                models = await client.get_models()
                return gr.Dropdown(choices=models, value=models[0][0] if models else None)
            except Exception as e:
                logger.error(f"Failed to load models: {e}")
                return gr.Dropdown(choices=[(None, f"Error: {str(e)}")], value=None)

        # Add error handling for the event handlers
        try:
            elevenlabs_key.change(fn=update_voices, inputs=elevenlabs_key, outputs=voice)
            openrouter_key.change(fn=update_models, inputs=openrouter_key, outputs=model)
            
            submit_btn.click(
                fn=create_podcast,
                inputs=[url_input, prompt, elevenlabs_key, voice, openrouter_key, model],
                outputs=[audio_output, status]
            )
        except Exception as e:
            logger.error(f"Failed to set up event handlers: {e}")
            raise

    logger.info("Gradio interface initialized successfully")
    return interface

if __name__ == '__main__':
    demo = create_ui()
    demo.launch()