base_script.py

"""
脚本基类 — 所有脚本的抽象基类

设计意图:
  - 统一脚本生命周期: setup → execute → teardown
  - 自动获取浏览器服务实例（通过 ServiceLocator）
  - 内置日志、截图、错误处理能力
  - 提供 hp/page 属性直接访问自愈浏览器页面
  - 脚本间可继承复用公共流程

生命周期:
  run() → setup() → execute() → teardown()
              ↑          ↑           ↑
           可选覆盖    必须实现      可选覆盖

用法:
  class MyScript(BaseScript):
      async def execute(self):
          await self.goto("https://example.com")
          await self.click("#login-btn")

约定:
  - 异常用 ScriptError 表示脚本逻辑错误
  - 用 ScriptAbort 表示主动中止（非错误），teardown 仍会执行
  - 子类在 setup() 中初始化资源，execute() 中执行业务逻辑
"""

from __future__ import annotations

import asyncio
from abc import ABC, abstractmethod
from typing import Any, Optional

from core.log_center import LogCenter
from core.service_locator import ServiceLocator


class ScriptError(Exception):
    """
    脚本执行异常。
    表示脚本逻辑层面的错误（如"需要点击的元素不存在"），
    和系统异常（如网络断开）区分开。
    """
    pass


class ScriptAbort(Exception):
    """
    脚本中止信号（非错误）。
    用于脚本内部主动退出（如"已达到每日上限，停止执行"），
    run() 会将其视为正常结束而非异常。
    """
    pass


class BaseScript(ABC):
    """
    脚本基类。

    子类需实现 execute() 方法。
    内置 hp (HealingPage, 自愈页面) 和 logger 供子类使用。
    自动从 ServiceLocator 解析浏览器服务，子类不需要关心浏览器初始化。

    Attributes:
        hp:     自愈 HealingPage 实例（延迟初始化）
        page:   原生 Playwright Page 实例（通过 hp 触发初始化）
        logger: 模块级 logger，日志名包含脚本名便于区分
        name:   脚本显示名称

    快捷方法分组（原子操作，供 AI 编排组合使用）:
      - 导航: goto, wait_for_load, get_url, get_title
      - 点击: click, hover, focus
      - 表单: fill, select_option, check, uncheck, upload_file
      - 输入: press_key
      - 等待: wait, sleep
      - 滚动: scroll_into_view
      - 数据: get_text, get_attribute, evaluate
      - 截图: screenshot
      - 框架: switch_frame, switch_to_default_frame
    """

    def __init__(self, script_name: str = "", _browser_svc=None):
        """
        Args:
            script_name: 脚本显示名称（日志中使用）
            _browser_svc: 浏览器服务实例（测试注入用，不传则从 ServiceLocator 获取）
        """
        self._name = script_name or self.__class__.__name__
        self.logger = LogCenter().get_logger(f"script.{self._name}")
        self._browser_svc = _browser_svc
        self._hp = None
        self._page = None
        self._current_frame = None

    # ------------------------------------------------------------------
    # 属性
    # ------------------------------------------------------------------

    @property
    def hp(self):
        """
        自愈 HealingPage 实例。

        延迟初始化：首次访问时从 _browser_svc 或 ServiceLocator 获取。
        测试时可以通过 _browser_svc 参数注入 mock 对象。
        """
        if self._hp is None:
            if self._browser_svc is not None:
                self._hp = self._browser_svc.healer_page
                self._page = self._browser_svc.page
            else:
                svc = ServiceLocator()
                self._browser_svc = svc.get("browser")
                self._hp = self._browser_svc.healer_page
                self._page = self._browser_svc.page
        return self._hp

    @property
    def page(self):
        """
        原生 Playwright Page 实例。

        如果先访问 page 后访问 hp，page 会触发 hp 的延迟初始化。
        推荐只使用 hp（自愈版本），原生 page 仅用于 wait_for_load_state 等
        playwright-healer 未包装的 API。
        """
        if self._page is None:
            _ = self.hp  # 触发延迟初始化
        return self._page

    @property
    def name(self) -> str:
        return self._name

    # ------------------------------------------------------------------
    # 生命周期
    # ------------------------------------------------------------------

    async def setup(self) -> None:
        """
        脚本初始化（可选覆盖）。

        在 execute() 之前调用。适合做:
          - 配置验证
          - 资源预加载
          - 条件检查（如检查是否已达到每日上限）
        """
        self.logger.info(f"[{self._name}] 开始初始化...")

    @abstractmethod
    async def execute(self) -> Any:
        """
        ★ 脚本核心逻辑 — 子类必须实现。

        在此方法中调用 self.goto/click/fill 等快捷操作。
        返回值会被 run() 返回给调用方。
        """
        ...

    async def teardown(self) -> None:
        """
        脚本清理（可选覆盖）。

        无论 execute() 成功还是失败（ScriptAbort 除外），teardown 都会执行。
        适合做:
          - 资源释放
          - 最终截图
          - 结果记录
        """
        self.logger.info(f"[{self._name}] 执行完成。")

    async def run(self) -> Any:
        """
        完整生命周期入口。

        执行顺序: setup() → execute() → teardown()
        异常处理:
          - ScriptAbort: 正常中止，不视为错误
          - 其他异常: 记录错误日志后重新抛出

        Returns:
            execute() 的返回值
        """
        self.logger.info(f"{'='*40}")
        self.logger.info(f"  ▶ 开始执行: {self._name}")
        self.logger.info(f"{'='*40}")
        try:
            await self.setup()
            result = await self.execute()
            await self.teardown()
            return result
        except ScriptAbort:
            self.logger.info(f"[{self._name}] 脚本已中止")
        except Exception as e:
            self.logger.error(f"[{self._name}] 执行失败: {e}", exc_info=True)
            raise

    # ------------------------------------------------------------------
    # 常用原子操作快捷方法
    #
    # 每个方法封装一个不可分割的页面操作。
    # 这些是"原子操作"——复杂业务逻辑应由 AI 编排图
    # 组合这些原子操作来自动生成。
    # ------------------------------------------------------------------

    async def goto(self, url: str, **kwargs) -> None:
        """导航到 URL"""
        self.logger.info(f"导航到: {url}")
        await self.hp.goto(url, **kwargs)

    async def click(self, selector: str, description: str = "") -> None:
        """点击元素"""
        desc = f" ({description})" if description else ""
        self.logger.info(f"点击: {selector}{desc}")
        await self.hp.click(selector, description or selector)

    async def fill(self, selector: str, value: str, description: str = "") -> None:
        """填写输入框"""
        desc = f" ({description})" if description else ""
        self.logger.info(f"填写: {selector}{desc} = '{value}'")
        await self.hp.fill(selector, description or selector, value)

    async def wait(self, selector: str, timeout: Optional[int] = None, **kwargs) -> None:
        """等待元素出现"""
        await self.hp.wait_for_selector(selector, timeout=timeout, **kwargs)

    async def wait_for_load(self) -> None:
        """等待页面加载完成（networkidle 状态）"""
        await self.page.wait_for_load_state("networkidle")

    async def sleep(self, seconds: float) -> None:
        """等待指定秒数（操作间隔、动画等待等）"""
        await asyncio.sleep(seconds)

    async def screenshot(self, name: str = "") -> str:
        """截图并返回文件路径"""
        timestamp = __import__("datetime").datetime.now().strftime("%H%M%S")
        label = name or f"{self._name}_{timestamp}"
        return str(await self._browser_svc.screenshot(label))

    # ------------------------------------------------------------------
    # 表单操作
    # ------------------------------------------------------------------

    async def select_option(self, selector: str, value: str, description: str = "") -> None:
        """选择下拉框选项"""
        desc = f" ({description})" if description else ""
        self.logger.info(f"选择: {selector}{desc} = '{value}'")
        await self.page.select_option(selector, value=value)

    async def check(self, selector: str, description: str = "") -> None:
        """勾选复选框或单选框"""
        desc = f" ({description})" if description else ""
        self.logger.info(f"勾选: {selector}{desc}")
        await self.page.check(selector)

    async def uncheck(self, selector: str, description: str = "") -> None:
        """取消勾选复选框"""
        desc = f" ({description})" if description else ""
        self.logger.info(f"取消勾选: {selector}{desc}")
        await self.page.uncheck(selector)

    async def upload_file(self, selector: str, file_path: str, description: str = "") -> None:
        """上传文件"""
        desc = f" ({description})" if description else ""
        self.logger.info(f"上传文件: {selector}{desc} = '{file_path}'")
        await self.page.set_input_files(selector, file_path)

    async def focus(self, selector: str, description: str = "") -> None:
        """聚焦到元素"""
        desc = f" ({description})" if description else ""
        self.logger.info(f"聚焦: {selector}{desc}")
        await self.page.focus(selector)

    # ------------------------------------------------------------------
    # 鼠标/键盘操作
    # ------------------------------------------------------------------

    async def hover(self, selector: str, description: str = "") -> None:
        """悬停在元素上"""
        desc = f" ({description})" if description else ""
        self.logger.info(f"悬停: {selector}{desc}")
        await self.hp.hover(selector, description or selector)

    async def press_key(self, selector: str, key: str, description: str = "") -> None:
        """在元素上按键（如 Enter, Tab, Escape）"""
        desc = f" ({description})" if description else ""
        self.logger.info(f"按键: {selector}{desc} → '{key}'")
        await self.page.press(selector, key)

    async def scroll_into_view(self, selector: str, description: str = "") -> None:
        """滚动到元素可见位置"""
        desc = f" ({description})" if description else ""
        self.logger.info(f"滚动到: {selector}{desc}")
        element = self.page.locator(selector)
        await element.scroll_into_view_if_needed()

    # ------------------------------------------------------------------
    # 数据提取
    # ------------------------------------------------------------------

    async def get_text(self, selector: str, timeout: Optional[int] = None) -> str:
        """获取元素文本内容"""
        text = await self.page.text_content(selector, timeout=timeout)
        return (text or "").strip()

    async def get_attribute(self, selector: str, attr: str,
                            timeout: Optional[int] = None) -> Optional[str]:
        """获取元素属性值"""
        return await self.page.get_attribute(selector, attr, timeout=timeout)

    async def evaluate(self, js: str, *args) -> Any:
        """执行 JS 并返回结果"""
        return await self.page.evaluate(js, *args)

    async def get_url(self) -> str:
        """获取当前页面 URL"""
        return self.page.url

    async def get_title(self) -> str:
        """获取当前页面标题"""
        return await self.page.title()

    # ------------------------------------------------------------------
    # 页面/iframe 切换
    # ------------------------------------------------------------------

    async def switch_frame(self, selector: str) -> None:
        """切换到指定 iframe"""
        self.logger.info(f"切换到 iframe: {selector}")
        frame = self.page.frame_locator(selector)
        self._current_frame = frame

    async def switch_to_default_frame(self) -> None:
        """切回主页面（退出 iframe）"""
        self.logger.info("切回主页面")
        self._current_frame = None