#!/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
import time
import urllib.parse as _urlparse_mod
from typing import Optional, Callable, Dict, Any, List
from urllib.parse import urlparse, urlunsplit, parse_qs, unquote
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
from app.services.semantic_resolver import get_semantic_resolver
from app.executors.ui_context import ExecutionContext

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",  # 兼容默认值，实际使用 self.base_url（配置可覆盖）
        "username": "admin@xty",
        "password": "Ubains@13579",
        "captcha": "csba",
    }

    # 已知被测系统主机集合（方案A·SUT地址跟随）：
    # 导航/接口步骤若指向这些主机，执行时自动替换为当前配置的 SUT 主机（self.base_url）。
    # 默认从 LOGIN_CONFIG 派生（https://192.168.5.44），覆盖全部存量用例；
    # 未来若切到其他被测系统并回切，可在此追加主机（如 "192.168.5.69"）。
    KNOWN_SUT_HOSTS = {urlparse(LOGIN_CONFIG["base_url"]).netloc}

    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 {}
        # 被测系统基础 URL：优先取执行配置（execution_service 传入 settings.TARGET_URL），
        # 其次回退到全局配置，保证切换目标系统后无需改代码。
        self.base_url: str = (
            self.config.get("base_url")
            or getattr(settings, "TARGET_URL", "")
            or self.LOGIN_CONFIG["base_url"]
        ).rstrip("/") or self.LOGIN_CONFIG["base_url"]
        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._semantic_resolver = get_semantic_resolver()

        # 内部状态
        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

        # 语义化执行上下文（execute_case 入口按用例重置，防跨用例污染）
        self._ctx: Optional[ExecutionContext] = None

        # 语义化执行：当前页面 scope（页面注册表 id，登录/导航后同步，仅记录）
        self._current_scope: Optional[str] = None

        # 取消请求标记：由 run_all_cases_sync 在创建执行器后设置执行 ID，
        # execute_step 在每次重试前检查 is_cancel_requested 并提前退出
        self._execution_id: Optional[str] = None
        self._cancel_requested: 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.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.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.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）执行后的状态同步。
        多信号检测：.block 卡片 / .home_nav_left 导航栏 / 登录表单消失 + URL 离开登录路由。
        如果检测到已登录，设置 _is_logged_in=True，
        确保下一个用例（auto_login=True）复用登录态而非重新登录。
        """
        try:
            current_url = self._page.url
            if "login" not in current_url.lower():
                # 多信号检测
                has_block = self._page.query_selector(".block")
                has_nav = self._page.query_selector(".home_nav_left")
                login_form = self._page.query_selector('input[placeholder*="手机号"]')
                on_login_route = (
                    "platform%2Flogin" in current_url
                    or current_url.rstrip("/").endswith("#/login")
                )

                if has_block or has_nav or (not login_form and not on_login_route):
                    self._is_logged_in = True
                    logger.info(
                        f"检测到已登录进入主页，设置登录态标记 "
                        f"(block={bool(has_block)}, nav={bool(has_nav)}, "
                        f"login_form={bool(login_form)}, login_route={on_login_route})"
                    )
                    # 语义化执行：登录后同步当前页面 scope
                    self._sync_current_scope()
        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

    # ==================== micro-app 穿透支持 ====================
    # @micro-zoe/micro-app 框架将子应用内容渲染在 <micro-app> → <micro-app-body>
    # 自定义元素中（非 iframe、非 shadow DOM）。Playwright 的标准
    # wait_for_selector/click 无法穿透自定义元素边界，需用 evaluate 在
    # micro-app-body 内部直接查询和操作。

    _MICRO_APP_JS = """(args) => {
        const {cssSel, hasText, action, value, visibleOnly} = args;
        // 文本归一化：\xa0（&nbsp;）等不可见空白在 innerText/textContent 中常见，
        // 与用例中的普通空格对比时需归一化后再匹配
        const norm = s => (s || '').replace(/[\\s\\u00a0]+/g, ' ');
        const bodies = document.querySelectorAll('micro-app micro-app-body');
        for (const body of bodies) {
            let elements;
            try {
                elements = cssSel && cssSel !== '*' ? body.querySelectorAll(cssSel) : [body];
            } catch(e) { continue; }
            for (const el of elements) {
                if (hasText && !norm(el.textContent).includes(hasText)) continue;
                if (visibleOnly && !(el.offsetWidth > 0 || el.offsetHeight > 0 || el.getClientRects().length > 0)) continue;
                if (action === 'exists') return true;
                if (action === 'text') return el.textContent || '';
                if (action === 'click') {
                    el.scrollIntoView({block: 'center'});
                    el.click();
                    return true;
                }
                if (action === 'fill') {
                    el.focus();
                    el.value = value;
                    el.dispatchEvent(new Event('input', {bubbles: true}));
                    el.dispatchEvent(new Event('change', {bubbles: true}));
                    return true;
                }
            }
        }
        return action === 'text' ? null : false;
    }"""

    @staticmethod
    def _split_micro_app_selector(selector: str):
        """
        拆分选择器：剥离 Playwright 伪选择器（:has-text 等），
        返回 (标准 CSS 选择器, 文本匹配条件)。XPath 选择器返回 None 表示不支持。

        Args:
            selector (str): 原始选择器

        Returns:
            tuple: (css_selector or None, has_text str)
        """
        # XPath 不走 micro-app 回退
        if selector.startswith('/') or selector.startswith('('):
            return None, ""
        text_match = re.search(r':has-text\((["\'])(.*?)\1\)', selector)
        has_text = text_match.group(2) if text_match else ""
        css_sel = re.sub(r':has-text\((["\']).*?\1\)', '', selector)
        css_sel = re.sub(r':visible\b', '', css_sel).strip()
        if css_sel == '':
            css_sel = '*'
        return css_sel, has_text

    def _do_in_micro_apps(
        self,
        action: str,
        selector: str,
        value: str = "",
        timeout: int = 5000,
        visible_only: bool = False,
    ):
        """
        在 <micro-app micro-app-body> 自定义元素内执行查找/取文本/点击/填充

        Args:
            action (str): exists / text / click / fill
            selector (str): 原始选择器（支持 :has-text() 伪选择器）
            value (str): fill 时的填充值
            timeout (int): 轮询超时毫秒
            visible_only (bool): 是否仅匹配可见元素

        Returns:
            exists/click/fill: bool；text: str 或 None
        """
        css_sel, has_text = self._split_micro_app_selector(selector)
        if css_sel is None:
            return False if action != 'text' else None

        deadline = time.time() + timeout / 1000
        while True:
            try:
                result = self._page.evaluate(
                    self._MICRO_APP_JS,
                    {"cssSel": css_sel, "hasText": has_text, "action": action,
                     "value": str(value), "visibleOnly": visible_only},
                )
                if result is not False and result is not None:
                    return result
            except Exception as e:
                logger.debug(f"micro-app evaluate 失败 {selector}: {e}")
            if time.time() >= deadline:
                break
            self._page.wait_for_timeout(500)

        logger.debug(f"⚠ micro-app 内未找到/操作失败: {selector} (action={action})")
        return False if action != 'text' else None

    def _wait_for_element(
        self,
        selector: str,
        timeout: Optional[int] = None,
        state: str = "visible",
    ) -> 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

        # 遍历 micro-app 自定义元素（@micro-zoe/micro-app 渲染在 micro-app-body）
        if self._do_in_micro_apps('exists', selector, timeout=actual_timeout,
                                  visible_only=(state in ("visible", "attached"))):
            logger.debug(f"⏳ 在 micro-app 中找到元素: {selector}")
            return True

        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 _sync_current_scope(self) -> None:
        """
        语义化执行：同步当前页面 scope（页面注册表 id）

        登录/导航成功后调用，仅记录 _current_scope，本阶段不参与元素定位。
        识别不到时置 None（零行为变化）。
        """
        try:
            rec = self._page_url_service.recognize_current_page(self._page)
            self._current_scope = rec["id"] if rec else None
            if rec:
                logger.debug(
                    f"[scope] 当前页面识别: {rec['id']} ({rec['name']}) confidence={rec['confidence']}"
                )
        except Exception as e:
            logger.debug(f"[scope] 页面识别异常: {e}")

    def _navigate_to_target_page(self, page_target: Dict[str, Any]) -> None:
        """
        Phase 4: URL 直达导航到目标页面

        直接 goto 目标页面 URL，跳过中间导航点击步骤。
        导航后等待页面内容加载完成。
        """
        try:
            logger.info(f"[页面直达] 正在导航到: {page_target['url']}")
            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}")
            # 语义化执行：导航成功后同步当前页面 scope
            self._sync_current_scope()
        except Exception as e:
            logger.warning(f"[页面直达] 导航到目标页面失败: {e}")

    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", {})

        # 语义化执行：按用例重置执行上下文（防 run_all_cases_sync 复用执行器跨用例污染）
        self._ctx = ExecutionContext(case.get("parameters") or {})
        self._ctx.set("caseName", case_name)
        self._ctx.set("caseId", case_id)

        # 合并配置
        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()
        navigated_to_target = False  # 是否已完成 URL 直达导航
        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._navigate_to_target_page(page_target)
                    navigated_to_target = True
                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:
                    # 兜底：Phase 4 未完成 URL 直达（登录检测失败等），
                    # 且当前 URL 既非目标页面也非登录页 → 强制直达目标 URL
                    if not navigated_to_target and page_target is not None:
                        current_url = self._page.url
                        target_url = page_target["url"]
                        on_login_route = (
                            "login" in current_url.lower()
                            or "platform%2Flogin" in current_url
                            or current_url.rstrip("/").endswith("#/login")
                        )
                        if not on_login_route and target_url not in current_url:
                            logger.warning(
                                f"[页面直达] 兜底: 步骤 {order} 处于跳过列表但未完成直达导航，"
                                f"强制导航到 {target_url} (当前 URL: {current_url})"
                            )
                            self._navigate_to_target_page(page_target)
                            navigated_to_target = True
                    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

                # finally/on_failure 步骤不在主流程执行，统一推迟到主流程结束后的
                # _run_finally_steps 后置块（否则会被执行两次——主循环一遍 + 后置一遍）
                if step.get("run_on", "always") in ("finally", "on_failure"):
                    logger.debug(f"步骤 {order} run_on={step.get('run_on')}，推迟到后置执行")
                    continue

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

                # 同步登录态：登录点击/导航后可能已进入主页，
                # 多信号检测确保 _is_logged_in 及时置位，触发 Phase 4 URL 直达
                if not self._is_logged_in and step_result.status == "passed":
                    # 登录类点击步骤多等待一小段时间，确保 SPA 跳转完成
                    if "登录" in step_result.name and step_result.action == "click":
                        self._page.wait_for_timeout(1500)
                        logger.debug(f"登录点击步骤 {order} 后等待 SPA 跳转并重新检测登录态")
                    self._detect_login_state()

                # Phase 4：用例内登录成功后，若已识别目标页面且尚未直达导航，执行 URL 直达
                if (
                    not navigated_to_target
                    and page_target is not None
                    and self._is_logged_in
                ):
                    logger.info(f"[页面直达] 步骤 {order} 后检测到登录成功，导航到目标页面: {page_target['name']}")
                    self._navigate_to_target_page(page_target)
                    navigated_to_target = True

                # 如果步骤失败，后续步骤标记为跳过
                # （单步重试已在 execute_step 内完成，这里不再整用例重试）
                # 注意：finally/on_failure 步骤不跳过，保留到主流程结束后的后置执行
                if step_result.status == "failed":
                    result.status = "failed"
                    result.error_message = step_result.error

                    # 后续步骤跳过（仅跳过 always/on_success）
                    for remaining in steps[len(result.steps_result):]:
                        run_on = remaining.get("run_on", "always")
                        if run_on in ("finally", "on_failure"):
                            continue
                        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()
            # 后置执行 finally/on_failure 步骤（数据自清理等）
            # 无论主流程成功/失败/异常都执行；失败不掩盖主结果（仅标注）
            self._run_finally_steps(steps, result, callback)
            logger.info(f"用例执行完成: {case_name}, 状态={result.status}, 耗时={result.duration:.2f}s")

        return result

    def _run_finally_steps(
        self,
        steps: List[Dict[str, Any]],
        result,
        callback: Optional[Callable[[StepResult], None]] = None,
    ) -> None:
        """
        主流程结束后执行 finally/on_failure 步骤（数据自建/自清理）

        - finally: 无论主流程成败都执行（清理残留数据）
        - on_failure: 仅主流程失败时执行（失败补救）

        finally 失败不覆盖主结果（仅标注"清理步骤失败"），避免清理异常掩盖真实结果。
        """
        main_failed = result.status != "passed"
        for step in steps:
            run_on = step.get("run_on", "always")
            if run_on == "finally":
                pass  # 无条件执行
            elif run_on == "on_failure" and main_failed:
                pass  # 主流程失败时执行
            else:
                continue
            try:
                sr = self.execute_step(step, callback=callback)
                result.steps_result.append(sr)
                if sr.status != "passed" and result.status in ("passed", None):
                    result.status = "failed"
                    result.error_message = (result.error_message or "") + "；清理步骤失败"
                    logger.warning(f"[清理] 后置步骤失败: {step.get('name', '')} → {sr.error}")
            except Exception as e:
                if result.status in ("passed", None):
                    result.status = "failed"
                    result.error_message = (result.error_message or "") + f"；清理步骤异常: {e}"
                logger.warning(f"[清理] 后置步骤异常: {step.get('name', '')} → {e}")

    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", {}) or {}

        # 语义化执行：渲染 params 中的模板变量（{__NOW__}/{__CTX:key__} 等）
        # 渲染返回新 dict，不写回 step，防重复渲染放大 NOW 等动态值
        if self._ctx is not None and isinstance(params, dict):
            rendered = self._ctx.render(params)
            if isinstance(rendered, dict):
                params = rendered

        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}")

        # ==================== 语义解析器（语义目标优先，走现有定位回退链） ====================
        # semantic 顶层字段存在时，解析为选择器链注入 params["selectors"] 高优先级，
        # 继续走 _do_click/_do_fill 全部成熟回退策略。语义解析成功则跳过元素映射表块。
        semantic_chain: List[str] = []
        semantic_target: Optional[Dict[str, Any]] = None
        if settings.SEMANTIC_ENABLED and self._semantic_resolver is not None:
            try:
                semantic_target = step.get("semantic")
                semantic_chain = self._semantic_resolver.resolve_step(
                    self._page, step, self._current_scope, self._ctx
                ) or []
                if semantic_chain:
                    existing_selectors = params.get("selectors")
                    if isinstance(existing_selectors, list) and existing_selectors:
                        params["selectors"] = list(semantic_chain) + list(existing_selectors)
                    else:
                        original = params.get("selector")
                        if original:
                            original = str(original)
                            params["selectors"] = list(semantic_chain) + [original]
                        else:
                            params["selectors"] = list(semantic_chain)
                    if not params.get("selector"):
                        params["selector"] = semantic_chain[0]
                    logger.info(f"[语义解析] 步骤 {step_result.order} {step_result.name} → 选择器链: {semantic_chain}")
            except Exception as e:
                logger.warning(f"[语义解析] 步骤 {step_result.order} 解析异常，回退原路径: {e}")
                semantic_chain = []

        # ==================== Phase 3: 元素映射表最高优先级 ====================
        # 查询元素映射表，如命中则注入到 params 中作为最高优先级选择器
        # （语义解析已命中时跳过，语义优先）
        # 优先级：v2 显式字典（scope + element_key/semantic.text 精确查找）→ legacy 评分路径
        step_name = step.get("name", "")
        step_action = step.get("action", "")
        if (not semantic_chain) and self._element_mapping_service.is_enabled() and step_name:
            mapping_selector: Optional[str] = None
            mapping_chain: Optional[List[str]] = None

            # v2 显式字典优先（无评分、纯 dict 精确查找）
            v2_key = step.get("element_key") or (step.get("semantic") or {}).get("text")
            if v2_key:
                try:
                    lookup = self._element_mapping_service.lookup_by_key(
                        self._current_scope, str(v2_key)
                    )
                except Exception as e:
                    logger.debug(f"[元素映射表 v2] 查找异常: {e}")
                    lookup = None
                if lookup and lookup.get("selectors"):
                    mapping_chain = list(lookup["selectors"])
                    mapping_selector = mapping_chain[0]
                    logger.info(f"[元素映射表 v2] 命中: {self._current_scope}/{v2_key} → {mapping_selector}")

            # legacy 评分路径（USE_FUZZY_MAPPING=false 时返回 None）
            if mapping_selector is None:
                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}")

            if mapping_selector:
                # 将映射表选择器注入到 params 中作为最高优先级
                existing_selectors = params.get("selectors")
                if isinstance(existing_selectors, list) and existing_selectors:
                    params["selectors"] = (mapping_chain or [mapping_selector]) + list(existing_selectors)
                else:
                    # 保留原始 selector 作为回退（避免映射表选择器不精确时无备选）
                    original = params.get("selector")
                    if original:
                        original = str(original)
                        # v2 链（多个候选）直接整链 + 原始回退
                        if mapping_chain and len(mapping_chain) > 1:
                            params["selectors"] = mapping_chain + [original]
                        # 注入顺序优化（P1）：当映射表命中"通用复选框选择器"（裸 .el-checkbox /
                        # .el-checkbox__inner / input[type=checkbox]，page.click 会命中页面第一个
                        # 复选框，如"短信"通知勾选框，点击后假通过但目标未选中），而原始选择器
                        # 自带目标文本（:has-text("X")，可走策略5 表格行点击回退精确命中目标行）时，
                        # 优先用原始选择器。注意：仅对复选框类通用映射生效，避免把
                        # div:visible:has-text("会议室列表") 这类宽泛文本选择器也提前。
                        elif mapping_selector in (".el-checkbox", ".el-checkbox__inner", "input[type=checkbox]") and ':has-text("' in original:
                            params["selectors"] = [original, mapping_selector]
                        else:
                            params["selectors"] = [mapping_selector, original]
                    else:
                        params["selectors"] = mapping_chain or [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
        # 取消请求标志（由 cancel_execution → request_cancel 在 execute_step 之间置位）
        self._cancel_requested = False

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

            # ★ 取消请求检查（步骤边界）：置位后立即中断，避免重试放大器
            try:
                from app.services.execution_service import is_cancel_requested

                if is_cancel_requested(self._execution_id or ""):
                    self._cancel_requested = True
            except Exception:
                pass
            if self._cancel_requested:
                logger.info(
                    f"步骤 {step_result.order} 检测到取消请求，中断执行 "
                    f"(case={self._ctx.case_name if self._ctx else ''})"
                )
                break

            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.base_url in url and
                        ("login" in url.lower() or url == self.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 = "✓ 截图保存"
                elif action == "api_call":
                    self._do_api_call(params)
                    step_result.log = f"✓ 接口调用: {params.get('api_case', {}).get('name', '') or 'api_call'}"
                else:
                    raise ValueError(f"不支持的动作类型: {action}")

                # 成功：退出重试循环
                step_result.status = "passed"
                last_error = None
                # save_as：从语义目标命中的元素提取值写入上下文变量
                # （用于后续步骤 {__CTX:key__} 引用，如自建数据的名称/编号）
                if semantic_target and semantic_chain and self._ctx is not None:
                    self._semantic_resolver.extract_save_as(
                        self._page, semantic_target, semantic_chain, self._ctx
                    )
                # 状态验证（防假通过）：verify 配置翻译成 _do_assert
                # must=true 失败 → 步骤真实 failed（不重试，避免重复点击副作用）
                try:
                    fallback_sel = semantic_chain[0] if semantic_chain else params.get("selector")
                    self._verify_step_state(step, params, fallback_sel)
                except AssertionError as ve:
                    step_result.status = "failed"
                    step_result.error = str(ve)
                    step_result.log = f"✗ 状态验证失败: {str(ve)}"
                    last_error = ve
                    break
                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"

        # 取消请求中断：标记步骤为 skipped（不走失败重试/截图），
        # 由 execute_case 在回填主结果时识别并保持 cancelled 语境
        if self._cancel_requested:
            step_result.status = "skipped"
            step_result.log = "⏹ 检测到取消请求，中断执行"

        # 结束计时
        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 _rebase_sut_url(self, url: str) -> str:
        """
        被测系统地址跟随（方案A·执行期重写）：
        若目标 URL 的主机属于 KNOWN_SUT_HOSTS（默认 https://192.168.5.44），
        则将其主机替换为当前配置的 self.base_url 主机（保留 scheme/路径/查询/hash）。

        作用：在系统配置中切换被测系统后，存量与新建用例的导航/接口步骤自动跟随，
        无需批量修改用例数据；非被测系统的外部地址不受影响。

        Args:
            url (str): 原始目标 URL

        Returns:
            str: 重写后的 URL（无需改写时原样返回）
        """
        try:
            parsed = urlparse(url)
            if parsed.scheme not in ("http", "https"):
                return url
            if parsed.netloc not in self.KNOWN_SUT_HOSTS:
                return url
            base = urlparse(self.base_url)
            if not base.netloc or base.netloc == parsed.netloc:
                return url  # 已指向当前配置地址，无需改写
            return urlunsplit(
                (parsed.scheme, base.netloc, parsed.path, parsed.query, parsed.fragment)
            )
        except Exception as e:
            logger.debug(f"[SUT跟随] URL 重写失败（保留原样）: {e}")
            return url

    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不能为空")
        # 系统配置切换被测系统后，导航地址自动跟随（硬编码旧 SUT 地址的存量用例同样生效）
        url = self._rebase_sut_url(url)
        current_url = self._page.url
        # 跨微应用导航（外层路由从 platform 切到 backend 微应用）必须整页加载，
        # 否则平台壳只追加 &backstage= 参数而停留在 Home，导致 micro-app 不加载。
        # 判断依据：目标外层路由（第一个 hash 段，如 /backend/backstage）与当前不同
        if self._is_micro_app_route_change(current_url, url):
            logger.debug(f"跨微应用路由切换，改用整页 goto: {url[:120]}")
            self._ensure_platform_shell_state(url)
            self._goto_full_page(url, params.get("wait_until", "domcontentloaded"))
            self._ensure_micro_app_loaded(url)
            self._verify_micro_app_stable(url)
            return
        # 如果只是 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

    @staticmethod
    def _outer_route(url: str) -> str:
        """提取外层 hash 路由的第一个路径段，如 '#/platform?...' -> '/platform'"""
        try:
            if "#" not in url:
                return ""
            hash_part = url.split("#", 1)[1]
            seg = hash_part.split("?", 1)[0]
            return seg
        except Exception:
            return ""

    def _is_micro_app_route_change(self, current: str, target: str) -> bool:
        """
        判断是否跨微应用路由切换（外层第一个路由段变化）。

        平台壳按外层路由决定加载哪个 micro-app：
          - #/platform...  -> platform 微应用
          - #/backend/...  -> backstage 微应用
          - #/monitor/...  -> monitor 微应用
        直接改 hash 无法触发平台壳重新解析外层路由（只会追加 &xxx= 参数而停留），
        必须整页加载。故外层路由段不同时返回 True。
        """
        try:
            cur = self._outer_route(current)
            tgt = self._outer_route(target)
            if not cur or not tgt:
                return False
            return cur != tgt
        except Exception:
            return False

    def _ensure_platform_shell_state(self, url: str) -> None:
        """
        跨微应用整页导航前，修复平台壳 localStorage 状态。

        仅当目标外层路由为 /backend/... 时注入：headless 登录后 projectCode='platform'
        会使平台壳路由解析崩溃（console: TypeError: this.menuList is not iterable），
        backstage 微应用无法挂载。探测验证 DevTools 正常会话 projectCode='ubains'，
        注入 navPayload / pageOpen 后整页加载可稳定渲染 backstage 菜单（素材管理/终端管理）。
        注入仅影响本会话页面状态，不触碰 SUT 数据。
        """
        try:
            if self._outer_route(url) != "/backend/backstage":
                return
            self._page.evaluate("""() => {
                localStorage.setItem('projectCode', 'ubains');
                localStorage.setItem('navPayload', JSON.stringify({
                    list: 'doorScreenPublish',
                    path: '/Backend/MediaPublish/MaterialManage'
                }));
                localStorage.setItem('pageOpen', 'true');
            }""")
            logger.debug("已注入平台壳 localStorage: projectCode=ubains / navPayload / pageOpen")
        except Exception as e:
            logger.debug(f"注入平台壳 localStorage 失败: {e}")

    def _goto_full_page(self, url: str, wait_until: str = "domcontentloaded") -> None:
        """整页导航到目标 URL，超时忽略（SPA hash 路由可能不触发完成事件）"""
        t0 = time.time()
        try:
            self._page.goto(url, wait_until=wait_until, timeout=15000)
        except PlaywrightTimeoutError:
            logger.info(f"[跨微应用导航] goto 超时（忽略）: {url[:120]}")
        except Exception as e:
            logger.info(f"[跨微应用导航] goto 异常（忽略）: {type(e).__name__}: {str(e)[:120]}")
        logger.info(f"[跨微应用导航] goto 完成 t={time.time()-t0:.1f}s, URL={self._page.url[:120]}")
        self._wait_for_loading_overlay()
        logger.info(f"[跨微应用导航] loading 遮罩处理完成 t={time.time()-t0:.1f}s")

    def _ensure_micro_app_loaded(self, url: str, timeout: int = 15000) -> None:
        """
        跨微应用整页导航后，等待目标 micro-app 内容真正加载。

        平台壳全页加载后可能先渲染默认页（如 Home 或外层路由第一屏），
        外层 hash 中的内层路由（?backstage=...）需要平台壳随后解析并派发。
        探测发现 backstage 落地可能停在 ...Backend 根页（无 .title1），
        需重新设置内层 hash 才能到达 MaterialManage 等子页面。

        轮询逻辑：
          - 阶段1：前 8s 等待 marker 出现在 URL 中 + micro-app-body 有内容
          - 阶段2：若超时且 marker 未出现，重置 window.location.hash 为原始目标 hash
          - 阶段3：继续等待剩余时间

        稳定性确认（防假就绪）：goto 后 URL 短暂含 marker 且内容就绪，但平台壳
        可能在 ~3s 后重写 URL 丢弃内层路由（回到 ...Backend 根页），导致此函数
        返回后内容丢失。因此每次"就绪"后需在确认窗口内连续复查（URL 仍含
        marker 且内容仍就绪）才返回；复查失败则视为被重写，进入 hash 重置。
        """
        # 目标 URL 中需要最终出现的关键路径片段（优先取内层路由最后一个路径段）
        # 同时保留原始 hash，供阶段2重新设置内层路由
        marker = ""
        original_hash = ""
        try:
            if "#" in url:
                original_hash = url.split("#", 1)[1]
                outer = url.split("#", 1)[1]
                # 可能内层编码在 query 里: ?backstage=%2F...%23%2F...
                qs_vals = []
                for v in parse_qs(outer).values():
                    qs_vals.extend(v)
                decoded = unquote(" ".join(qs_vals))
                segs = [s for s in decoded.replace("%23", "#").replace("%2F", "/").split("/") if s]
                if segs:
                    marker = segs[-1]
        except Exception:
            pass

        def _body_ready(texts: str) -> bool:
            """micro-app-body 文本是否代表真实页面内容（排除过渡/默认页）"""
            t = (texts or "").strip()
            # 排除空文本、"加载中..."过渡占位、"欢迎使用..."后台默认首页
            # （headless 下平台壳丢弃内层路由时会落到默认首页，文本非空但非目标页）
            return bool(t) and ("加载中" not in t) and ("欢迎使用" not in t)

        def _dump_state(tag: str) -> str:
            """dump 当前 URL + micro-app-body 文本摘要，用于诊断日志"""
            cur = self._page.url
            body = self._page.query_selector("micro-app micro-app-body")
            body_text = (body.inner_text() or "")[:80] if body else "NO-BODY"
            short_url = cur[:120].replace('https://192.168.5.44/#/', '')
            result = f"[{tag}] URL={short_url} body={body_text!r}"
            return result

        def _target_ready() -> bool:
            """当前 URL 含 marker 且 micro-app-body 内容就绪，返回 True。

            marker 比对策略：跨微应用导航后，平台壳会先把内层原始 hash（含
            MaterialManage）派发出来（此时 URL 含 marker），随后再把外层壳的
            当前路由（Backend 根页）写入 URL——两次派发之间抛出的 marker URL
            才是真实目标页。这里同时接受「marker 出现在 URL」或「marker 出现在
            micro-app-body 文本中」（body 命中即 URL 被重写但内容仍在）。
            """
            cur = self._page.url
            body = self._page.query_selector("micro-app micro-app-body")
            body_text = body.inner_text() if body else ""
            if marker and marker in body_text:
                return True
            ok = True if not marker else marker in cur
            if not ok:
                return False
            if not body:
                return False
            return _body_ready(body_text)

        def _confirm_stable(brief_check_s: float, confirm_s: float = 0.4) -> bool:
            """就绪后复查确认窗口，验证就绪状态未被平台壳重写破坏。

            平台壳可能在就绪后的 ~3s 窗口内重写 URL 丢弃内层路由。这里在
            confirm_s 的时间窗口内持续复查（多次采样，窗口覆盖越过后视为稳定）。
            复查失败返回 False，由调用方触发 hash 重置。
            """
            if not _target_ready():
                return False
            deadline_c = time.time() + confirm_s
            while time.time() < deadline_c:
                if not _target_ready():
                    return False
                self._page.wait_for_timeout(150)
            return _target_ready()

        # 阶段1：等待 marker 自然出现（页面自行到达目标内层路由）。
        # goto 后平台壳先派发内层原始 hash（URL 短暂含 marker），数秒后必被
        # 重写为外层壳路由 —— 阶段1 只给 ~5s 快速探测，命中即返回，未命中不恋战。
        phase1_deadline = time.time() + min(timeout * 0.4, 5000) / 1000  # timeout 毫秒 → 秒
        logger.info(f"[micro-app加载] 阶段1 开始 deadline={phase1_deadline - time.time():.1f}s")
        ok_cycles = 0
        while time.time() < phase1_deadline:
            try:
                logger.info(f"[micro-app加载] 阶段1 探查: {_dump_state('阶段1')}")
                if _confirm_stable(1.0, confirm_s=0.6):
                    ok_cycles += 1
                    if ok_cycles >= 2:
                        logger.debug(f"[micro-app加载] 内容就绪（阶段1，{ok_cycles} 轮稳定复查）")
                        return
                else:
                    ok_cycles = 0
            except Exception as e:
                logger.info(f"[micro-app加载] 阶段1 异常: {str(e)[:100]}")
            self._page.wait_for_timeout(1000)
        logger.info(f"[micro-app加载] 阶段1 超时, {_dump_state('阶段1超时')}")

        # 阶段2：平台壳在 goto 后必定重写 URL 丢弃内层路由，始终重置 hash
        # 无论 marker 是否在 URL 中，无条件重置以触发微应用解析目标路由
        if marker and original_hash:
            try:
                logger.info(f"[micro-app加载] 阶段2 重置 hash (无条件): #{original_hash[:100]}")
                self._page.evaluate(f'window.location.hash = "#{original_hash}"')
                self._page.wait_for_timeout(2000)
                logger.info(f"[micro-app加载] 阶段2 重置后: {_dump_state('阶段2')}")
            except Exception:
                pass

        # 阶段3：等待内核路由到达（重设 hash 后平台壳重新派发内层路由）。
        # 重设 hash 后可能仍需数秒加载 micro-app，这里给最多 remaining 秒轮询；
        # 若 URL 又被平台壳重写回根页，则再次重置 hash。
        try:
            ok_cycles = 0  # 连续稳定就绪的轮次
            remaining = max(timeout - 8000, 5000) / 1000  # 阶段3最长等待
            deadline = time.time() + remaining
            while time.time() < deadline:
                logger.info(f"[micro-app加载] 阶段3 探查: {_dump_state('阶段3')}")
                if _confirm_stable(1.5, confirm_s=0.6):
                    ok_cycles += 1
                    if ok_cycles >= 2:
                        logger.debug(f"[micro-app加载] 内容就绪（阶段3，{ok_cycles} 轮稳定复查）")
                        return
                else:
                    ok_cycles = 0
                    # URL 又被平台壳重写回根页（hash 设置未生效），再次重置一次
                    if marker and original_hash:
                        self._page.evaluate(f'window.location.hash = "#{original_hash}"')
                        self._page.wait_for_timeout(1500)
                self._page.wait_for_timeout(1000)
        except Exception:
            pass

        logger.debug(f"等待 micro-app 内容加载超时: {url[:120]}")

    def _verify_micro_app_stable(self, url: str, max_tries: int = 3) -> None:
        """
        `_ensure_micro_app_loaded` 返回后的最终稳定性确认。

        平台壳可能在就绪后重写 URL 丢弃内层路由（如 MaterialManage ->
        Backend 根页），使该函数返回时内容已丢。这里用短确认窗口连续复查：
          - URL 仍含 marker 且内容就绪 → 稳定，直接返回
          - 否则重新设置原始 hash 并重试（最多 max_tries 次）
        复查不抛异常（作为导航容错的最后兜底，避免单点失败）。
        """
        marker = ""
        original_hash = ""
        try:
            if "#" in url:
                original_hash = url.split("#", 1)[1]
                outer = url.split("#", 1)[1]
                qs_vals = []
                for v in parse_qs(outer).values():
                    qs_vals.extend(v)
                decoded = unquote(" ".join(qs_vals))
                segs = [s for s in decoded.replace("%23", "#").replace("%2F", "/").split("/") if s]
                if segs:
                    marker = segs[-1]
        except Exception:
            pass
        if not marker:
            return

        def _ready() -> bool:
            cur = self._page.url
            body = self._page.query_selector("micro-app micro-app-body")
            body_text = body.inner_text() if body else ""
            if marker:
                if marker in cur or marker in body_text:
                    t = body_text
                    return bool(t) and ("加载中" not in t) and ("欢迎使用" not in t)
                return False
            t = body_text
            return bool(t) and ("加载中" not in t) and ("欢迎使用" not in t)

        for attempt in range(1, max_tries + 1):
            if _ready():
                # 短确认窗口再复查一次，确认未被后续重写
                try:
                    self._page.wait_for_timeout(1200)
                except Exception:
                    pass
                if _ready():
                    logger.info(f"[最终确认] 微应用稳定（第 {attempt} 次确认通过）: {_dump_state('最终确认')}")
                    return
            # 不稳定：重新设置 hash 恢复内层路由
            try:
                self._page.evaluate(f'window.location.hash = "#{original_hash}"')
                self._page.wait_for_timeout(2000)
            except Exception:
                pass
            logger.debug(f"[最终确认] 第 {attempt} 次重试后仍需恢复内层路由")
        logger.debug(f"[最终确认] 微应用内容仍不稳定（已重试 {max_tries} 次），交给后续步骤容错")

    def _checkbox_is_checked(self, el) -> bool:
        """判断 Element UI 复选框是否已勾选（.is-checked 类），用于勾选后校验防假通过"""
        try:
            return bool(self._page.evaluate(
                "(el) => { const c = el.closest('.el-checkbox');"
                " return c ? c.classList.contains('is-checked') : false; }",
                el,
            ))
        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 选择器 ====================
            # ⚠️ 表格异步加载兜底：点击含 :has-text("X") 的选择器前，先等目标文本 X 出现（25s）。
            # 目标系统(192.168.5.44)列表/会议室表格异步加载极慢且不稳，直接 page.click 会因
            # "目标行暂不存在"失败（智能定位已验证：先等 has-text 文本渲染再点击可稳定命中）。
            # 对普通按钮（如 button:has-text("登录")），文本本就存在，等待立即返回，无额外开销。
            try:
                _tm = re.search(r'has-text\("([^"]+)"\)', selector)
                if _tm:
                    self._page.wait_for_selector(f':has-text("{_tm.group(1)}")', timeout=25000)
                    self._page.wait_for_timeout(300)
                    logger.debug(f"✓ 目标文本已渲染: {_tm.group(1)}")
            except Exception:
                pass

            # 等待元素可见（宽松等待，不阻塞后续操作）
            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}")

            # ==================== 策略5（提前）: Element UI 表格行选择回退 ====================
            # Element UI 表格的选择列复选框带 is-hidden class（隐藏列），直接点击
            # .el-checkbox:has-text("X") 会失败。从选择器中提取目标文本，
            # 优先等待表格懒加载渲染完成（SPA 可达 10-40s），直接 force 勾选目标行
            # checkbox 并校验 is-checked 状态，行点击仅作末位兜底（避免行点击不触发
            # 勾选导致的假通过）。放在策略2（force点击）之前执行。
            m = re.search(r':has-text\("([^"]+)"\)', selector)
            if m:
                target_text = m.group(1)
                checkbox_sel = f'.el-table__row:has-text("{target_text}") .el-checkbox'
                row_sel = f'.el-table__row:has-text("{target_text}")'

                # 1. 轮询等待表格行渲染（覆盖 SPA 表格懒加载 12s-130s+ 的波动；state=attached 覆盖 is-hidden）
                #    之前单次 30s 等待在慢加载场景不足导致失败，改为每 3s 探测一次、上限 150s。
                # 1.1 非表格场景快速路径：目标选择器本身（菜单/按钮/链接）可直接点击时，
                #     直接 force 点击返回，避免误入表格行轮询（150s）白等。checkbox 选择器
                #     除外——隐藏 checkbox force 点击不可靠，保留原表格行回退逻辑。
                if 'checkbox' not in selector.lower():
                    try:
                        self._page.wait_for_selector(selector, state='attached', timeout=2000)
                        self._page.click(selector, timeout=5000, force=True)
                        logger.debug(f"✓ 非表格场景 force 点击成功: {selector}")
                        self._wait_for_loading_overlay()
                        return
                    except Exception:
                        pass
                row_found = False
                row_wait_start = time.time()
                while time.time() - row_wait_start < 150:
                    try:
                        if len(self._page.query_selector_all(row_sel)) > 0:
                            row_found = True
                            logger.debug(f"✓ 表格行已渲染: {row_sel}")
                            break
                    except Exception as e5:
                        last_error = e5
                    self._page.wait_for_timeout(3000)
                if not row_found:
                    logger.debug(f"⚠ 表格行未在 150s 内渲染 {row_sel}: {last_error}")

                # 2. 轮询勾选目标行 checkbox（force 绕过 is-hidden；每次校验 is-checked 防假通过）
                if row_found:
                    cb_deadline = time.time() + 40
                    checked_ok = False
                    while time.time() < cb_deadline:
                        try:
                            cb_el = self._page.query_selector(checkbox_sel)
                            if cb_el:
                                # 方式A：force 点击 checkbox 本身
                                cb_el.click(force=True)
                                self._page.wait_for_timeout(300)
                                if self._checkbox_is_checked(cb_el):
                                    checked_ok = True
                                    logger.debug(f"✓ Element UI checkbox 直接勾选成功: {checkbox_sel}")
                                    break
                                # 方式B：JS 点击兜底（绕过事件/遮挡）
                                self._page.evaluate('(el) => el.click()', cb_el)
                                self._page.wait_for_timeout(300)
                                if self._checkbox_is_checked(cb_el):
                                    checked_ok = True
                                    logger.debug(f"✓ Element UI checkbox JS 勾选成功: {checkbox_sel}")
                                    break
                                logger.debug(f"⚠ checkbox 勾选后状态未变化，2s 后重试: {checkbox_sel}")
                        except Exception as e5:
                            last_error = e5
                            logger.debug(f"⚠ checkbox 勾选失败，重试中: {e5}")
                        self._page.wait_for_timeout(2000)
                    if checked_ok:
                        self._wait_for_loading_overlay()
                        return
                    logger.debug(f"⚠ checkbox 40s 内未勾选成功（假通过风险），继续后续策略: {checkbox_sel}")

                # 3. 行点击末位兜底（部分表格仅行点击可选中；保持原逻辑避免回归）
                for row_sel_candidate in (row_sel, f'tr:has-text("{target_text}")'):
                    try:
                        self._page.wait_for_selector(row_sel_candidate, state="attached", timeout=8000)
                        row_el = self._page.query_selector(row_sel_candidate)
                        if row_el:
                            row_el.scroll_into_view_if_needed()
                            self._page.wait_for_timeout(200)
                            row_el.click(force=True)
                            logger.debug(f"✓ Element UI 表格行点击成功: {row_sel_candidate}")
                            self._wait_for_loading_overlay()
                            return
                    except Exception as e5:
                        last_error = e5
                        logger.debug(f"⚠ 表格行点击失败 {row_sel_candidate}: {e5}")

            # 策略6（提前）: 若选择器是隐藏复选框，尝试点击其所在表格行
            # 覆盖 .el-checkbox 无 :has-text 的情况（如勾选用户 admin@xty12）
            # 也放在策略2之前，避免 force 点击假成功
            if selector in (".el-checkbox", ".el-checkbox__inner", "input[type=checkbox]"):
                try:
                    row_el = self._page.evaluate(
                        """(sel) => {
                            const el = document.querySelector(sel);
                            if (!el) return null;
                            const row = el.closest('tr, .el-table__row');
                            if (!row) return null;
                            row.click();
                            return true;
                        }""",
                        selector,
                    )
                    if row_el:
                        logger.debug(f"✓ 点击复选框所在表格行成功: {selector}")
                        self._wait_for_loading_overlay()
                        return
                except Exception as e6:
                    last_error = e6
                    logger.debug(f"⚠ 点击复选框所在行失败 {selector}: {e6}")

            # 策略2: 主页面force点击（忽略遮挡）
            # 注意：对 checkbox 类元素，force 点击隐藏 input 可能不触发框架事件，
            # 因此需要加 before/after 状态校验
            if not force:
                try:
                    # force 点击前：记录 checkbox 选中状态（仅限 checkbox 类元素 + 能爬到 tr）
                    checkbox_checked_before = None
                    if selector in (".el-checkbox", ".el-checkbox__inner", "input[type=checkbox]"):
                        try:
                            el = self._page.query_selector(selector)
                            if el:
                                checkbox_checked_before = self._page.evaluate(
                                    """(sel) => {
                                        const el = document.querySelector(sel);
                                        if (!el) return null;
                                        const row = el.closest('tr, .el-table__row');
                                        if (!row) return null;
                                        const cb = row.querySelector('.el-checkbox');
                                        return cb ? cb.classList.contains('is-checked') : null;
                                    }""",
                                    selector,
                                )
                        except Exception:
                            pass

                    self._page.click(selector, timeout=5000, force=True)
                    self._page.wait_for_timeout(300)

                    # force 点击后：校验 checkbox 状态是否变化
                    if checkbox_checked_before is not None:
                        try:
                            checked_after = self._page.evaluate(
                                """(sel) => {
                                    const el = document.querySelector(sel);
                                    if (!el) return null;
                                    const row = el.closest('tr, .el-table__row');
                                    if (!row) return null;
                                    const cb = row.querySelector('.el-checkbox');
                                    return cb ? cb.classList.contains('is-checked') : null;
                                }""",
                                selector,
                            )
                            if checked_after is not None and checked_after == checkbox_checked_before:
                                logger.debug(f"⚠ force 点击后 checkbox 状态未变化（假通过），继续尝试后续策略: {selector}")
                                # 不 return，继续走后续策略
                            else:
                                logger.debug(f"✓ 主页面force点击成功，checkbox 状态已变化: {selector}")
                                self._wait_for_loading_overlay()
                                return
                        except Exception:
                            # 校验失败不阻塞，视为 force 点击成功
                            logger.debug(f"✓ 主页面force点击成功（校验跳过）: {selector}")
                            self._wait_for_loading_overlay()
                            return
                    else:
                        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

            # 策略3.5: micro-app 中点击
            try:
                if self._do_in_micro_apps('click', selector, timeout=5000):
                    logger.debug(f"✓ micro-app 中点击成功: {selector}")
                    self._wait_for_loading_overlay()
                    return
            except Exception as e_ma:
                last_error = e_ma
                logger.debug(f"⚠ micro-app 中点击失败 {selector}: {e_ma}")

            # 策略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}")

            # 注：策略5（表格行点击）和策略6（复选框所在行点击）已前移到策略2（force点击）之前，
            # 以避免 force 点击隐藏 checkbox 假成功

        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)

                # 滚动到可见区域（确保元素不被遮挡，与 _do_click 一致）
                try:
                    self._page.locator(selector).first.scroll_into_view_if_needed(timeout=5000)
                except Exception:
                    pass  # 滚动失败不阻塞后续操作

                # 强制清空
                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

            # 在 micro-app 中尝试
            try:
                if self._do_in_micro_apps('fill', selector, value=str(value), timeout=5000):
                    logger.debug(f"✓ micro-app 中填充成功: {selector} = {value}")
                    return
            except Exception as e_ma:
                last_error = e_ma
                logger.debug(f"⚠ micro-app 中填充失败 {selector}: {e_ma}")

        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
            # micro-app 快速检查
            if self._do_in_micro_apps('exists', selector, timeout=1000):
                logger.debug(f"✓ micro-app 中等待元素成功（快速）: {selector}")
                return

        # 阶段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

            # 在 micro-app 中尝试
            if self._do_in_micro_apps('exists', selector, timeout=per_selector_timeout):
                logger.debug(f"✓ micro-app 中等待元素成功: {selector}")
                return

        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_api_call(self, params: dict):
        """
        复用 ApiTestExecutor 发送单次 HTTP 请求（数据自建/自清理）

        配置（params）：
            api_case: {method, url, headers, body, assertions, description}
            save_as: {"key": "$.data.id"} 从响应体 JSONPath 提取写上下文

        说明：
        - 同步 requests，无浏览器、无事件循环冲突（与 Playwright 独立）
        - api_case 已由 execute_step 顶部整步渲染（{__CTX:key__} 等），这里幂等兜底
        """
        from app.executors.api_test_executor import ApiTestExecutor, AssertionEngine

        api_case = params.get("api_case") or {}
        save_as = params.get("save_as")
        if self._ctx is not None:
            rendered = self._ctx.render(api_case)
            if isinstance(rendered, dict):
                api_case = rendered

        # 系统配置切换被测系统后，接口地址自动跟随（与导航步骤同一套重写规则）
        steps = api_case.get("steps") or {}
        if isinstance(steps, dict) and steps.get("url"):
            steps["url"] = self._rebase_sut_url(str(steps["url"]))

        # 被测系统接口需鉴权（无 token 返回 401 令牌不能为空）：
        # 目标为 TARGET_URL 且未显式指定鉴权头时，自动从浏览器会话注入
        # localStorage.meet_token —— finally 数据自清理复用主流程已建立的登录态。
        api_case = self._inject_session_auth(api_case)

        client_config = {
            "target": {"base_url": settings.TARGET_URL, "verify_ssl": False, "timeout": 30}
        }
        ex = ApiTestExecutor(client_config)
        try:
            ex.start()
            result = ex.execute_case({
                "id": "",
                "name": api_case.get("name", "api_call"),
                "steps": api_case.get("steps", {}),
            })
        finally:
            try:
                ex.stop()
            except Exception:
                pass

        if result.status != "passed":
            raise Exception(f"api_call 失败: {result.error or result.status}")

        # save_as：从响应体 JSONPath 提取写 ctx（如自建数据的 id）
        if save_as and self._ctx and result.step is not None:
            body = result.step.response_body or ""
            for key, path in save_as.items():
                try:
                    value = AssertionEngine._extract_json_path(body, str(path))
                    self._ctx.set(key, value)
                    logger.info(f"[api_call] save_as: {key} ← {value!r}")
                except Exception as e:
                    logger.warning(f"[api_call] save_as 提取失败 {key}={path}: {e}")

        return result

    def _inject_session_auth(self, api_case: dict) -> dict:
        """
        api_call 目标为被测系统且步骤未显式指定鉴权头时，
        自动从浏览器会话（localStorage.meet_token）注入 authorization。

        背景：5.44 被测系统接口无 token 时返回 401「令牌不能为空」；api_call
        （尤其 finally 数据自清理）复用主流程已建立的登录态最可靠，无需手动
        维护会过期的 token。步骤显式配置了 authorization 时尊重作者配置。
        """
        steps = api_case.get("steps") or {}
        if not isinstance(steps, dict):
            return api_case

        url = steps.get("url") or ""
        base = str(settings.TARGET_URL or "").rstrip("/")
        if not base:
            return api_case
        full_url = url if str(url).startswith("http") else f"{base}/{str(url).lstrip('/')}"
        if not full_url.startswith(base):
            return api_case  # 目标非被测系统，不注入

        headers = steps.get("headers") or {}
        if not isinstance(headers, dict):
            headers = {}
        if headers.get("authorization") or headers.get("Authorization"):
            return api_case  # 已显式指定，尊重作者配置

        if self._page is None:
            return api_case
        try:
            token = self._page.evaluate("localStorage.getItem('meet_token')")
        except Exception as e:
            logger.debug(f"[api_call] 浏览器会话 token 读取失败: {e}")
            token = None
        if not token:
            logger.warning("[api_call] 浏览器无 meet_token，无法注入鉴权（被测系统接口可能 401）")
            return api_case

        headers = dict(headers)
        headers["authorization"] = str(token)
        steps = dict(steps)
        steps["headers"] = headers
        # 用浏览器 token，不再走账号库登录（_do_api_call 的 client_config 无 accounts）
        steps.setdefault("auth_required", False)
        api_case = dict(api_case)
        api_case["steps"] = steps
        logger.info(f"[api_call] 自动注入浏览器会话 token (meet_token {str(token)[:24]}...)")
        return api_case

    def _verify_step_state(
        self,
        step: Dict[str, Any],
        params: dict,
        fallback_selector: Optional[str] = None,
    ) -> bool:
        """
        状态验证（防假通过）：把 verify 配置翻译成 _do_assert 调用，复用 10 种断言

        Args:
            step: 步骤定义（含可选 verify 顶层字段）
            params: 已渲染的步骤参数（取 selectors/selector 兜底）
            fallback_selector: 语义解析链首选择器（verify.selector 缺省时兜底）

        Returns:
            True 通过或非强制未通过（仅告警）；raise AssertionError 当 must=true 失败

        verify 配置格式：
            {"type": "text_visible", "selector": "...", "text": "预定会议", "must": true}
        """
        v = step.get("verify") or {}
        if not v:
            # ENFORCE_VERIFICATION：expected 非空且无 verify 的 click/fill 走默认校验
            if settings.ENFORCE_VERIFICATION and step.get("action") in ("click", "fill"):
                expected = step.get("expected") or ""
                if expected:
                    self._default_verify(expected)
            return True

        must = v.get("must", False)
        vtype = v.get("type", "text_visible")
        text = str(v.get("text") or v.get("expected") or "").strip()
        selector = v.get("selector")
        if not selector and fallback_selector:
            selector = fallback_selector

        assert_params = self._build_verify_assert(vtype, selector, text)
        if assert_params is None:
            logger.debug(f"[verify] 类型 {vtype} 缺少选择器，跳过校验")
            return True

        try:
            self._do_assert(assert_params, text)
            return True
        except Exception as e:
            if must:
                logger.warning(f"[verify] 强制校验失败: {vtype} {e}")
                raise AssertionError(f"状态验证失败 ({vtype}): {e}") from e
            logger.warning(f"[verify] 非强制校验未通过（仅告警）: {vtype} {e}")
            return True

    def _build_verify_assert(
        self, vtype: str, selector: Optional[str], text: str
    ) -> Optional[dict]:
        """verify 类型 → _do_assert params（复用 10 种断言类型）"""
        sels = [selector] if selector else None
        if vtype == "text_visible":
            return {"type": "contains", "selectors": sels, "expected": text} if sels else None
        if vtype == "element_visible":
            return {"type": "visible", "selectors": sels} if sels else None
        if vtype == "text_hidden":
            return {"type": "hidden", "selectors": sels} if sels else None
        if vtype == "element_exists":
            return {"type": "element_exists", "selectors": sels} if sels else None
        if vtype == "url_contains":
            return {"type": "url_contains", "expected": text}
        if vtype == "dialog":
            # 弹窗提示校验：message/dialog/notification 任一含文本即通过
            s = selector or f'.el-message:has-text("{text}")'
            return {"type": "element_exists", "selectors": [s, f'.el-dialog:has-text("{text}")', f'.el-notification:has-text("{text}")']}
        if vtype == "value":
            return {"type": "value", "selectors": sels, "expected": text} if sels else None
        if vtype == "count":
            return {"type": "count", "selectors": sels, "expected": text} if sels else None
        return {"type": "visible", "selectors": sels} if sels else None

    def _default_verify(self, expected: str) -> None:
        """ENFORCE_VERIFICATION：默认校验（弹窗/文本，仅告警不强制）"""
        selector = (
            f'.el-message:has-text("{expected}"), '
            f'.el-dialog:has-text("{expected}"), '
            f'.el-notification:has-text("{expected}")'
        )
        try:
            self._do_assert({"type": "element_exists", "selectors": [selector]}, expected)
        except Exception as e:
            logger.warning(f"[verify] 默认校验未通过（仅告警）: {e}")

    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")
        # 兼容旧用例 `url`（阶段 6 统一为 url_contains，老数据零影响）
        if assert_type == "url":
            assert_type = "url_contains"
        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
            # micro-app 回退
            text = self._do_in_micro_apps('text', selectors[0], timeout=5000)
            if text is not None and text.strip() == expected_value.strip():
                return
            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=5000)
                    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
            # micro-app 回退
            text = self._do_in_micro_apps('text', selectors[0], timeout=5000)
            if text is not None and expected_value in text:
                return
            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 中存在即可）
            # 使用 wait_for_selector 而非 query_selector，以支持 Playwright 伪选择器如 :has-text()
            if not selectors:
                raise AssertionError("断言 element_exists 缺少选择器")
            found = False
            for selector in selectors:
                try:
                    self._page.wait_for_selector(selector, timeout=5000, state='attached')
                    found = True
                    break
                except Exception:
                    # 也在 iframe 中尝试
                    for frame in self._page.frames:
                        if frame == self._page.main_frame:
                            continue
                        try:
                            frame.wait_for_selector(selector, timeout=5000, state='attached')
                            found = True
                            break
                        except Exception:
                            continue
                    if found:
                        break
                    # 再在 micro-app 中尝试
                    try:
                        if self._do_in_micro_apps('exists', selector, timeout=5000):
                            found = True
                            break
                    except Exception:
                        continue
                continue
            if not found:
                raise AssertionError(f"✗ 断言失败: 元素不存在: {selectors}")

        elif assert_type == "element_count_min":
            # 验证元素数量 >= min_count
            min_count = int(expected_value) if (expected_value is not None and str(expected_value).strip() != "") else 0
            total = 0
            for selector in selectors:
                try:
                    elements = self._page.query_selector_all(selector)
                    total += len(elements)
                except Exception:
                    continue
            # 主文档/iframe 命中不足时，轮询 micro-app 内元素计数兜底
            # （@micro-zoe/micro-app 内容渲染在 <micro-app-body> 自定义元素内，
            #  query_selector_all 无法穿透，需 evaluate 内部统计）
            if total < min_count:
                deadline = time.time() + 8
                while time.time() < deadline:
                    micro_count = self._do_in_micro_apps('count', selector, timeout=1000)
                    if isinstance(micro_count, int) and micro_count > 0:
                        total = max(total, micro_count)
                        if total >= min_count:
                            break
                    self._page.wait_for_timeout(800)
            if total < min_count:
                raise AssertionError(f"✗ 断言失败: 期望至少 {min_count} 个元素，实际 {total} 个: {selectors}")

        elif assert_type == "body_not_contains":
            # 页面/元素文本不应包含指定字符串（数据自清理校验，与 api_test_executor 语义对齐）
            # expected 为空时视为失败（防假通过）
            expected_str = str(params.get("expected") or expected or "").strip()
            if not expected_str:
                raise AssertionError("✗ 断言失败: body_not_contains 期望值不能为空")
            got_text = None
            if selectors:
                for selector in selectors:
                    try:
                        el = self._page.wait_for_selector(selector, timeout=self.timeout)
                        if el:
                            got_text = el.inner_text()
                            break
                    except Exception:
                        continue
                if got_text is None:
                    for selector in selectors:
                        for frame in self._page.frames:
                            if frame == self._page.main_frame:
                                continue
                            try:
                                el = frame.wait_for_selector(selector, timeout=5000)
                                if el:
                                    got_text = el.inner_text()
                                    break
                            except Exception:
                                continue
                        if got_text is not None:
                            break
                if got_text is None:
                    raise AssertionError(f"✗ 断言失败: body_not_contains 未找到元素: {selectors}")
            else:
                try:
                    got_text = self._page.inner_text("body")
                except Exception:
                    got_text = self._page.content()
            if expected_str in (got_text or ""):
                raise AssertionError(f"✗ 断言失败: 页面仍包含 [{expected_str}]")

        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)}")
