refactor(tests): restructure into layered POM architecture with Allure reporting

- Split framework into core/ (config, logger, db, driver factories, screenshot,
  allure integration) and pages/ (PO layer: base/admin/driver); flatten test
  files per platform (web/admin, app/driver_android, cross_end)
- Slim root conftest.py 656 -> ~100 lines: drop inline CN HTML report, add
  Allure severity mapping (p0/p1/p2 -> blocker/normal/minor) and
  environment.properties; support both tests.*-prefixed and bare imports
- Preserve TC-D2S-001/002 cross-end E2E cases with business logic unchanged,
  add allure epic/feature/story/step annotations
- Deduplicate: single makereport hook, shared failure-screenshot helpers,
  single SCREENSHOTS_DIR constant; deterministic auth-state role loading
- Move DB credentials from db_helper.py into test_config.yaml (database section)
- Move pytest.ini to repo root; switch reporting to allure-pytest
  (drop pytest-html/junitxml); update requirements.txt accordingly
- Delete deprecated scaffolding: test_freight_full_flow + 6 stale page
  objects, iOS/shipper empty shells, dead utils
- Fix flaky vxe-table action-row wait: wait for a visible td instead of
  hidden placeholder rows (fixed shipper-payment timeouts)
- Verified end-to-end against live env: both E2E cases pass (2 passed 3m12s);
  run artifacts (logs/screenshots/allure) included per repo convention

tests/core/__init__.py

@@ -0,0 +1 @@
+"""框架核心层 — 与业务无关的通用能力。"""

tests/core/allure_support.py

@@ -0,0 +1,41 @@
+"""Allure 集成辅助 — 截图附件 + 报告环境信息。
+
+allure-pytest 未安装时所有函数静默降级（截图仍保存到本地文件）。
+"""
+
+from pathlib import Path
+
+try:
+    import allure
+except ImportError:  # pragma: no cover
+    allure = None
+
+
+def attach_png(path: str, name: str = "screenshot") -> None:
+    """把 PNG 文件附加到 Allure 报告当前用例/步骤上。"""
+    if allure is None:
+        return
+    try:
+        allure.attach.file(path, name=name, attachment_type=allure.attachment_type.PNG)
+    except Exception:
+        pass  # 附件失败不影响用例主流程
+
+
+def write_environment_properties(results_dir: str, config: dict) -> None:
+    """向 allure-results 目录写入 environment.properties（报告环境页展示）。"""
+    if not results_dir:
+        return
+    admin = config.get("admin", {})
+    lines = [
+        f"base_url={admin.get('base_url', '')}",
+        f"browsers={','.join(admin.get('browsers', []))}",
+        f"skip_login={admin.get('skip_login', '')}",
+        "platform=Windows",
+        "project=运八网络货运平台",
+    ]
+    # Allure 按 ISO-8859-1 解析 properties，非 ASCII 内容转 \uXXXX 防乱码
+    content = "\n".join(lines) + "\n"
+    content = content.encode("ascii", "backslashreplace").decode("ascii")
+    out_dir = Path(results_dir)
+    out_dir.mkdir(parents=True, exist_ok=True)
+    (out_dir / "environment.properties").write_text(content, encoding="ascii")

tests/core/appium_support.py

@@ -0,0 +1,226 @@
+"""Appium 支持 — driver 工厂 + 登录态复用 + 失败截图。
+
+各 App 端 conftest 调用本模块即可，无需重复实现。
+"""
+
+from datetime import datetime
+
+import pytest
+
+from tests.core.allure_support import attach_png
+from tests.core.paths import SCREENSHOTS_DIR
+
+
+def create_driver(section_config: dict, platform: str):
+    """创建 Appium driver。
+
+    Args:
+        section_config: 配置中对应端的 section（如 config["driver_android"]）
+        platform: "android" | "ios"
+    """
+    try:
+        from appium import webdriver
+        from appium.options.android import UiAutomator2Options
+        from appium.options.ios import XCUITestOptions
+    except ImportError:
+        pytest.skip("Appium-Python-Client 未安装，跳过 App 测试。")
+
+    appium_host = section_config.get("appium_host", "http://localhost:4723")
+
+    if platform == "android":
+        options = UiAutomator2Options()
+        options.device_name = section_config["device_name"]
+        options.app_package = section_config["app_package"]
+        options.app_activity = section_config["app_activity"]
+        options.no_reset = section_config.get("no_reset", True)
+        options.new_command_timeout = section_config.get("new_command_timeout", 120)
+        options.automation_name = "UiAutomator2"
+    else:
+        options = XCUITestOptions()
+        options.device_name = section_config["device_name"]
+        options.bundle_id = section_config["bundle_id"]
+        options.no_reset = section_config.get("no_reset", True)
+        options.new_command_timeout = section_config.get("new_command_timeout", 120)
+        options.automation_name = "XCUITest"
+
+    return webdriver.Remote(appium_host, options=options)
+
+
+def make_ensure_logged_in(section_config: dict, config: dict, account_role: str):
+    """创建一个 ensure_logged_in 闭包。
+
+    策略：
+        1. 回到首页（优先 start_activity 跳转，fallback back 键循环）
+        2. 点击"我的"tab，检查"登录/注册"文本是否存在
+        3. 不存在 → 已登录；存在 → 未登录
+
+    Args:
+        section_config: App 端配置
+        config: 全局配置
+        account_role: 默认账号角色名（如 "driver"、"shipper"）
+    """
+    from selenium.webdriver.support.ui import WebDriverWait
+    from selenium.webdriver.support import expected_conditions as EC
+    from appium.webdriver.common.appiumby import AppiumBy
+
+    def _ensure_logged_in(driver, timeout_s: float = 10.0) -> None:
+        skip_login = section_config.get("skip_login", True)
+        wait = WebDriverWait(driver, timeout=timeout_s)
+
+        app_package = section_config.get("app_package", "")
+        app_activity = section_config.get("app_activity", "")
+
+        # ── 底部导航栏三个 tab 定位（使用 text 定位，更可靠）──
+        home_tab = (AppiumBy.ANDROID_UIAUTOMATOR, 'new UiSelector().text("首页")')
+        freight_tab = (AppiumBy.ANDROID_UIAUTOMATOR, 'new UiSelector().text("货源")')
+        mine_tab = (AppiumBy.ANDROID_UIAUTOMATOR, 'new UiSelector().text("我的")')
+
+        # ── 步骤1: 回到首页 ──
+        # 策略：优先用 start_activity 直接跳转到主 Activity（最快最可靠），
+        # 失败时 fallback 到 back 键循环
+        _navigate_to_home(driver, app_package, app_activity, home_tab, freight_tab, mine_tab)
+
+        # ── 步骤2: 点击"我的"tab ──
+        wait.until(EC.visibility_of_element_located(mine_tab))
+        driver.find_element(*mine_tab).click()
+        driver.implicitly_wait(1)
+
+        # ── 步骤3: 判断登录态 ──
+        login_check = (AppiumBy.ANDROID_UIAUTOMATOR, 'new UiSelector().text("登录/注册")')
+        try:
+            el = driver.find_element(*login_check)
+            if el.is_displayed():
+                if skip_login:
+                    raise RuntimeError(
+                        "APP 未登录！请在设备上手动完成一次登录（含验证码），"
+                        "之后重新运行用例即可。\n如需自动化登录，请设置 skip_login: false。"
+                    )
+                _perform_ui_login(driver, config["accounts"].get(account_role, {}), wait)
+                return
+        except RuntimeError:
+            raise
+        except Exception:
+            pass  # 没找到"登录/注册" → 已登录
+
+        # ── 步骤4: 回到首页 ──
+        try:
+            wait.until(EC.visibility_of_element_located(home_tab))
+            driver.find_element(*home_tab).click()
+        except Exception:
+            pass
+
+    return _ensure_logged_in
+
+
+def _navigate_to_home(driver, app_package: str, app_activity: str,
+                      home_tab, freight_tab, mine_tab) -> None:
+    """回到 APP 主页。
+
+    策略:
+        1) 优先用 start_activity 直接跳到主 Activity（深层页面中 back 不可靠）
+        2) fallback: back 键循环（适用于浅层页面）
+    """
+    # ── 策略1: 直接跳到主 Activity ──
+    if app_package and app_activity:
+        try:
+            driver.start_activity(app_package, app_activity)
+            driver.implicitly_wait(3)
+            if _all_tabs_visible(driver, home_tab, freight_tab, mine_tab):
+                return
+        except Exception:
+            pass
+
+    # ── 策略2: back 键循环 ──
+    for i in range(30):
+        if _all_tabs_visible(driver, home_tab, freight_tab, mine_tab):
+            return
+        try:
+            driver.back()
+            driver.implicitly_wait(1.5)
+        except Exception:
+            break
+
+    # 最终检查：如果仍然不在一级页面，抛异常而不是让后续步骤无意义超时
+    if not _all_tabs_visible(driver, home_tab, freight_tab, mine_tab):
+        try:
+            current = driver.current_activity
+        except Exception:
+            current = "unknown"
+        raise RuntimeError(
+            f"无法回到 APP 首页！当前 Activity: {current}。"
+            "请检查模拟器中 APP 是否正常运行，或手动将 APP 退到首页后重试。"
+        )
+
+
+def _all_tabs_visible(driver, home_tab, freight_tab, mine_tab) -> bool:
+    """检查底部三个导航 tab 是否全部可见。"""
+    return (
+        _is_displayed(driver, home_tab)
+        and _is_displayed(driver, freight_tab)
+        and _is_displayed(driver, mine_tab)
+    )
+
+
+def _is_displayed(driver, locator) -> bool:
+    """检查元素是否可见（不抛异常）。"""
+    try:
+        el = driver.find_element(*locator)
+        return el.is_displayed()
+    except Exception:
+        return False
+
+
+def _perform_ui_login(driver, account: dict, wait) -> None:
+    """执行 UI 自动登录（skip_login: false 时使用）。当前仅适配司机 APP。"""
+    from appium.webdriver.common.appiumby import AppiumBy
+
+    phone = account.get("phone", "")
+    password = account.get("password", "")
+
+    # 点击登录入口
+    login_entry = (AppiumBy.ANDROID_UIAUTOMATOR, 'new UiSelector().text("登录/注册")')
+    wait.until(lambda d: d.find_element(*login_entry))
+    driver.find_element(*login_entry).click()
+    driver.implicitly_wait(1)
+
+    # 填写手机号
+    phone_input = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/ed_phone")
+    wait.until(lambda d: d.find_element(*phone_input))
+    driver.find_element(*phone_input).send_keys(phone)
+
+    # 填写密码
+    password_input = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/ed_pwd")
+    driver.find_element(*password_input).send_keys(password)
+
+    # 点击登录
+    login_btn = (AppiumBy.ID, "com.arpa.ynchenggangdriver:id/tv_login")
+    driver.find_element(*login_btn).click()
+    driver.implicitly_wait(2)
+
+    # 回到首页
+    home_tab = (
+        AppiumBy.ANDROID_UIAUTOMATOR,
+        'new UiSelector().resourceId("com.arpa.ynchenggangdriver:id/fixed_bottom_navigation_icon").instance(0)',
+    )
+    try:
+        wait.until(lambda d: d.find_element(*home_tab))
+        driver.find_element(*home_tab).click()
+    except Exception:
+        pass
+
+
+def save_failure_screenshot(driver, section_config: dict, platform: str, request) -> str | None:
+    """用例失败时截图（存盘 + attach Allure），返回路径；未失败返回 None。"""
+    if not (
+        section_config.get("screenshot_on_failure", True)
+        and hasattr(request.node, "rep_call")
+        and request.node.rep_call.failed
+    ):
+        return None
+    scenario_dir = SCREENSHOTS_DIR / request.node.name
+    scenario_dir.mkdir(parents=True, exist_ok=True)
+    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
+    path = str(scenario_dir / f"FAIL_{platform}_{ts}.png")
+    driver.save_screenshot(path)
+    attach_png(path, name=f"失败截图-{platform}")
+    return path

tests/core/config_loader.py

@@ -0,0 +1,38 @@
+"""配置加载器。
+
+从 tests/config/test_config.yaml 读取配置。
+
+Usage:
+    from tests.core.config_loader import load_config
+
+    config = load_config()
+    admin_url = config["admin"]["base_url"]
+    admin_phone = config["accounts"]["super_admin"]["phone"]
+
+各端通过 fixture 直接引用对应 section:
+    def test_xxx(admin_config):
+        page.goto(admin_config["base_url"])
+"""
+
+from pathlib import Path
+from typing import Any, Dict
+
+import yaml
+
+_CONFIG_PATH = Path(__file__).parent.parent / "config" / "test_config.yaml"
+_cache: Dict[str, Any] | None = None
+
+
+def load_config(force_reload: bool = False) -> Dict[str, Any]:
+    """加载测试配置（带缓存，force_reload=True 可强制重载）。"""
+    global _cache
+    if _cache is not None and not force_reload:
+        return _cache
+
+    if not _CONFIG_PATH.exists():
+        raise FileNotFoundError(f"配置文件不存在: {_CONFIG_PATH}")
+
+    with open(_CONFIG_PATH, encoding="utf-8") as f:
+        _cache = yaml.safe_load(f)
+
+    return _cache

tests/core/data_helper.py

@@ -0,0 +1,39 @@
+"""测试数据生成辅助工具。
+
+Usage:
+    from tests.core.data_helper import generate_order_no, generate_phone
+
+    order_no = generate_order_no()
+    phone = generate_phone()
+"""
+
+import random
+import string
+from datetime import datetime
+
+
+def generate_order_no(prefix: str = "TEST") -> str:
+    """生成唯一订单号。"""
+    ts = datetime.now().strftime("%Y%m%d%H%M%S")
+    rand = "".join(random.choices(string.digits, k=4))
+    return f"{prefix}{ts}{rand}"
+
+
+def generate_phone() -> str:
+    """生成合法格式的测试手机号（138 开头）。"""
+    suffix = "".join(random.choices(string.digits, k=8))
+    return f"138{suffix}"
+
+
+def generate_shipment_no(prefix: str = "YB") -> str:
+    """生成运单号（格式: YB + 日期 + 随机数）。"""
+    ts = datetime.now().strftime("%Y%m%d")
+    rand = "".join(random.choices(string.digits, k=6))
+    return f"{prefix}{ts}{rand}"
+
+
+def generate_unique_id(prefix: str = "id") -> str:
+    """生成通用唯一 ID。"""
+    ts = datetime.now().strftime("%Y%m%d%H%M%S%f")
+    rand = "".join(random.choices(string.ascii_lowercase + string.digits, k=4))
+    return f"{prefix}_{ts}_{rand}"

tests/core/db_helper.py

@@ -0,0 +1,75 @@
+"""MySQL 数据库操作工具 — 测试中的数据库校验/数据修改。
+
+连接配置在 tests/config/test_config.yaml 的 database 段维护。
+
+Usage:
+    from tests.core.db_helper import update_risk_control_status
+
+    affected = update_risk_control_status("RHJC040026071600021457570057", status=1)
+"""
+
+import pymysql
+
+from tests.core.config_loader import load_config
+
+
+def _resolve_config(config: dict | None) -> dict:
+    """显式传入优先，否则读 yaml 的 database 段。"""
+    if config is not None:
+        return config
+    cfg = dict(load_config()["database"])
+    cfg.setdefault("charset", "utf8mb4")
+    return cfg
+
+
+def get_connection(config: dict | None = None):
+    """获取数据库连接。"""
+    return pymysql.connect(**_resolve_config(config))
+
+
+def execute_sql(sql: str, params: tuple | None = None, config: dict | None = None) -> int:
+    """执行 SQL（INSERT/UPDATE/DELETE），返回影响行数。"""
+    conn = get_connection(config)
+    try:
+        with conn.cursor() as cursor:
+            affected = cursor.execute(sql, params)
+            conn.commit()
+            return affected
+    finally:
+        conn.close()
+
+
+def query_sql(sql: str, params: tuple | None = None, config: dict | None = None) -> list[dict]:
+    """执行查询 SQL，返回结果列表。"""
+    conn = get_connection(config)
+    try:
+        with conn.cursor(pymysql.cursors.DictCursor) as cursor:
+            cursor.execute(sql, params)
+            return cursor.fetchall()
+    finally:
+        conn.close()
+
+
+def update_risk_control_status(order_code: str, status: int = 1, config: dict | None = None) -> int:
+    """更新运单风控状态。
+
+    执行 SQL:
+        UPDATE tms_order_detail_ext SET risk_control_status = <status>
+        WHERE order_detail_code IN (
+            SELECT `code` FROM `tms_order_detail` WHERE order_code = '<order_code>'
+        )
+
+    Args:
+        order_code: 运输单号（对应 tms_order_detail.order_code）
+        status: 风控状态值，默认 1
+        config: 数据库连接配置（不传则读 yaml database 段）
+    Returns:
+        影响行数
+    """
+    sql = (
+        "UPDATE tms_order_detail_ext SET risk_control_status = %s "
+        "WHERE order_detail_code IN ("
+        "  SELECT `code` FROM `tms_order_detail` WHERE order_code = %s"
+        ")"
+    )
+    return execute_sql(sql, (status, order_code), config)

tests/core/logger.py

@@ -0,0 +1,46 @@
+"""统一日志工具。
+
+Usage:
+    from tests.core.logger import get_logger
+
+    logger = get_logger(__name__)
+    logger.info("开始执行用例")
+    logger.error(f"用例失败: {e}")
+"""
+
+import logging
+import sys
+from datetime import datetime
+
+from tests.core.paths import LOGS_DIR
+
+LOGS_DIR.mkdir(parents=True, exist_ok=True)
+
+# 日志格式
+_FORMAT = "%(asctime)s [%(levelname)s] %(name)s — %(message)s"
+_DATE_FMT = "%Y-%m-%d %H:%M:%S"
+
+# 根 logger 配置（模块级，只初始化一次）
+_ROOT_LOGGER = logging.getLogger("tests")
+_ROOT_LOGGER.setLevel(logging.DEBUG)
+
+# 控制台 Handler
+_console_handler = logging.StreamHandler(sys.stdout)
+_console_handler.setLevel(logging.INFO)
+_console_handler.setFormatter(logging.Formatter(_FORMAT, datefmt=_DATE_FMT))
+
+# 文件 Handler（每次运行一个日志文件）
+_log_file = LOGS_DIR / f"test_run_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log"
+_file_handler = logging.FileHandler(_log_file, encoding="utf-8")
+_file_handler.setLevel(logging.DEBUG)
+_file_handler.setFormatter(logging.Formatter(_FORMAT, datefmt=_DATE_FMT))
+
+_ROOT_LOGGER.addHandler(_console_handler)
+_ROOT_LOGGER.addHandler(_file_handler)
+
+
+def get_logger(name: str) -> logging.Logger:
+    """获取以 `tests.` 为前缀的子 logger。"""
+    if not name.startswith("tests."):
+        name = f"tests.{name}"
+    return logging.getLogger(name)

tests/core/paths.py

@@ -0,0 +1,15 @@
+"""统一路径常量。
+
+约定: pytest 从项目根目录运行，output/ 产物目录相对于项目根。
+"""
+
+from pathlib import Path
+
+# 项目根目录 (tests/core/paths.py 上三级)
+PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
+
+# 产物目录（相对 cwd，与从项目根运行 pytest 的约定一致）
+OUTPUT_DIR = Path("output")
+LOGS_DIR = OUTPUT_DIR / "logs"
+SCREENSHOTS_DIR = OUTPUT_DIR / "screenshots"
+ALLURE_RESULTS_DIR = OUTPUT_DIR / "allure-results"

tests/core/playwright_support.py

@@ -0,0 +1,157 @@
+"""Playwright 支持 — 浏览器 Page 工厂 + 快速关闭 + 登录态加载 + 失败截图。
+
+快速关闭背景: 本机 Windows 环境下，Playwright 关闭浏览器或退出驱动时会先做
+最长约 30s 的优雅关闭等待（等 Chrome 进程自然退出），导致脚本/用例结束慢。
+fast_shutdown 强杀浏览器进程树 + node 驱动，让 p.stop() 立即返回。
+"""
+
+import os
+import subprocess
+from contextlib import contextmanager
+from datetime import datetime
+from pathlib import Path
+
+import pytest
+
+from tests.core.allure_support import attach_png
+from tests.core.paths import SCREENSHOTS_DIR
+
+# 本机 %LOCALAPPDATA%\ms-playwright 下 Chrome for Testing 存在 SxS 激活失败问题，
+# 浏览器迁移到了自定义目录；仅当该目录存在时兜底设置，不影响其他机器。
+_CUSTOM_BROWSERS_PATH = "C:/Users/23184/pw-playwright"
+if os.path.isdir(_CUSTOM_BROWSERS_PATH):
+    os.environ.setdefault("PLAYWRIGHT_BROWSERS_PATH", _CUSTOM_BROWSERS_PATH)
+
+
+# ═══════════════════════════════════════════════════════════════════
+# Page 工厂
+# ═══════════════════════════════════════════════════════════════════
+
+@contextmanager
+def admin_browser_page(admin_config: dict, browser_name: str = "chromium",
+                       role: str | None = None):
+    """创建管理端 Playwright Page（skip_login 时加载指定角色的登录态）。
+
+    Yields:
+        (page, loaded_role): loaded_role 为实际加载登录态的角色，未加载为 None
+
+    退出时自动 fast_shutdown，跳过本机 ~30s 优雅关闭等待。
+    """
+    try:
+        from playwright.sync_api import sync_playwright
+    except ImportError:
+        pytest.skip("Playwright 未安装。")
+
+    viewport = admin_config.get("viewport", {"width": 1920, "height": 1080})
+    timeout = admin_config.get("timeout_seconds", 120) * 1000
+    storage_state_dir = Path(admin_config.get("storage_state_dir", "tests/web/auth"))
+    skip_login = admin_config.get("skip_login", True)
+
+    with sync_playwright() as p:
+        browser = getattr(p, browser_name).launch(headless=True)
+
+        context_kwargs = {
+            "viewport": {"width": viewport["width"], "height": viewport["height"]},
+            "locale": "zh-CN",
+        }
+
+        loaded_role = None
+        if skip_login and role:
+            auth_file = storage_state_dir / f"{role}_{browser_name}.json"
+            if auth_file.exists():
+                context_kwargs["storage_state"] = str(auth_file)
+                loaded_role = role
+
+        context = browser.new_context(**context_kwargs)
+        page = context.new_page()
+        page.set_default_timeout(timeout)
+
+        try:
+            yield page, loaded_role
+        finally:
+            fast_shutdown(p)
+
+
+def pick_available_role(storage_state_dir: Path, browser_name: str) -> str | None:
+    """选择可用的登录态角色：优先 super_admin，否则按文件名排序取第一个。"""
+    if not storage_state_dir.exists():
+        return None
+    files = sorted(storage_state_dir.glob(f"*_{browser_name}.json"))
+    if not files:
+        return None
+    for f in files:
+        if f.stem == f"super_admin_{browser_name}":
+            return "super_admin"
+    return files[0].stem[: -len(f"_{browser_name}")]
+
+
+def save_web_failure_screenshot(page, request, tag: str) -> str | None:
+    """用例失败时整页截图（存盘 + attach Allure），返回路径；未失败返回 None。
+
+    Args:
+        page: Playwright Page
+        request: pytest request（依赖根 conftest 的 makereport hook 提供 rep_call）
+        tag: 文件名标识，如 "chromium" / "admin_chromium"
+    """
+    if not (hasattr(request.node, "rep_call") and request.node.rep_call.failed):
+        return None
+    scenario_dir = SCREENSHOTS_DIR / request.node.name
+    scenario_dir.mkdir(parents=True, exist_ok=True)
+    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
+    path = str(scenario_dir / f"FAIL_{tag}_{ts}.png")
+    try:
+        page.screenshot(path=path, full_page=True)
+    except Exception:
+        return None
+    attach_png(path, name=f"失败截图-{tag}")
+    return path
+
+
+# ═══════════════════════════════════════════════════════════════════
+# 快速关闭 (Windows)
+# ═══════════════════════════════════════════════════════════════════
+
+def _driver_proc(p):
+    """取 sync_playwright 对象内部的 node 驱动子进程对象。"""
+    try:
+        return p._impl_obj._connection._transport._proc
+    except Exception:
+        return None
+
+
+def kill_browser_tree(driver_pid: int) -> list:
+    """杀掉 node 驱动直接拉起的浏览器主进程树（chrome / chrome-headless-shell）。"""
+    query = (
+        "Get-CimInstance Win32_Process | Where-Object { "
+        f"$_.ParentProcessId -eq {driver_pid} -and "
+        "($_.Name -like 'chrome*' -or $_.Name -like 'headless*') } "
+        "| Select-Object -ExpandProperty ProcessId"
+    )
+    try:
+        ps = subprocess.run(
+            ["powershell", "-NoProfile", "-Command", query],
+            capture_output=True, text=True, timeout=20,
+        )
+    except Exception:
+        return []
+    pids = [x.strip() for x in ps.stdout.split() if x.strip().isdigit()]
+    for pid in pids:
+        subprocess.run(["taskkill", "/F", "/T", "/PID", pid], capture_output=True, timeout=20)
+    return pids
+
+
+def fast_shutdown(p) -> None:
+    """强杀浏览器进程树 + node 驱动，跳过 Playwright 的 30s 优雅关闭等待。
+
+    必须在浏览器已用完、且后续不再使用 p 的操作时调用。
+    """
+    proc = _driver_proc(p)
+    if proc is None:
+        return
+    try:
+        kill_browser_tree(proc.pid)
+    finally:
+        try:
+            proc.kill()
+        except Exception:
+            pass