browser.py

import subprocess
import asyncio
from contextlib import contextmanager
from typing import List, Optional
from threading import Lock
from smolagents.tools import Tool
from langchain_community.tools.playwright.utils import (
    create_async_playwright_browser,
)
from langchain_community.agent_toolkits import PlayWrightBrowserToolkit
import logging
from playwright.async_api import async_playwright

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

class BrowserManager:
    _instance = None
    _lock = Lock()
    _browser_tools: Optional[List[Tool]] = None
    _loop: Optional[asyncio.AbstractEventLoop] = None
    
    def __new__(cls):
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = super(BrowserManager, cls).__new__(cls)
        return cls._instance
    
    def __init__(self):
        if not hasattr(self, 'initialized'):
            # Run setup script
            subprocess.run(["bash", "scripts.sh"])
            self.initialized = True
    
    def _create_browser_tools(self):
        """Create browser tools in a new event loop"""
        # Create a new event loop for browser tools
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        
        try:
            logger.debug("Creating async browser...")
            # Create browsers in the new loop
            async def create_browser():
                playwright = await async_playwright().start()
                return await playwright.chromium.launch(headless=True)
            
            async_browser = loop.run_until_complete(create_browser())
            logger.debug("Async browser created successfully")
            
            # Create toolkit and tools
            logger.debug("Creating browser toolkit...")
            browser_toolkit = PlayWrightBrowserToolkit.from_browser(
                async_browser=async_browser,
                sync_browser=None  # Don't use sync browser
            )
            logger.debug("Browser toolkit created successfully")
            
            logger.debug("Converting tools...")
            tools = []
            for tool in browser_toolkit.get_tools():
                logger.debug(f"Converting tool: {tool.name}")
                try:
                    converted_tool = Tool.from_langchain(tool)
                    converted_tool._browser = async_browser
                    converted_tool._loop = loop
                    tools.append(converted_tool)
                    logger.debug(f"Successfully converted tool: {tool.name}")
                except Exception as e:
                    logger.error(f"Error converting tool {tool.name}: {e}")
                    raise
                
            logger.debug(f"Successfully created {len(tools)} browser tools")
            return tools
            
        except Exception as e:
            logger.error(f"Error creating browser tools: {e}")
            if loop.is_running():
                loop.stop()
            loop.close()
            raise
    
    @contextmanager
    def get_browser_tools(self):
        """Get browser tools in a context that ensures proper cleanup"""
        try:
            if self._browser_tools is None:
                with self._lock:
                    if self._browser_tools is None:
                        logger.debug("Creating new browser tools...")
                        self._browser_tools = self._create_browser_tools()
                        logger.debug("Browser tools created successfully")
            
            yield self._browser_tools
            
        except Exception as e:
            logger.error(f"Error in browser context: {e}")
            # Reset tools on error
            self._browser_tools = None
            raise
        finally:
            # Cleanup if needed
            if self._browser_tools:
                for tool in self._browser_tools:
                    if hasattr(tool, '_browser') and hasattr(tool, '_loop'):
                        try:
                            loop = tool._loop
                            if loop and not loop.is_closed():
                                logger.debug("Cleaning up browser...")
                                loop.run_until_complete(tool._browser.close())
                                loop.close()
                                logger.debug("Browser cleanup completed")
                        except Exception as e:
                            logger.error(f"Error during browser cleanup: {e}")
                self._browser_tools = None

# Create a singleton instance
browser_manager = BrowserManager()

# For backward compatibility, but prefer using browser_manager.get_browser_tools()
def get_browser_tools():
    """Get browser tools (use with context manager)"""
    return browser_manager.get_browser_tools()