fix(conftest,config_loader): 修复 get_caps 的 Capabilities 加载逻辑

- 新增 pytest_addoption 增加 "--caps_name" 获取配置文件中的设备/平台名称
- 修复 get_caps 的 Capabilities 加载逻辑
- 优化 其他优化 enums.py
- 删除 移除部分文档,代码,第三方包(loguru)

.gitignore

@@ -5,10 +5,25 @@ build/
 dist/
 wheels/
 *.egg-info
-
-# Virtual environments
-.venv
-node_modules
-
 .idea
+
+# --- 依赖与环境 ---
+.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 <your-repository-url>
+   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,29 @@
+# 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

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.full_screen_screenshot(name=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.driver.get_screenshot_as_png()
+
+        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,684 @@
+#!/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 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 core.enums import AppPlatform
+from core.settings import IMPLICIT_WAIT_TIMEOUT, EXPLICIT_WAIT_TIMEOUT, APPIUM_HOST, APPIUM_PORT, SCREENSHOT_DIR
+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.full_screen_screenshot(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 full_screen_screenshot(self, name: str | None = None) -> str:
+        """
+        截取当前完整屏幕内容 (自愈逻辑、异常报错首选)
+        :param name: 图片文件名
+        :return: 截图保存的路径
+        """
+        file_name = f"{name 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 element_screenshot(self, by: str, value: str, name: str | None = None) -> str:
+        """
+        截取特定元素的图像 (业务校验、UI对比首选)
+        :param by: 定位策略
+        :param value: 定位值
+        :param name: 图片文件名
+        :return: 截图保存的路径
+        """
+        file_name = f"{name or secrets.token_hex(8)}.png"
+        path = (SCREENSHOT_DIR / file_name).as_posix()
+
+        try:
+            by = by_converter(by)
+            # 核心：直接调用底层 find_element，
+            self.driver.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"
+    }
+}

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, with_appium
+import pytest
+
+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
 
 
-@with_appium
+# netstat -ano | findstr :4723
-def main(service):
+# taskkill /PID 12345 /F
-    print(f"正在测试，服务模式: {service.role}")
+
-    # 简单操作示例
+def _archive_logs():
-    driver = None
+    """
+    在测试开始前，归档上一次运行的日志文件。
+    此时没有任何句柄占用，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:
-        # 在自动化套件启动前执行
+            # 移动并重命名
-        # proc = start_appium_service()
+            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("未找到原始日志文件，跳过备份。")
 
-        # 配置Android设备参数
-        capabilities = dict(
-            platformName='Android',
-            automationName='uiautomator2',
-            deviceName='Android',
-            appPackage='com.android.settings',
-            appActivity='.Settings'
-        )
 
-        # 转换capabilities为Appium Options
+# 日志清理
-        options = UiAutomator2Options().load_capabilities(capabilities)
+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:
+        # 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}")
 
-        # 连接Appium服务器
-        # driver = webdriver.Remote('http://localhost:4723', options=options)
-        driver = webdriver.Remote('http://127.0.0.1:4723', options=options)
-        time.sleep(1)
-        print("当前Activity:", driver.current_activity)
     finally:
-        driver.quit()
+        print("Time-of-check to Time-of-use")
-        # 在自动化套件结束后执行
-        # stop_appium_service(proc)
-    print("Hello from AppAutoTest!")
 
 
 if __name__ == "__main__":

page_objects/wan_android_home.py

@@ -0,0 +1,64 @@
+#!/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.full_screen_screenshot("登陆成功")
+            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",
+    "openpyxl>=3.1.2",
+    "pytest-rerunfailures>=16.1",
+    "python-dotenv>=1.2.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,273 +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 functools
-import signal
-import subprocess
-import time
-import os
-import sys
-import http.client
-import socket
-import json
-from collections import namedtuple
-from pathlib import Path
-from enum import Enum
-
-# --- 核心配置 ---
-# 使用 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 = "未知状态"
-
-
-class ServiceRole(Enum):
-    """服务角色枚举：明确权责"""
-    MANAGED = "由脚本启动 (受控模式)"  # 需要负责清理
-    EXTERNAL = "复用外部服务 (共享模式)"  # 不负责清理
-    NULL = "无效服务 (空模式)"  # 占位或失败状态
-
-
-def get_appium_command():
-    """精确定位 Appium 执行文件，避免 npm 包装层"""
-    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，引导用户安装
-        print(f"\n❌ 错误: 在路径 {appium_bin} 未找到 Appium 执行文件。")
-        print("💡 请确保已在项目目录下执行过: npm install appium")
-        sys.exit(1)
-    # 返回执行列表（用于 shell=False）
-    return [str(appium_bin), "-p", str(APPIUM_PORT)]
-
-
-# 在全局或 start_appium_service 中获取命令
-APP_CMD_LIST = get_appium_command()
-
-
-def _cleanup_process_tree(process: subprocess.Popen = None):
-    """核心清理逻辑：针对方案一的跨平台递归关闭"""
-    if not process or process.poll() is not None:
-        return
-    print(f"\n🛑 正在关闭 Appium 进程树 (PID: {process.pid})...")
-    try:
-        if sys.platform == "win32":
-            # Windows 下使用 taskkill 强制关闭进程树 /T
-            subprocess.run(['taskkill', '/F', '/T', '/PID', str(process.pid)],
-                           stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
-        else:
-            # Unix/Mac: 获取进程组 ID 并发送终止信号
-            pgid = os.getpgid(process.pid)
-            os.killpg(pgid, signal.SIGTERM)
-
-        process.wait(timeout=5)
-        print("✅ 所有相关进程已安全退出。")
-    except Exception as e:
-        print(f"⚠️ 关闭进程时遇到意外: {e}")
-        try:
-            process.kill()
-        except Exception as e:
-            print(f"强制: {e}")
-            # 这里通常保持安静，因为我们已经尝试过清理了
-    print("✅ 服务已完全清理。")
-
-
-class AppiumService:
-    """Appium 服务上下文容器"""
-
-    def __init__(self, role: ServiceRole, process: subprocess.Popen = None):
-        self.role = role
-        self.process = process
-
-    def stop(self):
-        """统一停止接口：根据角色决定是否关闭进程"""
-        match self.role:
-            case ServiceRole.EXTERNAL:
-                print(f"--> [角色: {self.role.value}] 脚本退出，保留原服务运行。")
-                return
-            case ServiceRole.MANAGED:
-                _cleanup_process_tree(self.process)
-            case ServiceRole.NULL:
-                print(f"--> [角色: {self.role.value}] 无需清理。")
-
-    def __repr__(self):
-        return f"<AppiumService 角色='{self.role.value}' 端口={APPIUM_PORT}>"
-
-
-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.UNKNOWN
-    except Exception as e:
-        print(e)
-        return AppiumStatus.ERROR
-    finally:
-        if conn: conn.close()
-
-
-def start_appium_service() -> AppiumService:
-    """管理 Appium 服务的生命周期"""
-    process = None  # 1. 预先初始化变量，防止作用域错误
-    managed = False
-    # 轮询等待真正就绪
-    max_retries = 40
-
-    for i in range(max_retries):
-        status = get_appium_status()
-        match status:  # Python 3.10+ 的模式匹配
-            case AppiumStatus.READY:
-                if managed:
-                    # 安全打印 PID
-                    pid_str = f"PID: {process.pid}" if process else "EXTERNAL"
-                    print(f"✨ Appium 已经完全就绪! ({pid_str})")
-                    return AppiumService(ServiceRole.MANAGED, process)
-                else:
-                    print(f"--> [复用] 有效的 Appium 服务已在运行 (Port: {APPIUM_PORT})")
-                    print("      [注意] 脚本退出时将保留该服务，不会将其关闭。")
-                    return AppiumService(ServiceRole.EXTERNAL, process)
-            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:
-
-                if not managed:
-                    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(
-                            APP_CMD_LIST,
-                            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
-                        )
-                        managed = True
-                    except Exception as e:
-                        print(f"❌ 启动过程发生异常: {e}")
-                        sys.exit(1)
-                else:
-                    if process and process.poll() is not None:
-                        print("❌ Appium 进程启动后异常退出。")
-                        sys.exit(1)
-            case AppiumStatus.INITIALIZING:
-                if managed and process and process.poll() is not None:
-                    print("❌ Appium 驱动加载期间进程崩溃。")
-                    _cleanup_process_tree(process)
-                    sys.exit(1)
-                if i % 4 == 0:  # 每 2 秒提醒一次，避免刷屏
-                    print("⏳ Appium 正在加载驱动/插件，请稍候...")
-            case AppiumStatus.ERROR:
-                print("❌ 探测接口发生内部错误（可能是解析失败或严重网络异常），脚本终止。")
-                if managed and process:
-                    _cleanup_process_tree(process)
-                sys.exit(1)
-            case _:
-                print("Appium 启动异常")
-                sys.exit(1)
-
-        time.sleep(0.5)
-
-    print("❌ 启动超时：Appium 在规定时间内未完成初始化。")
-    _cleanup_process_tree(process)
-    sys.exit(1)
-
-
-def stop_appium_service(server: AppiumService):
-    # """安全关闭服务"""
-    server.stop()
-
-# --- 装饰器实现 ---
-def with_appium(func):
-    @functools.wraps(func)
-    def wrapper(*args, **kwargs):
-        service = start_appium_service()
-        try:
-            return func(service, *args, **kwargs)
-        finally:
-            service.stop()
-    return wrapper
-
-
-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