feat(driver): 增加元素高亮截图功能并重构截图矩阵

- 新增 `_get_highlighted_image` 核心逻辑，支持自动缩放 (Scale) 适配。
- 新增 `get_highlight_screenshot_as_file/bytes` 接口，支持诊断级高亮截图。
- 重构 `get_screenshot_as_file/bytes` 统一命名规范。
- 引入 `ImagePath` 与 `ImageBytes` 类型别名，增强代码语义化。
- 优化内存管理，使用 `io.BytesIO` 与 `img.load()` 确保在高频截图下的稳定性。

.gitignore

@@ -5,10 +5,25 @@ build/
 dist/
 wheels/
 *.egg-info
-
-# Virtual environments
-.venv
-node_modules
-
 .idea
-uv.lock
+
+# --- 依赖与环境 ---
+.venv
+venv/
+node_modules/
+uv.lock
+
+# --- 屏蔽outputs ---
+outputs/
+
+# --- Allure 报告 ---
+temp/
+reports/
+.allure/
+
+# --- pytest缓存 ---
+.pytest_cache/
+.allure_cache/
+
+# --- 配置文件 ---
+.env

README.md

@@ -1,2 +1,177 @@
-# AppAutoTest
+# AppAutoTest - 基于 Python 的 App 自动化测试框架
 
+## 1. 简介
+
+`AppAutoTest` 是一个基于 Python 构建的健壮且可扩展的移动应用自动化测试框架。它利用 Appium、Pytest 和 Allure 等行业标准工具，为测试
+Android 和 iOS 应用提供了全面的解决方案。该框架采用页面对象模型 (POM) 设计，以提高可维护性和可扩展性，并包含一套实用工具以简化测试开发和执行。
+
+## 2. 特性
+
+- **跨平台支持**: 原生支持 Android (通过 `UiAutomator2`) 和 iOS (通过 `XCUITest`)。
+- **页面对象模型 (POM)**: 强制分离 UI 元素定位器和测试逻辑，增强代码可读性和可维护性。
+- **丰富的测试报告**: 与 Allure 框架无缝集成，生成包含测试步骤、截图、日志和环境信息的详细交互式报告。
+- **自动服务管理**: 智能管理 Appium 服务器生命周期。如果本地服务器未运行，它可以自动启动，或者连接到现有的外部服务器。
+- **强大的驱动引擎**: 包含核心驱动封装 (`CoreDriver`)，通过内置显式等待、流式 API、高级日志记录和健壮的错误处理简化 Appium
+  操作。
+- **灵活配置**: 轻松管理设备能力 (`config/caps.yaml`)、特定环境参数和敏感数据 (`.env`)。
+- **可自定义等待条件**: 扩展 Selenium 的标准预期条件，提供用于复杂场景（如等待 Toast 消息或特定元素计数）的自定义可重用等待。
+- **高级日志与追踪**: 利用自定义装饰器 (`@step_trace`) 自动生成测试步骤的分层日志，包括执行时间、输入参数和状态。
+- **自动截图**: 在测试失败时自动捕获截图，并允许在任何步骤轻松将截图附加到 Allure 报告中以便快速调试。
+- **现代依赖管理**: 使用 `uv` (通过 `pyproject.toml`) 进行快速 Python 依赖解析，使用 `npm` (通过 `package.json`) 管理
+  Node.js 依赖。
+
+## 3. 项目结构
+
+框架遵循逻辑清晰且模块化的结构，以促进可扩展性和易于导航。
+
+```text
+AppAutoTest/
+├── config/
+│   └── caps.yaml             # 不同平台的 Appium capabilities 配置。
+├── core/
+│   ├── base_page.py          # 所有页面对象的抽象基类。
+│   ├── config_loader.py      # 加载配置文件 (caps, 环境设置)。
+│   ├── custom_expected_conditions.py # 定义复杂 UI 状态的自定义等待条件。
+│   ├── driver.py             # 核心 Appium 驱动封装，增强了操作和等待。
+│   ├── modules.py            # 通用枚举定义 (如 AppPlatform, Locator)。
+│   ├── run_appium.py         # 处理 Appium 服务器的生命周期。
+│   └── settings.py           # 全局框架设置 (路径, 超时等)。
+├── data/                     # 数据驱动测试的示例测试数据。
+├── docs/                     # 文档和说明。
+├── outputs/
+│   ├── logs/                 # 存储测试运行的日志文件。
+│   └── screenshots/          # 存储测试期间捕获的截图。
+├── page_objects/             # 对象类。
+├── reports/                  # 存储生成的 Allure HTML 报告。
+├── temp/                     # 原始 Allure 结果的临时目录。
+├── test_cases/               # 测试脚本。
+├── utils/
+│   ├── decorators.py         # 用于日志、截图等的自定义装饰器。
+│   ├── dirs_manager.py       # 确保所需目录存在的工具。
+│   ├── finder.py             # 定位策略转换工具。
+│   └── report_handler.py     # Allure 报告生成工具。
+├── .env                      # 存储环境变量 (如凭据)。Git 忽略此文件。
+├── conftest.py               # Pytest 的核心配置文件，用于 fixtures 和 hooks。
+├── main.py                   # 执行测试套件的主入口点。
+├── package.json              # 定义 Node.js 项目元数据和依赖 (Appium)。
+├── pytest.ini                # Pytest 配置文件 (日志, 标记, 路径)。
+├── pyproject.toml            # Python 项目元数据和依赖定义 (PEP 621)。
+└── README.md                 # 项目说明。
+```
+
+## 4. 前置条件
+
+- **Node.js**: 版本 20+ (如 `package.json` 中指定)。
+- **Python**: 版本 3.11+ (如 `pyproject.toml` 中指定)。
+- **Java JDK**: Appium 和 Android SDK 需要。
+- **Android SDK**: 构建和与 Android 应用交互需要。
+- **Allure Commandline**: 生成和查看 HTML 报告需要。
+> 注意：[Android SDK 环境配置指南](docs/AndroidSDK环境配置指南.md)
+## 5. 安装与运行指南
+
+按照以下步骤搭建测试环境。
+
+1. **克隆仓库:**
+   ```bash
+   git clone https://github.com/CNWeiWei/AppAutoTest
+   cd AppAutoTest
+   ```
+
+2. **安装 Node.js 依赖:**
+   此命令将在项目目录中本地安装 Appium 及其相关驱动，如 `package.json` 中所定义。
+   ```bash
+   npm install
+   ```
+
+3. **安装 Python 依赖:**
+   使用 `uv` 进行 Python 依赖管理。
+   ```bash
+   # 如果没有安装 uv，请先安装
+   # 使用 uv 安装项目依赖
+   uv sync
+   ```
+   不使用 `uv`进行 Python 依赖管理。
+   ```bash
+   #使用 pip 安装项目依赖
+   # 以可编辑模式安装（开发时推荐）
+    pip install -e .
+    
+    # 或以普通模式安装
+    pip install .
+   ```
+
+## 6. 如何运行测试
+
+框架提供了两种主要的测试执行方式。
+
+### 方法 1: 使用主入口点 (推荐)
+
+执行 `main.py` 脚本来运行整个测试套件。此脚本会自动处理所有运行前和运行后的任务。
+
+```bash
+python main.py
+```
+
+此命令将：
+
+1. 确保所有必要的输出目录已创建。
+2. 归档上次运行的日志文件。
+3. 启动 Appium 服务器 (或连接到现有的)。
+4. 通过 Pytest 运行 `test_cases/` 目录下的所有测试。
+5. 在 `reports/` 目录生成新的 Allure 报告。
+
+### 方法 2: 直接使用 Pytest
+
+为了更精细的控制，您可以直接从命令行调用 `pytest`。`conftest.py` 中定义的 fixtures 仍将管理 Appium 服务器和驱动会话。
+
+```bash
+# 运行特定的测试文件并显示详细输出
+pytest -v test_cases/test_wan_android_home.py
+
+# 运行测试并在第一个失败处停止
+pytest -x
+
+# 覆盖默认平台并指定设备 UDID
+pytest --platform IOS --udid <your_iphone_udid>
+```
+
+**可用的自定义命令行参数:**
+
+- `--platform`: 目标平台 (`Android` 或 `IOS`)。默认为 `Android`。
+- `--caps_name`: 设备/平台名称。
+- `--udid`: 目标设备的唯一设备标识符 (UDID)。
+- `--host`: Appium 服务器的主机地址。默认为 `127.0.0.1`。
+- `--port`: Appium 服务器的端口。默认为 `4723`。
+
+> 注意：[其他常用参数](./docs/常用参数.md)
+
+## 7. 测试报告
+
+测试运行后，原始 Allure 结果存储在 `temp/` 目录中。`main.py` 脚本会自动根据这些结果生成 HTML 报告。
+
+要查看最新报告，可以使用 Allure 命令行工具启动服务：
+
+```bash
+# 此命令将在默认 Web 浏览器中打开报告
+allure serve temp/
+```
+
+## 8. 核心概念解释
+
+#### 页面对象模型 (POM)
+
+- **`page_objects/`**: 此目录包含 POM 设计的核心。每个类 (例如 `HomePage`)
+  代表应用程序的一个屏幕或重要组件。它负责封装元素定位器和对这些元素执行操作的方法 (例如 `login(user, pwd)`)。
+- **`core/base_page.py`**: 这是所有页面对象的父类。它提供共享功能，如页面间导航 (`go_to`)、将截图附加到报告 (
+  `attach_screenshot_bytes`) 以及实现通用断言 (`assert_text`)。
+
+#### 驱动引擎
+
+- **`core/driver.py`**: 此类是标准 Appium `webdriver.Remote` 的强大封装。它通过集成内置的显式等待增强了基本操作，使测试更稳定，更不容易出现竞争条件。它还为高级移动手势（如
+  `swipe()`、`long_press()` 和 `smart_scroll()`）提供了流畅的接口。
+
+#### 自定义装饰器
+
+- **`utils/decorators.py`**: 此模块包含装饰器，可在不干扰逻辑的情况下向测试和页面方法添加强大的横切关注点。
+    - `@step_trace()`: 自动记录任何被装饰函数的进入、退出、参数和执行时间，创建一个清晰的分层日志，这对调试非常有价值。
+    - `@action_screenshot()`: 一个简单但有效的装饰器，在方法成功执行后自动截图，在测试报告中提供可视化的审计跟踪。

config/caps.yaml

@@ -0,0 +1,38 @@
+# Appium Capabilities Configuration
+android:
+  platformName: "Android"
+  automationName: "uiautomator2"
+  deviceName: "Android"
+  appPackage: "com.manu.wanandroid"
+  appActivity: "com.manu.wanandroid.ui.main.activity.MainActivity"
+  noReset: false
+  newCommandTimeout: 60
+  # udid: "emulator-5554" # Can be injected via CLI
+
+ios:
+  platformName: "iOS"
+  automationName: "XCUITest"
+  deviceName: "iPhone 14"
+  bundleId: "com.example.app"
+  autoAcceptAlerts: true
+  waitForQuiescence: false
+  newCommandTimeout: 60
+
+wan_android:
+  platformName: "Android"
+  automationName: "uiautomator2"
+  deviceName: "Android"
+  appPackage: "com.manu.wanandroid"
+  appActivity: "com.manu.wanandroid.ui.main.activity.MainActivity"
+  noReset: false
+  newCommandTimeout: 60
+  # udid: "emulator-5554" # Can be injected via CLI
+
+huawei:
+  platformName: "Android"
+  automationName: "uiautomator2"
+  deviceName: "Android"
+  appPackage: "com.android.settings"
+  appActivity: ".Settings"
+  noReset: false
+  newCommandTimeout: 60

conftest.py

@@ -0,0 +1,200 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+"""
+@author: CNWei,ChenWei
+@Software: PyCharm
+@contact: t6g888@163.com
+@file: conftest
+@date: 2026/1/16 10:52
+@desc: Pytest 核心配置与 Fixture 管理
+"""
+import logging
+import secrets
+from pathlib import Path
+from typing import Generator, Any
+
+import pytest
+import allure
+from dotenv import load_dotenv
+
+from core.run_appium import start_appium_service, stop_appium_service
+from core.driver import CoreDriver
+from core.settings import APPIUM_HOST, APPIUM_PORT
+from core.enums import AppPlatform
+from core.config_loader import get_caps
+
+load_dotenv()
+
+
+# 注册命令行参数
+def pytest_addoption(parser: Any) -> None:
+    """
+    注册自定义命令行参数
+    :param parser: Pytest 参数解析器对象
+    """
+    parser.addoption("--platform", action="store", default="android1", help="目标平台: Android or IOS")
+    parser.addoption("--caps_name", action="store", default=None, help="配置文件中的设备/平台名称")
+    parser.addoption("--udid", action="store", default=None, help="设备唯一标识")
+    parser.addoption("--host", action="store", default=APPIUM_HOST, help="Appium Server Host")
+    parser.addoption("--port", action="store", default=str(APPIUM_PORT), help="Appium Server Port")
+
+
+@pytest.fixture(scope="session")
+def appium_server(request: pytest.FixtureRequest) -> Generator[Any, None, None]:
+    """
+    第一层：管理 Appium Server 进程。
+    :param request: Pytest 请求对象
+    :return: Appium 服务进程实例
+    """
+    # 获取命令行参数
+    host = request.config.getoption("--host")
+    port = int(request.config.getoption("--port"))
+
+    service = start_appium_service(host, port)
+    yield service
+    stop_appium_service(service)
+
+
+@pytest.fixture(scope="session")
+def driver_session(request: pytest.FixtureRequest, appium_server: Any) -> Generator[CoreDriver, None, None]:
+    """
+    第二层：全局单例 Driver 管理 (Session Scope)。
+    负责创建和销毁 Driver，整个测试过程只启动一次 App。
+    :param request: Pytest 请求对象
+    :param appium_server: Appium 服务 fixture 依赖
+    :return: CoreDriver 实例
+    """
+    platform: AppPlatform = request.config.getoption("--platform")
+    # 配置名称(caps_name)（决定去 YAML 哪个节点拿数据，默认等于 platform）
+    caps_name = request.config.getoption("--caps_name") or platform
+    ud_id = request.config.getoption("--udid")
+    host = request.config.getoption("--host")
+    port = int(request.config.getoption("--port"))
+
+    # 1. 获取基础 Caps
+    caps = get_caps(caps_name)
+
+    # 2. 动态注入参数
+    if ud_id: caps["udid"] = ud_id
+
+    # 将最终生效的 caps 存入 pytest 配置，方便报告读取
+    request.config._final_caps = caps
+    request.config._caps_name = caps_name
+
+    # 3. 初始化 Driver
+    driver_helper = CoreDriver()
+    driver_helper.server_config(host=host, port=port)
+
+    try:
+        driver_helper.connect(platform=platform, caps=caps)
+    except Exception as e:
+        pytest.exit(f"无法初始化 Driver: {e}")
+
+    yield driver_helper
+
+    # 4. 清理
+    driver_helper.quit()
+
+
+@pytest.fixture(scope="function")
+def driver(driver_session: CoreDriver) -> Generator[Any, None, None]:
+    """
+    第三层：用例级 Driver 注入。
+    每个用例直接获取已存在的 Driver 实例。
+    可以在这里添加 reset_app() 逻辑，确保用例间独立性。
+    :param driver_session: CoreDriver 会话实例
+    :return: 原始 Appium Driver 对象 (webdriver.Remote)
+    """
+    # 可选：如果需要在每个用例前重置 App 状态
+    # driver_session.driver.reset() 
+    yield driver_session.driver
+
+
+def pytest_exception_interact(node: Any, call: Any, report: Any) -> None:
+    """
+    当测试用例抛出异常（断言失败或代码报错）时，Pytest 会调用这个钩子。
+    我们在这里手动把错误信息喂给 logging。
+    :param node: 测试节点
+    :param call: 调用信息
+    :param report: 测试报告
+    """
+    logger = logging.getLogger("pytest")
+
+    if report.failed:
+        # 获取详细的错误堆栈（包含 assert 的对比信息）
+        # long,short,no-locals
+        exc_info = call.excinfo.getrepr(style='short')
+        screenshot_name = f"异常截图_{secrets.token_hex(4)}"
+
+        logger.error(f"\n{'=' * 40} TEST FAILED {'=' * 40}\n"
+                     f"Node ID: {node.nodeid}\n"
+                     f"截图名称: {screenshot_name}\n"
+                     f"详细错误信息:\n{exc_info}"
+                     )
+
+        # 尝试获取 driver_session (CoreDriver 实例)
+        if "driver_session" in node.funcargs:
+            helper = node.funcargs["driver_session"]
+            try:
+                # 截图并附加到 Allure
+                screenshot_bytes = helper.driver.get_screenshot_as_png()
+                allure.attach(
+                    screenshot_bytes,
+                    name=screenshot_name,
+                    attachment_type=allure.attachment_type.PNG
+                )
+            except Exception as e:
+                logger.error(f"执行异常截图失败: {e}")
+        logger.error("=" * 93 + "\n")
+
+
+def pytest_sessionfinish(session: Any, exitstatus: int) -> None:
+    """
+    测试会话结束时，收集环境信息到 Allure 报告
+    :param session: Pytest 会话对象
+    :param exitstatus: 退出状态码
+    """
+    match exitstatus:
+        case pytest.ExitCode.OK:
+            logging.info("测试全部通过！")
+        case pytest.ExitCode.TESTS_FAILED:
+            logging.warning("部分测试用例执行失败，请检查报告。")
+        case pytest.ExitCode.INTERRUPTED:
+            logging.error("测试被用户手动中断（Ctrl+C）。")
+        case pytest.ExitCode.INTERNAL_ERROR:
+            logging.critical("Pytest 发生内部错误！")
+        case pytest.ExitCode.USAGE_ERROR:
+            logging.error("Pytest 命令行参数错误或用法不当。")
+        case pytest.ExitCode.NO_TESTS_COLLECTED:
+            logging.warning("未发现任何测试用例。")
+        case _:
+            logging.error(f"未知错误状态码: {exitstatus}")
+
+    report_dir = session.config.getoption("--alluredir")
+    final_caps = getattr(session.config, "_final_caps", {})
+    caps_name = getattr(session.config, "_caps_name", '')
+
+    if not report_dir:
+        return
+    report_path = Path(report_dir)
+    # 收集环境信息
+    env_info = {
+        "Platform": session.config.getoption("--platform"),
+        "UDID": final_caps.get("udid") or session.config.getoption("--udid") or "Not Specified",
+        "CapsName": caps_name,
+        "Host": session.config.getoption("--host"),
+        "Python": "3.11+"
+    }
+
+    try:
+        if not report_path.exists():
+            report_path.mkdir(parents=True, exist_ok=True)
+        # 生成 environment.properties 文件
+        env_file = report_path / "environment.properties"
+        with env_file.open("w", encoding="utf-8") as f:
+            for k, v in env_info.items():
+                f.write(f"{k}={v}\n")
+        logging.info("Allure 环境信息已生成。")
+    except Exception as e:
+        logging.error(f"无法写入环境属性: {e}")

core/__init__.py

@@ -0,0 +1,11 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+"""
+@author: CNWei,ChenWei
+@Software: PyCharm
+@contact: t6g888@163.com
+@file: __init__.py
+@date: 2026/1/16 10:49
+@desc: 
+"""

core/base_page.py

@@ -0,0 +1,148 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+"""
+@author: CNWei,ChenWei
+@Software: PyCharm
+@contact: t6g888@163.com
+@file: base_page
+@date: 2026/1/26 17:33
+@desc: 
+"""
+import logging
+from pathlib import Path
+from typing import Type, TypeVar, Optional
+
+import allure
+from appium import webdriver
+
+from core.driver import CoreDriver
+
+# 定义一个泛型，用于类型推断
+T = TypeVar('T', bound='BasePage')
+
+logger = logging.getLogger(__name__)
+
+
+class BasePage(CoreDriver):
+    # --- 全局通用的属性 ---
+    def __init__(self, driver: webdriver.Remote):
+        """
+        初始化 BasePage。
+
+        :param driver: Appium WebDriver 实例
+        """
+        super().__init__(driver)
+
+    # --- 所有页面通用的元动作 ---
+    def go_to(self, page_cls: Type[T]) -> T:
+        """
+        通用的页面跳转/实例化方法 (Page Factory)。
+
+        :param page_cls: 目标页面类 (BasePage 的子类)
+        :return: 目标页面的实例
+        """
+        logger.info(f"跳转到页面: {page_cls.__name__}")
+        return page_cls(self.driver)
+
+    def handle_permission_popups(self):
+        """
+        处理通用的系统权限弹窗。
+        遍历预定义的黑名单，尝试关闭出现的系统级弹窗（如权限申请、安装确认等）。
+        """
+        # 普适性黑名单
+        popup_blacklist = [
+            ("id", "com.android.packageinstaller:id/permission_allow_button"),
+            ("xpath", "//*[@text='始终允许']"),
+            ("xpath", "//*[@text='稍后提醒']"),
+            ("xpath", "//*[@text='以后再说']"),
+            ("id", "com.app:id/iv_close_global_ad"),
+            ("accessibility id", "Close"),  # iOS 常用
+        ]
+        self.clear_popups(popup_blacklist)
+
+    def handle_business_ads(self):
+        """
+        处理全 App 通用的业务广告弹窗。
+        针对应用启动后可能出现的全局广告进行关闭处理。
+        """
+        ads_blacklist = [("id", "com.app:id/global_ad_close")]
+        return self.clear_popups(ads_blacklist)
+
+    def save_and_attach_screenshot(self, label: str = "日志截图") -> None:
+        """
+        保存截图到本地并附加到 Allure 报告。
+        
+        :param label: 截图在报告中显示的名称
+        """
+        path_str = self.get_screenshot_as_file(filename=label)
+
+        if path_str:
+            img_path = Path(path_str)
+            if img_path.exists():
+                allure.attach.file(
+                    img_path,
+                    name=label,
+                    attachment_type=allure.attachment_type.PNG
+                )
+
+    def attach_screenshot_bytes(self, label: str = "日志截图") -> None:
+        """
+        直接获取内存中的截图数据并附加到 Allure 报告（不存本地文件）。
+        
+        :param label: 截图在报告中显示的名称
+        """
+        screenshot_bytes: bytes = self.get_screenshot_as_bytes()
+
+        allure.attach(
+            screenshot_bytes,
+            name=label,
+            attachment_type=allure.attachment_type.PNG
+        )
+
+    # --- 常用断言逻辑 ---
+    def assert_text(self, by: str, value: str, expected_text: str, timeout: Optional[float] = None) -> 'BasePage':
+        """
+        断言元素的文本内容是否符合预期。
+        
+        :param by: 定位策略。
+        :param value: 定位值。
+        :param expected_text: 期望的文本。
+        :param timeout: 等待元素可见的超时时间。
+        :return: self，支持链式调用。
+        :raises AssertionError: 如果文本不匹配。
+        """
+        # 1. 增强报告展示：将断言动作包装为一个清晰的步骤
+        step_name = f"断言校验 | 预期结果: '{expected_text}'"
+        with allure.step(step_name):
+            actual = self.get_text(by, value, timeout)
+            # 2. 动态附件：在报告中直观对比，方便后期排查
+            allure.attach(
+                f"预期值: {expected_text}\n实际值: {actual}",
+                name="文本对比结果",
+                attachment_type=allure.attachment_type.TEXT
+            )
+            # 3. 执行核心断言
+            # 如果断言失败，抛出的 AssertionError 会被 conftest.py 中的 Hook 捕获并截图
+            assert actual == expected_text, f"断言失败: 期望 {expected_text}, 实际 {actual}"
+            logger.info(f"断言通过: 文本匹配 '{actual}'")
+        return self
+
+    def assert_visible(self, by: str, value: str, msg: str = "元素可见性校验") -> 'BasePage':
+        """
+        断言元素是否可见。
+        
+        :param by: 定位策略
+        :param value: 定位值
+        :param msg: 断言描述信息
+        :return: self，支持链式调用
+        """
+        with allure.step(f"断言检查: {msg}"):
+            element = self.find_element(by, value)
+            is_displayed = element.is_displayed()
+
+            if is_displayed:
+                logger.info(f"断言通过: 元素 [{value}] 可见")
+
+            assert is_displayed, f"断言失败: 元素 [{value}] 不可见"
+        return self

core/config_loader.py

@@ -0,0 +1,62 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+"""
+@author: CNWei,ChenWei
+@Software: PyCharm
+@contact: t6g888@163.com
+@file: config_loader
+@date: 2026/1/16 10:52
+@desc: Pytest 核心配置与 Fixture 管理
+"""
+import logging
+from typing import Any, Optional
+from utils.data_loader import load_yaml
+from core.settings import CAPS_CONFIG_PATH, ENV_CONFIG, CURRENT_ENV
+
+logger = logging.getLogger(__name__)
+
+
+def get_env_config(env_name: Optional[str] = None) -> dict[str, str]:
+    """
+    获取当前运行环境的业务配置信息。
+
+    逻辑：
+    1. 优先使用传入的 `env_name`。
+    2. 若未传入，使用全局设置 `CURRENT_ENV`。
+    3. 若都为空，默认为 "test"。
+    4. 如果目标环境在配置中不存在，强制回退到 "test" 环境并记录警告。
+
+    :param env_name: 指定的环境名称 (e.g., "dev", "prod")，可选。
+    :return: 对应环境的配置字典。
+    """
+    target_env = env_name or CURRENT_ENV or "test"
+
+    if target_env not in ENV_CONFIG:
+        logger.warning(f"环境 '{target_env}' 未在配置中定义，将回退到 'test' 环境。")
+        return ENV_CONFIG.get("test", {})
+
+    return ENV_CONFIG[target_env]
+
+
+def get_caps(caps_name: str) -> dict[str, Any]:
+    """
+    从 YAML 配置文件加载指定的 Appium Capabilities。
+
+    :param caps_name: 配置文件中的设备/平台名称 (不区分大小写)，例如 "android_pixel"。
+    :return: 该设备对应的 Capabilities 字典。
+    :raises ValueError: 当指定的 caps_name 在配置文件中不存在时。
+    :raises RuntimeError: 当配置文件加载失败或格式错误时。
+    """
+    try:
+        all_caps = load_yaml(CAPS_CONFIG_PATH)
+        all_caps = {k.lower(): v for k, v in all_caps.items()}
+        caps_key = caps_name.lower()
+
+        if caps_key not in all_caps:
+            raise ValueError(f"在 {CAPS_CONFIG_PATH} 中找不到平台 '{caps_key}' 的配置")
+
+        return all_caps[caps_key]
+
+    except Exception as e:
+        raise RuntimeError(f"加载 Capabilities 失败 ({CAPS_CONFIG_PATH}): {e}")

core/custom_expected_conditions.py

@@ -0,0 +1,237 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+"""
+@author: CNWei,ChenWei
+@Software: PyCharm
+@contact: t6g888@163.com
+@file: custom_expected_conditions
+@date: 2026/1/22 16:13
+@desc: 自定义预期条件 (Expected Conditions)
+用于 WebDriverWait 的显式等待判断逻辑。
+"""
+
+import logging
+from typing import Any, Union
+
+from appium.webdriver.webdriver import WebDriver
+from selenium.webdriver.support import expected_conditions as EC
+from selenium.webdriver.common.by import By
+from selenium.webdriver.remote.webelement import WebElement
+from selenium.common.exceptions import StaleElementReferenceException, NoSuchElementException
+
+logger = logging.getLogger(__name__)
+
+"""
+    常用等待条件（expected_conditions）--来自EC模块
+        presence_of_element_located: 元素存在于DOM。
+        visibility_of_element_located: 元素可见。
+        element_to_be_clickable: 元素可点击。
+        title_contains: 页面标题包含特定文本。
+        text_to_be_present_in_element: 元素包含特定文本。
+"""
+
+
+# 自定义预期条件(Custom Expected Condition)
+class BaseCondition:
+    """
+    基础条件类：负责统一的 WebDriverWait 协议实现和异常拦截。
+    所有自定义的类形式 EC 都应继承此类
+    """
+
+    def __call__(self, driver: WebDriver):
+        """
+        WebDriverWait 调用的入口方法。
+        
+        :param driver: WebDriver 实例
+        :return: check 方法的返回值，或者在捕获异常时返回 False
+        """
+        try:
+            return self.check(driver)
+        except (NoSuchElementException, StaleElementReferenceException):
+            return False
+
+    def check(self, driver: WebDriver):
+        """
+        执行具体的检查逻辑，由子类实现。
+
+        :param driver: WebDriver 实例
+        :return: 判定成功返回对象或 True，失败返回 False
+        """
+        raise NotImplementedError("子类必须实现 check 方法")
+
+
+EC_MAPPING: dict[str, Any] = {}
+
+
+def register(name: str = None):
+    """
+    自定义预期条件注册装饰器：
+    1. @register() -> 使用函数名注册
+    2. @register("alias") -> 使用别名注册
+    3. register("name", func) -> 手动注入
+
+    :param name: 注册别名，默认为函数/类名
+    :return: 装饰器函数
+    """
+
+    def decorator(item):
+        reg_name = name or item.__name__
+        EC_MAPPING[reg_name] = item
+        return item
+
+    return decorator
+
+
+@register("toast_visible")
+class ToastVisible(BaseCondition):
+    """检查 Toast 消息是否可见"""
+
+    def __init__(self, text: str, partial: Union[str, bool] = True):
+        """
+        :param text: 期望的 Toast 文本
+        :param partial: 是否部分匹配 (默认 True)
+        """
+        self.text = text
+        # 处理从装饰器传来的字符串 "true"/"false"
+        if isinstance(partial, str):
+            self.partial = partial.lower() != "false"
+        else:
+            self.partial = partial
+
+    def check(self, driver: WebDriver):
+        # 注意：这里不再需要显式 try-except，BaseCondition 会处理
+        xpath = f"//*[contains(@text, '{self.text}')]" if self.partial else f"//*[@text='{self.text}']"
+        element = driver.find_element(By.XPATH, xpath)
+        return element if element.is_displayed() else False
+
+
+@register("attr_contains")
+class ElementHasAttribute(BaseCondition):
+    """检查元素的属性是否包含特定值"""
+
+    # 扁平化参数以支持字符串调用: "attr_contains:id,btn_id,checked,true"
+    def __init__(self, by: str, value: str, attribute: str, expect_value: str):
+        """
+        :param by: 定位策略
+        :param value: 定位值
+        :param attribute: 属性名
+        :param expect_value: 期望包含的属性值
+        """
+        self.locator = (by, value)
+        self.attribute = attribute
+        self.value = expect_value
+
+    def check(self, driver: WebDriver):
+        element = driver.find_element(*self.locator)
+        attr_value = element.get_attribute(self.attribute)
+        return element if (attr_value and self.value in attr_value) else False
+
+
+@register()
+class ElementCountAtLeast(BaseCondition):
+    """检查页面上匹配定位符的元素数量是否至少为 N 个"""
+
+    def __init__(self, by: str, value: str, count: Union[str, int]):
+        self.locator = (by, value)
+        # 确保字符串参数转为整数
+        self.count = int(count)
+
+    def check(self, driver: WebDriver) -> bool | list[WebElement]:
+        elements = driver.find_elements(*self.locator)
+        if len(elements) >= self.count:
+            return elements
+        return False
+
+
+@register()  # 使用函数名 is_element_present 注册
+def is_element_present(by: str, value: str):
+    """
+    检查元素是否存在于 DOM 中 (不一定可见)。
+
+    :param by: 定位策略
+    :param value: 定位值
+    :return: 判定函数
+    """
+    locator = (by, value)
+
+    def _predicate(driver):
+        try:
+            return driver.find_element(*locator)
+        except Exception as e:
+            logger.warning(f"{__name__}异常：{e}")
+            return False
+
+    return _predicate
+
+
+@register()
+def system_ready(api_client):
+    """
+    检查外部系统 (API) 是否就绪。
+
+    :param api_client: API 客户端实例
+    :return: 判定函数
+    """
+
+    def _predicate(_):  # 忽略传入的 driver
+
+        try:
+            return api_client.get_status() == "OK"
+        except Exception as e:
+            logger.warning(f"{__name__}异常：{e}")
+            return False
+
+    return _predicate
+
+
+def get_condition(method: Union[str, Any], *args, **kwargs):
+    """
+    智能获取预期条件：
+    1. 如果 method 是字符串，先查自定义 EC_MAPPING
+    2. 如果自定义里没有，去官方 selenium.webdriver.support.expected_conditions 找
+    3. 如果 method 本身就是 Callable (比如 EC.presence_of_element_located)，直接透传
+
+    :param method: 预期条件名称 (str) 或 可调用对象
+    :param args: 传递给条件的参数
+    :param kwargs: 传递给条件的关键字参数
+    :return: 实例化后的预期条件对象 (Callable)
+    """
+
+    # 情况 A: 如果传入的是官方 EC 对象或自定义函数实例，直接返回
+    if callable(method) and not isinstance(method, type):
+        return method
+
+    # 情况 B: 如果传入的是字符串别名
+    if isinstance(method, str):
+        # 1. 尝试从自定义映射查找
+        if method in EC_MAPPING:
+            target = EC_MAPPING[method]
+        # 2. 尝试从官方 EC 库查找
+        elif hasattr(EC, method):
+            target = getattr(EC, method)
+        else:
+            raise ValueError(f"找不到预期条件: {method}. 请检查拼写或是否已注册。")
+
+        # 实例化并返回 (无论是类还是闭包工厂)
+        return target(*args, **kwargs)
+
+    # 情况 C: 传入的是类名本身
+    if isinstance(method, type):
+        return method(*args, **kwargs)
+
+    raise TypeError(f"不支持的条件类型: {type(method)}")
+
+
+if __name__ == "__main__":
+    # print(EC_MAPPING)
+    cond1 = get_condition("toast_visible", "保存成功")
+    print(cond1)
+    # 调用闭包生成的条件
+    # cond2 = get_condition("is_element_present", (By.ID, "submit"))
+    # print(cond2)
+    cond3 = get_condition(EC.presence_of_element_located, (By.ID, "submit"))
+    print(cond3)
+    cond4 = get_condition("system_ready", (By.ID, "submit"))
+    print(cond4)
+    # WebDriverWait(driver, 10).until(cond1)

core/driver.py

@@ -0,0 +1,813 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+"""
+@author: CNWei,ChenWei
+@Software: PyCharm
+@contact: t6g888@163.com
+@file: driver
+@date: 2026/1/16 10:49
+@desc: Appium 核心驱动封装，提供统一的 API 用于 Appium 会话管理和元素操作。
+"""
+import logging
+import io
+import secrets  # 原生库，用于生成安全的随机数
+from typing import Optional, Type, TypeVar, Union, Callable, Any
+from time import sleep
+
+from appium import webdriver
+from appium.options.android import UiAutomator2Options
+from appium.options.ios import XCUITestOptions
+from appium.options.common.base import AppiumOptions
+from appium.webdriver.webdriver import ExtensionBase
+from appium.webdriver.webelement import WebElement
+from appium.webdriver.client_config import AppiumClientConfig
+
+from selenium.common import TimeoutException, StaleElementReferenceException, NoSuchElementException
+from selenium.webdriver.support.ui import WebDriverWait
+from selenium.webdriver.support import expected_conditions as EC
+from selenium.webdriver.common.action_chains import ActionChains
+from selenium.webdriver.common.actions import interaction
+from selenium.webdriver.common.actions.action_builder import ActionBuilder
+from selenium.webdriver.common.actions.pointer_input import PointerInput
+
+from PIL import Image, ImageDraw
+
+from core.enums import AppPlatform
+from core.settings import IMPLICIT_WAIT_TIMEOUT, EXPLICIT_WAIT_TIMEOUT, APPIUM_HOST, APPIUM_PORT, SCREENSHOT_DIR
+from core.types import ImagePath, ImageBytes
+from utils.finder import by_converter
+from utils.decorators import resolve_wait_method
+
+logger = logging.getLogger(__name__)
+
+T = TypeVar("T")
+
+
+class CoreDriver:
+    def __init__(self, driver: Optional[webdriver.Remote] = None):
+        """
+        初始化 CoreDriver 实例。
+        从 settings.py 加载默认的 Appium 服务器主机和端口。
+        """
+        self.driver = driver
+        self._current_implicit_timeout = IMPLICIT_WAIT_TIMEOUT
+        self._host = APPIUM_HOST
+        self._port = APPIUM_PORT
+
+    @property
+    def server_url(self) -> str:
+        """
+        动态构造 URL，确保 server_config 修改后立即生效。
+        """
+        return f"http://{self._host}:{self._port}"
+
+    def server_config(self, host: str = APPIUM_HOST, port: int = APPIUM_PORT) -> 'CoreDriver':
+        """
+        配置服务器信息。支持链式调用。
+        :param host: ip
+        :param port: 端口
+        :return: 返回 CoreDriver 实例自身，支持链式调用。
+        """
+        self._host = host
+        self._port = port
+        logger.info(f"Appium Server 指向 -> {self._host}:{self._port}")
+        return self
+
+    @staticmethod
+    def _make_options(platform: str | AppPlatform, caps: dict) -> AppiumOptions:
+        """
+        根据平台生成对应的 Options
+        :param platform: 目标平台 ('android' 或 'ios')，支持 AppPlatform 枚举或字符串。
+        :param caps: Appium capabilities 字典。
+        :return: AppiumOptions
+        """
+        match platform:
+            case AppPlatform.ANDROID.value:
+                logger.info(f"正在初始化 Android 会话...")
+                return UiAutomator2Options().load_capabilities(caps)
+
+            case AppPlatform.IOS.value:
+                logger.info(f"正在初始化 iOS 会话...")
+                return XCUITestOptions().load_capabilities(caps)
+
+            case _:
+                # 优化：不再默认返回 Android，而是显式报错 (Fail Fast)
+                msg = f"不支持的平台类型: [{platform}]。当前仅支持: [android, ios]"
+                logger.error(msg)
+                raise ValueError(msg)
+
+    def connect(self, platform: str | AppPlatform, caps: dict,
+                extensions: list[Type[ExtensionBase]] | None = None,
+                client_config: AppiumClientConfig | None = None) -> 'CoreDriver':
+        """
+        连接到 Appium 服务器并创建一个新的会话。
+
+        :param platform: 目标平台 ('android' 或 'ios')，支持 AppPlatform 枚举或字符串。
+        :param caps: Appium capabilities 字典。
+        :param extensions: Appium 驱动扩展列表。
+        :param client_config: Appium 客户端配置。
+        :return: 返回 CoreDriver 实例自身，支持链式调用。
+        :raises ValueError: 如果平台不受支持。
+        :raises ConnectionError: 如果无法连接到 Appium 服务。
+        """
+        # 1. 统一格式化平台名称
+        platform_name = platform.value if isinstance(platform, AppPlatform) else platform.lower().strip()
+
+        # 2. 预校验：如果已经有 driver 正在运行，先清理（防止 Session 冲突）
+        if self.driver:
+            logger.warning("发现旧的 Driver 实例尚未关闭，正在强制重置...")
+            self.quit()
+
+        # 3. 匹配平台并加载 Options
+        options: AppiumOptions = self._make_options(platform_name, caps)
+
+        try:
+
+            # 4. 创建连接
+            self.driver = webdriver.Remote(
+                command_executor=self.server_url,
+                options=options,
+                extensions=extensions,
+                client_config=client_config
+            )
+
+            logger.info(f"已成功连接到 {platform_name.upper()} 设备 (SessionID: {self.driver.session_id})")
+            return self
+
+        except Exception as e:
+            logger.error(f"驱动连接失败！底层错误信息: {e}")
+            # 确保失败后清理现场
+            self.driver = None
+            raise ConnectionError(f"无法连接到 Appium 服务，请检查端口 {self._port} 或设备状态。") from e
+
+    # --- 核心操作 ---
+    def find_element(self, by: str, value: str, timeout: Optional[float] = None) -> WebElement:
+        """
+        内部通用查找（显式等待）
+        :param by: 定位策略
+        :param value: 定位值
+        :param timeout: 等待超时时间 (秒)。如果为 None, 则使用全局默认超时.
+        :return: WebElement.
+        """
+        by = by_converter(by)
+        mark = (by, value)
+        method = EC.presence_of_element_located(mark)
+        return self.explicit_wait(method, timeout)
+
+    def find_elements(self, by: str, value: str, timeout: Optional[float] = None) -> list[WebElement]:
+        """
+        内部通用查找（显式等待）
+        :param by: 定位策略
+        :param value: 定位值
+        :param timeout: 等待超时时间 (秒)。如果为 None, 则使用全局默认超时.
+        :return: list[WebElement].
+        """
+        by = by_converter(by)
+        mark = (by, value)
+        method = EC.presence_of_all_elements_located(mark)
+        return self.explicit_wait(method, timeout)
+
+    def delay(self, timeout: int | float) -> 'CoreDriver':
+        """
+        强制等待（线程阻塞）。
+
+        应谨慎使用，主要用于等待非 UI 元素的异步操作或调试。
+        :param timeout: 等待时间（秒）。
+        :return: self
+        """
+        sleep(timeout)
+        return self
+
+    def implicit_wait(self, timeout: float = IMPLICIT_WAIT_TIMEOUT) -> None:
+        """
+        设置全局隐式等待时间。
+        在每次 find_element 时生效，直到元素出现或超时。
+        :param timeout: 超时时间
+        :return:
+        """
+        self.driver.implicitly_wait(timeout)
+        self._current_implicit_timeout = timeout  # 记录等待时间
+
+    @resolve_wait_method
+    def explicit_wait(self, method: Union[Callable[[webdriver.Remote], T], str], timeout: Optional[float] = None) -> \
+            Union[T, WebElement]:
+        """
+        执行显式等待，直到满足某个条件或超时。
+
+        使用示例:
+        1. 使用原生 Selenium EC:
+           driver.explicit_wait(EC.presence_of_element_located((By.ID, "el_id")))
+
+        2. 使用自定义字符串别名 (支持参数传递):
+           # 检查 Toast 消息
+           driver.explicit_wait("toast_visible:登录成功")
+           # 检查元素属性 (格式: key:by,value,attr,expect)
+           driver.explicit_wait("attr_contains:id,btn_submit,checked,true")
+           # 检查元素数量
+           driver.explicit_wait("count_at_least:xpath,//android.widget.TextView,3")
+
+        :param method: EC等待条件(Callable) 或 自定义等待条件的名称(str)
+        :param timeout: 超时时间 (秒)。如果为 None, 则使用全局默认超时.
+        :return: 等待条件的执行结果 (通常是 WebElement 或 bool)
+        """
+        wait_timeout = timeout if timeout is not None else EXPLICIT_WAIT_TIMEOUT
+
+        try:
+            # 获取函数名称用于日志，兼容 lambda 和普通函数
+            func_name = getattr(method, '__name__', repr(method))
+            logger.info(f"执行显式等待: {func_name}, 最大超时时长: {wait_timeout}s")
+            return WebDriverWait(self.driver, wait_timeout).until(method)
+        except TimeoutException:
+            logger.error(f"等待超时: {wait_timeout}s 内未满足条件 {method}")
+            raise
+        except TypeError as te:
+            logger.error(f"显示等待异常: {te}")
+            # self.driver.quit()
+            raise te
+
+    def page_load_timeout(self, timeout: Optional[float] = None) -> None:
+        """
+        设置页面加载超时时间。
+        :param timeout: 超时时间 (秒)。如果为 None, 则使用全局默认超时.
+        """
+        wait_timeout = timeout if timeout is not None else EXPLICIT_WAIT_TIMEOUT
+        self.driver.set_page_load_timeout(wait_timeout)
+
+    def click(self, by: str, value: str, timeout: Optional[float] = None) -> 'CoreDriver':
+        """
+        查找元素并执行点击操作。
+        内置显式等待，确保元素可点击。
+        :param by: 定位策略。
+        :param value: 定位值。
+        :param timeout: 等待超时时间。
+        :return: self
+        """
+        by = by_converter(by)
+        mark = (by, value)
+        logger.info(f"点击: {mark}")
+        method = EC.element_to_be_clickable(mark)
+        self.explicit_wait(method, timeout).click()
+        return self
+
+    def clear(self, by: str, value: str, timeout: Optional[float] = None) -> 'CoreDriver':
+        """
+        查找元素并清空其内容。
+        内置显式等待，确保元素可见。
+        :param by: 定位策略。
+        :param value: 定位值。
+        :param timeout: 等待超时时间。
+        :return: self
+        """
+        by = by_converter(by)
+        mark = (by, value)
+        logger.info(f"清空输入框: {mark}")
+        method = EC.visibility_of_element_located(mark)
+        self.explicit_wait(method, timeout).clear()
+        return self
+
+    def input(self, by: str, value: str, text: str, sensitive: bool = False,
+              timeout: Optional[float] = None) -> 'CoreDriver':
+        """
+        查找元素并输入文本。
+        内置显式等待，确保元素可见。
+        :param by: 定位策略。
+        :param value: 定位值。
+        :param text: 要输入的文本。
+        :param sensitive: 是否为敏感信息（如密码），如果是，日志中将掩码显示。
+        :param timeout: 等待超时时间。
+        :return: self
+        """
+        by = by_converter(by)
+        mark = (by, value)
+        display_text = "******" if sensitive else text
+        logger.info(f"输入文本到 {mark}: '{display_text}'")
+        method = EC.visibility_of_element_located(mark)
+        self.explicit_wait(method, timeout).send_keys(text)
+        return self
+
+    def is_visible(self, by: str, value: str) -> bool | None:
+        """
+        判断元素是否可见
+        :param by: 定位策略。
+        :param value: 定位值。
+        :return: bool
+        """
+        # 禁用隐式等待
+        original_timeout = self._current_implicit_timeout
+        try:
+            self.implicit_wait(0)
+
+            by = by_converter(by)
+            elements = self.driver.find_elements(by, value)
+
+            if elements:
+                return elements[0].is_displayed()
+            # 2. 元素存在于 DOM 中，还需要判断它在 UI 上是否真正可见（宽/高 > 0 且未隐藏）
+            return False
+        except (StaleElementReferenceException, NoSuchElementException):
+            # 这些属于预料中的“不可见”情况
+            return False
+        except Exception as e:
+            _ = e
+            return False
+        finally:
+            # 恢复原来的隐式等待时间
+            self.implicit_wait(original_timeout)
+
+    def wait_until_visible(self, by: str, value: str, timeout: Optional[float] = None) -> bool:
+        """
+        等待元素出现
+        :param by: 定位策略。
+        :param value: 定位值。
+        :param timeout: 等待超时时间。
+        :return: bool
+        """
+        try:
+            by = by_converter(by)
+            mark = (by, value)
+            method = EC.visibility_of_element_located(mark)
+            self.explicit_wait(method, timeout)
+            return True
+        except TimeoutException:
+            return False
+
+    def wait_until_not_visible(self, by: str, value: str, timeout: Optional[float] = None) -> bool:
+        """
+        等待元素消失
+        :param by: 定位策略。
+        :param value: 定位值。
+        :param timeout: 等待超时时间。
+        :return: bool
+        """
+        try:
+            by = by_converter(by)
+            mark = (by, value)
+            method = EC.invisibility_of_element_located(mark)
+            self.explicit_wait(method, timeout)
+            return True
+        except TimeoutException:
+            return False
+
+    def get_text(self, by: str, value: str, timeout: Optional[float] = None) -> str:
+        """
+        获取元素文本
+        :param by: 定位策略。
+        :param value: 定位值。
+        :param timeout: 等待超时时间。
+        :return:获取到的文本
+        """
+        by = by_converter(by)
+        mark = (by, value)
+        method = EC.visibility_of_element_located(mark)
+
+        text = self.explicit_wait(method, timeout).text
+        logger.info(f"获取到的文本: {text}")
+        return text
+
+    def get_attribute(self, by: str, value: str, name: str, timeout: Optional[float] = None) -> str:
+        """
+        获取元素属性
+        :param by: 定位策略。
+        :param value: 定位值。
+        :param timeout: 等待超时时间。
+        :param name: 属性名称 (如 'checked', 'enabled', 'resource-id')
+        """
+        by = by_converter(by)
+        mark = (by, value)
+        method = EC.presence_of_element_located(mark)
+        element = self.explicit_wait(method, timeout)
+        attr_value = element.get_attribute(name)
+        logger.info(f"获取属性 {name} of {mark}: {attr_value}")
+        return attr_value
+
+    def clear_popups(self, black_list: list = None, max_rounds: int = 5) -> bool:
+        """
+        显式清理弹窗函数。
+        说明：
+        1. 快速扫描：使用 is_visible (不等待) 确认弹窗是否存在。
+        2. 动作处理：发现后点击，并触发 wait_until_not_visible (异步等待消失)。
+        3. 自适应退出：当整轮扫描无障碍物时，立即返回，不浪费时间。
+        4. 异常存证：若点击失败或发生错误，自动截图。
+        :param black_list: 允许传入当前页面特有的弹窗定位 [(by, value), ...]（如某个活动的特殊广告）
+        :param max_rounds: 最大扫描轮数
+        :return: active: bool(清理过一个弹窗都将返回 True)
+        """
+        if not black_list:
+            logger.warning("未提供黑名单列表，跳过清理动作。")
+            return False
+
+        list_len = len(black_list)
+        active = False
+
+        logger.info(f"开始执行显式弹窗清理，待检查项: {list_len} 个")
+        for round_idx in range(max_rounds):
+            skip_count = 0
+            for by, value in black_list:
+
+                if not self.is_visible(by, value):
+                    skip_count += 1
+                    continue
+
+                logger.info(f"当前权重{skip_count},第 {round_idx + 1} 轮：待清理弹窗 -> {value}")
+                try:
+                    elements = self.find_elements(by, value, timeout=0.5)  # 使用极短超时
+                    if not elements:
+                        skip_count += 1
+                    else:
+                        elements[0].click()
+                        active = True
+
+                        # 消失得快返回得就快，最多等 1.5s
+                        if self.wait_until_not_visible(by, value, timeout=1.5):
+                            logger.info(f"弹窗已成功消失")
+                        else:
+                            logger.warning(f"弹窗点击后仍存在")
+
+                except Exception as e:
+                    safe_val = secrets.token_hex(8)
+                    file_name = f"popup_fail_{round_idx}_{safe_val}.png"
+
+                    logger.error(f"清理弹窗尝试点击时失败[{safe_val}:{value}]: {e}")
+                    self.get_screenshot_as_file(file_name)
+                    raise e
+            if skip_count == list_len:
+                break
+        return active
+
+    def back(self) -> 'CoreDriver':
+        """
+        模拟设备返回键操作。
+        :return: self
+        """
+        self.driver.back()
+        return self
+
+    @property
+    def session_id(self) -> Any | None:
+        """获取当前 Appium 会话的 Session ID。"""
+        return self.driver.session_id
+
+    # --- 移动端特有：方向滑动 ---
+    def swipe_by_coordinates(self, start_x: int, start_y: int, end_x: int, end_y: int,
+                             duration: int = 1000) -> 'CoreDriver':
+        """
+        基于绝对坐标的滑动 (W3C Actions 底层实现)
+        :param start_x: 起点 X
+        :param start_y: 起点 Y
+        :param end_x: 终点 X
+        :param end_y: 终点 Y
+        :param duration: 滑动持续时间 (ms)
+        :return: self
+        """
+        actions = ActionChains(self.driver)
+        # 覆盖默认的鼠标输入为触摸输入
+        actions.w3c_actions = ActionBuilder(self.driver, mouse=PointerInput(interaction.POINTER_TOUCH, "touch"))
+
+        actions.w3c_actions.pointer_action.move_to_location(start_x, start_y)
+        actions.w3c_actions.pointer_action.pointer_down()
+        actions.w3c_actions.pointer_action.pause(duration / 1000)
+        actions.w3c_actions.pointer_action.move_to_location(end_x, end_y)
+        actions.w3c_actions.pointer_action.release()
+        actions.perform()
+        return self
+
+    def swipe(self, direction: str = "up", duration: int = 1000) -> 'CoreDriver':
+        """
+        封装方向滑动
+        :param direction: 滑动方向 up/down/left/right (指手指滑动的方向)
+        :param duration: 滑动持续时间 (ms)
+        :return: self
+        """
+        # 每次获取屏幕尺寸以适应旋转
+        size = self.driver.get_window_size()
+        w, h = size['width'], size['height']
+
+        # 定义滑动坐标 (中心区域，避开边缘，防止误触系统操作)
+        coords = {
+            "up": (w * 0.5, h * 0.8, w * 0.5, h * 0.2),
+            "down": (w * 0.5, h * 0.2, w * 0.5, h * 0.8),
+            "left": (w * 0.9, h * 0.5, w * 0.1, h * 0.5),
+            "right": (w * 0.1, h * 0.5, w * 0.9, h * 0.5)
+        }
+        start_x, start_y, end_x, end_y = coords.get(direction.lower(), coords["up"])
+        logger.info(f"执行滑动: {direction} ({start_x}, {start_y}) -> ({end_x}, {end_y})")
+
+        return self.swipe_by_coordinates(start_x, start_y, end_x, end_y, duration)
+
+    def long_press(self, element: Optional[WebElement] = None, x: Optional[int] = None, y: Optional[int] = None,
+                   duration: int = 2000) -> 'CoreDriver':
+        """
+        长按封装：支持传入元素或坐标。
+        如果传入 element，则计算其中心点坐标进行长按。
+        如果传入 x, y，则直接在坐标处长按。
+
+        :param element: 目标元素 (WebElement)，可选。
+        :param x: 绝对坐标 X，可选。
+        :param y: 绝对坐标 Y，可选。
+        :param duration: 长按持续时间 (ms)，默认 2000ms。
+        :return: self
+        :raises ValueError: 如果既未传入 element 也未传入坐标 (x, y)。
+        """
+        if element:
+            rect = element.rect
+            x = rect['x'] + rect['width'] // 2
+            y = rect['y'] + rect['height'] // 2
+
+        if x is None or y is None:
+            raise ValueError("Long press requires an element or (x, y) coordinates.")
+
+        # 复用 swipe_by_coordinates，当起点和终点一致时，即为长按效果。
+        # 逻辑：Move -> Down -> Pause -> Move(原地) -> Release
+        return self.swipe_by_coordinates(x, y, x, y, duration)
+
+    def drag_and_drop(self, source_el: WebElement, target_el: WebElement, duration: int = 1000) -> 'CoreDriver':
+        """
+        将 source_el 拖拽到 target_el。
+        计算两个元素的中心点，执行从源元素中心到目标元素中心的拖拽操作。
+
+        :param source_el: 源元素 (WebElement)。
+        :param target_el: 目标元素 (WebElement)。
+        :param duration: 拖拽过程持续时间 (ms)，默认 1000ms。
+        :return: self
+        """
+        s_rect = source_el.rect
+        t_rect = target_el.rect
+
+        sx, sy = s_rect['x'] + s_rect['width'] // 2, s_rect['y'] + s_rect['height'] // 2
+        tx, ty = t_rect['x'] + t_rect['width'] // 2, t_rect['y'] + t_rect['height'] // 2
+
+        logger.info(f"执行拖拽: ({sx}, {sy}) -> ({tx}, {ty})")
+        return self.swipe_by_coordinates(sx, sy, tx, ty, duration)
+
+    def smart_scroll(self, element: WebElement, direction: str = "down") -> 'CoreDriver':
+        """
+        智能滚动：自动识别平台并调用最稳定的原生滚动脚本
+        :param element: 需要滚动的容器元素 (如 ScrollView, RecyclerView, TableView)
+        :param direction: 滚动方向 'up', 'down', 'left', 'right'
+        :return: self
+        """
+        platform = self.driver.capabilities.get('platformName', '').lower()
+        match platform:
+            case AppPlatform.ANDROID.value:
+                # Android UiAutomator2 专用滚动手势
+                self.driver.execute_script('mobile: scrollGesture', {
+                    'elementId': element.id,
+                    'direction': direction,
+                    'percent': 1.0
+                })
+            case AppPlatform.IOS.value:
+                # iOS XCUITest 专用滚动 (默认为 iOS 处理)
+                self.driver.execute_script("mobile: scroll", {
+                    "elementId": element.id,
+                    "direction": direction
+                })
+
+        return self
+
+    def swipe_by_percent(self, start_xp: float, start_yp: float, end_xp: float, end_yp: float,
+                         duration: int = 1000) -> 'CoreDriver':
+        """
+        按屏幕比例滑动。
+        坐标值为屏幕宽高的百分比 (0.0 - 1.0)。
+
+        :param start_xp: 起点 X 比例 (如 0.5)。
+        :param start_yp: 起点 Y 比例 (如 0.8)。
+        :param end_xp: 终点 X 比例 (如 0.5)。
+        :param end_yp: 终点 Y 比例 (如 0.2)。
+        :param duration: 滑动持续时间 (ms)，默认 1000ms。
+        :return: self
+        """
+        size = self.driver.get_window_size()
+        w, h = size['width'], size['height']
+
+        return self.swipe_by_coordinates(
+            int(w * start_xp),
+            int(h * start_yp),
+            int(w * end_xp),
+            int(h * end_yp),
+            duration
+        )
+
+    # --- 上下文切换 (Context Switch) ---
+    def switch_to_context(self, context_name: str) -> 'CoreDriver':
+        """
+        切换到指定的上下文 (Context)。
+        :param context_name: 上下文名称 (如 'NATIVE_APP', 'WEBVIEW_com.example.app')
+        :return: self
+        """
+        logger.info(f"尝试切换到上下文: {context_name}")
+        try:
+            self.driver.switch_to.context(context_name)
+            logger.info(f"成功切换到上下文: {context_name}")
+        except Exception as e:
+            logger.error(f"切换上下文失败: {e}")
+            available_contexts = self.driver.contexts
+            logger.info(f"当前可用上下文: {available_contexts}")
+            raise e
+        return self
+
+    def switch_to_webview(self) -> 'CoreDriver':
+        """
+        自动切换到第一个可用的 WebView 上下文。
+        :return: self
+        """
+        logger.info("尝试自动切换到 WebView 上下文...")
+        contexts = self.driver.contexts
+        for context in contexts:
+            if 'WEBVIEW' in context:
+                return self.switch_to_context(context)
+
+        logger.warning("未找到 WebView 上下文，保持在当前上下文。")
+        return self
+
+    def switch_to_native(self) -> 'CoreDriver':
+        """
+        切换回原生应用上下文 (NATIVE_APP)。
+        :return: self
+        """
+        return self.switch_to_context('NATIVE_APP')
+
+    def get_screenshot_as_bytes(self) -> ImageBytes:
+        """截取当前完整屏幕并以字节流形式返回。
+
+        与 get_screenshot_as_file 类似，但它不将图片写入磁盘，而是直接返回
+        PNG 格式的二进制数据。适用于需要动态处理或传输截图的场景。
+
+        :return: PNG 格式的图像字节流 (bytes)。如果获取失败，则返回空字节串 b""。
+        """
+        try:
+            return self.driver.get_screenshot_as_png()
+        except Exception as e:
+            logger.error(f"获取全屏字节流失败: {e}")
+            return b""
+
+    def get_screenshot_as_file(self, filename: str | None = None) -> ImagePath:
+        """截取当前完整屏幕并保存为文件。
+
+        这是一个通用的全屏截图功能，常用于记录测试执行的某个特定状态、
+        在发生未知错误时捕获现场，或用于视觉回归测试的基准图像。
+
+        :param filename: 保存的文件名 (不含扩展名)。如果为 None，将生成一个随机文件名。
+        :return: 保存后的图像文件的绝对路径 (str)。如果保存失败，则返回空字符串。
+        """
+        file_name = f"{filename or secrets.token_hex(8)}.png"
+        path = (SCREENSHOT_DIR / file_name).as_posix()
+
+        try:
+            # 核心：save_screenshot 是底层原生方法，不依赖任何元素定位
+            self.driver.save_screenshot(path)
+            logger.info(f"全屏截图已保存: {path}")
+            return path
+        except Exception as e:
+            logger.error(f"全屏截图失败: {e}")
+            return ""
+
+    def _get_highlighted_image(self, by: str, value: str, color: str = "red") -> Image.Image:
+        """[内部核心] 截取屏幕，高亮指定元素，并返回 PIL Image 对象。
+
+        此方法是高亮截图功能的基础。它首先获取全屏截图，然后定位指定元素，
+        计算其在截图上的物理像素位置（处理分辨率缩放），最后在元素周围绘制一个
+        矩形框。
+
+        如果元素定位失败或在绘制过程中发生任何异常，为了不中断主流程，
+        该方法会记录警告并返回未经修改的原始截图。
+
+        :param by: 元素定位策略 (如 'id', 'xpath')。
+        :param value: 元素定位值。
+        :param color: 高亮框的颜色 (CSS 颜色名称，如 'red', 'blue')，默认为 'red'。
+        :return: 一个 PIL.Image.Image 对象。成功时为带高亮框的截图，失败时为原始截图。
+        """
+        # 1. 获取基础截图
+        screenshot_png = self.driver.get_screenshot_as_png()
+        # 2. 显式管理内存流
+        with io.BytesIO(screenshot_png) as stream:
+            img = Image.open(stream)
+            # 强制加载进入内存，确保 stream 关闭后 img 对象依然可用
+            img.load()
+
+        try:
+            # 3. 定位元素并计算缩放 (逻辑像素 vs 物理像素)
+            el = self.find_element(by, value)
+
+            window_size = self.driver.get_window_size()
+
+            scale = img.size[0] / (window_size['width'] or 1)
+
+            # 4. 计算坐标
+            loc, sz = el.location, el.size
+            rect = [
+                (loc['x'] - 2) * scale,
+                (loc['y'] - 2) * scale,
+                (loc['x'] + sz['width'] + 2) * scale,
+                (loc['y'] + sz['height'] + 2) * scale
+            ]
+
+            # 5. 绘制高亮
+            draw = ImageDraw.Draw(img)
+            line_width = max(int(3 * scale), 5)
+            draw.rectangle(rect, outline=color, width=line_width)
+            return img
+        except Exception as e:
+            # 如果元素找不到或其他异常，记录并返回原图，确保流程不中断
+            logger.warning(f"无法为元素({by}={value})绘制高亮，返回原图。原因: {e}")
+            return img
+
+    def get_highlight_screenshot_as_bytes(self, by: str, value: str, color: str = "red") -> ImageBytes:
+        """获取高亮元素的截图，并以字节流形式返回。
+
+        此方法适用于需要将截图数据直接在内存中处理或通过网络传输的场景，
+        例如，将其嵌入到测试报告中或发送到图像识别服务，而无需先保存到磁盘。
+
+        :param by: 元素定位策略。
+        :param value: 元素定位值。
+        :param color: 高亮框的颜色，默认为 'red'。
+        :return: PNG 格式的图像字节流 (bytes)。
+        """
+        img = self._get_highlighted_image(by, value, color)
+        with io.BytesIO() as img_byte_arr:
+            img.save(img_byte_arr, format='PNG')
+            return img_byte_arr.getvalue()
+
+    def get_highlight_screenshot_as_file(self, by: str, value: str, filename: str | None = None,
+                                         color: str = "red") -> ImagePath:
+        """获取高亮元素的截图，并将其保存为文件。
+
+        当需要在调试、错误分析或生成可视化报告时，将带有上下文（高亮元素）的
+        截图持久化到磁盘时，应使用此方法。
+
+        :param by: 元素定位策略。
+        :param value: 元素定位值。
+        :param filename: 保存的文件名 (不含扩展名)。如果为 None，将生成一个随机文件名。
+        :param color: 高亮框的颜色，默认为 'red'。
+        :return: 保存后的图像文件的绝对路径 (str)。如果保存失败，则返回空字符串。
+        """
+        file_name = f"{filename or secrets.token_hex(8)}.png"
+        path = (SCREENSHOT_DIR / file_name).as_posix()
+        try:
+            img = self._get_highlighted_image(by, value, color)
+            img.save(path)
+            return path
+        except Exception as e:
+            logger.error(f"全屏截图失败: {e}")
+            return ""
+
+    def get_element_screenshot_as_bytes(self, by: str, value: str) -> ImageBytes:
+        """截取指定元素的图像，并以字节流形式返回。
+
+        此方法只捕获元素自身矩形区域内的图像，不包含屏幕的其他部分。
+        非常适合用于元素级别的图像对比、OCR 识别或在报告中精确展示某个控件。
+
+        :param by: 元素定位策略。
+        :param value: 元素定位值。
+        :return: 元素截图的 PNG 格式图像字节流 (bytes)。如果失败，则返回空字节串 b""。
+        """
+        try:
+            el = self.find_element(by, value)
+            return el.screenshot_as_png
+        except Exception as e:
+            logger.error(f"获取元素字节流失败: {e}")
+            return b""
+
+    def get_element_screenshot_as_file(self, by: str, value: str, filename: str | None = None) -> ImagePath:
+        """截取指定元素的图像，并将其保存为文件。
+
+        与 get_element_screenshot_as_bytes 功能相同，但将结果直接持久化到磁盘。
+        适用于需要对特定 UI 控件进行存档或后续分析的场景。
+
+        :param by: 元素定位策略。
+        :param value: 元素定位值。
+        :param filename: 保存的文件名 (不含扩展名)。如果为 None，将生成一个随机文件名。
+        :return: 保存后的图像文件的绝对路径 (str)。如果保存失败，则返回空字符串。
+        """
+        file_name = f"{filename or secrets.token_hex(8)}.png"
+        path = (SCREENSHOT_DIR / file_name).as_posix()
+
+        try:
+            # 核心：直接调用 find_element，
+            self.find_element(by, value).screenshot(path)
+            logger.info(f"元素截图已保存: {path}")
+            return path
+        except Exception as e:
+            logger.error(f"元素截图失败: {e}")
+            return ""
+
+    @property
+    def is_alive(self) -> bool:
+        """判断当前驱动会话是否仍然存活。"""
+        return self.driver is not None and self.driver.session_id is not None
+
+    def quit(self):
+        """安全关闭 Appium 驱动并断开连接。"""
+        if self.driver:
+            try:
+                # 获取 session_id 用于日志追踪
+                sid = self.session_id
+                self.driver.quit()
+                logger.info(f"已安全断开连接 (Session: {sid})")
+            except Exception as e:
+                logger.warning(f"断开连接时发生异常 (可能服务已预先关闭): {e}")
+            finally:
+                self.driver = None
+        else:
+            logger.debug("没有正在运行的 Driver 实例需要关闭。")

core/enums.py

@@ -0,0 +1,58 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+"""
+@author: CNWei,ChenWei
+@Software: PyCharm
+@contact: t6g888@163.com
+@file: enums
+@date: 2026/2/27 17:05
+@desc: 
+"""
+from enum import Enum
+
+
+class AppiumStatus(Enum):
+    """Appium 服务状态枚举"""
+    READY = "服务已启动"  # 服务和驱动都加载完成 (HTTP 200 + ready: true)
+    INITIALIZING = "驱动正在加载"  # 服务已响应但驱动仍在加载 (HTTP 200 + ready: false)
+    CONFLICT = "端口被其他程序占用"  # 端口被其他非 Appium 程序占用
+    OFFLINE = "服务未启动"  # 服务未启动
+    ERROR = "内部错误"
+    UNKNOWN = "未知状态"
+
+
+class ServiceRole(Enum):
+    """服务角色枚举：定义服务的所有权和生命周期"""
+    MANAGED = "托管模式"  # 由本脚本启动，负责清理
+    EXTERNAL = "共享模式"  # 复用现有服务，不负责清理
+    NULL = "空模式"  # 无效或未初始化的服务
+
+
+class AppPlatform(Enum):
+    """
+    定义支持的移动应用平台枚举。
+    """
+    ANDROID = "android"
+    IOS = "ios"
+
+
+class Locator(str, Enum):
+    """
+    定义元素定位策略枚举。
+    继承 str 以便直接作为参数传递给 Selenium/Appium 方法。
+    """
+    # --- 原有 Selenium 支持 ---
+    ID = "id"
+    NAME = "name"
+    CLASS = "class"
+    TAG = "tag"
+    LINK_TEXT = "link_text"
+    PARTIAL_LINK_TEXT = "partial_link_text"
+    CSS = "css"
+    XPATH = "xpath"
+    # --- Appium 特有支持 ---
+    ACCESSIBILITY_ID = "accessibility_id"
+    AID = "aid"  # 简写
+    ANDROID_UIAUTOMATOR = "android_uiautomator"
+    IOS_PREDICATE = "ios_predicate"

core/modules.py

@@ -0,0 +1,11 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+"""
+@author: CNWei,ChenWei
+@Software: PyCharm
+@contact: t6g888@163.com
+@file: modules
+@date: 2026/1/20 11:54
+@desc: 
+"""

core/run_appium.py

@@ -0,0 +1,343 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+"""
+@author: CNWei,ChenWei
+@Software: PyCharm
+@contact: t6g888@163.com
+@file: run_appium
+@date: 2026/1/12 10:21
+@desc: 
+"""
+import functools
+import logging
+import signal
+import subprocess
+import time
+import os
+import sys
+import http.client
+import socket
+import json
+
+from typing import List
+
+from core.settings import BASE_DIR, APPIUM_HOST, APPIUM_PORT, MAX_RETRIES
+from core.enums import AppiumStatus, ServiceRole
+
+logger = logging.getLogger(__name__)
+
+
+# 使用 npm run 确保 APPIUM_HOME=. 变量和本地版本生效
+# NODE_CMD = f"npm run appium -- -p {APPIUM_PORT}"
+# --- 1. 异常体系设计 (精确分类) ---
+class AppiumStartupError(Exception):
+    """所有 Appium 启动相关异常的基类"""
+    pass
+
+
+class AppiumPortConflictError(AppiumStartupError):
+    """端口被其他非 Appium 程序占用"""
+    pass
+
+
+class AppiumProcessCrashError(AppiumStartupError):
+    """进程启动后立即崩溃或在运行中意外退出"""
+    pass
+
+
+class AppiumTimeoutError(AppiumStartupError):
+    """服务在规定时间内未进入 READY 状态"""
+    pass
+
+
+class AppiumInternalError(AppiumStartupError):
+    """探测接口返回了无法解析的数据或网络异常"""
+    pass
+
+
+def resolve_appium_command(host: str, port: int | str) -> List[str]:
+    """
+       解析 Appium 可执行文件的绝对路径。
+       优先查找项目 node_modules 下的本地安装版本，避免 npm 包装层带来的信号传递问题。
+
+       :return: 用于 subprocess 的命令列表
+       :raises SystemExit: 如果找不到 Appium 执行文件
+       """
+    bin_name = "appium.cmd" if sys.platform == "win32" else "appium"
+    appium_bin = BASE_DIR / "node_modules" / ".bin" / bin_name
+
+    if not appium_bin.exists():
+        # 报错提示：找不到本地 Appium，引导用户安装
+        logger.error(f"\n错误: 在路径 {appium_bin} 未找到 Appium 执行文件。")
+        logger.info("请确保已在项目目录下执行过: npm install appium")
+        sys.exit(1)
+    # 返回执行列表（用于 shell=False）--address 127.0.0.1 --port 4723 appium -a 127.0.0.1 -p 4723
+    return [str(appium_bin), "--address", host, "--port", str(port)]
+
+
+# 移除全局调用，防止 import 时因找不到 appium 而直接退出
+# APP_CMD_LIST = resolve_appium_command()
+
+def _cleanup_process_tree(process: subprocess.Popen = None) -> None:
+    """
+    核心清理逻辑：跨平台递归终止进程树。
+
+    :param process: subprocess.Popen 对象
+    """
+    if not process or process.poll() is not None:
+        return
+    pid = process.pid
+    logger.info(f"正在关闭 Appium 进程树 (PID: {pid})...")
+    try:
+        if sys.platform == "win32":
+            # Windows 下使用 taskkill 强制关闭进程树 /T
+            subprocess.run(
+                ['taskkill', '/F', '/T', '/PID', str(pid)],
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.DEVNULL
+            )
+        else:
+            # Unix/Linux: 获取进程组 ID 并发送终止信号
+            pgid = os.getpgid(pid)
+            os.killpg(pgid, signal.SIGTERM)
+            try:
+                process.wait(timeout=5)
+            except subprocess.TimeoutExpired:
+                # 5秒没关掉，强制杀掉整个进程组
+                # 如果 SIGTERM 不起作用，直接 SIGKILL 整个组
+                logger.warning(f"进程 {pid} 未响应 SIGTERM，发送 SIGKILL...")
+                os.killpg(pgid, signal.SIGKILL)
+        logger.info("Appium 服务已停止,相关进程已安全退出。")
+    except Exception as e:
+        logger.error(f"停止服务时发生异常: {e}")
+        try:
+            process.kill()
+            logger.warning(f"已强制杀死进程: {e}")
+        except Exception as e:
+            logger.debug(f"强制杀死进程失败 (可能已退出): {e}")
+            # 这里通常保持安静，因为我们已经尝试过清理了
+    logger.info("服务已完全清理。")
+
+
+class AppiumService:
+    """Appium 服务实例封装，用于管理服务生命周期"""
+
+    def __init__(self, role: ServiceRole, host: str, port: int | str, process: subprocess.Popen = None):
+        self.role = role
+        self.host = host
+        self.port = port
+        self.process = process
+
+    def __repr__(self):
+        return f"<AppiumService 角色='{self.role.value}' 地址=http://{self.host}:{self.port}>"
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.stop()
+
+    def stop(self):
+        """统一停止接口：根据角色决定是否关闭进程"""
+        match self.role:
+            case ServiceRole.EXTERNAL:
+                logger.info(f"--> [角色: {self.role.value}] 脚本退出，保留外部服务运行。")
+                return
+            case ServiceRole.MANAGED:
+                logger.info(f"正在关闭托管的 Appium 服务 (PID: {self.process.pid})...")
+                _cleanup_process_tree(self.process)
+                self.process = None
+            case ServiceRole.NULL:
+                logger.info(f"--> [角色: {self.role.value}] 无需执行清理。")
+
+
+def _check_port_availability(host: str, port: int) -> AppiumStatus:
+    """辅助函数：检查端口是否被占用"""
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        try:
+            sock.settimeout(1)
+            # 尝试绑定端口，如果成功说明端口空闲
+            sock.bind((host, port))
+            return AppiumStatus.OFFLINE  # 真正未启动
+        except OSError:
+            # 绑定失败，说明端口被占用，但此前 HTTP 请求失败，说明不是 Appium
+            return AppiumStatus.CONFLICT  # 端口被占用但没响应 HTTP
+
+
+def get_appium_status(host: str, port: int | str) -> AppiumStatus:
+    """
+    检测 Appium 服务当前状态。
+
+    逻辑：
+    1. 尝试 HTTP 连接 /status 接口。
+    2. 如果连接成功，检查 ready 字段。
+    3. 如果连接被拒绝，尝试绑定端口以确认端口是否真正空闲。
+
+    :return: AppiumStatus 枚举
+    """
+    connection = None
+    port = int(port)  # 确保端口是整数
+    try:
+        # 1. 端口开启，尝试获取 Appium 状态接口
+        connection = http.client.HTTPConnection(host, port, timeout=2)
+        connection.request("GET", "/status")
+        response = connection.getresponse()
+
+        if response.status != 200:
+            return AppiumStatus.CONFLICT
+
+        data = json.loads(response.read().decode('utf-8'))
+
+        # 2. 解析 Appium 3.x 标准响应结构
+        # 即使服务响应了，也要看驱动是否加载完成 (ready 字段)
+        is_ready = data.get("value", {}).get("ready")
+        return AppiumStatus.READY if is_ready else AppiumStatus.INITIALIZING
+
+    except (socket.error, ConnectionRefusedError):
+        # 3. 如果通信拒绝，检查端口是否被占用
+        return _check_port_availability(host, port)
+    except (http.client.HTTPException, json.JSONDecodeError):
+        return AppiumStatus.UNKNOWN
+    except Exception as e:
+        logger.error(f"状态检测异常: {e}")
+        return AppiumStatus.ERROR
+    finally:
+        if connection: connection.close()
+
+
+def start_appium_service(host: str = APPIUM_HOST, port: int | str = APPIUM_PORT) -> AppiumService:
+    """
+    管理 Appium 服务的生命周期
+    如果服务未启动，则启动本地服务；如果已启动，则复用。
+
+    :return: AppiumService 对象
+    """
+    process = None  # 1. 预先初始化变量，防止作用域错误
+    is_managed = False
+    # 轮询等待真正就绪
+    # 延迟获取命令，确保只在真正需要启动服务时检查环境
+    cmd_args = resolve_appium_command(host, port)
+    try:
+        for i in range(MAX_RETRIES):
+            status = get_appium_status(host, port)
+            match status:  # Python 3.10+ 的模式匹配
+                case AppiumStatus.READY:
+                    if is_managed:
+                        # 安全打印 PID
+                        pid_info = f"PID: {process.pid}" if process else "EXTERNAL"
+                        logger.info(f"Appium 服务启动成功! ({pid_info})")
+                        return AppiumService(ServiceRole.MANAGED, host, port, process)
+                    else:
+                        logger.info(f"--> [复用] 有效的 Appium 服务已在运行 (http://{host}:{port})")
+                        logger.info("--> [注意] 脚本退出时将保留该服务，不会将其关闭。")
+                        return AppiumService(ServiceRole.EXTERNAL, host, port, process)
+                case AppiumStatus.CONFLICT:
+                    _handle_port_conflict(port)
+                case AppiumStatus.OFFLINE:
+                    if not is_managed:
+                        process = _spawn_appium_process(cmd_args)
+                        is_managed = True
+
+                    elif process and process.poll() is not None:
+                        raise AppiumProcessCrashError("Appium 进程启动后异常退出。")
+
+                case AppiumStatus.INITIALIZING:
+                    if is_managed and process and process.poll() is not None:
+                        raise AppiumProcessCrashError("Appium 在初始化期间崩溃。")
+
+                    if i % 4 == 0:  # 每 2 秒提醒一次，避免刷屏
+                        logger.info("Appium 正在加载驱动/插件，请稍候...")
+                case AppiumStatus.ERROR:
+                    raise AppiumInternalError("探测接口发生内部错误（可能是解析失败或严重网络异常），脚本终止。")
+                case _:
+                    raise AppiumInternalError("Appium 启动异常")
+
+            time.sleep(0.5)
+
+        raise AppiumTimeoutError("启动超时：Appium 在规定时间内未完成初始化。")
+    except AppiumStartupError as e:
+        if process:
+            logger.warning(f"检测到启动阶段异常，正在回收进程资源 (PID: {process.pid})...")
+            _cleanup_process_tree(process)
+        # 将清理后的异常继续向上传递给装饰器
+        raise e
+
+
+def _handle_port_conflict(port: int | str) -> AppiumStartupError:
+    hand = f"端口 {port} 被其他程序占用,建议清理命令：\n"
+    if sys.platform == "win32":
+        raise AppiumPortConflictError(
+            f"{hand}CMD: for /f \"tokens=5\" %a in ('netstat -aon ^| findstr :{port}') do taskkill /F /PID %a")
+    else:
+        raise AppiumPortConflictError(f"{hand}Unix: lsof -ti:{port} | xargs kill -9")
+
+
+def _spawn_appium_process(cmd_args: List[str]) -> subprocess.Popen:
+    """启动 Appium 子进程"""
+    logger.info(f"正在启动本地 Appium 服务...")
+
+    # 注入环境变量，确保 Appium 寻找项目本地的驱动
+    env_vars = os.environ.copy()
+    env_vars["APPIUM_HOME"] = str(BASE_DIR)
+
+    try:
+        return subprocess.Popen(
+            cmd_args,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+            env=env_vars,
+            cwd=BASE_DIR,
+            # Windows 和 Linux/Mac 的处理方式不同
+            creationflags=subprocess.CREATE_NEW_PROCESS_GROUP if sys.platform == "win32" else 0,
+            preexec_fn=os.setsid if sys.platform != "win32" else None
+        )
+    except Exception as e:
+        logger.error(f"启动过程发生异常: {e}")
+        sys.exit(1)
+
+
+def stop_appium_service(service: AppiumService):
+    """
+    停止 Appium 服务。
+    如果已经使用了 with 语句，通常不需要手动调用此函数。
+    :param service: AppiumService 对象
+    """
+    if service:
+        service.stop()
+
+
+# --- 装饰器实现 ---
+def managed_appium(host: str = APPIUM_HOST, port: int | str = APPIUM_PORT):
+    def decorator(func):
+        @functools.wraps(func)
+        def wrapper(*args, **kwargs):
+            # 1. start_appium_service 返回 AppiumService 实例
+            # 2. with 触发 __enter__
+            # 3. 结束或异常触发 __exit__ -> stop()
+            with start_appium_service(host, port) as service:
+                logger.debug(f"装饰器环境就绪: {service}")
+                return func(*args, **kwargs)
+
+        return wrapper
+
+    return decorator
+
+
+if __name__ == "__main__":
+    # 使用示例：作为一个上下文管理器或简单的生命周期示例
+    appium_service = None
+    try:
+        appium_service = start_appium_service()
+        print(f"\n[项目路径] {BASE_DIR}")
+        print(f"[服务状态] Appium 运行中...)")
+        print("[操作提示] 按 Ctrl+C 停止服务...")
+
+        # 保持运行，直到手动停止（在实际测试框架中，这里会被替换为测试执行逻辑）
+        while True:
+            time.sleep(1)
+
+    except KeyboardInterrupt:
+        print("\n收到停止信号...")
+    finally:
+        stop_appium_service(appium_service)

core/settings.py

@@ -0,0 +1,65 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+"""
+@author: CNWei,ChenWei
+@Software: PyCharm
+@contact: t6g888@163.com
+@file: settings
+@date: 2026/1/19 16:54
+@desc: 全局配置管理
+"""
+
+import os
+from pathlib import Path
+
+# 项目根目录 (core 的上一级)
+BASE_DIR = Path(__file__).resolve().parents[1]
+
+# --- 目录配置 ---
+OUTPUT_DIR = BASE_DIR / "outputs"
+SCREENSHOT_DIR = OUTPUT_DIR / "screenshots"
+LOG_DIR = OUTPUT_DIR / "logs"
+LOG_BACKUP_DIR = LOG_DIR / "backups"
+ALLURE_TEMP = BASE_DIR / "temp"
+REPORT_DIR = BASE_DIR / "reports"
+CONFIG_DIR = BASE_DIR / "config"
+DATA_DIR = BASE_DIR / "data"
+
+# 需要初始化的目录列表
+REQUIRED_DIRS = [LOG_DIR, LOG_BACKUP_DIR, ALLURE_TEMP, SCREENSHOT_DIR]
+
+# --- 文件路径 ---
+LOG_SOURCE = LOG_DIR / "pytest.log"
+CAPS_CONFIG_PATH = CONFIG_DIR / "caps.yaml"
+
+# --- 启动 Appium 最大尝试次数 ---
+MAX_RETRIES = 40
+
+# --- 核心配置 ---
+IMPLICIT_WAIT_TIMEOUT = 10
+EXPLICIT_WAIT_TIMEOUT = 10
+
+# 默认 Appium Server 地址 (可通过命令行参数覆盖)
+APPIUM_HOST = "127.0.0.1"
+APPIUM_PORT = 4723
+
+# --- 环境配置 (Environment Switch) ---
+CURRENT_ENV = os.getenv("APP_ENV", "test")
+
+# base_url：业务接口的入口地址。主要用于通过 API 快速构造测试数据（前置条件）或查询数据库/接口状态来验证 App 操作是否生效（数据断言）
+# source_address：后端服务器的物理/源站地址。通常用于绕过负载均衡或 DNS 直接指定访问特定的服务器节点。
+ENV_CONFIG = {
+    "test": {
+        "base_url": "https://test.example.com",
+        "source_address": "192.168.1.100"
+    },
+    "uat": {
+        "base_url": "https://uat.example.com",
+        "source_address": "192.168.1.200"
+    },
+    "prod": {
+        "base_url": "https://www.example.com",
+        "source_address": "10.0.0.1"
+    }
+}

core/types.py

@@ -0,0 +1,18 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+"""
+@author: CNWei,ChenWei
+@Software: PyCharm
+@contact: t6g888@163.com
+@file: types
+@date: 2026/3/3 17:10
+@desc: 
+"""
+from typing import TypeAlias
+
+# 标识这是一个指向磁盘图片的字符串路径
+ImagePath: TypeAlias = str
+# 标识这是一个图片的二进制字节流
+ImageBytes: TypeAlias = bytes
+

docs/AndroidSDK环境配置指南.md

@@ -0,0 +1,106 @@
+# Android SDK 环境配置指南
+
+本指南将引导您完成 Appium for Android 测试所需的基础环境——Android SDK 的安装与配置。
+
+## 1. 创建 SDK 目录
+
+首先，在您的电脑上选择一个合适的位置（例如 `D:\`），创建一个新目录作为 Android SDK 的根目录。
+
+```
+D:\android_sdk
+```
+
+## 2. 安装命令行工具 (Command Line Tools)
+
+1. 访问 Android Studio [官方下载页面](https://developer.android.google.cn/studio?hl=zh-cn#command-line-tools-only)
+   ，在页面底部找到 **"仅命令行工具"** 部分，下载适用于您操作系统的软件包（例如 `commandlinetools-win-xxxxxxx_latest.zip`）。
+   ![image-20260110142023709.png](assets/image-20260110142023709.png)
+2. 解压下载的 `zip` 文件。
+
+3. 将解压后得到的 `cmdline-tools` 目录移动到您在第一步创建的 `android_sdk` 目录中。
+
+4. 进入 `android_sdk/cmdline-tools` 目录，在内部创建一个名为 `latest` 的新目录。
+
+5. 将 `cmdline-tools` 目录内的所有内容（包括 `bin` 目录、`lib` 目录、`NOTICE.txt` 和 `source.properties` 文件）移动到刚刚创建的
+   `latest` 目录中。
+
+   完成后的目录结构应如下所示：
+
+   ```
+   android_sdk/
+   ├── cmdline-tools/
+   │   └── latest/        ← 必须有这个 'latest' 目录
+   │       ├── bin/
+   │       │   └── sdkmanager
+   │       ├── lib/
+   │       └── ...
+   ```
+
+## 3. 安装平台工具 (Platform-Tools)
+
+1. 访问 SDK
+   平台工具版本说明页面，下载最新的[平台工具包](https://developer.android.google.cn/tools/releases/platform-tools?hl=zh-cn#downloads.html)
+   （例如 `platform-tools-latest-windows.zip`）。
+   ![image-20260110141432210.png](assets/image-20260110141432210.png)
+2. 解压下载的 `zip` 文件。
+
+3. 将解压后得到的 `platform-tools` 目录整个移动到 `android_sdk` 根目录下。
+
+   此时，您的 `android_sdk` 目录结构应如下：
+
+   ```
+   android_sdk/
+   ├── cmdline-tools/
+   ├── platform-tools/
+   └── ...
+   ```
+
+## 4. 配置环境变量
+
+为了让系统能够识别 `adb`、`sdkmanager` 等命令，需要配置以下环境变量：
+
+1. **新建 `ANDROID_HOME`**:
+    * 变量名: `ANDROID_HOME`
+    * 变量值: `D:\android_sdk` (替换为您的实际路径)
+
+2. **编辑 `Path` 变量**:
+   在系统变量的 `Path` 中，追加以下两项：
+   ```
+   %ANDROID_HOME%/tools
+   %ANDROID_HOME%/tools/bin
+   %ANDROID_HOME%/platform-tools
+   %ANDROID_HOME%/cmdline-tools/latest/bin
+   ```
+   ![image-20260110144326458.png](assets/image-20260110144326458.png)
+   ![image-20260110144556568.png](assets/image-20260110144556568.png)
+
+> **提示**: 关于环境变量的更多详细设置，请参阅
+> Android [官方文档](https://developer.android.google.cn/tools/variables?hl=zh-cn)。
+
+## 5. 验证安装
+
+打开一个新的命令行窗口（CMD 或 PowerShell），输入以下命令：
+
+```sh
+adb --version
+```
+
+如果成功显示 Android Debug Bridge 的版本信息，则说明 `platform-tools` 已配置成功。
+
+## 6. (可选) 配置 Bundletool
+
+`bundletool` 是 Google 用于处理 Android App Bundles (`.aab` 文件) 的工具。如果 `appium-doctor` 提示
+`bundletool.jar cannot be found` 警告，您可以按以下步骤处理：
+
+1. **下载**: 从 bundletool GitHub Releases 页面下载最新的 `bundletool-all-x.x.x.jar` 文件。
+
+2. **创建目录**: 在您的 `ANDROID_HOME` 目录下（例如 `D:\android_sdk`）创建一个名为 `bundle-tool` 的新文件夹。
+
+3. **重命名并移动**: 将下载的 `.jar` 文件重命名为 `bundletool.jar`，然后将其移动到 `bundle-tool` 文件夹中。
+
+4. **配置 PATH**: 将 `bundle-tool` 目录也添加到系统的 `Path` 环境变量中：
+   ```
+   %ANDROID_HOME%\bundle-tool
+   ```
+
+完成以上步骤后，重新运行 `appium-doctor`，相关警告应会消失。

docs/常用参数.md

@@ -0,0 +1,79 @@
+# 常用参数说明
+
+## 设备能力配置 (Capabilities)
+
+```python
+ANDROID_CAPS = {
+    "platformName": "Android",
+    "automationName": "uiautomator2",
+    "deviceName": "Android",
+    "appPackage": "com.android.settings",
+    "appActivity": ".Settings",
+    "noReset": False
+}
+```
+
+| 字段名称           | 字段含义解释                                      | 示例值说明                                                                                 |
+|----------------|---------------------------------------------|---------------------------------------------------------------------------------------|
+| platformName   | 测试的平台/操作系统。这是必填项，告诉自动化框架（如Appium）目标是什么系统。   | "Android" 表示这是一个Android设备。如果是iOS设备，则为 "iOS"。                                          |
+| automationName | 使用的自动化驱动引擎。指定底层用哪个工具来驱动设备进行UI交互。            | "uiautomator2" 是当前主流的Android驱动框架，比旧的 "UiAutomator" 更稳定和高效。                            |
+| deviceName     | 设备标识/名称。用于在同时连接多台设备时指定目标。                   | "Android" 是一个通用标识。在实际测试中，通常用 adb devices 获取的真实设备序列号（如 emulator-5554）来替换它，以确保连接到正确的设备。 |
+| appPackage     | 要测试的应用程序包名。这是应用的唯一标识，就像它在Android系统中的“身份证号”。 | "com.android.settings" 是Android系统“设置”应用的包名。测试你自己的应用时，需替换为你应用的包名。                      |
+| appActivity    | 要启动的应用内具体页面。它指定了应用启动后打开的第一个界面（Activity）。    | ".Settings" 是“设置”应用的主界面。前面的点. 表示它是 appPackage 下的一个相对路径。                               |
+
+获取应用的 appPackage 有几种常用方法
+
+```shell
+adb shell pm list packages | findstr your_package_name
+# adb shell pm list packages -3 仅列出用户安装的第三方应用的包名
+```
+
+获取应用的 appActivity 有几种常用方法
+
+```shell
+# 1，使用 aapt 工具分析 APK 文件（需有安装包）/build-tools/{version}/aapt
+aapt dump badging your_app.apk | findstr launchable-activity
+# 输出结果：launchable-activity: name='com.example.myapp.MainActivity'
+# name= 后面的值 'com.example.myapp.MainActivity' 就是你需要的主 Activity
+
+# 2，通过 ADB 命令获取（需应用已安装）
+adb shell dumpsys window | findstr mCurrentFocus
+# 输出结果：mCurrentFocus=Window{... u0 com.example.myapp/com.example.myapp.MainActivity}
+# / 后面的部分 com.example.myapp.MainActivity 就是当前 Activity
+```
+
+常用补充字段：
+noReset：True/False。是否在会话开始前重置应用状态（例如清除应用数据）。设置为 True 可以避免每次测试都重新登录。
+
+platformVersion：指定设备的Android系统版本（如 "11.0"）。虽然不是必须，但指定后能增强兼容性。
+
+unicodeKeyboard 和 resetKeyboard：用于处理中文输入等特殊字符输入。
+
+newCommandTimeout：设置Appium服务器等待客户端发送新命令的超时时间（秒），默认为60秒。在长时间操作中可能需要增加。
+
+## allure核心属性表
+
+| 属性                  | 说明                                   | 用法示例        |      
+|---------------------|--------------------------------------|-------------|
+| @allure.epic        | 顶层分类（如：APP项目名称）                      | 定义在测试类/项目上  |  
+| @allure.feature     | 功能模块（如：登录模块、交易模块）                    | 定义在测试类上     |     
+| @allure.story       | 用户场景（如：成功登录、账号锁定）                    | 定义在测试方法上    |    
+| @allure.title       | 测试用例标题（支持动态显示）                       | 替换方法名显示在报告中 | 
+| @allure.severity    | "严重程度（BLOCKER, CRITICAL, NORMAL...）" | 用于筛选高优先级用例  |  
+| @allure.description | 详细描述（支持 Markdown）                    | 解释测试背景或前提条件 | 
+| @allure.link        | 外部链接（Bug系统、需求文档）                     | 快速点击跳转      |      
+| @allure.issue       | 缺陷链接（通常会自动带上 ISSUE 前缀）               | 追踪已知 Bug    |
+
+Pytest 原生高频参数 和 你的自定义参数。
+
+| **分类**    | **长参数**          | **短参数/别名** | **作用说明**                           |
+|-----------|------------------|------------|------------------------------------|
+| **基础运行**  | `--verbose`      | `-v`       | 打印详细运行过程（显示用例名称）                   |
+|           | (无)              | `-s`       | 允许在控制台打印代码里的 `print` 内容            |
+| **调试控制**  | `--exitfirst`    | `-x`       | 遇到第一个失败的用例就立即停止测试                  |
+|           | `--maxfail=n`    | (无)        | 累计失败 `n` 个用例后停止运行                  |
+|           | `--last-failed`  | `--lf`     | 只运行上次运行失败的用例                       |
+| **自定义参数** | **`--platform`** | **`-P`**   | 指定运行平台 \[Android\/IOS\/别名\]\(可自定义) |
+|           | **`--udid`**     | **`-U`**   | 指定手机唯一标识 (可自定义)                    |
+|           | **`--host`**     | **`-H`**   | 指定 Appium 服务器地址 (可自定义)             |
+| **报告相关**  | `--alluredir`    | (无)        | 指定 Allure 原始数据的存放路径                |

main.py

@@ -1,42 +1,100 @@
-import time
+#!/usr/bin/env python
+# coding=utf-8
 
-from appium import webdriver
-from appium.options.android import UiAutomator2Options
+"""
+@author: CNWei,ChenWei
+@Software: PyCharm
+@contact: t6g888@163.com
+@file: main
+@date: 2026/1/13 16:54
+@desc:
+"""
+import shutil
+import datetime
+from pathlib import Path
 
-from run_appium import start_appium_service, stop_appium_service
+import pytest
 
-# 在自动化套件启动前执行
-proc = start_appium_service()
+from core.settings import LOG_SOURCE, LOG_BACKUP_DIR, ALLURE_TEMP
+from core.enums import AppPlatform
+from utils.dirs_manager import ensure_dirs_ok
+from utils.report_handler import generate_allure_report
 
-# 配置Android设备参数
-capabilities = dict(
-    platformName='Android',
-    automationName='uiautomator2',
-    deviceName='Android',
-    appPackage='com.android.settings',
-    appActivity='.Settings'
-)
 
-# 转换capabilities为Appium Options
-options = UiAutomator2Options().load_capabilities(capabilities)
+# netstat -ano | findstr :4723
+# taskkill /PID 12345 /F
 
-# 连接Appium服务器
-# driver = webdriver.Remote('http://localhost:4723', options=options)
-driver = webdriver.Remote('http://127.0.0.1:4723', options=options)
+def _archive_logs():
+    """
+    在测试开始前，归档上一次运行的日志文件。
+    此时没有任何句柄占用，move 操作是 100% 安全的。
+    """
+    # 4. 备份日志 (无论测试是否崩溃都执行)
+    if LOG_SOURCE.exists() and LOG_SOURCE.stat().st_size > 0:
+        now = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
+        backup_path = LOG_BACKUP_DIR / f"pytest_{now}.log"
+        try:
+            # 移动并重命名
+            shutil.move(str(LOG_SOURCE), str(backup_path))
+            print(f"已自动归档上次运行的日志: {backup_path}")
+            # shutil.copy2(LOG_SOURCE, backup_path)
+            # print(f"日志已备份至: {backup_path}")
+            _clean_old_logs(LOG_BACKUP_DIR)
+        except Exception as e:
+            print(f"归档旧日志失败 (可能被外部编辑器打开): {e}")
+    else:
+        print("未找到原始日志文件，跳过备份。")
+
+
+# 日志清理
+def _clean_old_logs(backup_dir, keep_count=10):
+    files = sorted(Path(backup_dir).glob("pytest_*.log"), key=lambda p: p.stat().st_mtime)
+    while len(files) > keep_count:
+        file_to_remove = files.pop(0)
+        try:
+            file_to_remove.unlink(missing_ok=True)
+        except OSError as e:
+            print(f"清理旧日志失败 {file_to_remove}: {e}")
+
+
+def _clean_temp_dirs():
+    """
+    可选：如果你想在测试前清理掉旧的临时文件
+    """
+    if ALLURE_TEMP.exists():
+        shutil.rmtree(ALLURE_TEMP)
+        # 加上 ignore_errors 是为了防止文件被占用导致整个测试无法启动
+        shutil.rmtree(ALLURE_TEMP, ignore_errors=True)
+    ALLURE_TEMP.mkdir(parents=True, exist_ok=True)
 
 
 def main():
-    # 简单操作示例
     try:
-        time.sleep(1)
-        print("当前Activity:", driver.current_activity)
+        # 1. 创建目录
+        ensure_dirs_ok()
+
+        # 2. 处理日志
+        _archive_logs()
+
+        # 3. 执行 Pytest
+
+        args = [
+            "test_cases",
+            "-x",  # 注意：-x 表示遇到错误立即停止，如果是全量回归建议去掉 -x
+            "-v",
+            f"--alluredir={ALLURE_TEMP}",
+            f"--platform={AppPlatform.ANDROID.value}",
+            "--caps_name=wan_android"
+        ]
+        pytest.main(args)
+
+        # 4. 生成报告
+        generate_allure_report()
+    except Exception as e:
+        print(f"自动化测试执行过程中发生异常: {e}")
+
     finally:
-        driver.quit()
-        # 在自动化套件结束后执行
-        stop_appium_service(proc)
-    print("Hello from AppAutoTest!")
-
-
+        print("Time-of-check to Time-of-use")
 
 
 if __name__ == "__main__":

page_objects/wan_android_home.py

@@ -0,0 +1,66 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+"""
+@author: CNWei,ChenWei
+@Software: PyCharm
+@contact: t6g888@163.com
+@file: wan_android_home
+@date: 2026/1/30 17:18
+@desc: 
+"""
+import logging
+
+import allure
+from appium import webdriver
+
+from core.base_page import BasePage
+
+logger = logging.getLogger(__name__)
+
+
+class HomePage(BasePage):
+    # 定位参数
+    menu = ("accessibility id", "开启")
+    home = ("id", 'com.manu.wanandroid:id/largeLabel')
+    project = ("-android uiautomator", 'new UiSelector().text("项目")')
+    system = ("-android uiautomator", 'new UiSelector().text("体系")')
+
+    tv_name = ("id", "com.manu.wanandroid:id/tvName")
+
+    account = ("-android uiautomator", 'new UiSelector().text("账号")')
+    pass_word = ("-android uiautomator", 'new UiSelector().text("密码")')
+
+    login_button = ("accessibility id", '登录')
+
+    def __init__(self, driver: webdriver.Remote):
+        super().__init__(driver)
+
+    @allure.step("点击 “侧边栏”")
+    def click_open(self):
+        if self.wait_until_visible(*self.menu, timeout=1):
+            self.click(*self.menu)
+            self.attach_screenshot_bytes("侧边栏截图")
+        with allure.step("准备登录"):
+            self.click(*self.tv_name)
+
+    @allure.step("登录账号：{1}")
+    def login(self, username, password):
+        """执行登录业务逻辑"""
+        account_element_id = self.find_element(*self.account).id
+        account_input = {"elementId": account_element_id, "text": username}
+
+        pwd_element_id = self.find_element(*self.pass_word).id
+        pass_word_input = {"elementId": pwd_element_id, "text": password}
+
+        if self.wait_until_visible(*self.login_button):
+            self.click(*self.account).driver.execute_script('mobile: type', account_input)
+            self.click(*self.pass_word).driver.execute_script('mobile: type', pass_word_input)
+
+        self.click(*self.login_button)
+
+        if self.wait_until_visible(*self.tv_name):
+            self.get_screenshot_as_file("登陆成功")
+            self.get_highlight_screenshot_as_file(*self.tv_name, "登陆成功-高亮")
+            self.get_element_screenshot_as_file(*self.tv_name)
+            self.long_press(x=636, y=117, duration=300)

page_objects/wan_android_project.py

@@ -0,0 +1,41 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+"""
+@author: CNWei,ChenWei
+@Software: PyCharm
+@contact: t6g888@163.com
+@file: wan_android_project
+@date: 2026/1/30 17:37
+@desc: 
+"""
+import logging
+
+import allure
+from appium import webdriver
+
+from core.base_page import BasePage
+from utils.decorators import StepTracer
+
+logger = logging.getLogger(__name__)
+
+
+class ProjectPage(BasePage):
+    # 定位参数
+    project_title = ("-android uiautomator", 'new UiSelector().text("项目")')
+    pro_table_title = ("-android uiautomator", 'new UiSelector().text("完整项目")')
+
+    def __init__(self, driver: webdriver.Remote):
+        super().__init__(driver)
+
+    @allure.step("切换到“项目”页面")
+    def switch_to_project(self):
+        self.click(*self.project_title).attach_screenshot_bytes()
+
+    @allure.step("滑动切换“项目”内容")
+    @StepTracer("页面滑动")
+    def slide_views(self):
+        with allure.step("向左滑动3次"):
+            with StepTracer("开始划了"):
+                for _ in range(3):
+                    self.swipe("left")

pyproject.toml

@@ -1,11 +1,17 @@
 [project]
 name = "appautotest"
 version = "0.1.0"
-description = "Add your description here"
+description = "App自动化测试框架"
 readme = "README.md"
 requires-python = ">=3.11"
 dependencies = [
+    "allure-pytest==2.13.5",
     "appium-python-client>=5.2.4",
+    "pytest>=8.3.5",
+    "PyYAML>=6.0.1",
+    "pytest-rerunfailures>=16.1",
+    "python-dotenv>=1.2.1",
+    "pillow>=12.1.1",
 ]
 
 [[tool.uv.index]]

pytest.ini

@@ -0,0 +1,22 @@
+[pytest]
+addopts = -q --show-capture=no --reruns 2 --reruns-delay 1
+
+# 1. 开启实时控制台日志
+log_cli = True
+log_cli_level = INFO
+log_cli_format = %(asctime)s %(levelname)-5s [%(name)s]  - %(message)s
+log_cli_date_format = %H:%M:%S
+
+# 2. 开启日志文件记录
+log_file = outputs/logs/pytest.log
+log_file_level = INFO
+log_file_format = %(asctime)s %(levelname)-5s [%(name)s] %(module)s.%(funcName)s:%(lineno)d  - %(message)s
+log_file_date_format = %Y-%m-%d %H:%M:%S
+
+# 3. 基础配置
+# 解决中文测试用例显示为乱码（Unicode）的问题
+disable_test_id_escaping_and_forfeit_all_rights_to_community_support = True
+
+# 限制 Pytest 搜索范围，提升启动速度
+testpaths = test_cases
+python_files = test_*.py

run_appium.py

@@ -1,188 +0,0 @@
-#!/usr/bin/env python
-# coding=utf-8
-
-"""
-@author: CNWei,ChenWei
-@Software: PyCharm
-@contact: t6g888@163.com,chenwei@zygj.com
-@file: test
-@date: 2026/1/12 10:21
-@desc: 
-"""
-import subprocess
-import time
-import os
-import sys
-import http.client
-import socket
-import json
-from pathlib import Path
-from enum import Enum, auto
-
-# --- 核心配置 ---
-# 使用 pathlib 获取当前脚本所在的绝对路径
-BASE_DIR = Path(__file__).resolve().parent
-
-APPIUM_HOST = "127.0.0.1"
-APPIUM_PORT = 4723
-# 使用 npm run 确保 APPIUM_HOME=. 变量和本地版本生效
-NODE_CMD = f"npm run appium -- -p {APPIUM_PORT}"
-
-
-class AppiumStatus(Enum):
-    """Appium 服务状态枚举"""
-    READY = "服务已启动"  # 服务和驱动都加载完成 (HTTP 200 + ready: true)
-    INITIALIZING = "驱动正在加载"  # 服务已响应但驱动仍在加载 (HTTP 200 + ready: false)
-    CONFLICT = "端口被其他程序占用"  # 端口被其他非 Appium 程序占用
-    OFFLINE = "服务未启动"  # 服务未启动
-    ERROR = "内部错误"
-    UNKNOWN = "未知状态"
-
-
-def get_appium_status() -> AppiumStatus:
-    """深度探测 Appium 状态"""
-    conn = None
-    try:
-        # 1. 端口开启，尝试获取 Appium 状态接口
-        conn = http.client.HTTPConnection(APPIUM_HOST, APPIUM_PORT, timeout=2)
-        conn.request("GET", "/status")
-        response = conn.getresponse()
-
-        if response.status != 200:
-            return AppiumStatus.CONFLICT
-
-        data = json.loads(response.read().decode())
-
-        # 2. 解析 Appium 3.x 标准响应结构
-        # 即使服务响应了，也要看驱动是否加载完成 (ready 字段)
-        is_ready = data.get("value", {}).get("ready")
-        return AppiumStatus.READY if is_ready else AppiumStatus.INITIALIZING
-
-
-    except (socket.error, ConnectionRefusedError):
-        # 3. 如果通信拒绝，检查端口是否真的空闲
-        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
-            try:
-                s.settimeout(1)
-                s.bind((APPIUM_HOST, APPIUM_PORT))
-                return AppiumStatus.OFFLINE  # 真正未启动
-            except OSError:
-                return AppiumStatus.CONFLICT  # 端口被占用但没响应 HTTP
-    except (http.client.HTTPException, json.JSONDecodeError):
-        return AppiumStatus.CONFLICT
-    except Exception as e:
-        print(e)
-        return AppiumStatus.ERROR
-    finally:
-        if conn: conn.close()
-
-
-def start_appium_service():
-    """管理 Appium 服务的生命周期"""
-    # if check_before_start():
-    #     return None
-    process = None  # 1. 预先初始化变量，防止作用域错误
-    status = get_appium_status()
-    match status:  # Python 3.10+ 的模式匹配
-        case AppiumStatus.READY:
-            print(f"--> [复用] 有效的 Appium 服务已在运行 (Port: {APPIUM_PORT})")
-            return None
-        case AppiumStatus.INITIALIZING:
-            print("⏳ Appium 正在初始化，等待...")
-        case AppiumStatus.CONFLICT:
-            print(f"\n[!] 错误: 端口 {APPIUM_PORT} 被非 Appium 程序占用。")
-            print("=" * 60)
-            print("请手动执行以下命令释放端口后重试：")
-            if sys.platform == "win32":
-                print(
-                    f"  CMD: for /f \"tokens=5\" %a in ('netstat -aon ^| findstr :{APPIUM_PORT}') do taskkill /F /PID %a")
-            else:
-                print(f"  Terminal: lsof -ti:{APPIUM_PORT} | xargs kill -9")
-            print("=" * 60)
-            sys.exit(1)
-        case AppiumStatus.OFFLINE:
-            print("🔌 Appium 未启动")
-            print(f"🚀 正在准备启动本地 Appium 服务 (Port: {APPIUM_PORT})...")
-
-            # 注入环境变量，确保 Appium 寻找项目本地的驱动
-            env_vars = os.environ.copy()
-            env_vars["APPIUM_HOME"] = str(BASE_DIR)
-
-            try:
-                process = subprocess.Popen(
-                    NODE_CMD,
-                    shell=True,
-                    stdout=subprocess.DEVNULL,
-                    stderr=subprocess.DEVNULL,
-                    env=env_vars,
-                    cwd=BASE_DIR
-                )
-
-            except Exception as e:
-                print(f"❌ 启动过程发生异常: {e}")
-                sys.exit(1)
-        case _:
-            print("Appium 启动异常")
-            sys.exit(1)
-    # 轮询等待真正就绪
-    max_retries = 40
-    for i in range(max_retries):
-        status = get_appium_status()
-
-        if status == AppiumStatus.READY:
-            # 安全打印 PID
-            pid_str = f"PID: {process.pid}" if process else "EXTERNAL"
-            print(f"✨ Appium 已经完全就绪! ({pid_str})")
-            return process
-
-        if status == AppiumStatus.ERROR:
-            print("❌ 探测接口发生内部错误（可能是解析失败或严重网络异常），脚本终止。")
-            if process: process.terminate()
-            sys.exit(1)
-
-        if status == AppiumStatus.INITIALIZING:
-            if i % 4 == 0:
-                print("...Appium 正在加载驱动/插件，请稍候...")
-
-        if status == AppiumStatus.OFFLINE:
-            # 仅当进程是我们启动的（process 不为 None）才检查崩溃(None: 程序正常运行，非None: 程序异常)
-            if process and process.poll() is not None:
-                print("❌ Appium 进程启动后异常退出。")
-                sys.exit(1)
-
-        time.sleep(0.5)
-
-    print("❌ 启动超时：Appium 在规定时间内未完成初始化。")
-    if process: process.terminate()
-    sys.exit(1)
-
-
-def stop_appium_service(process):
-    """安全关闭服务"""
-    if process and process.poll() is None:
-        print(f"\n🛑 正在关闭 Appium 服务 (PID: {process.pid})...")
-        process.terminate()
-        try:
-            process.wait(timeout=5)
-        except subprocess.TimeoutExpired:
-            process.kill()
-        print("✅ 服务已安全退出。")
-
-
-if __name__ == "__main__":
-    # 使用示例：作为一个上下文管理器或简单的生命周期示例
-    appium_proc = None
-    try:
-        appium_proc = start_appium_service()
-        print(f"\n[项目路径] {BASE_DIR}")
-        print("\n[提示] 现在可以手动或通过其他脚本运行测试用例。")
-        print("[提示] 按下 Ctrl+C 可停止由本脚本启动的服务。")
-
-        # 保持运行，直到手动停止（在实际测试框架中，这里会被替换为测试执行逻辑）
-        while True:
-            time.sleep(1)
-
-    except KeyboardInterrupt:
-        pass
-    finally:
-        stop_appium_service(appium_proc)

test_cases/test_wan_android_home.py

@@ -0,0 +1,58 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+"""
+@author: CNWei,ChenWei
+@Software: PyCharm
+@contact: t6g888@163.com
+@file: test_wan_android_home
+@date: 2026/1/30 17:42
+@desc: 
+"""
+import logging
+import os
+
+import allure
+
+from page_objects.wan_android_home import HomePage
+from page_objects.wan_android_project import ProjectPage
+
+# 配置日志
+logger = logging.getLogger(__name__)
+
+
+@allure.epic("测试用例示例")
+@allure.feature("登录模块")
+class TestWanAndroidHome:
+    @allure.story("常规登录场景")
+    @allure.title("使用合法账号登录成功")
+    @allure.severity(allure.severity_level.BLOCKER)
+    @allure.description("""
+        验证用户在正常网络环境下：
+        1. 处理初始化弹窗和广告
+        2. 输入正确账号密码后成功登录
+        """)
+    @allure.link("https://docs.example.com/login_spec", name="登录业务说明文档")
+    @allure.issue("BUG-1001", "已知偶发：我是一个bug")
+    def test_api_demos_success(self, driver):
+        """
+        测试场景：使用正确的用户名和密码登录成功
+        """
+        home = HomePage(driver)
+
+        # 执行业务逻辑
+        home.click_open()
+
+        home.login(os.getenv("USER_NAME"), os.getenv("PASS_WORD"))
+
+        # 断言部分使用 allure.step 包装，使其在报告中也是一个可读的步骤
+        with allure.step("断言"):
+            assert os.getenv("USER_NAME") == 'admintest123456'
+
+        # 页面跳转
+        with allure.step("验证页面跳转"):
+            project = home.go_to(ProjectPage)
+            project.switch_to_project()
+            project.assert_text(*project.pro_table_title, expected_text='完整项目')
+
+        project.delay(5)

test_cases/test_wan_android_project.py

@@ -0,0 +1,49 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+"""
+@author: CNWei,ChenWei
+@Software: PyCharm
+@contact: t6g888@163.com
+@file: test_wan_android_project
+@date: 2026/1/30 17:42
+@desc: 
+"""
+import logging
+
+import allure
+
+from page_objects.wan_android_project import ProjectPage
+
+# 配置日志
+logger = logging.getLogger(__name__)
+
+
+@allure.epic("测试用例示例")
+@allure.feature("项目模块")
+class TestWanAndroidProject:
+    @allure.story("项目切换场景")
+    @allure.title("切换项目页面")
+    @allure.severity(allure.severity_level.NORMAL)
+    @allure.description("""
+        验证滑动切换项目页面展示：
+        1. XXXXX
+        2. XXXXXXXXXXX
+        """)
+    @allure.link("https://docs.example.com/login_spec", name="项目业务说明文档")
+    @allure.issue("BUG-1002", "已知偶发：我是一个bug")
+    def test_project_slide_success(self, driver):
+        """
+        测试场景：验证页面滑动
+        """
+        project = ProjectPage(driver)
+
+        # 执行业务逻辑
+        project.switch_to_project()
+        project.slide_views()
+
+        # 断言部分使用 allure.step 包装，使其在报告中也是一个可读的步骤
+        with allure.step("断言"):
+            assert 1 == 1
+
+        project.delay(5)

tests/test_finder_converter.py

@@ -0,0 +1,89 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+"""
+@author: CNWei,ChenWei
+@Software: PyCharm
+@contact: t6g888@163.com
+@file: test_finder_converter
+@date: 2026/1/20 15:40
+@desc: 测试 utils/finder.py 中的 FinderConverter 逻辑
+"""
+
+import pytest
+from appium.webdriver.common.appiumby import AppiumBy
+from utils.finder import by_converter, register_custom_finder, converter
+
+
+class TestFinderConverter:
+
+    def setup_method(self):
+        """每个测试用例开始前重置 converter 状态"""
+        converter.clear_custom_finders()
+
+    def teardown_method(self):
+        """每个测试用例结束后重置 converter 状态"""
+        converter.clear_custom_finders()
+
+    @pytest.mark.parametrize("input_by, expected", [
+        ("id", "id"),
+        ("xpath", "xpath"),
+        ("link text", "link text"),
+        ("aid", AppiumBy.ACCESSIBILITY_ID),
+        ("class", AppiumBy.CLASS_NAME),
+        ("css", AppiumBy.CSS_SELECTOR),
+        ("uiautomator", AppiumBy.ANDROID_UIAUTOMATOR),
+        ("predicate", AppiumBy.IOS_PREDICATE),
+        ("chain", AppiumBy.IOS_CLASS_CHAIN),
+    ])
+    def test_standard_and_shortcuts(self, input_by, expected):
+        """测试标准定位方式和简写"""
+        assert by_converter(input_by) == expected
+
+    @pytest.mark.parametrize("input_by, expected", [
+        ("ID", "id"),
+        (" Id ", "id"),
+        ("accessibility_id", AppiumBy.ACCESSIBILITY_ID),
+        ("accessibility-id", AppiumBy.ACCESSIBILITY_ID),
+        ("-ios class chain", AppiumBy.IOS_CLASS_CHAIN),
+        (" -Ios-Class-Chain ", AppiumBy.IOS_CLASS_CHAIN),
+        ("UI_AUTOMATOR", AppiumBy.ANDROID_UIAUTOMATOR),
+    ])
+    def test_normalization(self, input_by, expected):
+        """测试归一化容错 (大小写、空格、下划线、横杠)"""
+        assert by_converter(input_by) == expected
+
+    def test_custom_registration(self):
+        """测试自定义注册功能"""
+        register_custom_finder("my_text", "-android uiautomator")
+        assert by_converter("my_text") == "-android uiautomator"
+
+        # 测试注册后归一化依然生效
+        assert by_converter("MY_TEXT") == "-android uiautomator"
+
+    def test_reset_functionality(self):
+        """测试重置功能"""
+        register_custom_finder("temp_key", "xpath")
+        assert by_converter("temp_key") == "xpath"
+
+        converter.clear_custom_finders()
+
+        with pytest.raises(ValueError, match="Unsupported locator strategy"):
+            by_converter("temp_key")
+
+    def test_invalid_strategy(self):
+        """测试不支持的定位策略"""
+        with pytest.raises(ValueError, match="Unsupported locator strategy"):
+            by_converter("unknown_strategy")
+
+    def test_invalid_types(self):
+        """测试非法类型输入"""
+        with pytest.raises(ValueError, match="Invalid selector type"):
+            by_converter(123)  # type: ignore
+
+        with pytest.raises(ValueError, match="Invalid selector type"):
+            by_converter(None)  # type: ignore
+
+
+if __name__ == "__main__":
+    pytest.main(["-v", __file__])

utils/__init__.py

@@ -0,0 +1,11 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+"""
+@author: CNWei,ChenWei
+@Software: PyCharm
+@contact: t6g888@163.com
+@file: __init__.py
+@date: 2026/1/16 09:06
+@desc: 
+"""

utils/data_loader.py

@@ -0,0 +1,28 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+"""
+@author: CNWei,ChenWei
+@Software: PyCharm
+@contact: t6g888@163.com
+@file: data_loader
+@date: 2026/1/27 10:00
+@desc: 数据加载工具
+"""
+import yaml
+from pathlib import Path
+from typing import Any
+
+
+def load_yaml(file_path: Path | str) -> dict[str, Any] | list[Any]:
+    """
+    加载 YAML 文件
+    :param file_path: 文件路径
+    :return: 数据字典或列表
+    """
+    path = Path(file_path)
+    if not path.exists():
+        raise FileNotFoundError(f"YAML file not found: {path}")
+
+    with open(path, "r", encoding="utf-8") as f:
+        return yaml.safe_load(f)

utils/decorators.py

@@ -0,0 +1,259 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+"""
+@author: CNWei,ChenWei
+@Software: PyCharm
+@contact: t6g888@163.com
+@file: decorators
+@date: 2026/1/15 11:30
+@desc:
+"""
+
+import logging
+import time
+import inspect
+from functools import wraps
+from typing import Union, Callable
+from contextvars import ContextVar
+from contextlib import ContextDecorator
+
+from core.custom_expected_conditions import get_condition
+
+logger = logging.getLogger(__name__)
+
+# 定义一个上下文变量，初始值为 0
+indent_var = ContextVar("indent_level", default=0)
+
+
+class StepTracer(ContextDecorator):
+    """
+    一个结合了上下文管理器和装饰器功能的追踪器。
+
+    主要职责是记录一个代码块或函数的执行情况，包括：
+    - 开始和结束的日志，并根据上下文进行缩进，形成层级结构。
+    - 计算并记录执行耗时。
+    - 捕获并记录执行期间发生的异常。
+
+    可作为上下文管理器使用:
+    with StepTracer("处理数据"):
+        ...
+
+    也可作为装饰器的一部分（通过 step_trace 工厂函数）。
+    """
+
+    def __init__(self, step_desc, source='wrapper', func_info=None):
+        """
+        初始化 StepTracer。
+        :param step_desc: 对当前步骤或操作的描述。
+        :param source: 日志记录器的名称。
+        :param func_info: 关联的函数信息，用于日志输出。
+        """
+        self.step_desc = step_desc
+        self.logger = logging.getLogger(source)
+        self.func_info = func_info
+        self.start_t = None
+
+    def __enter__(self):
+        """
+        进入上下文，记录步骤开始，并增加日志缩进层级。
+        """
+        # 1. 获取当前层级并计算前缀
+        level = indent_var.get()
+        # 使用 "  " (空格) 或 "│ " 作为缩进符号
+        self.prefix = "│  " * level
+
+        self.start_t = time.perf_counter()
+        info = f" | 方法: {self.func_info}" if self.func_info else ""
+        # self.logger.info(f"[步骤开始] | {self.step_desc}{info}")
+        self.logger.info(f"{self.prefix}┌── [步骤开始] | {self.step_desc}{info}")
+
+        # 2. 进入下一层，层级 +1
+        indent_var.set(level + 1)
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        """
+        退出上下文，记录步骤结束（成功或失败）、耗时，并恢复日志缩进层级。
+        如果发生异常，会记录异常信息但不会抑制它，异常会继续向上传播。
+        """
+        # 3. 恢复层级，层级 -1
+        level = indent_var.get() - 1
+        indent_var.set(level)
+
+        duration = time.perf_counter() - self.start_t
+
+        prefix = "│  " * level
+
+        if exc_type:
+            # 异常发生
+            # self.logger.error(
+            #     f"[步骤失败] {self.step_desc} | 耗时: {duration:.2f}s | 异常: {exc_type.__name__}"
+            # )
+            self.logger.error(
+                f"{prefix}└── [步骤失败] {self.step_desc} | 耗时: {duration:.2f}s | 异常: {exc_type.__name__}"
+            )
+        else:
+            # 执行成功
+            # self.logger.info(f"[步骤成功] {self.step_desc} | 耗时: {duration:.2f}s")
+            self.logger.info(
+                f"{prefix}└── [步骤成功] {self.step_desc} | 耗时: {duration:.2f}s"
+            )
+        # return False 确保异常继续向上抛出，不拦截异常
+        return False
+
+
+def resolve_wait_method(func):
+    """
+    装饰器：将字符串形式的等待条件解析为可调用的 Expected Condition (EC) 对象。
+
+    这个装饰器用于修饰那些接受一个 `method` 参数的方法（通常是等待方法）。
+    如果 `method` 是一个字符串，它会尝试将其解析为一个预定义的 `expected_conditions`。
+
+    字符串格式支持:
+    - "key": 直接映射到一个无参数的 EC。
+    - "key:arg1,arg2": 映射到一个需要参数的 EC，参数以逗号分隔。
+
+    解析逻辑委托给 `core.custom_expected_conditions.get_condition` 函数。
+    如果解析失败，会记录错误并重新抛出异常。
+
+    Args:
+        func (Callable): 被装饰的函数。
+
+    Returns:
+        Callable: 包装后的函数。
+    """
+
+    @wraps(func)
+    def wrapper(self, method: Union[Callable, str], *args, **kwargs):
+        if isinstance(method, str):
+            # 解析格式 "key:arg1,arg2" 或 仅 "key"
+            ec_name = method
+            ec_args = []
+
+            if ":" in method:
+                ec_name, params = method.split(":", 1)
+                if params:
+                    ec_args = params.split(",")
+
+            # 委托给 core.custom_expected_conditions.get_condition 处理
+            try:
+                logger.info(f"解析命名等待条件: '{ec_name}' 参数: {ec_args}")
+                method = get_condition(ec_name, *ec_args)
+            except Exception as e:
+                logger.error(f"解析等待条件 '{method}' 失败: {e}")
+                raise e
+
+        return func(self, method, *args, **kwargs)
+
+    return wrapper
+
+
+def action_screenshot(func):
+    """
+    装饰器：在被装饰方法成功执行后，自动触发截图。
+
+    主要用于UI自动化测试中，记录关键业务操作执行后的页面状态。
+    它会调用被装饰对象（通常是 Page Object）的 `attach_screenshot_bytes` 方法。
+
+    注意:
+    - 截图操作在原方法成功返回后执行。
+    - 被装饰的实例 (`self`) 必须拥有 `attach_screenshot_bytes` 方法。
+    - 如果截图失败，会记录一个警告日志，但不会影响主流程。
+
+    Args:
+        func (Callable): 被装饰的函数。
+
+    Returns:
+        Callable: 包装后的函数。
+    """
+
+    @wraps(func)
+    def wrapper(self, *args, **kwargs):
+        # 1. 正常执行原方法
+        result = func(self, *args, **kwargs)
+
+        # 2. 执行成功后立即截图（如果你希望在操作后的状态截图）
+        if hasattr(self, "attach_screenshot_bytes"):
+            try:
+                class_name = self.__class__.__name__
+                func_name = func.__name__
+                msg = f"操作记录_{class_name}_{func_name}"
+                # 传入当前方法名作为截图备注
+                self.attach_screenshot_bytes(msg)
+            except Exception as e:
+                logger.warning(f"装饰器执行截图失败: {e}")
+
+        return result
+
+    return wrapper
+
+
+def _format_params(func, *args, **kwargs):
+    """
+    辅助函数：格式化函数调用的参数，以便清晰地记录日志。
+
+    它会检查函数的签名，并执行以下操作：
+    1. 过滤掉实例方法或类方法中的 `self` 或 `cls` 参数。
+    2. 将位置参数和关键字参数格式化为一个可读的字符串。
+
+    Args:
+        func (Callable): 目标函数。
+        *args: 传递给函数的位置参数。
+        **kwargs: 传递给函数的关键字参数。
+
+    Returns:
+        str: 格式化后的参数字符串，例如 "arg1, kwarg='value'"。
+    """
+    sig = inspect.signature(func)
+    params = list(sig.parameters.values())
+    display_args = args[1:] if params and params[0].name in ('self', 'cls') else args
+
+    # 格式化参数显示，方便阅读
+    args_repr = [repr(a) for a in display_args]
+    kwargs_repr = [f"{k}={v!r}" for k, v in kwargs.items()]
+    all_params = ", ".join(args_repr + kwargs_repr)
+    return all_params
+
+
+def step_trace(step_desc="", source='wrapper'):
+    """
+    装饰器工厂：创建一个通用的执行追踪装饰器。
+
+    此装饰器利用 `StepTracer` 上下文管理器来提供结构化的日志，记录
+    函数的调用、参数、执行耗时和任何发生的异常。
+
+    功能包括:
+    - 自动格式化并记录函数的输入参数（会智能过滤 `self` 和 `cls`）。
+    - 使用 `StepTracer` 生成带缩进的层级式日志，清晰展示调用栈。
+    - 记录每个被追踪步骤的开始、成功/失败状态以及执行耗时。
+    - 捕获并记录异常，但不会抑制异常，保证上层逻辑可以处理。
+
+    用法:
+        @step_trace("执行用户登录操作")
+        def login(username, password):
+            ...
+
+    Args:
+        step_desc (str, optional): 对被装饰函数所执行操作的描述。
+                                   如果为空，日志中将只显示函数信息。
+        source (str, optional): 指定日志记录器的名称。默认为 'wrapper'。
+
+    Returns:
+        Callable: 一个可以装饰函数的装饰器。
+    """
+
+    def decorator(func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            # 1. 提取参数显示逻辑 (抽离出来可以作为工具函数)
+            all_params = _format_params(func, *args, **kwargs)
+            func_name = f"{func.__module__}.{func.__name__}"
+
+            # 2. 使用上下文管理器
+            with StepTracer(step_desc, source, f"{func_name}({all_params})"):
+                return func(*args, **kwargs)
+
+        return wrapper
+
+    return decorator

utils/dirs_manager.py

@@ -0,0 +1,32 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+"""
+@author: CNWei,ChenWei
+@Software: PyCharm
+@contact: t6g888@163.com
+@file: dirs_manager
+@date: 2026/2/3 10:52
+@desc: 
+"""
+
+from pathlib import Path
+
+from core.settings import REQUIRED_DIRS
+
+
+def ensure_dirs_ok():
+    """
+    统一管理项目目录的创建逻辑
+    """
+    for folder in REQUIRED_DIRS:
+        # 使用 exist_ok=True 避免并发冲突
+        folder.mkdir(parents=True, exist_ok=True)
+
+
+def ensure_dir(path: Path) -> Path:
+    """确保路径存在并返回路径本身"""
+    if not isinstance(path, Path):
+        path = Path(path)
+    path.mkdir(parents=True, exist_ok=True)
+    return path

utils/finder.py

@@ -0,0 +1,164 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+"""
+@author: CNWei,ChenWei
+@Software: PyCharm
+@contact: t6g888@163.com
+@file: finder
+@date: 2026/1/20 15:40
+@desc: 
+"""
+from typing import Literal, Final
+
+from appium.webdriver.common.appiumby import AppiumBy
+
+ByType = Literal[
+    # By(selenium)
+    "id", "xpath", "link text", "partial link text", "name", "tag name", "class name", "css selector",
+        # AppiumBy
+    '-ios predicate string',
+    '-ios class chain',
+    '-android uiautomator',
+    '-android viewtag',
+    '-android datamatcher',
+    '-android viewmatcher',
+    'accessibility id',
+    '-image',
+    '-custom',
+    '-flutter semantics label',
+    '-flutter type',
+    '-flutter key',
+    '-flutter text',
+    '-flutter text containing',
+        # 自定义常用简写 (Shortcuts)
+    "aid", "class", "css", "uiautomator", "predicate", "chain",
+]
+
+class FinderConverter:
+    """
+    定位查找转换工具类
+    提供策略的归一化处理、简写映射及动态自定义注册
+    """
+
+    # 预设的常用简写
+    _BUILTIN_SHORTCUTS: Final = {
+        "aid": AppiumBy.ACCESSIBILITY_ID,
+        "class": AppiumBy.CLASS_NAME,
+        "css": AppiumBy.CSS_SELECTOR,
+        "uiautomator": AppiumBy.ANDROID_UIAUTOMATOR,
+        "predicate": AppiumBy.IOS_PREDICATE,
+        "chain": AppiumBy.IOS_CLASS_CHAIN,
+    }
+
+    def __init__(self):
+        self._finder_map: dict[str, str] = {}
+        self._map_cache: dict[str, str] = {}
+        self._initialize()
+
+    @staticmethod
+    def _normalize(text: str) -> str:
+        """
+        统一清洗逻辑：转小写、去除空格、下划线、横杠
+        """
+        if not isinstance(text, str):
+            raise TypeError(f"Locator strategy must be a string, got {type(text).__name__} instead.")
+        return text.lower().strip().replace('_', '').replace(' ', '').replace('-', '')
+
+    def _initialize(self) -> None:
+        """初始化基础映射表"""
+        # 1. 动态加载 AppiumBy 常量值
+        for attr_name in dir(AppiumBy):
+            if attr_name.startswith("_"):
+                continue
+
+            attr_value = getattr(AppiumBy, attr_name)
+            if isinstance(attr_value, str):
+                # "class name" -> classname,"class_name" -> classname
+                self._finder_map[self._normalize(attr_value)] = attr_value
+
+        # 2. 加载内置简写（会覆盖同名的策略）
+        self._finder_map.update(self._BUILTIN_SHORTCUTS)
+
+        # 3. 备份初始状态
+        self._map_cache = self._finder_map.copy()
+
+    def convert(self, by_value: ByType | str) -> str:
+        """
+        将模糊或简写的定位方式转换为 Appium 标准定位字符串
+        :raises ValueError: 当定位方式不支持时抛出
+        """
+        if not by_value or not isinstance(by_value, str):
+            raise ValueError(f"Invalid selector type: {type(by_value)}. Expected a string.")
+
+        clean_key = self._normalize(by_value)
+        target = self._finder_map.get(clean_key)
+
+        if target is None:
+            raise ValueError(f"Unsupported locator strategy: '{by_value}'.")
+        return target
+
+    def register_custom_finder(self, alias: str, target: str) -> None:
+        """注册自定义定位策略"""
+        self._finder_map[self._normalize(alias)] = target
+
+    def clear_custom_finders(self) -> None:
+        """重置回初始官方/内置状态"""
+        self._finder_map = self._map_cache.copy()
+
+    def get_all_finders(self) -> list[str]:
+        """返回当前所有支持的策略 key（用于调试）"""
+        return list(self._finder_map.keys())
+
+
+# 导出单例，方便直接使用
+converter = FinderConverter()
+by_converter = converter.convert
+register_custom_finder = converter.register_custom_finder
+
+__all__=["by_converter", "register_custom_finder"]
+
+if __name__ == '__main__':
+    # 1. 测试标准转换与内置简写
+    print(f"ID 转换: {by_converter('id')}")                  # 输出: id
+    print(f"AID 简写转换: {by_converter('aid')}")           # 输出: accessibility id
+    print(f"CSS 简写转换: {by_converter('css')}")           # 输出: css selector
+
+    # 2. 测试强大的归一化容错 (空格、下划线、横杠、大小写)
+    print(f"类链容错: {by_converter(' -Ios-Class-Chain ')}") # 输出: -ios class chain
+    print(f"UIAutomator 容错: {by_converter('UI_AUTOMATOR')}") # 输出: -android uiautomator
+
+    # 3. 测试自定义注册
+    register_custom_finder("my_text", "-android uiautomator")
+    print(f"自定义注册测试: {by_converter('my_text')}")      # 输出: -android uiautomator
+
+    # 4. 测试重置功能
+    converter.clear_custom_finders()
+    print("已重置自定义查找器")
+    try:
+        by_converter("my_text")
+    except ValueError as e:
+        print(f"验证重置成功 (捕获异常): {e}")
+
+    # 5. 查看当前全量支持的归一化后的 Key
+    print(f"当前支持的策略总数: {len(converter.get_all_finders())}")
+    print(f"前 5 个策略示例: {converter.get_all_finders()[:5]}")
+    
+    # 6. 增加类型非法测试
+    print("\n--- 异常类型测试 ---")
+    try:
+        by_converter(123)  # 传入数字
+    except ValueError as e:
+        print(f"验证类型拦截成功: {e}")
+
+    try:
+        by_converter(None) # 传入 None
+    except ValueError as e:
+        print(f"验证空值拦截成功: {e}")
+
+    # 7. 验证不支持的策略 (验证 if target is None 分支)
+    print("\n--- 验证不支持的策略 ---")
+    try:
+        by_converter("unknown_strategy")
+    except ValueError as e:
+        print(f"验证不支持策略拦截成功: {e}")

utils/logger.py

@@ -0,0 +1,12 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+"""
+@author: CNWei,ChenWei
+@Software: PyCharm
+@contact: t6g888@163.com
+@file: logger
+@date: 2026/1/15 11:30
+@desc: 
+"""
+

utils/report_handler.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+"""
+@author: CNWei,ChenWei
+@Software: PyCharm
+@contact: t6g888@163.com
+@file: report_handler
+@date: 2026/2/3 13:51
+@desc: 
+"""
+import logging
+import subprocess
+import shutil
+
+from core.settings import ALLURE_TEMP, REPORT_DIR
+
+logger = logging.getLogger(__name__)
+
+
+def generate_allure_report() -> bool:
+    """
+    将 JSON 原始数据转换为 HTML 报告
+    """
+    if not ALLURE_TEMP.exists() or not any(ALLURE_TEMP.iterdir()):
+        logger.warning("未发现 Allure 测试数据，跳过报告生成。")
+        return False
+
+    # 检查环境是否有 allure 命令行工具
+    if not shutil.which("allure"):
+        logger.error("系统未安装 Allure 命令行工具，请先安装：https://allurereport.org/docs/")
+        return False
+
+    try:
+        logger.info("正在生成 Allure HTML 报告...")
+        # --clean 会清理掉 REPORT_DIR 里的旧报告
+        subprocess.run(
+            f'allure generate "{ALLURE_TEMP}" -o "{REPORT_DIR}" --clean',
+            shell=True,
+            check=True,
+            capture_output=True,
+            text=True
+        )
+        logger.info(f"Allure 报告已生成至: {REPORT_DIR}")
+        return True
+    except subprocess.CalledProcessError as e:
+        logger.error(f"Allure 报告生成失败: {e.stderr}")
+        return False