#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称：playwright_executor.py
模块描述：Playwright 测试执行器，负责按步骤执行 Web 自动化测试

作者：czj
创建日期：2026-07-09
最后修改：2026-07-11
"""

import logging
import os
import re
import asyncio
from typing import Optional, Callable, Dict, Any, List
from datetime import datetime
from dataclasses import dataclass, field

from playwright.sync_api import (
    sync_playwright,
    Browser,
    BrowserContext,
    Page,
    Playwright,
)
from playwright._impl._errors import TimeoutError as PlaywrightTimeoutError

from app.config import settings
from app.utils.selector_mapper import SelectorMapper
from app.services.element_mapping_service import get_element_mapping_service
from app.services.page_url_service import get_page_url_service

logger = logging.getLogger(__name__)


# ==================== 数据类 ====================

@dataclass
class StepResult:
    """
    步骤执行结果

    Attributes:
        order (int): 步骤顺序
        name (str): 步骤名称
        action (str): 动作类型
        status (str): 状态 (pending/running/passed/failed/skipped)
        duration (float): 耗时（秒）
        screenshot (Optional[str]): 截图路径
        log (str): 执行日志
        error (Optional[str]): 错误信息
        start_time (datetime): 开始时间
        end_time (datetime): 结束时间
    """

    order: int
    name: str = ""
    action: str = ""
    status: str = "pending"
    duration: float = 0.0
    screenshot: Optional[str] = None
    log: str = ""
    error: Optional[str] = None
    start_time: Optional[datetime] = None
    end_time: Optional[datetime] = None


@dataclass
class CaseExecutionResult:
    """
    用例执行结果

    Attributes:
        case_id (str): 用例ID
        case_name (str): 用例名称
        status (str): 总体状态
        duration (float): 总耗时
        steps_result (List[StepResult]): 各步骤结果
        error_message (Optional[str]): 错误信息
        retry_count (int): 重试次数
    """
    case_id: str
    case_name: str = ""
    status: str = "pending"
    duration: float = 0.0
    steps_result: List[StepResult] = field(default_factory=list)
    error_message: Optional[str] = None
    retry_count: int = 0


# ==================== 执行器 ====================

class PlaywrightExecutor:
    """
    Playwright 测试执行器（同步版本）

    负责执行测试用例中的步骤，支持多种浏览器操作和断言验证。
    每个步骤自动截图，支持实时回调通知。

    使用同步 Playwright API 以避免 Windows 上 asyncio 子进程的限制。

    功能：
    - 页面导航 (navigate)
    - 元素点击 (click)
    - 输入框填充 (fill)
    - 下拉框选择 (select)
    - 鼠标悬停 (hover)
    - 元素等待 (wait)
    - 断言验证 (assert)
    - 滚动操作 (scroll)
    - 按键操作 (press)
    - 截图 (screenshot)
    - 复选框操作 (check/uncheck)
    - 自动登录 (auto_login) - 新增

    Example:
        >>> executor = PlaywrightExecutor()
        >>> executor.start()
        >>> result = executor.execute_case(case, callback=on_step_complete)
        >>> print(result.status)
        'passed'
        >>> executor.stop()
    """

    # 登录凭据（从环境变量或配置获取）
    LOGIN_CONFIG = {
        "base_url": "https://192.168.5.44",
        "username": "admin@xty",
        "password": "Ubains@13579",
        "captcha": "csba",
    }

    def __init__(self, config: Optional[Dict[str, Any]] = None):
        """
        初始化执行器

        Args:
            config (Optional[dict]): 执行配置
                - timeout (int): 超时时间（毫秒），默认30000
                - screenshot (bool): 是否截图，默认True
                - headless (bool): 是否无头模式，默认False
                - screenshot_dir (str): 截图目录
                - retry (int): 失败重试次数，默认0
                - ignore_https_errors (bool): 是否忽略 HTTPS 错误，默认True
                - auto_login (bool): 是否自动登录，默认False
        """
        self.config = config or {}
        self.timeout: int = self.config.get("timeout", 30000)
        self.screenshot_enabled: bool = self.config.get("screenshot", True)
        # headless 默认值取自全局配置 settings.PLAYWRIGHT_HEADLESS（由环境变量 PLAYWRIGHT_HEADLESS 控制）。
        # 容器内无 X Server，必须 headless=True；原默认 False（headed）会导致容器内启动崩溃。
        self.headless: bool = self.config.get("headless", settings.PLAYWRIGHT_HEADLESS)
        self.screenshot_dir: str = self.config.get(
            "screenshot_dir", settings.SCREENSHOT_DIR
        )
        self.retry_count: int = self.config.get("retry", 0)
        self.ignore_https_errors: bool = self.config.get("ignore_https_errors", True)
        self.auto_login: bool = self.config.get("auto_login", False)
        self.step_retry_count: int = self.config.get("step_retry_count", 2)

        # SelectorMapper 实例（用于 page_key + element_key 解析）
        self._selector_mapper: SelectorMapper = SelectorMapper()

        # 元素映射表服务（最高优先级选择器来源，Phase 3）
        self._element_mapping_service = get_element_mapping_service()

        # 页面识别服务（URL 直达导航，跳过中间点击步骤）
        self._page_url_service = get_page_url_service()

        # 内部状态
        self._playwright: Optional[Playwright] = None
        self._browser: Optional[Browser] = None
        self._context: Optional[BrowserContext] = None
        self._page: Optional[Page] = None
        self._is_running: bool = False
        self._is_logged_in: bool = False

        # 确保截图目录存在
        os.makedirs(self.screenshot_dir, exist_ok=True)

    def start(self) -> None:
        """
        启动执行器

        初始化 Playwright 浏览器实例，创建浏览器上下文和页面。
        使用同步 Playwright API 以避免 Windows 上 asyncio 子进程限制。

        Raises:
            RuntimeError: 当启动失败时抛出
        """
        try:
            # 在 Windows 上，当从 run_in_executor 调用时，需要绕过 asyncio 检测
            # Playwright sync_playwright().start() 内部会检测是否有事件循环
            # 关键：必须确保当前线程没有关联任何 asyncio 事件循环
            # asyncio.get_event_loop() 可能返回主线程的循环，所以必须用
            # asyncio.get_running_loop() 检测或直接设置 None
            import threading
            if threading.current_thread() is not threading.main_thread():
                # 非主线程（run_in_executor 环境）
                # 将当前线程的事件循环设为 None，这样 Playwright 的检测就不会发现 asyncio 循环
                # 注意：不能设为新的 event_loop，因为新循环也可能被 Playwright 检测到
                asyncio.set_event_loop(None)

            self._playwright = sync_playwright().start()

            # 启动浏览器
            # 关键修复（P1 卡点）：容器内 Linux headless Chromium 的 UA 含 "HeadlessChrome"，
            # 被测站点会据此返回不同页面/拦截登录表单渲染（导致 input:visible=0）。
            # 解决：隐藏自动化特征 + 伪装成正常桌面 Chrome。
            self._browser = self._playwright.chromium.launch(
                headless=self.headless,
                args=[
                    "--no-sandbox",
                    "--disable-setuid-sandbox",
                    # 隐藏自动化特征，避免 navigator.webdriver=true 被站点检测
                    "--disable-blink-features=AutomationControlled",
                    "--disable-automation",
                    # 容器环境兼容（共享内存不足会导致渲染异常/崩溃）
                    "--disable-dev-shm-usage",
                    "--disable-gpu",
                ],
            )

            # 创建浏览器上下文（忽略 HTTPS 证书错误 + 伪装 UA/locale）
            # 伪装成正常桌面 Chrome 的 User-Agent（去掉 HeadlessChrome 标识），
            # 统一设为 zh-CN locale 与 Asia/Shanghai 时区，避免容器默认 locale
            # 触发被测站点的不同分支。
            fake_ua = (
                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                "AppleWebKit/537.36 (KHTML, like Gecko) "
                "Chrome/126.0.0.0 Safari/537.36"
            )
            self._context = self._browser.new_context(
                ignore_https_errors=self.ignore_https_errors,
                viewport={"width": 1920, "height": 1080},
                user_agent=fake_ua,
                locale="zh-CN",
                timezone_id="Asia/Shanghai",
            )

            # 进一步隐藏 webdriver 痕迹（部分站点通过 navigator.webdriver 检测）
            try:
                self._context.add_init_script(
                    "Object.defineProperty(navigator, 'webdriver', {get: () => undefined})"
                )
            except Exception as e:
                logger.debug(f"add_init_script 失败（忽略）: {e}")

            # 创建页面
            self._page = self._context.new_page()
            self._page.set_default_timeout(self.timeout)

            self._is_running = True
            logger.info("Playwright 执行器启动成功")

        except Exception as e:
            logger.error(f"执行器启动失败: {str(e)}")
            raise RuntimeError(f"执行器启动失败: {str(e)}")

    def do_login(self) -> bool:
        """
        执行自动登录

        使用预设的账号密码登录目标系统。

        Returns:
            bool: 登录是否成功
        """
        if self._is_logged_in:
            return True

        try:
            logger.info("开始自动登录...")

            # 访问登录页 - 使用 domcontentloaded 避免 SPA 页面 networkidle 超时
            self._page.goto(self.LOGIN_CONFIG["base_url"], wait_until="domcontentloaded", timeout=30000)
            self._page.wait_for_timeout(3000)  # 等待页面渲染

            # 等待登录表单加载
            try:
                self._page.wait_for_selector('input[placeholder*="手机号"], input[type="password"]', timeout=10000)
                logger.debug("登录表单已加载")
            except Exception as e:
                logger.warning(f"等待登录表单超时: {e}")

            # 填写表单
            inputs = self._page.query_selector_all('input:visible')
            for inp in inputs:
                t = inp.get_attribute('type') or 'text'
                ph = inp.get_attribute('placeholder') or ''
                if '手机' in ph or '用户' in ph or '邮箱' in ph or 'account' in ph.lower():
                    inp.fill(self.LOGIN_CONFIG["username"])
                    logger.debug(f"填写用户名: {self.LOGIN_CONFIG['username']}")
                elif t == 'password':
                    inp.fill(self.LOGIN_CONFIG["password"])
                    logger.debug("填写密码")
                elif '验证' in ph:
                    inp.fill(self.LOGIN_CONFIG["captcha"])
                    logger.debug(f"填写验证码: {self.LOGIN_CONFIG['captcha']}")

            # === 先勾选协议复选框（在登录页面上），再点登录 ===
            # 避免先点登录再弹协议弹窗
            # 注意：el-checkbox 的原始 <input type="checkbox"> 视觉上隐藏（is_visible()=False），
            # 但 check(force=True) 可强制勾选，故不能依赖 is_visible() 判断。
            for cb_sel in ['.el-checkbox__original', 'input[type="checkbox"]', '.agreement input']:
                try:
                    cb = self._page.query_selector(cb_sel)
                    if cb:
                        try:
                            if not cb.is_checked():
                                cb.check(force=True)
                                logger.debug(f"勾选协议复选框: {cb_sel}")
                                self._page.wait_for_timeout(500)
                        except Exception as e:
                            logger.debug(f"复选框操作失败（可能不需要）: {e}")
                        break
                except Exception:
                    continue

            # 点击登录按钮
            buttons = self._page.query_selector_all('button')
            for btn in buttons:
                text = (btn.inner_text() or '').strip()
                if '登录' in text:
                    btn.click()
                    logger.debug("点击登录按钮")
                    break

            self._page.wait_for_timeout(3000)

            # === 处理可能残留的协议弹窗（勾选协议后应无弹窗，但安全处理） ===
            try:
                self._page.wait_for_selector('.el-dialog:visible', timeout=5000)
                logger.debug("协议弹窗已出现（可能勾选协议后仍有弹窗）")
            except Exception as e:
                logger.debug("无协议弹窗（勾选协议后自动跳过）")

            # === 多选择器 + 多种点击方式回退 ===
            # 注意：页面存在多个 .el-dialog（含隐藏的登录弹窗/用户协议弹窗），
            # query_selector 返回 DOM 中第一个匹配，可能是隐藏弹窗的按钮；
            # 故先遍历所有【可见】的"确定/同意"按钮（最鲁棒），再走精确选择器兜底。
            confirm_texts = ("确定", "同意", "阅读并同意")
            clicked = False

            # 方式1：遍历所有按钮，点击可见的确认按钮
            for btn in self._page.query_selector_all("button"):
                try:
                    text = (btn.inner_text() or "").strip()
                    if text in confirm_texts and btn.is_visible():
                        btn.click(timeout=5000)
                        logger.debug(f"协议确认-可见按钮点击成功: text='{text}'")
                        clicked = True
                        break
                except Exception:
                    continue

            # 方式2：精确选择器 + 多种点击方式兜底
            confirm_selectors = [
                ".el-dialog .el-dialog__body button:has-text('确定')",
                ".el-dialog button.el-button--primary:has-text('确定')",
                ".el-dialog button:has-text('确定')",
                ".el-dialog button:has-text('同意')",
                ".el-dialog button:has-text('阅读并同意')",
                "[role='dialog'] button:has-text('确定')",
            ]
            for sel in confirm_selectors:
                if clicked:
                    break
                try:
                    btn = self._page.query_selector(sel)
                    if btn:
                        # 方式2a：普通点击
                        try:
                            btn.click(timeout=5000)
                            logger.debug(f"协议确认-普通点击成功: {sel}")
                            clicked = True
                            break
                        except Exception as e1:
                            logger.debug(f"协议确认-普通点击失败: {e1}")
                        # 方式2b：dispatch_event
                        try:
                            btn.dispatch_event('click')
                            logger.debug(f"协议确认-dispatch_event成功: {sel}")
                            clicked = True
                            break
                        except Exception as e2:
                            logger.debug(f"协议确认-dispatch_event失败: {e2}")
                        # 方式2c：JS click
                        try:
                            self._page.evaluate('(el) => el.click()', btn)
                            logger.debug(f"协议确认-JS click成功: {sel}")
                            clicked = True
                            break
                        except Exception as e3:
                            logger.debug(f"协议确认-JS click失败: {e3}")
                except Exception as e:
                    logger.debug(f"选择器 {sel} 查询失败: {e}")

            if not clicked:
                logger.warning("所有协议弹窗点击方式均失败，尝试最后手段：键盘 Enter")
                try:
                    self._page.keyboard.press('Enter')
                    logger.debug("协议确认-键盘 Enter")
                except Exception as e:
                    logger.warning(f"键盘 Enter 也失败: {e}")

            # === 等待弹窗消失 ===
            try:
                self._page.wait_for_selector('.el-dialog', state='hidden', timeout=5000)
                logger.debug("协议弹窗已关闭")
            except Exception as e:
                logger.debug(f"等待弹窗关闭超时（可能已关闭）: {e}")

            self._page.wait_for_timeout(2000)

            # 改进的登录成功判断：检查页面是否有登录表单或用户名显示
            # 登录成功标志：页面不再有登录表单，或者显示了用户名
            # 注意：SPA 登录成功后会跳转 /#/platform?...，此时原登录页 DOM 可能仍残留
            # （隐藏的 .login 表单），因此以 URL 是否离开登录路由为强判据。
            login_form_present = self._page.query_selector('input[placeholder*="手机号"]')
            username_displayed = self._page.query_selector('text=admin@xty, text=' + self.LOGIN_CONFIG["username"])
            current_url = self._page.url
            # SPA 登录路由：/#/platform?platform=%2Fplatform%2F%23%2Flogin 或 /#/login
            on_login_route = (
                "platform%2Flogin" in current_url
                or current_url.rstrip("/").endswith("#/login")
            )

            if not login_form_present or username_displayed or not on_login_route:
                self._is_logged_in = True
                logger.info(f"自动登录成功, URL: {current_url}")
                return True
            else:
                logger.warning(f"自动登录失败, 仍停留在登录页 URL={current_url}")
                return False

        except Exception as e:
            logger.error(f"自动登录异常: {str(e)}")
            return False

    def reset_login_state(self) -> None:
        """
        重置登录态：清除 cookies + storage，重新访问登录页

        用于用例间隔离，保证每个需要登录页的用例都从初始状态开始。
        当上一个用例登录成功进入主页后，下一个用例需要验证登录页面时，
        通过本方法清除登录态，确保回到登录页面。

        实测验证：清除 cookies + localStorage/sessionStorage 后重新访问，
        被系统会跳转到登录页（URL 含 #/login），登录表单可用。
        """
        try:
            logger.info("重置登录态...")
            # 清除 cookies
            self._context.clear_cookies()
            # 清除 localStorage / sessionStorage
            try:
                self._page.evaluate('() => { try { localStorage.clear(); sessionStorage.clear(); } catch(e) {} }')
            except Exception:
                pass
            # 重新访问登录页（用 domcontentloaded 避免 networkidle 超时）
            try:
                self._page.goto(self.LOGIN_CONFIG["base_url"], wait_until="domcontentloaded", timeout=15000)
            except PlaywrightTimeoutError:
                logger.debug("重置登录态 goto 超时（忽略）")
            self._page.wait_for_timeout(2000)
            # 显式等待登录表单加载，避免页面未就绪导致后续步骤超时
            try:
                self._page.wait_for_selector(
                    'input[placeholder*="手机号"], input[type="password"]',
                    timeout=15000,
                )
            except Exception as e:
                logger.warning(f"重置后等待登录表单超时: {e}")
            # 重置登录状态标志
            self._is_logged_in = False
            logger.info("登录态已重置，回到登录页")
        except Exception as e:
            logger.error(f"重置登录态失败: {e}")

    def _navigate_to_home(self) -> None:
        """
        导航到主页

        在已登录状态下，确保下一个用例从主页开始操作。
        上一个用例执行完可能停留在子页面，直接点击菜单可能失败。
        """
        try:
            current_url = self._page.url
            # 如果已经在主页（常用功能页），无需导航
            # 主页 URL 格式: https://xxx/#/platform?platform=...Home
            if "Home" in current_url and "#" in current_url:
                logger.debug("已在主页，无需导航")
                return

            logger.info(f"导航到主页... (当前URL: {current_url})")
            # 直接导航到主页 URL（最可靠的方式）
            self._page.goto(
                self.LOGIN_CONFIG["base_url"] + "/#/platform?platform=%2Fplatform%2F%23%2FHome",
                wait_until="domcontentloaded"
            )
            self._page.wait_for_timeout(2000)
            # 等待常用功能卡片加载
            try:
                self._page.wait_for_selector(".block", timeout=10000)
                logger.debug("主页常用功能已加载")
            except Exception:
                logger.warning("等待主页 .block 元素超时")
            logger.info("已导航到主页")
        except Exception as e:
            logger.warning(f"导航到主页失败（不影响后续执行）: {e}")

    def _detect_logout(self) -> None:
        """
        检测是否已退出登录

        点击操作后检查页面是否回到了登录页（URL 含 login 或出现登录表单）。
        如果检测到退出登录，重置 _is_logged_in 标记，
        确保下一个用例的登录步骤不会被智能跳过。
        """
        try:
            current_url = self._page.url
            login_form = self._page.query_selector('input[placeholder*="手机号"], input[type="password"]')
            if "login" in current_url.lower() or login_form:
                self._is_logged_in = False
                logger.info("检测到退出登录，重置登录态标记")
        except Exception:
            pass

    def _detect_login_state(self) -> None:
        """
        检测是否已登录

        用于手动登录步骤（auto_login=False）执行后的状态同步。
        判断依据：URL 不含 login，且页面出现常用功能卡片(.block)或主页特征。
        如果检测到已登录，设置 _is_logged_in=True，
        确保下一个用例（auto_login=True）复用登录态而非重新登录。
        """
        try:
            current_url = self._page.url
            # 不在登录页 + 页面有 .block（常用功能卡片）= 已登录进入主页
            if "login" not in current_url.lower():
                has_block = self._page.query_selector(".block")
                if has_block:
                    self._is_logged_in = True
                    logger.info("检测到已登录进入主页，设置登录态标记")
        except Exception:
            pass

    # ==================== 选择器与等待辅助 ====================

    # ==================== 语义定位器解析（Phase 3）====================

    @staticmethod
    def _parse_semantic_selector(selector: str):
        """
        解析语义定位器格式

        支持格式：
        - role:button[name="登录"] -> ('role', 'button', {'name': '登录'})
        - role:button -> ('role', 'button', {})
        - text:新建会议 -> ('text', '新建会议', {})
        - placeholder:请输入名称 -> ('placeholder', '请输入名称', {})
        - label:用户名 -> ('label', '用户名', {})
        - testid:submit-btn -> ('testid', 'submit-btn', {})

        Args:
            selector (str): 语义选择器字符串

        Returns:
            tuple: (format_type, value, options) 或 None（非语义选择器）
        """
        if selector.startswith('role:'):
            # 解析 role:button[name="登录"]
            role_part = selector[5:]
            name_match = re.search(r'\[name=["\']([^"\']+)["\']\]', role_part)
            if name_match:
                role_name = role_part[:role_part.index('[')]
                name_value = name_match.group(1)
                return ('role', role_name, {'name': name_value})
            else:
                return ('role', role_part, {})

        elif selector.startswith('text:'):
            return ('text', selector[5:], {})

        elif selector.startswith('placeholder:'):
            return ('placeholder', selector[12:], {})

        elif selector.startswith('label:'):
            return ('label', selector[6:], {})

        elif selector.startswith('testid:'):
            return ('testid', selector[7:], {})

        return None  # 非语义选择器

    def _get_semantic_locator(self, page_or_frame, selector: str):
        """
        根据语义选择器获取 Playwright Locator 对象

        Args:
            page_or_frame: Page 或 Frame 对象
            selector (str): 语义选择器字符串

        Returns:
            Locator 对象 或 None
        """
        parsed = self._parse_semantic_selector(selector)
        if not parsed:
            return None

        fmt, value, options = parsed

        try:
            if fmt == 'role':
                return page_or_frame.get_by_role(value, **options)
            elif fmt == 'text':
                return page_or_frame.get_by_text(value)
            elif fmt == 'placeholder':
                return page_or_frame.get_by_placeholder(value)
            elif fmt == 'label':
                return page_or_frame.get_by_label(value)
            elif fmt == 'testid':
                return page_or_frame.get_by_test_id(value)
        except Exception as e:
            logger.debug(f"语义定位器获取失败: {selector}, {e}")

        return None

    def _resolve_selectors(self, params: Dict[str, Any]) -> List[str]:
        """
        解析参数中的选择器链

        优先级：
        1. params.selectors（多选择器回退链）
        2. params.selector（单选择器，向后兼容）
        3. params.page_key + params.element_key（调 SelectorMapper 取完整链）

        Args:
            params (dict): 步骤参数

        Returns:
            List[str]: 选择器列表（可能为空）
        """
        selectors = params.get("selectors")
        if isinstance(selectors, list) and selectors:
            return [str(s) for s in selectors if s]

        # 单选择器向后兼容
        single = params.get("selector")
        if single:
            return [str(single)]

        # SelectorMapper 解析
        page_key = params.get("page_key")
        element_key = params.get("element_key")
        if page_key and element_key:
            return self._selector_mapper.resolve_selectors(page_key, element_key)

        return []

    def _find_element_in_frames(self, selector: str, timeout: int = 5000, state: str = "visible"):
        """
        在主页面和所有 iframe 中查找元素

        Args:
            selector (str): 选择器
            timeout (int): 超时毫秒
            state (str): 状态 visible/hidden/attached/detached

        Returns:
            tuple: (frame, element) 或 (None, None)
        """
        # 先尝试主页面
        try:
            self._page.wait_for_selector(selector, state=state, timeout=timeout)
            return self._page, None
        except Exception:
            pass

        # 遍历 iframe
        for frame in self._page.frames:
            if frame == self._page.main_frame:
                continue
            try:
                frame.wait_for_selector(selector, state=state, timeout=timeout)
                return frame, None
            except Exception:
                continue

        return None, None

    def _wait_for_element(
        self,
        selector: str,
        timeout: Optional[int] = None,
        state: str = "visible",
    ) -> bool:
        """
        显式等待元素到达指定状态（支持 iframe）

        失败不抛异常，返回 False，由调用方决定后续行为。

        Args:
            selector (str): 选择器
            timeout (Optional[int]): 超时毫秒，None 用默认
            state (str): 状态 visible/hidden/attached/detached

        Returns:
            bool: 是否在超时前达到状态
        """
        actual_timeout = timeout or self.timeout
        try:
            # 先尝试主页面
            self._page.wait_for_selector(selector, state=state, timeout=actual_timeout)
            return True
        except Exception:
            pass

        # 遍历 iframe
        for frame in self._page.frames:
            if frame == self._page.main_frame:
                continue
            try:
                frame.wait_for_selector(selector, state=state, timeout=actual_timeout)
                logger.debug(f"⏳ 在 iframe 中找到元素: {selector}")
                return True
            except Exception:
                continue

        logger.debug(f"⏳ 等待元素未就绪: {selector} state={state}")
        return False

    def _safe_is_visible(self, selector: str) -> bool:
        """安全判断元素是否可见，不抛异常"""
        try:
            return self._page.is_visible(selector)
        except Exception:
            return False

    def _wait_for_loading_overlay(self, timeout: int = 10000) -> None:
        """
        等待加载遮罩两阶段：先短暂等出现，再等消失

        用于导航、点击触发列表加载等场景，避免在遮罩期间操作。
        """
        overlay_selectors = [
            ".el-loading-mask",
            ".loading",
            "[class*='loading']",
        ]
        # 第一阶段：短暂等待出现（很短，避免无遮罩场景空等）
        appeared = False
        for sel in overlay_selectors:
            try:
                self._page.wait_for_selector(sel, state="visible", timeout=800)
                appeared = True
                break
            except Exception:
                continue

        if not appeared:
            return

        # 第二阶段：等待所有遮罩隐藏
        try:
            for sel in overlay_selectors:
                try:
                    self._page.wait_for_selector(sel, state="hidden", timeout=timeout)
                except Exception:
                    continue
        except Exception:
            pass

    def _post_action_wait(self, action_type: str, override: Optional[int] = None) -> None:
        """
        动作完成后短等待，保证 UI 响应

        Args:
            action_type (str): 动作类型
            override (Optional[int]): 用户自定义延时（毫秒），非空则覆盖默认
        """
        if override is not None:
            try:
                self._page.wait_for_timeout(int(override))
            except Exception:
                pass
            return

        defaults = {
            "navigate": 2000,
            "click": 500,
            "fill": 300,
            "select": 500,
            "hover": 300,
            "check": 300,
            "uncheck": 300,
            "press": 200,
            "dialog": 800,
            "scroll": 200,
        }
        delay = defaults.get(action_type, 300)
        try:
            self._page.wait_for_timeout(delay)
        except Exception:
            pass

    def execute_case(
        self,
        case: Dict[str, Any],
        callback: Optional[Callable[[StepResult], None]] = None,
    ) -> CaseExecutionResult:
        """
        执行测试用例

        按顺序执行用例中所有步骤，支持失败重试。
        如果配置了auto_login，会在执行前自动登录。

        Args:
            case (Dict): 用例数据，包含 id, name, steps, config 等字段
            callback (Optional[Callable]): 步骤完成回调函数，用于实时推送

        Returns:
            CaseExecutionResult: 用例执行结果

        Example:
            >>> case = {
            ...     "id": "case_001",
            ...     "name": "正常登录测试",
            ...     "steps": [
            ...         {"order": 1, "name": "打开登录页面", "action": "navigate", "params": {"url": "/login"}}
            ...     ]
            ... }
            >>> result = executor.execute_case(case)
        """
        if not self._is_running:
            self.start()

        case_id = case.get("id", "")
        case_name = case.get("name", "")
        steps = case.get("steps", [])
        case_config = case.get("config", {})

        # 合并配置
        if case_config:
            self.timeout = case_config.get("timeout", self.timeout)
            self.screenshot_enabled = case_config.get("screenshot", self.screenshot_enabled)

        # 用例间登录态重置：保证登录模块用例独立执行
        # 当 reset_login=True 时，清除上一个用例遗留的登录态，回到登录页
        if case_config.get("reset_login", False):
            self.reset_login_state()

        # 如果启用了自动登录且尚未登录，先执行登录
        # 优先使用用例自身的 auto_login 配置，其次使用执行器全局配置
        need_auto_login = case_config.get("auto_login", self.auto_login)
        if need_auto_login and not self._is_logged_in:
            login_success = self.do_login()
            if not login_success:
                logger.warning("自动登录失败，可能影响需要登录态的测试步骤")
        elif need_auto_login and self._is_logged_in:
            # 已登录但需要确保从主页开始（上一个用例可能停留在子页面）
            self._navigate_to_home()
            logger.debug(f"跳过登录: auto_login={need_auto_login}, is_logged_in={self._is_logged_in}")
        else:
            logger.debug(f"跳过登录: auto_login={need_auto_login}, is_logged_in={self._is_logged_in}")

        result = CaseExecutionResult(case_id=case_id, case_name=case_name)
        start_time = datetime.now()

        # ==================== Phase 4: 页面识别 + URL 直达导航 ====================
        # 扫描用例步骤，识别目标页面，直达 URL 跳过中间导航点击步骤
        page_target = None
        skip_orders: set = set()
        if self._page_url_service.is_enabled():
            page_target = self._page_url_service.recognize_page(steps)
            if page_target:
                skip_orders = set(page_target["skip_orders"])
                logger.info(f"[页面直达] 识别到目标页面: {page_target['name']} (置信度={page_target['confidence']})")
                logger.info(f"[页面直达] 直达 URL: {page_target['url']}")
                logger.info(f"[页面直达] 将跳过中间导航步骤: {sorted(skip_orders)}")

                # 如果已登录，直接导航到目标页面 URL
                if self._is_logged_in:
                    logger.info(f"[页面直达] 已登录，直接导航到目标页面...")
                    self._page.goto(page_target["url"], wait_until="domcontentloaded", timeout=30000)
                    self._page.wait_for_timeout(3000)
                    self._wait_for_loading_overlay()
                    logger.info(f"[页面直达] 导航完成，当前 URL: {self._page.url}")
                else:
                    logger.warning("[页面直达] 尚未登录，将在登录后导航到目标页面")

        # 给步骤总数注入 params（用于日志显示）
        step_count = len(steps)
        for step in steps:
            step_params = step.get("params", {}) or {}
            step_params["_total_steps"] = step_count

        try:
            logger.info(f"开始执行用例: {case_name} (id={case_id})")

            for step in steps:
                order = step.get("order", 0)

                # 如果是被标记为跳过的中间导航步骤，跳过执行
                if order in skip_orders:
                    step_result = StepResult(
                        order=order,
                        name=step.get("name", ""),
                        action=step.get("action", ""),
                        status="skipped",
                        log=f"⏭ 页面直达，跳过中间导航步骤",
                    )
                    result.steps_result.append(step_result)
                    if callback:
                        try:
                            callback(step_result)
                        except Exception:
                            pass
                    logger.info(f"⏭ 跳过步骤 {order}: {step.get('name', '')}")
                    continue

                # 执行步骤
                step_result = self.execute_step(step, callback=callback)
                result.steps_result.append(step_result)

                # 如果步骤失败，后续步骤标记为跳过
                # （单步重试已在 execute_step 内完成，这里不再整用例重试）
                if step_result.status == "failed":
                    result.status = "failed"
                    result.error_message = step_result.error

                    # 后续步骤跳过
                    for remaining in steps[len(result.steps_result):]:
                        skipped = StepResult(
                            order=remaining.get("order", 0),
                            name=remaining.get("name", ""),
                            action=remaining.get("action", ""),
                            status="skipped",
                            log="⚠ 前置步骤失败，跳过执行",
                        )
                        result.steps_result.append(skipped)
                    break

            # 所有步骤通过
            if result.status == "pending":
                result.status = "passed"

        except Exception as e:
            result.status = "failed"
            result.error_message = f"执行异常: {str(e)}"
            logger.error(f"用例执行异常: {case_name}, {str(e)}")

        finally:
            result.duration = (datetime.now() - start_time).total_seconds()
            logger.info(f"用例执行完成: {case_name}, 状态={result.status}, 耗时={result.duration:.2f}s")

        return result

    def execute_step(
        self,
        step: Dict[str, Any],
        callback: Optional[Callable[[StepResult], None]] = None,
    ) -> StepResult:
        """
        执行单个测试步骤（支持多选择器回退 + 单步重试）

        Args:
            step (Dict): 步骤定义
                - order (int): 步骤顺序
                - name (str): 步骤名称
                - action (str): 动作类型
                - params (dict): 动作参数
                - expected (str): 预期结果
                - selectors (list): 多选择器回退链（可选）
                - page_key (str): 页面标识（可选，与 element_key 配合）
                - element_key (str): 元素标识（可选）
                - force (bool): 强制点击（可选）
                - wait_after (int): 操作后延时（可选）
            callback (Optional[Callable]): 步骤完成回调

        Returns:
            StepResult: 步骤执行结果
        """
        step_result = StepResult(
            order=step.get("order", 0),
            name=step.get("name", ""),
            action=step.get("action", ""),
            status="running",
            start_time=datetime.now(),
        )

        start_time = datetime.now()
        action = step.get("action", "")
        params = step.get("params", {})
        expected = step.get("expected", "")
        force = step.get("force", params.get("force", False))
        wait_after = step.get("wait_after", params.get("wait_after", None))

        logger.info(f"⏳ 步骤 {step_result.order}/{params.get('_total_steps', '?')}: {step_result.name}")

        # ==================== Phase 3: 元素映射表最高优先级 ====================
        # 查询元素映射表，如命中则注入到 params 中作为最高优先级选择器
        step_name = step.get("name", "")
        step_action = step.get("action", "")
        if self._element_mapping_service.is_enabled() and step_name:
            mapping_result = self._element_mapping_service.get_selector(step_name, step_action)
            if mapping_result:
                mapping_selector = mapping_result["selector"]
                logger.info(f"[元素映射表] 命中: {step_name} → {mapping_selector}")
                # 将映射表选择器注入到 params 中作为最高优先级
                existing_selectors = params.get("selectors")
                if isinstance(existing_selectors, list) and existing_selectors:
                    params["selectors"] = [mapping_selector] + list(existing_selectors)
                else:
                    # 保留原始 selector 作为回退（避免映射表选择器不精确时无备选）
                    original = params.get("selector")
                    if original:
                        params["selectors"] = [mapping_selector, str(original)]
                    else:
                        params["selectors"] = [mapping_selector]
                if not params.get("selector"):
                    params["selector"] = mapping_selector

        # 单步重试循环
        attempt = 0
        max_attempts = self.step_retry_count + 1  # 至少执行一次
        last_error: Optional[Exception] = None

        while attempt < max_attempts:
            attempt += 1
            step_result.status = "running"
            step_result.error = None

            try:
                logger.debug(f"⏳ 执行步骤 {step_result.order}: {step_result.name} (action={action}, attempt={attempt}/{max_attempts})")

                # 智能处理：如果已登录且步骤是导航到登录页或首页，跳过
                # 但退出登录步骤不跳过（通过 params.logout 或步骤名含"退出"判断）
                if (action == "navigate" and self._is_logged_in
                        and not params.get("logout", False)
                        and "退出" not in step_result.name):
                    url = params.get("url", "")
                    if (self.LOGIN_CONFIG["base_url"] in url and
                        ("login" in url.lower() or url == self.LOGIN_CONFIG["base_url"] or url.endswith("/"))):
                        step_result.status = "passed"
                        step_result.log = f"✓ 已登录，跳过导航: {url}"
                        logger.info(f"✓ 已登录，跳过步骤 {step_result.order}: {step_result.name}")
                        break

                # 根据动作类型分发执行
                if action == "navigate":
                    url = params.get("url", "")
                    # 退出登录：导航到登录页时，清除 cookies 强制登出
                    # （SPA hash 路由在已登录时访问 #/login 会被重定向回主页）
                    if "login" in url.lower() or params.get("logout", False):
                        logger.info(f"退出登录: name={step_result.name}, url={url}")
                        try:
                            self.reset_login_state()
                        except Exception as re:
                            logger.error(f"退出登录 reset 失败: {re}")
                            raise
                        step_result.log = f"✓ 退出登录，回到登录页"
                        step_result.status = "passed"
                        break
                    self._do_navigate(params)
                    step_result.log = f"✓ 导航到: {url}"
                    # 导航到登录页后重置登录态标记
                    if self._is_logged_in and "login" in url.lower():
                        self._is_logged_in = False
                        logger.info("导航到登录页，重置登录态标记")
                    # 检测是否已登录：导航后若进入主页，同步登录态标记
                    if not self._is_logged_in:
                        self._detect_login_state()
                elif action == "click":
                    self._do_click(params, force=force)
                    step_result.log = f"✓ 点击元素: {params.get('selector', '') or '多选择器'}"
                    # 检测是否退出登录：点击后页面回到登录页
                    if self._is_logged_in:
                        self._detect_logout()
                    # 检测是否已登录：点击后若进入主页，同步登录态标记
                    if not self._is_logged_in:
                        self._detect_login_state()
                elif action == "fill":
                    self._do_fill(params, force=force)
                    step_result.log = f"✓ 填充输入框: {params.get('selector', '') or '多选择器'} = {params.get('value', '')}"
                elif action == "select":
                    self._do_select(params, force=force)
                    step_result.log = f"✓ 选择下拉框: {params.get('selector', '') or '多选择器'} = {params.get('value', '')}"
                elif action == "hover":
                    self._do_hover(params)
                    step_result.log = f"✓ 悬停元素: {params.get('selector', '') or '多选择器'}"
                elif action == "wait":
                    self._do_wait(params)
                    step_result.log = f"✓ 等待元素: {params.get('selector', '') or '多选择器'}"
                elif action == "scroll":
                    self._do_scroll(params)
                    step_result.log = f"✓ 滚动页面: {params.get('direction', 'down')}"
                elif action == "press":
                    self._do_press(params)
                    step_result.log = f"✓ 按键: {params.get('key', '')}"
                elif action == "check":
                    self._do_check(params, check=True, force=force)
                    step_result.log = f"✓ 勾选: {params.get('selector', '') or '多选择器'}"
                elif action == "uncheck":
                    self._do_check(params, check=False, force=force)
                    step_result.log = f"✓ 取消勾选: {params.get('selector', '') or '多选择器'}"
                elif action == "assert":
                    self._do_assert(params, expected)
                    step_result.log = f"✓ 断言验证: {params.get('type', 'visible')} - 通过"
                elif action == "screenshot":
                    self._do_screenshot(params)
                    step_result.log = "✓ 截图保存"
                else:
                    raise ValueError(f"不支持的动作类型: {action}")

                # 成功：退出重试循环
                step_result.status = "passed"
                last_error = None
                break

            except Exception as e:
                last_error = e
                step_result.status = "failed"
                step_result.error = str(e)
                step_result.log = f"✗ 执行失败: {str(e)}"
                logger.warning(f"⚠ 步骤 {step_result.order} 尝试 {attempt}/{max_attempts} 失败: {str(e)}")

                # 如果还有重试机会，短暂等待后重试
                if attempt < max_attempts:
                    try:
                        self._page.wait_for_timeout(1000)
                    except Exception:
                        pass
                else:
                    # 达到最大重试次数，最终失败
                    step_result.log = f"✗ 步骤失败（已重试 {self.step_retry_count} 次）: {str(e)}"
                    logger.error(f"✗ 步骤 {step_result.order} 最终失败: {str(e)}")

        # 最终状态设置
        if last_error:
            step_result.status = "failed"
        elif step_result.status != "passed":
            step_result.status = "failed"

        # 结束计时
        step_result.end_time = datetime.now()
        step_result.duration = (step_result.end_time - start_time).total_seconds()

        # 成功后执行后置等待
        if step_result.status == "passed":
            self._post_action_wait(action, wait_after)

        # 截图（无论成功与否，如果开启了截图）
        if self.screenshot_enabled:
            try:
                screenshot_name = f"step_{step_result.order}_{datetime.now().strftime('%H%M%S')}.png"
                screenshot_path = os.path.join(self.screenshot_dir, screenshot_name)
                self._page.screenshot(path=screenshot_path)
                step_result.screenshot = screenshot_path
            except Exception as e:
                logger.error(f"✗ 截图失败: {str(e)}")

        # 回调通知
        if callback:
            try:
                callback(step_result)
            except Exception as e:
                logger.error(f"✗ 回调函数执行失败: {str(e)}")

        # 图标日志
        if step_result.status == "passed":
            logger.info(f"✓ 步骤 {step_result.order} 完成 (耗时 {step_result.duration:.2f}s)")
        else:
            logger.error(f"✗ 步骤 {step_result.order} 失败: {step_result.error}")

        return step_result

    # ==================== 动作实现 ====================

    def _do_navigate(self, params: dict) -> None:
        """
        页面导航

        Args:
            params (dict): {"url": str, "wait_until": str (可选)}
        """
        url = params.get("url", "")
        if not url:
            raise ValueError("导航URL不能为空")
        current_url = self._page.url
        # 如果只是 hash 变化（同源），用 JS 修改 location 更可靠，
        # 避免 goto 等待 navigation 事件在 SPA hash 路由上超时
        if self._is_same_origin_hash_change(current_url, url):
            try:
                # 提取 hash 部分设置
                hash_part = url.split("#", 1)[1] if "#" in url else ""
                if hash_part:
                    self._page.evaluate(f'window.location.hash = "#{hash_part}"')
                    self._page.wait_for_timeout(1500)
                    logger.debug(f"hash 导航完成: #{hash_part}")
                    self._wait_for_loading_overlay()
                    return
            except Exception as e:
                logger.debug(f"hash 导航失败，回退到 goto: {e}")

        # 默认用 domcontentloaded 避免 SPA 持续网络活动导致 networkidle 超时
        wait_until = params.get("wait_until", "domcontentloaded")
        try:
            self._page.goto(url, wait_until=wait_until, timeout=15000)
        except PlaywrightTimeoutError:
            # hash 路由或 SPA 导航可能不触发 goto 完成事件，忽略超时
            logger.debug(f"导航 goto 超时（忽略）: {url}")
        # 等待加载遮罩消失
        self._wait_for_loading_overlay()

    def _is_same_origin_hash_change(self, current: str, target: str) -> bool:
        """判断目标 URL 是否与当前同源且仅 hash 不同"""
        try:
            # 提取 scheme+host 部分
            def base(u):
                return u.split("#")[0].split("?")[0].rstrip("/")
            return base(current) == base(target) and "#" in target
        except Exception:
            return False

    def _do_click(self, params: dict, force: bool = False) -> None:
        """
        点击元素（多选择器回退 + 语义定位器 + 强制点击 + JS点击回退）

        点击策略：
        1. 语义定位器点击（role:/text:/placeholder:等）
        2. CSS/XPath 普通点击 → force点击 → JS点击
        3. iframe 中点击

        Args:
            params (dict): 支持 {"selector": str} / {"selectors": [str]} / {"page_key","element_key"}
            force (bool): 是否强制点击（忽略覆盖层）
        """
        selectors = self._resolve_selectors(params)
        if not selectors:
            raise ValueError("选择器不能为空")

        last_error: Optional[Exception] = None

        for selector in selectors:
            # ==================== Phase 6: frame 路径选择器 ====================
            if selector.startswith('iframe') and ' >> ' in selector:
                frame_selector, element_selector = selector.split(' >> ', 1)

                # 优先使用语义定位器
                parsed_inner = self._parse_semantic_selector(element_selector)
                if parsed_inner:
                    try:
                        frame_loc = self._page.frame_locator(frame_selector)
                        locator = self._get_semantic_locator(frame_loc, element_selector)
                        if locator:
                            locator.first.click(timeout=5000, force=force)
                            logger.debug(f"[OK] iframe semantic click: {selector}")
                            self._wait_for_loading_overlay()
                            return
                    except Exception as e:
                        last_error = e
                        logger.debug(f"[WARN] iframe semantic click failed {selector}: {e}")

                # CSS 选择器在 iframe 中
                try:
                    frame_loc = self._page.frame_locator(frame_selector)
                    frame_loc.locator(element_selector).first.click(timeout=5000, force=force)
                    logger.debug(f"[OK] frame path click: {selector}")
                    self._wait_for_loading_overlay()
                    return
                except Exception as e:
                    last_error = e
                    logger.debug(f"[WARN] frame path click failed {selector}: {e}")

                # 回退: 遍历所有 iframe 尝试
                for frame in self._page.frames:
                    if frame == self._page.main_frame:
                        continue
                    try:
                        frame.wait_for_selector(element_selector, state="visible", timeout=3000)
                        frame.click(element_selector, timeout=5000, force=True)
                        logger.debug(f"[OK] iframe traversal click: {element_selector}")
                        self._wait_for_loading_overlay()
                        return
                    except Exception as e3:
                        last_error = e3
                        continue

                continue

            # ==================== Phase 3: 语义定位器优先 ====================
            parsed = self._parse_semantic_selector(selector)
            if parsed:
                # 语义定位器：使用 Playwright 原生 API
                try:
                    locator = self._get_semantic_locator(self._page, selector)
                    if locator:
                        locator.first.scroll_into_view_if_needed(timeout=5000)
                        self._page.wait_for_timeout(200)
                        locator.first.click(timeout=5000, force=force)
                        logger.debug(f"✓ 语义定位器点击成功: {selector}")
                        self._wait_for_loading_overlay()
                        return
                except Exception as e:
                    last_error = e
                    logger.debug(f"⚠ 语义定位器点击失败 {selector}: {e}")

                # 语义定位器在 iframe 中尝试
                for frame in self._page.frames:
                    if frame == self._page.main_frame:
                        continue
                    try:
                        locator = self._get_semantic_locator(frame, selector)
                        if locator:
                            locator.first.click(timeout=5000, force=True)
                            logger.debug(f"✓ iframe 语义定位器点击成功: {selector}")
                            self._wait_for_loading_overlay()
                            return
                    except Exception as e3:
                        last_error = e3
                        continue

                # 语义定位器失败，继续尝试下一个选择器
                continue

            # ==================== CSS/XPath 选择器 ====================
            # 等待元素可见（宽松等待，不阻塞后续操作）
            self._wait_for_element(selector, timeout=10000, state="visible")

            # 滚动到可见区域（确保元素不被遮挡）
            try:
                el = self._page.query_selector(selector)
                if el:
                    el.scroll_into_view_if_needed()
                    self._page.wait_for_timeout(200)
            except Exception:
                pass  # 滚动失败不阻塞后续操作

            # 策略1: 主页面普通点击
            try:
                self._page.click(selector, timeout=5000, force=force)
                logger.debug(f"✓ 主页面点击成功: {selector}")
                self._wait_for_loading_overlay()
                return
            except Exception as e:
                last_error = e
                logger.debug(f"⚠ 主页面点击失败 {selector}: {e}")

            # 策略2: 主页面force点击（忽略遮挡）
            if not force:
                try:
                    self._page.click(selector, timeout=5000, force=True)
                    logger.debug(f"✓ 主页面force点击成功: {selector}")
                    self._wait_for_loading_overlay()
                    return
                except Exception as e2:
                    last_error = e2
                    logger.debug(f"⚠ 主页面force点击失败 {selector}: {e2}")

            # 策略3: iframe 中点击
            for frame in self._page.frames:
                if frame == self._page.main_frame:
                    continue
                try:
                    frame.wait_for_selector(selector, state="visible", timeout=5000)
                    frame.click(selector, timeout=5000, force=True)
                    logger.debug(f"✓ iframe 中点击成功: {selector}")
                    self._wait_for_loading_overlay()
                    return
                except Exception as e3:
                    last_error = e3
                    continue

            # 策略4: JS 点击回退（处理元素被遮挡或不可操作的情况）
            try:
                el = self._page.query_selector(selector)
                if el:
                    self._page.evaluate('(el) => el.click()', el)
                    logger.debug(f"✓ JS点击成功: {selector}")
                    self._wait_for_loading_overlay()
                    return
            except Exception as e4:
                last_error = e4
                logger.debug(f"⚠ JS点击失败 {selector}: {e4}")

        raise Exception(f"✗ 无法点击元素（已尝试 {len(selectors)} 个选择器）: {selectors} 最后错误: {last_error}")

    def _do_fill(self, params: dict, force: bool = False) -> None:
        """
        填充输入框（多选择器回退 + 语义定位器 + 强制清空 + 支持 iframe）

        先 click(count=3, force=True) 全选 → Ctrl+A → Delete → 等待 100ms → fill(value)

        Args:
            params (dict): {"selector"|"selectors"|"page_key","element_key": str, "value": str}
            force (bool): 是否强制点击
        """
        selectors = self._resolve_selectors(params)
        if not selectors:
            raise ValueError("选择器不能为空")
        value = params.get("value", "")

        last_error: Optional[Exception] = None

        for selector in selectors:
            # ==================== Phase 6: frame 路径选择器 ====================
            if selector.startswith('iframe') and ' >> ' in selector:
                frame_selector, element_selector = selector.split(' >> ', 1)

                # 优先使用语义定位器
                parsed_inner = self._parse_semantic_selector(element_selector)
                if parsed_inner:
                    try:
                        frame_loc = self._page.frame_locator(frame_selector)
                        locator = self._get_semantic_locator(frame_loc, element_selector)
                        if locator:
                            locator.first.click(timeout=5000, force=True)
                            self._page.keyboard.press("Control+A")
                            self._page.keyboard.press("Delete")
                            self._page.wait_for_timeout(100)
                            locator.first.fill(str(value), timeout=5000)
                            logger.debug(f"[OK] iframe semantic fill: {selector} = {value}")
                            return
                    except Exception as e:
                        last_error = e
                        logger.debug(f"[WARN] iframe semantic fill failed {selector}: {e}")

                # CSS 选择器在 iframe 中
                try:
                    frame_loc = self._page.frame_locator(frame_selector)
                    frame_loc.locator(element_selector).first.click(timeout=5000, force=True)
                    self._page.keyboard.press("Control+A")
                    self._page.keyboard.press("Delete")
                    self._page.wait_for_timeout(100)
                    frame_loc.locator(element_selector).first.fill(str(value), timeout=5000)
                    logger.debug(f"[OK] frame path fill: {selector} = {value}")
                    return
                except Exception as e:
                    last_error = e
                    logger.debug(f"[WARN] frame path fill failed {selector}: {e}")

                continue

            # ==================== Phase 3: 语义定位器优先 ====================
            parsed = self._parse_semantic_selector(selector)
            if parsed:
                # 语义定位器：使用 Playwright 原生 API
                try:
                    locator = self._get_semantic_locator(self._page, selector)
                    if locator:
                        locator.first.click(timeout=5000, force=True)
                        self._page.keyboard.press("Control+A")
                        self._page.keyboard.press("Delete")
                        self._page.wait_for_timeout(100)
                        locator.first.fill(str(value), timeout=5000)
                        logger.debug(f"✓ 语义定位器填充成功: {selector} = {value}")
                        return
                except Exception as e:
                    last_error = e
                    logger.debug(f"⚠ 语义定位器填充失败 {selector}: {e}")

                # 语义定位器在 iframe 中尝试
                for frame in self._page.frames:
                    if frame == self._page.main_frame:
                        continue
                    try:
                        locator = self._get_semantic_locator(frame, selector)
                        if locator:
                            locator.first.click(timeout=5000, force=True)
                            frame.keyboard.press("Control+A")
                            frame.keyboard.press("Delete")
                            frame.wait_for_timeout(100)
                            locator.first.fill(str(value), timeout=5000)
                            logger.debug(f"✓ iframe 语义定位器填充成功: {selector} = {value}")
                            return
                    except Exception as e2:
                        last_error = e2
                        continue

                # 语义定位器失败，继续尝试下一个选择器
                continue

            # ==================== CSS/XPath 选择器 ====================
            # 先在主页面尝试
            try:
                self._page.wait_for_selector(selector, state="visible", timeout=10000)

                # 强制清空
                self._page.click(selector, click_count=3, force=True, timeout=5000)
                self._page.keyboard.press("Control+A")
                self._page.keyboard.press("Delete")
                self._page.wait_for_timeout(100)

                # 填充
                self._page.fill(selector, str(value), timeout=5000)
                logger.debug(f"✓ 主页面填充成功: {selector} = {value}")
                return
            except Exception as e:
                last_error = e
                logger.debug(f"⚠ 主页面填充失败 {selector}: {e}")

            # 在 iframe 中尝试
            for frame in self._page.frames:
                if frame == self._page.main_frame:
                    continue
                try:
                    frame.wait_for_selector(selector, state="visible", timeout=5000)
                    frame.click(selector, click_count=3, force=True, timeout=5000)
                    frame.keyboard.press("Control+A")
                    frame.keyboard.press("Delete")
                    frame.wait_for_timeout(100)
                    frame.fill(selector, str(value), timeout=5000)
                    logger.debug(f"✓ iframe 中填充成功: {selector} = {value}")
                    return
                except Exception as e2:
                    last_error = e2
                    continue

        raise Exception(f"✗ 无法填充元素（已尝试 {len(selectors)} 个选择器）: {selectors} 最后错误: {last_error}")

    def _do_select(self, params: dict, force: bool = False) -> None:
        """
        选择下拉框（多选择器回退）

        Args:
            params (dict): {"selector"|"selectors": str|[str], "value": str}
            force (bool): 保留参数以保持接口一致
        """
        selectors = self._resolve_selectors(params)
        if not selectors:
            raise ValueError("选择器不能为空")
        value = params.get("value", "")

        last_error: Optional[Exception] = None
        for selector in selectors:
            try:
                self._wait_for_element(selector, timeout=10000, state="visible")
                self._page.select_option(selector, str(value), timeout=5000)
                logger.debug(f"✓ 下拉选择成功: {selector} = {value}")
                return
            except Exception as e:
                last_error = e
                logger.debug(f"⚠ 下拉选择失败 {selector}: {e}")
                continue

        raise Exception(f"✗ 无法选择下拉框（已尝试 {len(selectors)} 个选择器）: {selectors} 最后错误: {last_error}")

    def _do_hover(self, params: dict) -> None:
        """
        鼠标悬停（多选择器回退）

        Args:
            params (dict): {"selector"|"selectors": str|[str]}
        """
        selectors = self._resolve_selectors(params)
        if not selectors:
            raise ValueError("选择器不能为空")

        last_error: Optional[Exception] = None
        for selector in selectors:
            try:
                self._wait_for_element(selector, timeout=10000, state="visible")
                self._page.hover(selector, timeout=5000)
                logger.debug(f"✓ 悬停成功: {selector}")
                return
            except Exception as e:
                last_error = e
                logger.debug(f"⚠ 悬停失败 {selector}: {e}")
                continue

        raise Exception(f"✗ 无法悬停元素（已尝试 {len(selectors)} 个选择器）: {selectors} 最后错误: {last_error}")

    def _do_wait(self, params: dict) -> None:
        """
        等待元素出现（多选择器回退 + 支持 iframe）

        多选择器策略：先快速检查每个选择器是否已存在（1s超时），
        如果都不存在，再按顺序等待，每个选择器分配总超时的均等份额。

        Args:
            params (dict): {"selector"|"selectors": str|[str], "timeout": int}
        """
        selectors = self._resolve_selectors(params)
        if not selectors:
            raise ValueError("选择器不能为空")
        total_timeout = params.get("timeout", self.timeout)

        # 阶段1：快速检查 — 每个选择器 1s 超时，看元素是否已存在
        for selector in selectors:
            try:
                self._page.wait_for_selector(selector, timeout=1000)
                logger.debug(f"✓ 主页面等待元素成功（快速）: {selector}")
                return
            except Exception:
                pass
            # iframe 快速检查
            for frame in self._page.frames:
                if frame == self._page.main_frame:
                    continue
                try:
                    frame.wait_for_selector(selector, timeout=1000)
                    logger.debug(f"✓ iframe 中等待元素成功（快速）: {selector}")
                    return
                except Exception:
                    continue

        # 阶段2：按顺序等待 — 每个选择器分配均等超时
        per_selector_timeout = max(total_timeout // len(selectors), 5000)
        last_error: Optional[Exception] = None
        for selector in selectors:
            # 先在主页面尝试
            try:
                self._page.wait_for_selector(selector, timeout=per_selector_timeout)
                logger.debug(f"✓ 主页面等待元素成功: {selector}")
                return
            except Exception as e:
                last_error = e
                logger.debug(f"⚠ 主页面等待元素失败 {selector}: {e}")

            # 在 iframe 中尝试
            for frame in self._page.frames:
                if frame == self._page.main_frame:
                    continue
                try:
                    frame.wait_for_selector(selector, timeout=per_selector_timeout)
                    logger.debug(f"✓ iframe 中等待元素成功: {selector}")
                    return
                except Exception as e2:
                    last_error = e2
                    continue

        raise Exception(f"✗ 等待元素超时（已尝试 {len(selectors)} 个选择器，每个 {per_selector_timeout}ms）: {selectors} 最后错误: {last_error}")

    def _do_scroll(self, params: dict) -> None:
        """
        滚动页面

        Args:
            params (dict): {"direction": "down"/"up", "distance": int}
        """
        direction = params.get("direction", "down")
        distance = params.get("distance", 300)

        if direction == "down":
            self._page.evaluate(f"window.scrollBy(0, {distance})")
        elif direction == "up":
            self._page.evaluate(f"window.scrollBy(0, -{distance})")
        else:
            raise ValueError(f"不支持的滚动方向: {direction}")

    def _do_press(self, params: dict) -> None:
        """
        按键操作

        Args:
            params (dict): {"key": str, "selector": Optional[str]}
        """
        key = params.get("key", "")
        selector = params.get("selector", "")

        if selector:
            self._page.press(selector, key)
        else:
            self._page.keyboard.press(key)

    def _do_check(self, params: dict, check: bool = True, force: bool = False) -> None:
        """
        勾选/取消勾选（多选择器回退 + force）

        Args:
            params (dict): {"selector"|"selectors": str|[str]}
            check (bool): True=勾选, False=取消勾选
            force (bool): 是否强制点击
        """
        selectors = self._resolve_selectors(params)
        if not selectors:
            raise ValueError("选择器不能为空")

        last_error: Optional[Exception] = None
        for selector in selectors:
            try:
                self._wait_for_element(selector, timeout=10000, state="visible")
                if check:
                    self._page.check(selector, timeout=5000, force=force)
                else:
                    self._page.uncheck(selector, timeout=5000, force=force)
                logger.debug(f"✓ {'勾选' if check else '取消勾选'}成功: {selector}")
                return
            except Exception as e:
                last_error = e
                logger.debug(f"⚠ {'勾选' if check else '取消勾选'}失败 {selector}: {e}")
                continue

        raise Exception(f"✗ 无法勾选/取消勾选（已尝试 {len(selectors)} 个选择器）: {selectors} 最后错误: {last_error}")

    def _do_screenshot(self, params: dict) -> None:
        """
        手动截图

        Args:
            params (dict): {"path": Optional[str], "full_page": bool}
        """
        path = params.get("path", os.path.join(self.screenshot_dir, f"manual_{datetime.now().strftime('%H%M%S')}.png"))
        full_page = params.get("full_page", False)
        self._page.screenshot(path=path, full_page=full_page)

    def _do_assert(self, params: dict, expected: str = "") -> None:
        """
        执行断言验证（多选择器回退用于 visible/text/contains/value/count）

        支持多种断言类型：visible, hidden, text, contains, value, count, url, title

        Args:
            params (dict): 断言参数
            expected (str): 期望描述

        Raises:
            AssertionError: 断言失败时抛出
        """
        assert_type = params.get("type", "visible")
        selectors = self._resolve_selectors(params)
        expected_value = params.get("expected", "")
        # 调试日志：记录断言参数（排查 min_count 空值问题）
        if assert_type in ("element_count_min", "count"):
            logger.info(f"[DEBUG _do_assert] type={assert_type}, params={params}, expected={repr(expected)}, min_count={repr(params.get('min_count'))}")

        if assert_type == "visible":
            if not selectors:
                raise AssertionError("断言 visible 缺少选择器")
            for selector in selectors:
                if self._wait_for_element(selector, timeout=self.timeout, state="visible"):
                    return
            raise AssertionError(f"✗ 断言失败: 所有选择器均不可见: {selectors}")

        elif assert_type == "hidden":
            if not selectors:
                raise AssertionError("断言 hidden 缺少选择器")
            for selector in selectors:
                try:
                    self._page.wait_for_selector(selector, state="hidden", timeout=self.timeout)
                    return
                except Exception:
                    continue
            raise AssertionError(f"✗ 断言失败: 元素仍然可见: {selectors}")

        elif assert_type == "text":
            if not selectors:
                raise AssertionError("断言 text 缺少选择器")
            last_err: Optional[Exception] = None
            for selector in selectors:
                try:
                    element = self._page.wait_for_selector(selector, timeout=self.timeout)
                    if element:
                        text = element.text_content() or ""
                        if text.strip() == expected_value.strip():
                            return
                        last_err = AssertionError(f"断言失败: 期望文本 '{expected_value}'，实际 '{text.strip()}'")
                except Exception as e:
                    last_err = e
                    continue
            raise AssertionError(f"✗ 断言 text 失败: {selectors} 错误: {last_err}")

        elif assert_type == "contains":
            if not selectors:
                raise AssertionError("断言 contains 缺少选择器")
            last_err = None
            for selector in selectors:
                try:
                    element = self._page.wait_for_selector(selector, timeout=self.timeout)
                    if element:
                        text = element.text_content() or ""
                        if expected_value in text:
                            return
                        last_err = AssertionError(f"文本不包含 '{expected_value}'，实际 '{text}'")
                except Exception as e:
                    last_err = e
                    continue
            raise AssertionError(f"✗ 断言 contains 失败: {selectors} 错误: {last_err}")

        elif assert_type == "value":
            if not selectors:
                raise AssertionError("断言 value 缺少选择器")
            last_err = None
            for selector in selectors:
                try:
                    element = self._page.wait_for_selector(selector, timeout=self.timeout)
                    if element:
                        actual_value = element.input_value()
                        if actual_value == expected_value:
                            return
                        last_err = AssertionError(f"期望值 '{expected_value}'，实际 '{actual_value}'")
                except Exception as e:
                    last_err = e
                    continue
            raise AssertionError(f"✗ 断言 value 失败: {selectors} 错误: {last_err}")

        elif assert_type == "count":
            if not selectors:
                raise AssertionError("断言 count 缺少选择器")
            # 健壮处理空值/非数字
            try:
                expected_count = int(expected_value) if (expected_value is not None and str(expected_value).strip() != "") else 0
            except (ValueError, TypeError):
                expected_count = 0
            total = 0
            for selector in selectors:
                try:
                    elements = self._page.query_selector_all(selector)
                    total += len(elements)
                except Exception:
                    continue
            if total != expected_count:
                raise AssertionError(f"✗ 断言失败: 期望数量 {expected_count}，实际 {total}")

        elif assert_type == "url_contains":
            current_url = self._page.url
            if expected_value not in current_url:
                raise AssertionError(f"✗ 断言失败: 当前URL '{current_url}' 不包含 '{expected_value}'")

        elif assert_type == "title":
            current_title = self._page.title()
            if expected_value not in current_title:
                raise AssertionError(f"✗ 断言失败: 当前标题 '{current_title}' 不包含 '{expected_value}'")

        elif assert_type == "element_exists":
            # 验证元素存在（不要求可见，只要 DOM 中存在即可）
            if not selectors:
                raise AssertionError("断言 element_exists 缺少选择器")
            found = False
            for selector in selectors:
                try:
                    el = self._page.query_selector(selector)
                    if el is not None:
                        found = True
                        break
                except Exception:
                    continue
            if not found:
                raise AssertionError(f"✗ 断言失败: 元素不存在: {selectors}")

        elif assert_type == "element_count_min":
            # 验证元素数量 >= min_count
            if not selectors:
                raise AssertionError("断言 element_count_min 缺少选择器")
            # 支持从 params.min_count 或 expected 中获取最小数量
            # 健壮处理：空字符串/None/非数字值统一回退为 0
            min_count = params.get("min_count")
            if min_count is None or str(min_count).strip() == "":
                min_count = expected_value
            try:
                min_count = int(min_count) if (min_count is not None and str(min_count).strip() != "") else 0
            except (ValueError, TypeError):
                min_count = 0
            total = 0
            for selector in selectors:
                try:
                    elements = self._page.query_selector_all(selector)
                    total += len(elements)
                except Exception:
                    continue
            if total < min_count:
                raise AssertionError(f"✗ 断言失败: 期望至少 {min_count} 个元素，实际 {total} 个: {selectors}")

        else:
            raise ValueError(f"不支持的断言类型: {assert_type}")

    def stop(self) -> None:
        """
        停止执行器

        清理浏览器资源。
        """
        try:
            if self._page:
                self._page.close()
            if self._context:
                self._context.close()
            if self._browser:
                self._browser.close()
            if self._playwright:
                self._playwright.stop()

            self._is_running = False
            logger.info("Playwright 执行器已停止")

        except Exception as e:
            logger.error(f"执行器停止失败: {str(e)}")
