cmautobuy/client/tools/test_pdd_home_deeplink.py

"""验证通用 PDD 商品链接最终进入哪个 Android 页面。

本工具只打开固定网址并读取当前页面，不点击购买、下单或支付控件。存在已经进入
不可逆阶段、尚未完成核单的采购任务时，会在连接手机之前拒绝运行。
"""

from __future__ import annotations

import argparse
import sqlite3
import subprocess
import sys
import time
import xml.etree.ElementTree as ET
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Optional, Sequence


CLIENT_ROOT = Path(__file__).resolve().parents[1]
if str(CLIENT_ROOT) not in sys.path:
    sys.path.insert(0, str(CLIENT_ROOT))

from src.pdd_page_classifier import (  # noqa: E402
    PAGE_EXTERNAL,
    PAGE_HOME,
    classify_pdd_page,
)


TEST_URL = "https://mobile.yangkeduo.com/goods.html"
DEFAULT_DATABASE = CLIENT_ROOT / "data" / "client.db"
PDD_PACKAGE_NAME = "com.xunmeng.pinduoduo"
_BROWSER_PACKAGE_MARKERS = (
    "browser", "chrome", "edge", "ucmobile", "qqbrowser",
)
PAGE_NAMES = {
    "captcha": "安全验证页",
    "goods": "商品详情页",
    "loading": "加载页",
    "login_required": "登录页",
    "network_error": "网络错误页",
    "order_confirmation": "订单确认页",
    "payment": "支付页",
    "risk_control": "风控页",
    "unknown": "无法识别的 PDD 页面",
}


@dataclass(frozen=True)
class LinkDiagnostic:
    """浏览器打开链接后的只读诊断结果。"""

    category: str
    detail: str
    current_package: str
    page_kind: str


def find_blocking_purchase_tasks(database_path: Path) -> tuple[str, ...]:
    """只读查询尚未完成核单的不可逆采购任务。"""

    path = Path(database_path).resolve()
    if not path.is_file():
        raise FileNotFoundError(f"找不到 Client 数据库：{path}")
    connection = sqlite3.connect(
        f"file:{path.as_posix()}?mode=ro", uri=True, timeout=5.0
    )
    try:
        connection.execute("PRAGMA query_only = ON")
        rows = connection.execute(
            "SELECT t.remote_task_id FROM pdd_tasks t"
            " JOIN task_runs r ON r.id = ("
            "   SELECT r2.id FROM task_runs r2"
            "   WHERE r2.task_id = t.id ORDER BY r2.id DESC LIMIT 1"
            " )"
            " WHERE t.task_type = 'purchase'"
            " AND r.irreversible_action_at IS NOT NULL"
            " AND COALESCE(t.current_step, '') <> 'reconcile_completed'"
            " ORDER BY t.id"
        ).fetchall()
    finally:
        connection.close()
    return tuple(str(row[0]) for row in rows)


def classify_result(xml_data: str, current_package: str) -> LinkDiagnostic:
    """结合当前包名和控件树，生成容易阅读的中文结论。"""

    observation = classify_pdd_page(ET.fromstring(xml_data), current_package)
    if observation.kind == PAGE_HOME:
        return LinkDiagnostic(
            "PDD 首页", "已识别首页底部导航及选中的首页入口", current_package,
            observation.kind,
        )
    if observation.kind == PAGE_EXTERNAL:
        if any(
            marker in current_package.casefold()
            for marker in _BROWSER_PACKAGE_MARKERS
        ):
            return LinkDiagnostic(
                "浏览器", "链接仍停留在 Android 浏览器", current_package,
                observation.kind,
            )
        return LinkDiagnostic(
            "无法判断",
            "链接停留在非 PDD 应用或系统应用选择页面",
            current_package,
            observation.kind,
        )
    if current_package == PDD_PACKAGE_NAME or observation.pdd_hierarchy:
        return LinkDiagnostic(
            "PDD 其他页面",
            PAGE_NAMES.get(observation.kind, observation.kind),
            current_package,
            observation.kind,
        )
    return LinkDiagnostic(
        "无法判断", "当前包名和控件树都不足以确定页面", current_package,
        observation.kind,
    )


def run_diagnostic(
    device_address: str,
    database_path: Path,
    wait_seconds: float,
    *,
    adb_runner: Callable[..., Any] = subprocess.run,
    connector: Optional[Callable[[str], Any]] = None,
    sleeper: Callable[[float], None] = time.sleep,
) -> LinkDiagnostic:
    """先执行本地安全检查，再打开固定网址并读取一次页面。"""

    blocking_tasks = find_blocking_purchase_tasks(database_path)
    if blocking_tasks:
        task_ids = "、".join(blocking_tasks)
        raise RuntimeError(
            f"存在尚未完成核单的不可逆采购任务：{task_ids}；"
            "根据安全规则，本次没有连接或操作手机"
        )

    address = str(device_address or "").strip()
    if not address:
        raise ValueError("必须提供 Android 设备地址")
    adb_runner(
        ["adb", "-s", address, "get-state"],
        check=True,
        capture_output=True,
        text=True,
    )
    adb_runner(
        [
            "adb", "-s", address, "shell", "am", "start",
            "-a", "android.intent.action.VIEW", "-d", TEST_URL,
        ],
        check=True,
        capture_output=True,
        text=True,
    )
    sleeper(wait_seconds)

    if connector is None:
        import uiautomator2 as u2

        connector = u2.connect
    device = connector(address)
    current = device.app_current()
    current_package = str(current.get("package") or "")
    xml_data = str(device.dump_hierarchy(compressed=False))
    return classify_result(xml_data, current_package)


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="测试浏览器打开 PDD 通用商品链接后进入的页面"
    )
    parser.add_argument("--device", required=True, help="ADB 设备地址")
    parser.add_argument(
        "--wait-seconds", type=float, default=5.0,
        help="打开链接后等待页面跳转的秒数，默认 5 秒",
    )
    parser.add_argument(
        "--database", type=Path, default=DEFAULT_DATABASE,
        help="Client 数据库路径",
    )
    return parser


def main(argv: Optional[Sequence[str]] = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    if not 0 <= args.wait_seconds <= 30:
        parser.error("--wait-seconds 必须在 0 到 30 之间")
    try:
        result = run_diagnostic(
            args.device, args.database, args.wait_seconds
        )
    except Exception as exc:
        print(f"测试未执行：{exc}", file=sys.stderr)
        return 2

    print(f"测试网址：{TEST_URL}")
    print(f"当前包名：{result.current_package or '未知'}")
    print(f"页面分类：{result.category}")
    print(f"判断说明：{result.detail}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())