diff --git a/client/test/test_pdd_home_deeplink.py b/client/test/test_pdd_home_deeplink.py
new file mode 100644
index 0000000..c92ffcf
--- /dev/null
+++ b/client/test/test_pdd_home_deeplink.py
@@ -0,0 +1,116 @@
+"""PDD 通用商品链接诊断脚本测试;不连接真实手机。"""
+
+from pathlib import Path
+import sqlite3
+import tempfile
+import unittest
+
+from tools.test_pdd_home_deeplink import (
+ TEST_URL,
+ classify_result,
+ find_blocking_purchase_tasks,
+ run_diagnostic,
+)
+
+
+class FakeDevice:
+ def app_current(self):
+ return {"package": "com.android.chrome"}
+
+ def dump_hierarchy(self, compressed=False):
+ del compressed
+ return ''
+
+
+def create_database(path: Path, *, step: str = "", irreversible: bool = False) -> None:
+ connection = sqlite3.connect(path)
+ try:
+ connection.executescript(
+ "CREATE TABLE pdd_tasks ("
+ " id INTEGER PRIMARY KEY, remote_task_id TEXT,"
+ " task_type TEXT, current_step TEXT);"
+ "CREATE TABLE task_runs ("
+ " id INTEGER PRIMARY KEY, task_id INTEGER,"
+ " irreversible_action_at TEXT);"
+ )
+ connection.execute(
+ "INSERT INTO pdd_tasks VALUES (1, 'PUR-TEST', 'purchase', ?)",
+ (step,),
+ )
+ connection.execute(
+ "INSERT INTO task_runs VALUES (1, 1, ?)",
+ ("2026-08-10T10:00:00Z" if irreversible else None,),
+ )
+ connection.commit()
+ finally:
+ connection.close()
+
+
+class PddHomeDeeplinkToolTest(unittest.TestCase):
+ def test_unreconciled_irreversible_task_is_blocked_before_adb(self):
+ with tempfile.TemporaryDirectory() as directory:
+ database = Path(directory) / "client.db"
+ create_database(
+ database, step="reconcile_purchase", irreversible=True
+ )
+ adb_calls = []
+
+ with self.assertRaises(RuntimeError) as raised:
+ run_diagnostic(
+ "192.168.0.173:5555",
+ database,
+ 0,
+ adb_runner=lambda *args, **kwargs: adb_calls.append(
+ (args, kwargs)
+ ),
+ )
+
+ self.assertIn("PUR-TEST", str(raised.exception))
+ self.assertEqual(adb_calls, [])
+
+ def test_reconciled_task_no_longer_blocks_diagnostic(self):
+ with tempfile.TemporaryDirectory() as directory:
+ database = Path(directory) / "client.db"
+ create_database(
+ database, step="reconcile_completed", irreversible=True
+ )
+
+ self.assertEqual(find_blocking_purchase_tasks(database), ())
+
+ def test_safe_run_opens_only_fixed_url_and_reports_browser(self):
+ with tempfile.TemporaryDirectory() as directory:
+ database = Path(directory) / "client.db"
+ create_database(database)
+ adb_calls = []
+
+ result = run_diagnostic(
+ "USB-001",
+ database,
+ 0,
+ adb_runner=lambda command, **_kwargs: adb_calls.append(command),
+ connector=lambda _address: FakeDevice(),
+ sleeper=lambda _seconds: None,
+ )
+
+ self.assertEqual(result.category, "浏览器")
+ self.assertEqual(len(adb_calls), 2)
+ self.assertEqual(adb_calls[1][-1], TEST_URL)
+
+ def test_home_tree_is_reported_as_pdd_home(self):
+ xml_data = """
+
+
+
+
+
+
+ """
+
+ result = classify_result(xml_data, "com.xunmeng.pinduoduo")
+
+ self.assertEqual(result.category, "PDD 首页")
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/client/tools/test_pdd_home_deeplink.py b/client/tools/test_pdd_home_deeplink.py
new file mode 100644
index 0000000..8832a9a
--- /dev/null
+++ b/client/tools/test_pdd_home_deeplink.py
@@ -0,0 +1,211 @@
+"""验证通用 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())