Agent/tools/browserUseTools.py

"""
Browser-Use Tools Adapter
浏览器工具适配器

将 browser-use 库的工具适配到 Agent 框架中。
基于 browser-use 的 Action 定义实现了以下工具：
- ExtractAction: 内容提取
- SearchAction: 网页搜索
- NavigateAction: 页面导航
- ClickElementAction: 元素点击
- InputTextAction: 文本输入
- DoneAction: 任务完成
- SwitchTabAction: 标签切换
- CloseTabAction: 关闭标签
- ScrollAction: 页面滚动
- SendKeysAction: 键盘操作
- UploadFileAction: 文件上传
- GetDropdownOptionsAction: 获取下拉选项
- SelectDropdownOptionAction: 选择下拉选项

所有工具都使用 @tool() 装饰器自动注册到框架的工具注册表中。
"""

import sys
import os
from typing import Optional, List

# 将项目根目录添加到 Python 路径
# 这样可以正确导入 agent 模块
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

# 导入框架的工具装饰器和结果类
# tool: 用于注册工具的装饰器
# ToolResult: 工具执行结果的标准返回格式
from agent.tools import tool, ToolResult


# ============================================================
# 核心浏览器导航工具 (Core Browser Navigation Tools)
# 对应 browser-use 的 NavigateAction 和 GoBackEvent
# ============================================================

@tool()
async def navigate_to_url(url: str, new_tab: bool = False, uid: str = "") -> ToolResult:
    """
    导航到指定的 URL
    Navigate to a specific URL

    这个工具使用 Playwright 启动浏览器并导航到指定的网址。
    可以选择在新标签页中打开，或在当前标签页中打开。

    Args:
        url: 要访问的 URL 地址
        new_tab: 是否在新标签页中打开（默认 False）
        uid: 用户 ID（由框架自动注入，工具内部使用）

    Returns:
        ToolResult: 包含导航结果的工具返回对象
            - title: 操作标题
            - output: 成功打开的页面标题
            - long_term_memory: 简短的操作记录（用于 LLM 长期记忆）
            - metadata: 包含 url、title、new_tab 的元数据

    Example:
        navigate_to_url("https://www.baidu.com")
        navigate_to_url("https://www.google.com", new_tab=True)
    """
    try:
        # 导入 Playwright 异步 API
        from playwright.async_api import async_playwright

        # 使用异步上下文管理器启动 Playwright
        async with async_playwright() as p:
            # 启动 Chromium 浏览器（headless=False 表示显示浏览器窗口）
            browser = await p.chromium.launch(headless=False)
            # 创建浏览器上下文（类似于一个独立的浏览器会话）
            context = await browser.new_context()

            # 根据 new_tab 参数决定是否创建新标签页
            if new_tab:
                page = await context.new_page()
            else:
                # 使用现有标签页，如果没有则创建新的
                page = await context.pages()[0] if context.pages() else await context.new_page()

            # 导航到指定 URL
            await page.goto(url)
            # 等待页面完全加载（网络空闲状态）
            await page.wait_for_load_state("networkidle")

            # 获取页面标题
            title = await page.title()

            # 返回成功结果
            return ToolResult(
                title=f"Navigated to {url}",
                output=f"Successfully opened page: {title}",
                long_term_memory=f"Navigated to {url}",  # 简短记录，节省 token
                metadata={"url": url, "title": title, "new_tab": new_tab}
            )
    except Exception as e:
        # 捕获所有异常并返回错误结果
        return ToolResult(
            title="Navigation failed",
            output="",
            error=f"Failed to navigate to {url}: {str(e)}",
            long_term_memory=f"Navigation to {url} failed"
        )


@tool()
async def go_back(uid: str = "") -> ToolResult:
    """
    返回到上一个页面
    Go back to the previous page

    模拟浏览器的"后退"按钮功能。

    Args:
        uid: 用户 ID（由框架自动注入）

    Returns:
        ToolResult: 包含返回操作结果的工具返回对象

    Note:
        如果当前页面是历史记录的第一页，此操作可能会失败。
    """
    try:
        from playwright.async_api import async_playwright

        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=False)
            context = await browser.new_context()
            page = await context.pages()[0] if context.pages() else await context.new_page()

            # 执行后退操作
            await page.go_back()
            # 等待页面加载完成
            await page.wait_for_load_state("networkidle")

            return ToolResult(
                title="Went back",
                output="Successfully navigated back",
                long_term_memory="Navigated back to previous page"
            )
    except Exception as e:
        return ToolResult(
            title="Go back failed",
            output="",
            error=f"Failed to go back: {str(e)}",
            long_term_memory="Go back failed"
        )


# ============================================================
# 元素交互工具 (Element Interaction Tools)
# 对应 browser-use 的 ClickElementAction, InputTextAction, SendKeysAction
# ============================================================

@tool()
async def click_element(index: Optional[int] = None, coordinate_x: Optional[int] = None,
                       coordinate_y: Optional[int] = None, uid: str = "") -> ToolResult:
    """
    通过索引或坐标点击页面元素
    Click an element by index or coordinates

    支持两种点击方式：
    1. 通过坐标点击：提供 coordinate_x 和 coordinate_y
    2. 通过元素索引点击：提供 index（需要配合 DOM 状态使用）

    Args:
        index: 元素索引（从浏览器状态中获取，1-based）
        coordinate_x: 相对于视口左边缘的水平坐标（像素）
        coordinate_y: 相对于视口顶部的垂直坐标（像素）
        uid: 用户 ID（由框架自动注入）

    Returns:
        ToolResult: 包含点击操作结果的工具返回对象

    Example:
        # 通过坐标点击
        click_element(coordinate_x=100, coordinate_y=200)
        # 通过索引点击
        click_element(index=5)

    Note:
        - 必须提供 index 或 (coordinate_x, coordinate_y) 中的一种
        - 坐标点击更可靠，索引点击需要维护 DOM 状态映射
    """
    try:
        from playwright.async_api import async_playwright

        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=False)
            context = await browser.new_context()
            page = await context.pages()[0] if context.pages() else await context.new_page()

            # 方式1：通过坐标点击
            if coordinate_x is not None and coordinate_y is not None:
                await page.mouse.click(coordinate_x, coordinate_y)
                return ToolResult(
                    title="Clicked coordinate",
                    output=f"Clicked at ({coordinate_x}, {coordinate_y})",
                    long_term_memory=f"Clicked coordinate ({coordinate_x}, {coordinate_y})"
                )
            # 方式2：通过索引点击（需要 DOM 状态映射）
            elif index is not None:
                # 注意：这里需要 DOM 状态来将索引映射到实际的 CSS 选择器
                # 当前实现为占位符，实际使用时需要维护 DOM 状态
                return ToolResult(
                    title="Click by index",
                    output=f"Clicked element at index {index}",
                    long_term_memory=f"Clicked element {index}"
                )
            else:
                # 参数错误：必须提供一种点击方式
                return ToolResult(
                    title="Invalid parameters",
                    output="",
                    error="Must provide either index or coordinates",
                    long_term_memory="Click failed: invalid parameters"
                )
    except Exception as e:
        return ToolResult(
            title="Click failed",
            output="",
            error=f"Failed to click: {str(e)}",
            long_term_memory="Click failed"
        )


@tool()
async def input_text(index: int, text: str, clear: bool = True, uid: str = "") -> ToolResult:
    """
    在指定元素中输入文本
    Input text into an element

    Args:
        index: 元素索引（从浏览器状态中获取，0-based）
        text: 要输入的文本内容
        clear: 是否先清除现有文本（默认 True）
        uid: 用户 ID（由框架自动注入）

    Returns:
        ToolResult: 包含输入操作结果的工具返回对象

    Example:
        # 清除后输入
        input_text(index=0, text="Hello World", clear=True)
        # 追加输入
        input_text(index=0, text=" More text", clear=False)

    Note:
        当前实现使用通用键盘输入方式，实际使用时需要配合 DOM 状态
        将索引映射到具体的输入框选择器。
    """
    try:
        from playwright.async_api import async_playwright

        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=False)
            context = await browser.new_context()
            page = await context.pages()[0] if context.pages() else await context.new_page()

            # 注意：这里需要 DOM 状态来将索引映射到实际的输入框选择器
            # 当前使用通用键盘输入方式
            if clear:
                # 先全选（Ctrl+A）再输入，实现清除效果
                await page.keyboard.press("Control+A")
            # 输入文本
            await page.keyboard.type(text)

            return ToolResult(
                title="Input text",
                output=f"Input text into element {index}",
                long_term_memory=f"Input text into element {index}",
                metadata={"index": index, "clear": clear}
            )
    except Exception as e:
        return ToolResult(
            title="Input failed",
            output="",
            error=f"Failed to input text: {str(e)}",
            long_term_memory="Input text failed"
        )


@tool()
async def send_keys(keys: str, uid: str = "") -> ToolResult:
    """
    发送键盘按键或快捷键
    Send keyboard keys or shortcuts

    支持发送单个按键、组合键和快捷键。

    Args:
        keys: 要发送的按键字符串
              - 单个按键: "Enter", "Escape", "PageDown", "Tab"
              - 组合键: "Control+o", "Shift+Tab", "Alt+F4"
              - 功能键: "F1", "F2", ..., "F12"
        uid: 用户 ID（由框架自动注入）

    Returns:
        ToolResult: 包含按键操作结果的工具返回对象

    Example:
        send_keys("Enter")           # 回车键
        send_keys("Control+o")       # Ctrl+O 打开文件
        send_keys("PageDown")        # 向下翻页
        send_keys("Escape")          # ESC 键

    Note:
        按键名称遵循 Playwright 的键盘 API 规范。
        参考: https://playwright.dev/python/docs/api/class-keyboard
    """
    try:
        from playwright.async_api import async_playwright

        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=False)
            context = await browser.new_context()
            page = await context.pages()[0] if context.pages() else await context.new_page()

            # 发送按键
            await page.keyboard.press(keys)

            return ToolResult(
                title="Sent keys",
                output=f"Sent keys: {keys}",
                long_term_memory=f"Sent keys: {keys}"
            )
    except Exception as e:
        return ToolResult(
            title="Send keys failed",
            output="",
            error=f"Failed to send keys: {str(e)}",
            long_term_memory="Send keys failed"
        )


# ============================================================
# Content Extraction Tools
# ============================================================

@tool()
async def extract_content(query: str, extract_links: bool = False,
                         start_from_char: int = 0, uid: str = "") -> ToolResult:
    """
    Extract content from the current page based on a query

    Args:
        query: What to extract from the page
        extract_links: Whether to extract links (default: False, saves tokens)
        start_from_char: Start extraction from specific character (for long content)
        uid: User ID (auto-injected)

    Returns:
        Extracted content
    """
    try:
        from playwright.async_api import async_playwright

        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=False)
            context = await browser.new_context()
            page = await context.pages()[0] if context.pages() else await context.new_page()

            # Extract text content
            content = await page.content()
            text_content = await page.inner_text("body")

            # Apply start_from_char if specified
            if start_from_char > 0:
                text_content = text_content[start_from_char:]

            # Extract links if requested
            links = []
            if extract_links:
                link_elements = await page.query_selector_all("a[href]")
                for elem in link_elements[:50]:  # Limit to 50 links
                    href = await elem.get_attribute("href")
                    text = await elem.inner_text()
                    if href:
                        links.append({"text": text, "href": href})

            output = f"Query: {query}\n\nContent:\n{text_content[:2000]}"
            if extract_links and links:
                output += f"\n\nLinks found: {len(links)}"

            return ToolResult(
                title=f"Extracted: {query}",
                output=output,
                long_term_memory=f"Extracted content for query: {query}",
                include_output_only_once=True,
                metadata={"query": query, "links": links if extract_links else []}
            )
    except Exception as e:
        return ToolResult(
            title="Extraction failed",
            output="",
            error=f"Failed to extract content: {str(e)}",
            long_term_memory="Content extraction failed"
        )


# ============================================================
# Search Tools
# ============================================================

@tool()
async def search_web(query: str, engine: str = "duckduckgo", uid: str = "") -> ToolResult:
    """
    Search the web using a search engine

    Args:
        query: Search query
        engine: Search engine to use (duckduckgo, google, bing) - default: duckduckgo
        uid: User ID (auto-injected)

    Returns:
        Search results
    """
    try:
        from playwright.async_api import async_playwright

        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=False)
            context = await browser.new_context()
            page = await context.new_page()

            # Navigate to search engine
            if engine == "google":
                await page.goto(f"https://www.google.com/search?q={query}")
            elif engine == "bing":
                await page.goto(f"https://www.bing.com/search?q={query}")
            else:  # duckduckgo
                await page.goto(f"https://duckduckgo.com/?q={query}")

            await page.wait_for_load_state("networkidle")

            # Extract search results
            results_text = await page.inner_text("body")

            await browser.close()

            return ToolResult(
                title=f"Search: {query}",
                output=f"Search results from {engine}:\n{results_text[:2000]}",
                long_term_memory=f"Searched {engine} for: {query}",
                include_output_only_once=True,
                metadata={"query": query, "engine": engine}
            )
    except Exception as e:
        return ToolResult(
            title="Search failed",
            output="",
            error=f"Search failed: {str(e)}",
            long_term_memory=f"Search for '{query}' failed"
        )


# ============================================================
# Scroll Tools
# ============================================================

@tool()
async def scroll_page(down: bool = True, pages: float = 1.0,
                     index: Optional[int] = None, uid: str = "") -> ToolResult:
    """
    Scroll the page or a specific element

    Args:
        down: True to scroll down, False to scroll up
        pages: Number of pages to scroll (0.5=half page, 1=full page, 10=to bottom/top)
        index: Optional element index to scroll within specific element
        uid: User ID (auto-injected)

    Returns:
        Scroll result
    """
    try:
        from playwright.async_api import async_playwright

        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=False)
            context = await browser.new_context()
            page = await context.pages()[0] if context.pages() else await context.new_page()

            # Calculate scroll amount
            viewport_height = page.viewport_size["height"] if page.viewport_size else 800
            scroll_amount = int(viewport_height * pages)

            if down:
                await page.mouse.wheel(0, scroll_amount)
                direction = "down"
            else:
                await page.mouse.wheel(0, -scroll_amount)
                direction = "up"

            return ToolResult(
                title=f"Scrolled {direction}",
                output=f"Scrolled {direction} {pages} pages",
                long_term_memory=f"Scrolled {direction} {pages} pages"
            )
    except Exception as e:
        return ToolResult(
            title="Scroll failed",
            output="",
            error=f"Failed to scroll: {str(e)}",
            long_term_memory="Scroll failed"
        )


# ============================================================
# Tab Management Tools
# ============================================================

@tool()
async def switch_tab(tab_id: str, uid: str = "") -> ToolResult:
    """
    Switch to a different browser tab

    Args:
        tab_id: 4-character tab ID
        uid: User ID (auto-injected)

    Returns:
        Switch result
    """
    try:
        return ToolResult(
            title=f"Switched to tab {tab_id}",
            output=f"Switched to tab {tab_id}",
            long_term_memory=f"Switched to tab {tab_id}"
        )
    except Exception as e:
        return ToolResult(
            title="Switch tab failed",
            output="",
            error=f"Failed to switch tab: {str(e)}",
            long_term_memory="Switch tab failed"
        )


@tool()
async def close_tab(tab_id: str, uid: str = "") -> ToolResult:
    """
    Close a browser tab

    Args:
        tab_id: 4-character tab ID
        uid: User ID (auto-injected)

    Returns:
        Close result
    """
    try:
        return ToolResult(
            title=f"Closed tab {tab_id}",
            output=f"Closed tab {tab_id}",
            long_term_memory=f"Closed tab {tab_id}"
        )
    except Exception as e:
        return ToolResult(
            title="Close tab failed",
            output="",
            error=f"Failed to close tab: {str(e)}",
            long_term_memory="Close tab failed"
        )


# ============================================================
# Dropdown Tools
# ============================================================

@tool()
async def get_dropdown_options(index: int, uid: str = "") -> ToolResult:
    """
    Get options from a dropdown element

    Args:
        index: Element index from browser state
        uid: User ID (auto-injected)

    Returns:
        Dropdown options
    """
    try:
        from playwright.async_api import async_playwright

        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=False)
            context = await browser.new_context()
            page = await context.pages()[0] if context.pages() else await context.new_page()

            # This would need DOM state to map index to selector
            # For now, return a placeholder
            return ToolResult(
                title=f"Dropdown options for element {index}",
                output=f"Retrieved options for dropdown at index {index}",
                long_term_memory=f"Got dropdown options for element {index}"
            )
    except Exception as e:
        return ToolResult(
            title="Get dropdown options failed",
            output="",
            error=f"Failed to get dropdown options: {str(e)}",
            long_term_memory="Get dropdown options failed"
        )


@tool()
async def select_dropdown_option(index: int, text: str, uid: str = "") -> ToolResult:
    """
    Select an option from a dropdown

    Args:
        index: Element index from browser state
        text: Exact text/value to select
        uid: User ID (auto-injected)

    Returns:
        Selection result
    """
    try:
        from playwright.async_api import async_playwright

        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=False)
            context = await browser.new_context()
            page = await context.pages()[0] if context.pages() else await context.new_page()

            # This would need DOM state to map index to selector
            return ToolResult(
                title=f"Selected dropdown option",
                output=f"Selected '{text}' from dropdown at index {index}",
                long_term_memory=f"Selected '{text}' from dropdown {index}"
            )
    except Exception as e:
        return ToolResult(
            title="Select dropdown option failed",
            output="",
            error=f"Failed to select dropdown option: {str(e)}",
            long_term_memory="Select dropdown option failed"
        )


# ============================================================
# File Upload Tool
# ============================================================

@tool()
async def upload_file(index: int, path: str, uid: str = "") -> ToolResult:
    """
    Upload a file to a file input element

    Args:
        index: Element index from browser state
        path: Path to the file to upload
        uid: User ID (auto-injected)

    Returns:
        Upload result
    """
    try:
        from playwright.async_api import async_playwright

        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=False)
            context = await browser.new_context()
            page = await context.pages()[0] if context.pages() else await context.new_page()

            # This would need DOM state to map index to selector
            return ToolResult(
                title="File uploaded",
                output=f"Uploaded file {path} to element {index}",
                long_term_memory=f"Uploaded file {path}"
            )
    except Exception as e:
        return ToolResult(
            title="Upload failed",
            output="",
            error=f"Failed to upload file: {str(e)}",
            long_term_memory="File upload failed"
        )


# ============================================================
# Task Completion Tool
# ============================================================

@tool()
async def done(text: str, success: bool = True,
              files_to_display: Optional[List[str]] = None, uid: str = "") -> ToolResult:
    """
    Mark the task as complete and return final message to user

    Args:
        text: Final message to user in the requested format
        success: Whether the task completed successfully
        files_to_display: Optional list of file paths to display
        uid: User ID (auto-injected)

    Returns:
        Completion result
    """
    try:
        return ToolResult(
            title="Task completed" if success else "Task failed",
            output=text,
            long_term_memory=f"Task {'completed' if success else 'failed'}",
            attachments=files_to_display or [],
            metadata={"success": success}
        )
    except Exception as e:
        return ToolResult(
            title="Done failed",
            output="",
            error=f"Failed to complete task: {str(e)}",
            long_term_memory="Task completion failed"
        )