ck_pro/ck_web/playwright_utils.py

#
# 内置Playwright实现的WebEnv
# 替换HTTP API架构，直接使用Playwright Python API

import os
import sys
import time
import base64
import json
import uuid
import asyncio
import threading
import subprocess
from typing import Dict, List, Optional, Any
from contextlib import asynccontextmanager
from playwright.async_api import async_playwright, Browser, BrowserContext, Page
from playwright.sync_api import sync_playwright, Browser as SyncBrowser, BrowserContext as SyncBrowserContext, Page as SyncPage

from ..agents.utils import KwargsInitializable, rprint, zwarn, zlog
from .utils import WebState, MyMarkdownify


class PlaywrightBrowserPool:
    """Playwright浏览器池管理器"""

    def __init__(self, max_browsers: int = 16, headless: bool = True, logger=None):
        self.max_browsers = max_browsers
        self.headless = headless
        self.logger = logger
        self.browsers: Dict[str, Dict] = {}
        self.playwright = None
        self.browser_type = None
        self._lock = threading.Lock()
        
    def start(self):
        """启动Playwright和浏览器池"""
        if self.playwright is None:
            # 简单直接的启动方式
            try:
                # Force Playwright to look for browsers in the same path used by postBuild
                os.environ["PLAYWRIGHT_BROWSERS_PATH"] = os.environ.get("PLAYWRIGHT_BROWSERS_PATH", "/home/user/.cache/ms-playwright")
                path = os.environ["PLAYWRIGHT_BROWSERS_PATH"]
                if self.logger:
                    self.logger.info("[PW_CHECK] PLAYWRIGHT_BROWSERS_PATH=%s", path)
                # If the path does not exist (build hook didn't run), install Chrome at runtime (non-root)
                if not os.path.isdir(path):
                    if self.logger:
                        self.logger.warning("[PW_SETUP] %s missing; installing Chromium via Playwright...", path)
                    try:
                        subprocess.run([sys.executable, "-m", "playwright", "install", "chromium"], check=True)
                    except Exception as ie:
                        if self.logger:
                            self.logger.error("[PW_SETUP] Runtime install failed: %s", ie)
                        raise RuntimeError(f"Runtime install of Playwright Chromium failed: {ie}")
                    # Re-check
                    if not os.path.isdir(path):
                        raise RuntimeError(f"Playwright install reported success but path still missing: {path}")
                else:
                    # optional: show a peek into the directory
                    try:
                        entries = sorted(os.listdir(path))[:5]
                        if self.logger:
                            self.logger.info("[PW_CHECK] %s entries=%s", path, entries)
                    except Exception as ie:
                        if self.logger:
                            self.logger.warning("[PW_CHECK] listdir failed: %s", ie)
                self.playwright = sync_playwright().start()
            except Exception as e:
                if self.logger:
                    self.logger.error("[PLAYWRIGHT_POOL] Failed to start Playwright: %s", e)
                raise RuntimeError(f"Cannot start Playwright: {e}")

            # 使用Chromium（临时方案，避免 Chrome 在 Space 上需 root 的依赖校验）
            self.browser_type = self.playwright.chromium

            # Ensure we skip host requirement validation during any runtime install
            os.environ.setdefault("PLAYWRIGHT_SKIP_VALIDATE_HOST_REQUIREMENTS", "1")

            if self.logger:
                self.logger.info("[PLAYWRIGHT_POOL] Started with max_browsers=%d (Chromium headless)", self.max_browsers)

    def stop(self):
        """停止所有浏览器和Playwright"""
        with self._lock:
            for browser_id, browser_info in self.browsers.items():
                try:
                    if browser_info.get('context'):
                        browser_info['context'].close()
                    if browser_info.get('browser'):
                        browser_info['browser'].close()
                except Exception as e:
                    if self.logger:
                        self.logger.warning("[PLAYWRIGHT_POOL] Error closing browser %s: %s", browser_id, e)

            self.browsers.clear()

            if self.playwright:
                self.playwright.stop()
                self.playwright = None

            if self.logger:
                self.logger.info("[PLAYWRIGHT_POOL] Stopped")
    
    def get_browser(self, storage_state=None, geo_location=None) -> str:
        """获取浏览器实例，返回browser_id"""
        with self._lock:
            # 检查是否有可用的浏览器槽位
            if len(self.browsers) >= self.max_browsers:
                # 清理不活跃的浏览器
                self._cleanup_inactive_browsers()
                
                if len(self.browsers) >= self.max_browsers:
                    raise RuntimeError(f"Browser pool exhausted (max: {self.max_browsers})")
            
            browser_id = str(uuid.uuid4())
            
            try:
                # 启动新浏览器 - 使用Chrome headless模式
                launch_args = [
                    '--no-sandbox',
                    '--disable-dev-shm-usage',
                    '--disable-gpu',
                    '--disable-web-security',
                    '--disable-features=VizDisplayCompositor',
                    '--disable-background-timer-throttling',
                    '--disable-backgrounding-occluded-windows',
                    '--disable-renderer-backgrounding'
                ]

                # Docker环境不再需要特殊参数 - 移除不必要的环境变量检查
                # launch_args.extend([
                #     '--disable-dev-shm-usage',
                #     '--no-first-run',
                #     '--no-default-browser-check'
                # ])

                # 尝试使用Chrome，如果失败则回退到Chromium
                browser = self.browser_type.launch(
                    headless=self.headless,
                    args=launch_args
                )

                # 创建浏览器上下文 - 使用真实Chrome用户代理
                context_options = {
                    'viewport': {'width': 1024, 'height': 768},
                    'locale': 'en-US',
                    'geolocation': geo_location or {'latitude': 40.4415, 'longitude': -80.0125},
                    'permissions': ['geolocation'],
                    'user_agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
                    'extra_http_headers': {
                        'Accept-Language': 'en-US,en;q=0.9',
                        'Accept-Encoding': 'gzip, deflate, br',
                        'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8'
                    }
                }
                
                if storage_state:
                    context_options['storage_state'] = storage_state
                
                context = browser.new_context(**context_options)
                
                self.browsers[browser_id] = {
                    'browser': browser,
                    'context': context,
                    'pages': {},
                    'last_activity': time.time(),
                    'status': 'active'
                }
                
                if self.logger:
                    self.logger.info("[PLAYWRIGHT_POOL] Created browser %s", browser_id)
                
                return browser_id
                
            except Exception as e:
                if self.logger:
                    self.logger.error("[PLAYWRIGHT_POOL] Failed to create browser: %s", e)
                raise
    
    def close_browser(self, browser_id: str):
        """关闭指定浏览器"""
        with self._lock:
            if browser_id in self.browsers:
                browser_info = self.browsers[browser_id]
                try:
                    if browser_info.get('context'):
                        browser_info['context'].close()
                    if browser_info.get('browser'):
                        browser_info['browser'].close()
                    
                    del self.browsers[browser_id]
                    
                    if self.logger:
                        self.logger.info("[PLAYWRIGHT_POOL] Closed browser %s", browser_id)
                        
                except Exception as e:
                    if self.logger:
                        self.logger.warning("[PLAYWRIGHT_POOL] Error closing browser %s: %s", browser_id, e)
    
    def get_browser_context(self, browser_id: str) -> Optional[SyncBrowserContext]:
        """获取浏览器上下文"""
        browser_info = self.browsers.get(browser_id)
        if browser_info:
            browser_info['last_activity'] = time.time()
            return browser_info.get('context')
        return None
    
    def _cleanup_inactive_browsers(self):
        """清理不活跃的浏览器"""
        current_time = time.time()
        inactive_threshold = 3600  # 1小时不活跃则清理
        
        inactive_browsers = []
        for browser_id, browser_info in self.browsers.items():
            if current_time - browser_info['last_activity'] > inactive_threshold:
                inactive_browsers.append(browser_id)
        
        for browser_id in inactive_browsers:
            self.close_browser(browser_id)
            if self.logger:
                self.logger.info("[PLAYWRIGHT_POOL] Cleaned up inactive browser %s", browser_id)

    def get_status(self):
        """获取浏览器池状态"""
        with self._lock:
            active_count = len([b for b in self.browsers.values() if b['status'] == 'active'])
            return {
                'active': active_count,
                'total': len(self.browsers),
                'available': self.max_browsers - len(self.browsers),
                'max_browsers': self.max_browsers
            }


class PlaywrightWebEnv(KwargsInitializable):
    """基于Playwright的内置WebEnv实现"""

    def __init__(self, settings=None, starting=True, starting_target_url=None, logger=None, **kwargs):
        # 基础配置 - 从TOML配置读取
        if settings and hasattr(settings, 'web') and hasattr(settings.web, 'env_builtin'):
            self.max_browsers = settings.web.env_builtin.max_browsers
            self.headless = settings.web.env_builtin.headless
            self.web_timeout = settings.web.env_builtin.web_timeout
            self.screenshot_boxed = settings.web.env_builtin.screenshot_boxed
            self.target_url = settings.web.env_builtin.target_url
        else:
            # Fallback defaults if no settings provided
            self.max_browsers = 16
            self.headless = True
            self.web_timeout = 600
            self.screenshot_boxed = True
            self.target_url = "https://www.bing.com/"

        self.logger = logger

        # Playwright相关
        self.browser_pool = None
        self.current_browser_id = None
        self.current_page_id = None

        # 状态管理
        self.state: WebState = None

        super().__init__(**kwargs)

        # 创建浏览器池
        self._create_browser_pool()

        if starting:
            self.start(starting_target_url)
    
    def _create_browser_pool(self):
        """创建浏览器池"""
        self.browser_pool = PlaywrightBrowserPool(
            max_browsers=self.max_browsers,
            headless=self.headless,
            logger=self.logger
        )
        self.browser_pool.start()
    
    def start(self, target_url=None):
        """启动web环境"""
        self.stop()  # 先停止现有环境
        
        target_url = target_url if target_url is not None else self.target_url
        
        # Google到Bing的重定向（保持与原有逻辑一致）
        if 'www.google.com' in target_url and 'www.google.com/maps' not in target_url:
            target_url = target_url.replace('www.google.com', 'www.bing.com')
        
        self.init_state(target_url)
    
    def stop(self):
        """停止web环境"""
        if self.current_browser_id and self.browser_pool:
            self.browser_pool.close_browser(self.current_browser_id)
            self.current_browser_id = None
            self.current_page_id = None

        if self.state is not None:
            self.state = None

    def __del__(self):
        """析构函数"""
        self.stop()
        if self.browser_pool:
            self.browser_pool.stop()

    def get_state(self, export_to_dict=True, return_copy=True):
        """获取当前状态"""
        assert self.state is not None, "Current state is None, should first start it!"
        if export_to_dict:
            ret = self.state.to_dict()
        elif return_copy:
            ret = self.state.copy()
        else:
            ret = self.state
        return ret

    def get_target_url(self):
        """获取目标URL"""
        return self.target_url

    def init_state(self, target_url: str):
        """初始化浏览器状态"""
        if self.logger:
            self.logger.info("[PLAYWRIGHT_INIT] Starting browser initialization")
            self.logger.info("[PLAYWRIGHT_INIT] Target_URL: %s", target_url)

        # 获取浏览器实例
        self.current_browser_id = self.browser_pool.get_browser()

        if self.logger:
            self.logger.info("[PLAYWRIGHT_INIT] Browser_Created: %s", self.current_browser_id)

        # 打开页面
        self.current_page_id = self._open_page(target_url)

        if self.logger:
            self.logger.info("[PLAYWRIGHT_INIT] Page_Opened: %s", self.current_page_id)

        # 创建状态对象
        curr_step = 0
        self.state = WebState(
            browser_id=self.current_browser_id,
            page_id=self.current_page_id,
            target_url=target_url,
            curr_step=curr_step,
            total_actual_step=curr_step
        )

        # 获取初始页面信息
        results = self._get_accessibility_tree_results()
        self.state.update(**results)

        if self.logger:
            actual_url = getattr(self.state, 'step_url', 'unknown')
            self.logger.info("[PLAYWRIGHT_INIT] State_Initialized: Actual_URL: %s", actual_url)
            if actual_url != target_url:
                self.logger.warning("[PLAYWRIGHT_INIT] URL_Mismatch: Expected: %s | Actual: %s", target_url, actual_url)

    def _open_page(self, target_url: str) -> str:
        """打开新页面"""
        context = self.browser_pool.get_browser_context(self.current_browser_id)
        if not context:
            raise RuntimeError(f"Browser context not found for {self.current_browser_id}")

        page = context.new_page()
        page_id = str(uuid.uuid4())

        # 设置下载处理
        page.on("download", self._handle_download)

        # 导航到目标URL
        try:
            page.goto(target_url, wait_until="domcontentloaded", timeout=30000)

            # 存储页面引用
            browser_info = self.browser_pool.browsers[self.current_browser_id]
            browser_info['pages'][page_id] = page

            if self.logger:
                actual_url = page.url
                self.logger.info("[PLAYWRIGHT_PAGE] Opened: %s -> %s", target_url, actual_url)

            return page_id

        except Exception as e:
            if self.logger:
                self.logger.error("[PLAYWRIGHT_PAGE] Failed to open %s: %s", target_url, e)
            raise

    def _handle_download(self, download):
        """处理文件下载"""
        try:
            # 生成下载文件路径
            download_path = f"./downloads/{download.suggested_filename}"
            os.makedirs(os.path.dirname(download_path), exist_ok=True)

            # 保存文件
            download.save_as(download_path)

            # 更新状态中的下载文件列表
            if self.state and hasattr(self.state, 'downloaded_file_path'):
                if download_path not in self.state.downloaded_file_path:
                    self.state.downloaded_file_path.append(download_path)

            if self.logger:
                self.logger.info("[PLAYWRIGHT_DOWNLOAD] Saved: %s", download_path)

        except Exception as e:
            if self.logger:
                self.logger.error("[PLAYWRIGHT_DOWNLOAD] Failed: %s", e)

    def _get_current_page(self) -> Optional[SyncPage]:
        """获取当前页面对象"""
        if not self.current_browser_id or not self.current_page_id:
            return None

        browser_info = self.browser_pool.browsers.get(self.current_browser_id)
        if not browser_info:
            return None

        return browser_info['pages'].get(self.current_page_id)

    def _get_accessibility_tree_results(self) -> Dict[str, Any]:
        """获取可访问性树和页面信息"""
        page = self._get_current_page()
        if not page:
            return self._get_default_results()

        try:
            # 获取基本页面信息
            current_url = page.url
            html_content = page.content()

            # 处理HTML为Markdown
            html_md = self._process_html(html_content)

            # 获取可访问性树
            accessibility_tree = self._get_accessibility_tree(page)

            # 获取截图
            screenshot_b64 = self._take_screenshot(page)

            # 检查Cookie弹窗
            has_cookie_popup = self._check_cookie_popup(page)

            results = {
                "current_accessibility_tree": accessibility_tree,
                "step_url": current_url,
                "html_md": html_md,
                "snapshot": "",  # 可以添加accessibility snapshot
                "boxed_screenshot": screenshot_b64,
                "downloaded_file_path": getattr(self.state, 'downloaded_file_path', []),
                "get_accessibility_tree_succeed": True,
                "current_has_cookie_popup": has_cookie_popup,
                "expanded_part": None
            }

            return results

        except Exception as e:
            if self.logger:
                self.logger.error("[PLAYWRIGHT_AXTREE] Failed to get page info: %s", e)
            return self._get_default_results()

    def _get_default_results(self) -> Dict[str, Any]:
        """获取默认结果（错误情况下）"""
        return {
            "current_accessibility_tree": "**Warning**: The accessibility tree is currently unavailable.",
            "step_url": "",
            "html_md": "",
            "snapshot": "",
            "boxed_screenshot": "",
            "downloaded_file_path": [],
            "get_accessibility_tree_succeed": False,
            "current_has_cookie_popup": False,
            "expanded_part": None
        }

    def _process_html(self, html_content: str) -> str:
        """处理HTML内容为Markdown"""
        if not html_content.strip():
            return ""
        try:
            return MyMarkdownify.md_convert(html_content)
        except Exception as e:
            if self.logger:
                self.logger.warning("[PLAYWRIGHT_HTML] Failed to convert HTML: %s", e)
            return ""

    def _get_accessibility_tree(self, page: SyncPage) -> str:
        """获取可访问性树"""
        try:
            # 使用Playwright的accessibility API
            snapshot = page.accessibility.snapshot()
            if snapshot:
                return self._format_accessibility_tree(snapshot)
            else:
                return "No accessibility tree available"
        except Exception as e:
            if self.logger:
                self.logger.warning("[PLAYWRIGHT_AXTREE] Failed to get accessibility tree: %s", e)
            return "**Warning**: Failed to get accessibility tree"

    def _format_accessibility_tree(self, snapshot: Dict, level: int = 0) -> str:
        """格式化可访问性树为文本"""
        lines = []
        indent = "  " * level

        # 获取节点信息
        role = snapshot.get('role', 'unknown')
        name = snapshot.get('name', '')
        value = snapshot.get('value', '')

        # 构建节点描述
        node_desc = f"{indent}[{level}] {role}"
        if name:
            node_desc += f" \"{name}\""
        if value:
            node_desc += f" value=\"{value}\""

        lines.append(node_desc)

        # 递归处理子节点
        children = snapshot.get('children', [])
        for child in children:
            lines.extend(self._format_accessibility_tree(child, level + 1).split('\n'))

        return '\n'.join(lines)

    def _take_screenshot(self, page: SyncPage) -> str:
        """截取页面截图并返回base64编码"""
        try:
            screenshot_bytes = page.screenshot(full_page=False)
            return base64.b64encode(screenshot_bytes).decode('utf-8')
        except Exception as e:
            if self.logger:
                self.logger.warning("[PLAYWRIGHT_SCREENSHOT] Failed: %s", e)
            return ""

    def _check_cookie_popup(self, page: SyncPage) -> bool:
        """检查是否有Cookie弹窗"""
        try:
            # 常见的Cookie弹窗选择器
            cookie_selectors = [
                '[id*="cookie"]',
                '[class*="cookie"]',
                '[id*="consent"]',
                '[class*="consent"]',
                'button:has-text("Accept")',
                'button:has-text("Allow")',
                'button:has-text("Agree")'
            ]

            for selector in cookie_selectors:
                elements = page.query_selector_all(selector)
                if elements:
                    return True

            return False
        except Exception as e:
            if self.logger:
                self.logger.warning("[PLAYWRIGHT_COOKIE] Cookie popup check failed: %s", e)
            return False

    def step_state(self, action_string: str) -> str:
        """执行浏览器动作"""
        if self.logger:
            self.logger.info("[PLAYWRIGHT_ACTION] Step_State_Start: %s", action_string)

        # 解析动作
        action = self._parse_action(action_string)

        # 更新状态
        self.state.curr_step += 1
        self.state.total_actual_step += 1
        self.state.update(action=action, action_string=action_string, error_message="")

        # 执行动作
        if not action["action_name"]:
            error_msg = f"The action you previously choose is not well-formatted: {action_string}"
            self.state.error_message = error_msg
            return error_msg

        try:
            success = self._perform_action(action)

            if not success:
                error_msg = f"The action you have chosen cannot be executed: {action_string}"
                self.state.error_message = error_msg
                if self.logger:
                    self.logger.error("[PLAYWRIGHT_ACTION] Failed: %s", action_string)
                return error_msg
            else:
                # 获取新状态
                if self.logger:
                    self.logger.info("[PLAYWRIGHT_ACTION] Success: %s", action_string)

                results = self._get_accessibility_tree_results()
                self.state.update(**results)
                return f"Browser step: {action_string}"

        except Exception as e:
            error_msg = f"Browser error: {e}"
            self.state.error_message = error_msg
            if self.logger:
                self.logger.error("[PLAYWRIGHT_ACTION] Exception: %s", e)
            return error_msg

    def _parse_action(self, action_string: str) -> Dict[str, Any]:
        """解析动作字符串"""
        action = {
            "action_name": "",
            "target_id": None,
            "target_element_type": "",
            "target_element_name": "",
            "action_value": "",
            "need_enter": True
        }

        action_string = action_string.strip()

        # 解析不同类型的动作
        if action_string.startswith("click"):
            action["action_name"] = "click"
            # 解析 click [id] name 格式
            import re
            match = re.match(r'click\s+\[(\d+)\]\s*(.*)', action_string)
            if match:
                action["target_id"] = int(match.group(1))
                action["target_element_name"] = match.group(2).strip()
                action["target_element_type"] = "clickable"

        elif action_string.startswith("type"):
            action["action_name"] = "type"
            # 解析 type [id] content 格式
            import re
            match = re.match(r'type\s+\[(\d+)\]\s+(.*?)(?:\[NOENTER\])?$', action_string)
            if match:
                action["target_id"] = int(match.group(1))
                action["action_value"] = match.group(2).strip()
                action["target_element_type"] = "textbox"
                action["need_enter"] = "[NOENTER]" not in action_string

        elif action_string in ["scroll_up", "scroll up"]:
            action["action_name"] = "scroll_up"

        elif action_string in ["scroll_down", "scroll down"]:
            action["action_name"] = "scroll_down"

        elif action_string == "wait":
            action["action_name"] = "wait"

        elif action_string == "goback":
            action["action_name"] = "goback"

        elif action_string == "restart":
            action["action_name"] = "restart"

        elif action_string.startswith("goto"):
            action["action_name"] = "goto"
            # 解析 goto url 格式
            parts = action_string.split(None, 1)
            if len(parts) > 1:
                action["action_value"] = parts[1].strip()

        elif action_string.startswith("stop"):
            action["action_name"] = "stop"

        elif action_string.startswith("save"):
            action["action_name"] = "save"

        elif action_string.startswith("screenshot"):
            action["action_name"] = "screenshot"
            parts = action_string.split()
            if len(parts) > 1:
                action["action_value"] = " ".join(parts[1:])

        return action

    def _perform_action(self, action: Dict[str, Any]) -> bool:
        """执行具体的浏览器动作"""
        page = self._get_current_page()
        if not page:
            return False

        action_name = action["action_name"]

        try:
            if action_name == "click":
                return self._perform_click(page, action)

            elif action_name == "type":
                return self._perform_type(page, action)

            elif action_name == "scroll_up":
                page.keyboard.press("PageUp")
                return True

            elif action_name == "scroll_down":
                page.keyboard.press("PageDown")
                return True

            elif action_name == "wait":
                time.sleep(5)
                return True

            elif action_name == "goback":
                page.go_back(wait_until="domcontentloaded")
                return True

            elif action_name == "restart":
                page.goto(self.target_url, wait_until="domcontentloaded")
                return True

            elif action_name == "goto":
                url = action.get("action_value", "")
                if url:
                    page.goto(url, wait_until="domcontentloaded")
                    return True
                return False

            elif action_name in ["stop", "save", "screenshot"]:
                # 这些动作由上层处理
                return True

            else:
                if self.logger:
                    self.logger.warning("[PLAYWRIGHT_ACTION] Unknown action: %s", action_name)
                return False

        except Exception as e:
            if self.logger:
                self.logger.error("[PLAYWRIGHT_ACTION] Error executing %s: %s", action_name, e)
            return False

    def _perform_click(self, page: SyncPage, action: Dict[str, Any]) -> bool:
        """执行点击动作"""
        target_id = action.get("target_id")
        if target_id is None:
            return False

        try:
            # 使用简化的选择器策略
            # 在实际实现中，需要维护元素ID到选择器的映射
            # 这里使用一个简化的实现

            # 尝试通过data-testid或其他属性查找元素
            selectors = [
                f'[data-testid="{target_id}"]',
                f'[data-id="{target_id}"]',
                f'#{target_id}',
                f'*:nth-child({target_id})'
            ]

            element = None
            for selector in selectors:
                try:
                    element = page.query_selector(selector)
                    if element:
                        break
                except:
                    continue

            if element:
                element.click()
                return True
            else:
                # 如果找不到特定元素，尝试通过可访问性树查找
                return self._click_by_accessibility_tree(page, target_id)

        except Exception as e:
            if self.logger:
                self.logger.error("[PLAYWRIGHT_CLICK] Error: %s", e)
            return False

    def _perform_type(self, page: SyncPage, action: Dict[str, Any]) -> bool:
        """执行输入动作"""
        target_id = action.get("target_id")
        text = action.get("action_value", "")
        need_enter = action.get("need_enter", True)

        if target_id is None:
            return False

        try:
            # 类似点击，查找输入元素
            selectors = [
                f'[data-testid="{target_id}"]',
                f'[data-id="{target_id}"]',
                f'#{target_id}',
                'input[type="text"]',
                'input[type="search"]',
                'textarea'
            ]

            element = None
            for selector in selectors:
                try:
                    element = page.query_selector(selector)
                    if element and element.is_visible():
                        break
                except:
                    continue

            if element:
                element.click()  # 先点击获得焦点
                element.clear()  # 清空现有内容
                element.type(text)  # 输入文本

                if need_enter:
                    element.press("Enter")

                return True
            else:
                return self._type_by_accessibility_tree(page, target_id, text, need_enter)

        except Exception as e:
            if self.logger:
                self.logger.error("[PLAYWRIGHT_TYPE] Error: %s", e)
            return False

    def _click_by_accessibility_tree(self, page: SyncPage, target_id: int) -> bool:
        """通过可访问性树查找并点击元素"""
        try:
            # 获取所有可点击元素
            clickable_elements = page.query_selector_all('button, a, [role="button"], [onclick], input[type="submit"], input[type="button"]')

            if target_id < len(clickable_elements):
                clickable_elements[target_id].click()
                return True

            return False
        except Exception as e:
            if self.logger:
                self.logger.error("[PLAYWRIGHT_CLICK_AX] Error: %s", e)
            return False

    def _type_by_accessibility_tree(self, page: SyncPage, target_id: int, text: str, need_enter: bool) -> bool:
        """通过可访问性树查找并输入文本"""
        try:
            # 获取所有输入元素
            input_elements = page.query_selector_all('input[type="text"], input[type="search"], input[type="email"], input[type="password"], textarea')

            if target_id < len(input_elements):
                element = input_elements[target_id]
                element.click()
                element.clear()
                element.type(text)

                if need_enter:
                    element.press("Enter")

                return True

            return False
        except Exception as e:
            if self.logger:
                self.logger.error("[PLAYWRIGHT_TYPE_AX] Error: %s", e)
            return False

    def sync_files(self):
        """同步下载的文件（内置实现中文件已经直接保存到本地）"""
        # 在内置实现中，文件下载已经通过_handle_download直接处理
        # 这里只需要确保状态中的文件路径是正确的
        if self.logger:
            downloaded_files = getattr(self.state, 'downloaded_file_path', [])
            self.logger.info("[PLAYWRIGHT_SYNC] Downloaded files: %s", downloaded_files)
        return True