browser_agent.py

#!/usr/bin/env python3
"""
Browser Agent with Gemini 3 Flash Preview + Playwright
Human-in-the-loop browser automation for appointments, logins, etc.
"""

import asyncio
import base64
import sys
import os
from playwright.async_api import async_playwright

try:
    from google import genai
    from google.genai import types
except ImportError:
    print("Error: google-genai not installed. Run: pip install google-genai")
    sys.exit(1)


MODEL_ID = "gemini-2.0-flash"  # Gemini 3 Flash Preview


class BrowserAgent:
    def __init__(self, api_key: str):
        self.client = genai.Client(api_key=api_key)
        self.browser = None
        self.page = None
        self.context = None

    async def start(self, url: str, headless: bool = False):
        """Launch browser and navigate to URL."""
        self.playwright = await async_playwright().start()
        self.browser = await self.playwright.chromium.launch(headless=headless)
        self.context = await self.browser.new_context(
            viewport={"width": 1280, "height": 800}
        )
        self.page = await self.context.new_page()
        await self.page.goto(url, wait_until="domcontentloaded")
        print(f"\n🌐 Navigated to: {url}")

    async def screenshot(self) -> bytes:
        """Take a screenshot of current page."""
        return await self.page.screenshot(type="png", full_page=False)

    async def ask_gemini(self, screenshot_bytes: bytes, user_context: str = "") -> str:
        """Send screenshot to Gemini and get description."""

        prompt = """You are a browser assistant. Analyze this screenshot and:

1. Describe what you see on the page (main elements, forms, buttons, navigation)
2. If there's a login form, identify the username/password fields
3. If there's an appointment/booking interface, describe available options
4. Suggest what actions the user might want to take

Be concise but helpful. Focus on actionable elements."""

        if user_context:
            prompt += f"\n\nUser's previous instruction: {user_context}"

        # Create the image part
        image_part = types.Part.from_bytes(
            data=screenshot_bytes,
            mime_type="image/png"
        )

        response = self.client.models.generate_content(
            model=MODEL_ID,
            contents=[prompt, image_part]
        )

        return response.text

    async def execute_action(self, action: str) -> str:
        """Execute an action based on user input."""
        action = action.strip()

        # Parse commands
        if action.lower().startswith("goto ") or action.lower().startswith("go to "):
            url = action.split(" ", 2)[-1].strip()
            if not url.startswith("http"):
                url = "https://" + url
            await self.page.goto(url, wait_until="domcontentloaded")
            return f"Navigated to {url}"

        elif action.lower().startswith("click "):
            target = action[6:].strip()
            try:
                # Try multiple selectors
                element = None
                for selector_fn in [
                    lambda: self.page.get_by_text(target, exact=False).first,
                    lambda: self.page.get_by_role("button", name=target).first,
                    lambda: self.page.get_by_role("link", name=target).first,
                    lambda: self.page.locator(f"text={target}").first,
                ]:
                    try:
                        element = selector_fn()
                        if await element.is_visible(timeout=1000):
                            break
                    except:
                        continue

                if element:
                    await element.click()
                    await self.page.wait_for_load_state("domcontentloaded")
                    return f"Clicked on '{target}'"
                else:
                    return f"Could not find element: {target}"
            except Exception as e:
                return f"Error clicking: {e}"

        elif action.lower().startswith("type "):
            # Format: type "field name" text to type
            # or: type text (types in focused element)
            parts = action[5:].strip()
            if parts.startswith('"'):
                # Extract field name and text
                end_quote = parts.find('"', 1)
                if end_quote > 0:
                    field = parts[1:end_quote]
                    text = parts[end_quote+1:].strip()
                    try:
                        await self.page.get_by_placeholder(field).first.fill(text)
                        return f"Typed in '{field}'"
                    except:
                        try:
                            await self.page.get_by_label(field).first.fill(text)
                            return f"Typed in '{field}'"
                        except Exception as e:
                            return f"Could not find field '{field}': {e}"
            else:
                # Just type in focused element
                await self.page.keyboard.type(parts)
                return "Typed text"

        elif action.lower().startswith("login "):
            # Format: login username password
            parts = action.split()
            if len(parts) >= 3:
                username = parts[1]
                password = parts[2]

                # Try common login form patterns
                try:
                    # Find username field
                    for sel in ["username", "email", "user", "login", "Username", "Email"]:
                        try:
                            field = self.page.get_by_placeholder(sel).first
                            if await field.is_visible(timeout=500):
                                await field.fill(username)
                                break
                        except:
                            continue
                    else:
                        # Try input types
                        await self.page.locator('input[type="text"], input[type="email"]').first.fill(username)

                    # Find password field
                    await self.page.locator('input[type="password"]').first.fill(password)

                    # Find submit button
                    for btn_text in ["Log in", "Login", "Sign in", "Submit", "Enter"]:
                        try:
                            btn = self.page.get_by_role("button", name=btn_text).first
                            if await btn.is_visible(timeout=500):
                                await btn.click()
                                break
                        except:
                            continue
                    else:
                        await self.page.locator('button[type="submit"], input[type="submit"]').first.click()

                    await self.page.wait_for_load_state("domcontentloaded")
                    return "Login attempted"
                except Exception as e:
                    return f"Login error: {e}"
            else:
                return "Usage: login <username> <password>"

        elif action.lower() == "back":
            await self.page.go_back()
            return "Went back"

        elif action.lower() == "forward":
            await self.page.go_forward()
            return "Went forward"

        elif action.lower() == "refresh":
            await self.page.reload()
            return "Refreshed page"

        elif action.lower().startswith("scroll "):
            direction = action[7:].strip().lower()
            if direction == "down":
                await self.page.keyboard.press("PageDown")
                return "Scrolled down"
            elif direction == "up":
                await self.page.keyboard.press("PageUp")
                return "Scrolled up"
            else:
                return "Use: scroll up/down"

        elif action.lower().startswith("select "):
            # Format: select "dropdown" option
            parts = action[7:].strip()
            if parts.startswith('"'):
                end_quote = parts.find('"', 1)
                if end_quote > 0:
                    dropdown = parts[1:end_quote]
                    option = parts[end_quote+1:].strip()
                    try:
                        await self.page.get_by_label(dropdown).first.select_option(label=option)
                        return f"Selected '{option}' from '{dropdown}'"
                    except Exception as e:
                        return f"Error selecting: {e}"
            return "Usage: select \"dropdown name\" option"

        elif action.lower() == "wait":
            await asyncio.sleep(2)
            return "Waited 2 seconds"

        elif action.lower().startswith("press "):
            key = action[6:].strip()
            await self.page.keyboard.press(key)
            return f"Pressed {key}"

        else:
            return f"Unknown command: {action}\n\nAvailable commands:\n" \
                   "  goto <url>       - Navigate to URL\n" \
                   "  click <text>     - Click element by text\n" \
                   "  type \"field\" text - Type in a field\n" \
                   "  login user pass  - Fill login form\n" \
                   "  scroll up/down   - Scroll page\n" \
                   "  back/forward     - Navigate history\n" \
                   "  refresh          - Reload page\n" \
                   "  select \"dropdown\" option - Select from dropdown\n" \
                   "  press <key>      - Press keyboard key\n" \
                   "  wait             - Wait 2 seconds\n" \
                   "  help             - Show this help\n" \
                   "  exit             - Close browser and quit"

    async def run_loop(self):
        """Main interaction loop."""
        print("\n" + "="*60)
        print("🤖 Browser Agent Ready")
        print("="*60)
        print("Commands: goto, click, type, login, scroll, back, forward, refresh, exit")
        print("Or describe what you want to do and I'll help!")
        print("="*60)

        while True:
            try:
                # Take screenshot and analyze
                print("\n📸 Analyzing page...")
                screenshot = await self.screenshot()
                description = await self.ask_gemini(screenshot)

                print("\n" + "-"*50)
                print("🔍 GEMINI SAYS:")
                print("-"*50)
                print(description)
                print("-"*50)

                # Get user input
                print("\n💬 What would you like to do?")
                user_input = input(">>> ").strip()

                if not user_input:
                    continue

                if user_input.lower() in ["exit", "quit", "q"]:
                    print("👋 Closing browser...")
                    break

                if user_input.lower() == "help":
                    await self.execute_action("help")
                    continue

                # Check if it's a direct command
                cmd_prefixes = ["goto", "go to", "click", "type", "login",
                               "scroll", "back", "forward", "refresh",
                               "select", "press", "wait"]

                is_command = any(user_input.lower().startswith(p) for p in cmd_prefixes)

                if is_command:
                    result = await self.execute_action(user_input)
                    print(f"\n✅ {result}")
                else:
                    # Natural language - ask Gemini what to do
                    print("\n🤔 Let me figure out how to do that...")

                    action_prompt = f"""Based on this screenshot and the user's request: "{user_input}"

What specific action should I take? Respond with ONE command from this list:
- goto <url>
- click <visible text or button name>
- type "field placeholder" text to enter
- scroll up/down
- press <key like Enter, Tab, etc>

Just respond with the command, nothing else."""

                    action_response = await self.ask_gemini(screenshot, action_prompt)
                    print(f"💡 Suggested action: {action_response.strip()}")

                    confirm = input("Execute this? (y/n): ").strip().lower()
                    if confirm in ["y", "yes", ""]:
                        result = await self.execute_action(action_response.strip())
                        print(f"\n✅ {result}")
                    else:
                        print("Skipped.")

            except KeyboardInterrupt:
                print("\n👋 Interrupted. Closing...")
                break
            except Exception as e:
                print(f"\n❌ Error: {e}")
                continue

        await self.close()

    async def close(self):
        """Clean up resources."""
        if self.browser:
            await self.browser.close()
        if hasattr(self, 'playwright'):
            await self.playwright.stop()


async def main():
    # Get API key
    api_key = os.environ.get("GEMINI_API_KEY")

    if not api_key:
        print("="*50)
        print("🔑 GEMINI API KEY REQUIRED")
        print("="*50)
        print("Set it with: export GEMINI_API_KEY='your-key-here'")
        print("Get your key at: https://aistudio.google.com/apikey")
        print("="*50)
        api_key = input("Or enter API key now: ").strip()
        if not api_key:
            print("No API key provided. Exiting.")
            return

    # Get starting URL
    if len(sys.argv) > 1:
        start_url = sys.argv[1]
    else:
        print("\n🌐 Enter starting URL (or press Enter for Google):")
        start_url = input(">>> ").strip()
        if not start_url:
            start_url = "https://www.google.com"
        elif not start_url.startswith("http"):
            start_url = "https://" + start_url

    # Run the agent
    agent = BrowserAgent(api_key)
    await agent.start(start_url)
    await agent.run_loop()


if __name__ == "__main__":
    asyncio.run(main())