Compare commits
10
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 QiuSW
|
1dc83086a0 | ||
|
QiuSW
|
ad55e77bda | ||
|
QiuSW
|
652eca7953 | ||
|
QiuSW
|
cb646b4974 | ||
|
QiuSW
|
9452debf66 | ||
|
QiuSW
|
4adeb1b37f | ||
|
QiuSW
|
2ea28c2626 | ||
|
QiuSW
|
7040bb61d8 | ||
|
QiuSW
|
e7a4be1b9b | ||
|
QiuSW
|
4281b06711 |
@@ -0,0 +1,75 @@
|
||||
"""打开已验证的拼多多商品直链并采集只读本地证据。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
from pathlib import Path
|
||||
import sys
|
||||
|
||||
|
||||
CLIENT_ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(CLIENT_ROOT / "src"))
|
||||
|
||||
from cmbuyer_client.device.adb import AdbClient, DeviceConnectionError, SubprocessAdbRunner
|
||||
from cmbuyer_client.device.baseline import NoReconnectUiautomatorConnector
|
||||
from cmbuyer_client.pdd.product_open import ProductOpenCapturer, ProductOpenError
|
||||
from cmbuyer_client.pdd.product_url import ProductUrlError, parse_product_url
|
||||
|
||||
|
||||
def parse_arguments(argv: list[str] | None = None) -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(description="打开 canonical 拼多多商品链接并采集只读证据。")
|
||||
parser.add_argument("--serial", required=True, help="ADB device serial；禁止自动选择。")
|
||||
parser.add_argument("--url", required=True, help="唯一允许的 goods.html?goods_id= 直链。")
|
||||
parser.add_argument("--output-dir", required=True, type=Path, help="新建的本地证据目录；不得覆盖已有目录。")
|
||||
parser.add_argument("--timeout", type=float, default=10.0, help="ADB 和只读 RPC 超时（秒）。")
|
||||
parser.add_argument("--adb", default="adb", help="adb 可执行文件路径。")
|
||||
return parser.parse_args(argv)
|
||||
|
||||
|
||||
def validate_arguments(arguments: argparse.Namespace) -> None:
|
||||
if not arguments.serial.strip():
|
||||
raise ValueError("必须显式提供非空 --serial。")
|
||||
if arguments.timeout <= 0:
|
||||
raise ValueError("--timeout 必须大于 0。")
|
||||
parse_product_url(arguments.url)
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
arguments = parse_arguments(argv)
|
||||
try:
|
||||
validate_arguments(arguments)
|
||||
link = parse_product_url(arguments.url)
|
||||
except (ValueError, ProductUrlError) as error:
|
||||
print(f"失败：{error}", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
try:
|
||||
import adbutils
|
||||
import uiautomator2 as u2
|
||||
except ImportError:
|
||||
print("失败：缺少 uiautomator2；请在采购工具虚拟环境中运行。", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
client = AdbClient(SubprocessAdbRunner(arguments.adb), timeout_seconds=arguments.timeout)
|
||||
connector = NoReconnectUiautomatorConnector(
|
||||
adbutils.AdbClient(socket_timeout=arguments.timeout).device_list,
|
||||
u2.connect,
|
||||
)
|
||||
capturer = ProductOpenCapturer(client, connector, timeout_seconds=arguments.timeout)
|
||||
try:
|
||||
result = capturer.open_and_capture(arguments.serial, link.canonical_url, arguments.output_dir)
|
||||
except (DeviceConnectionError, ProductOpenError) as error:
|
||||
# 不打印 ADB 输出、serial、Activity、XML 或页面正文。
|
||||
print(f"商品打开取证失败：{error}", file=sys.stderr)
|
||||
return 1
|
||||
except OSError:
|
||||
print("商品打开取证失败：无法创建或发布本地证据目录。", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
print(f"商品打开取证完成：{result.output_directory}")
|
||||
print(f"manifest：{result.manifest_path}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -48,6 +48,10 @@ class DuplicatePhysicalDeviceError(DeviceConnectionError):
|
||||
"""同一物理手机通过多个 ADB 通道同时在线。"""
|
||||
|
||||
|
||||
class IntentLaunchUnconfirmedError(DeviceConnectionError):
|
||||
"""`am start -W` 没有给出可确认的启动成功结果。"""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CommandResult:
|
||||
"""可注入命令执行器的最小、可离线构造结果。"""
|
||||
@@ -57,6 +61,14 @@ class CommandResult:
|
||||
returncode: int = 0
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class IntentLaunchSummary:
|
||||
"""不含 Activity、页面内容或 ADB 输出的受限启动摘要。"""
|
||||
|
||||
status: str
|
||||
returncode: int
|
||||
|
||||
|
||||
class CommandRunner(Protocol):
|
||||
"""运行 ADB 子命令的可替换边界。"""
|
||||
|
||||
@@ -195,6 +207,42 @@ class AdbClient:
|
||||
result = self._run_checked(("devices", "-l"))
|
||||
return parse_adb_devices(result.stdout)
|
||||
|
||||
def start_pdd_view_intent(self, serial: str, goods_id: str) -> IntentLaunchSummary:
|
||||
"""以参数数组启动唯一允许的拼多多 ACTION_VIEW Intent。
|
||||
|
||||
这里刻意不提供任意 shell 或任意 package 的执行接口。调用方必须先完成
|
||||
``inspect`` 和应用版本核验；本方法在本层从纯数字 ``goods_id`` 重建 URL，调用方不能
|
||||
把另一个 URL 直接交给 ADB。本方法既不点击控件，也不解析 Activity 或页面文本。
|
||||
"""
|
||||
|
||||
selected_serial = _require_serial(serial)
|
||||
if (
|
||||
not isinstance(goods_id, str)
|
||||
or not goods_id
|
||||
or any(character < "0" or character > "9" for character in goods_id)
|
||||
):
|
||||
raise ValueError("goods_id 必须是纯数字")
|
||||
canonical_url = f"https://mobile.yangkeduo.com/goods.html?goods_id={goods_id}"
|
||||
result = self._run_checked(
|
||||
(
|
||||
"-s",
|
||||
selected_serial,
|
||||
"shell",
|
||||
"am",
|
||||
"start",
|
||||
"-W",
|
||||
"-a",
|
||||
"android.intent.action.VIEW",
|
||||
"-d",
|
||||
canonical_url,
|
||||
"-p",
|
||||
"com.xunmeng.pinduoduo",
|
||||
)
|
||||
)
|
||||
if not any(line.strip() == "Status: ok" for line in result.stdout.splitlines()):
|
||||
raise IntentLaunchUnconfirmedError("商品链接启动结果无法确认，已停止后续取证。")
|
||||
return IntentLaunchSummary(status="ok", returncode=result.returncode)
|
||||
|
||||
def _physical_identity(self, device: AdbDevice) -> frozenset[str]:
|
||||
serialno = self._getprop(device.serial, "ro.serialno")
|
||||
boot_serialno = self._getprop(device.serial, "ro.boot.serialno")
|
||||
|
||||
@@ -0,0 +1,15 @@
|
||||
"""拼多多链接的受限打开与只读取证。
|
||||
|
||||
此包不提供页面选择器、输入、滑动、下单或支付能力。
|
||||
"""
|
||||
|
||||
from .product_open import ProductOpenCapturer, ProductOpenResult
|
||||
from .product_url import ProductUrl, ProductUrlError, parse_product_url
|
||||
|
||||
__all__ = [
|
||||
"ProductOpenCapturer",
|
||||
"ProductOpenResult",
|
||||
"ProductUrl",
|
||||
"ProductUrlError",
|
||||
"parse_product_url",
|
||||
]
|
||||
@@ -0,0 +1,262 @@
|
||||
"""安全打开 canonical 商品链接后的只读取证。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass
|
||||
from datetime import UTC, datetime
|
||||
from hashlib import sha256
|
||||
import json
|
||||
from math import isfinite
|
||||
import os
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
from time import monotonic, sleep
|
||||
from typing import Any, Protocol
|
||||
from uuid import uuid4
|
||||
|
||||
from adbutils.errors import AdbTimeout
|
||||
from uiautomator2.exceptions import HTTPTimeoutError
|
||||
|
||||
from ..device.adb import AdbClient, DeviceConnectionError, DeviceInspection, IntentLaunchSummary
|
||||
from ..device.baseline import (
|
||||
HIERARCHY_PARAMS,
|
||||
PDD_PACKAGE,
|
||||
SCREENSHOT_PARAMS,
|
||||
_save_base64_screenshot,
|
||||
_sha256_file,
|
||||
_validate_hierarchy,
|
||||
)
|
||||
from .product_url import ProductUrl, parse_product_url
|
||||
|
||||
|
||||
EXPECTED_PDD_VERSION = "8.17.0"
|
||||
|
||||
|
||||
class ProductOpenError(RuntimeError):
|
||||
"""商品打开或证据发布未完整完成。"""
|
||||
|
||||
|
||||
class ProductVersionMismatchError(ProductOpenError):
|
||||
"""运行时拼多多版本不是经取证允许的版本。"""
|
||||
|
||||
|
||||
class ProductPackageMismatchError(ProductOpenError):
|
||||
"""Intent 后在有限时间内未观察到拼多多前台包。"""
|
||||
|
||||
|
||||
class ProductOpenTimeoutError(ProductOpenError):
|
||||
"""商品打开后的只读取证超时。"""
|
||||
|
||||
|
||||
class ProductScreenshotCaptureError(ProductOpenError):
|
||||
"""Intent 后截图不能作为完整 PNG 证据保存。"""
|
||||
|
||||
|
||||
class ProductHierarchyCaptureError(ProductOpenError):
|
||||
"""Intent 后完整节点树不能作为有效 XML 证据保存。"""
|
||||
|
||||
|
||||
class ProductOpenUiDevice(Protocol):
|
||||
"""本任务所需的只读 uiautomator2 接口；故意没有任何 UI 操作方法。"""
|
||||
|
||||
def app_info(self, package_name: str) -> dict[str, Any]:
|
||||
"""读取应用元数据。"""
|
||||
|
||||
def app_current(self) -> dict[str, Any]:
|
||||
"""读取当前前台应用元数据。"""
|
||||
|
||||
def jsonrpc_call(self, method: str, params: Any = None, timeout: float = 10) -> Any:
|
||||
"""调用只读取证所需的公开 JSON-RPC 方法。"""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ProductOpenResult:
|
||||
"""已原子发布的商品打开证据位置。"""
|
||||
|
||||
output_directory: Path
|
||||
manifest_path: Path
|
||||
screenshot_path: Path
|
||||
hierarchy_path: Path
|
||||
|
||||
|
||||
class ProductOpenCapturer:
|
||||
"""以 fail-closed 顺序打开已重建链接，并在打开后只读留证。
|
||||
|
||||
本类不判断商品页、Activity、文案或控件；打开后只确认当前 package，随后采集截图与
|
||||
完整节点树。任何失败都不会发布半成品证据目录。
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
adb_client: AdbClient,
|
||||
connector: Callable[[str], ProductOpenUiDevice],
|
||||
timeout_seconds: float,
|
||||
foreground_poll_interval_seconds: float = 0.2,
|
||||
monotonic_clock: Callable[[], float] = monotonic,
|
||||
sleep_function: Callable[[float], None] = sleep,
|
||||
) -> None:
|
||||
if not _is_positive_finite(timeout_seconds):
|
||||
raise ValueError("timeout_seconds 必须是大于 0 的有限数值")
|
||||
if not _is_positive_finite(foreground_poll_interval_seconds):
|
||||
raise ValueError("foreground_poll_interval_seconds 必须是大于 0 的有限数值")
|
||||
self._adb_client = adb_client
|
||||
self._connector = connector
|
||||
self._timeout_seconds = timeout_seconds
|
||||
self._foreground_poll_interval_seconds = foreground_poll_interval_seconds
|
||||
self._monotonic_clock = monotonic_clock
|
||||
self._sleep_function = sleep_function
|
||||
|
||||
def open_and_capture(self, serial: str, product_url: str, output_directory: Path) -> ProductOpenResult:
|
||||
"""完成唯一允许的 Intent 打开及其后的只读取证。"""
|
||||
|
||||
# 公共入口只接收原始字符串并每次重新解析，不能由调用方构造不一致的值对象伪造 manifest。
|
||||
link = parse_product_url(product_url)
|
||||
target = Path(output_directory)
|
||||
_validate_new_target(target)
|
||||
|
||||
staging: Path | None = None
|
||||
try:
|
||||
# inspect 必须先于连接和 Intent，复用 T-101 的显式 serial、重复物理设备拒绝逻辑。
|
||||
inspection = self._adb_client.inspect(serial)
|
||||
device = self._connector(serial)
|
||||
pdd_version = _require_expected_version(device.app_info(PDD_PACKAGE))
|
||||
|
||||
# 版本精确匹配是 Intent 的前置条件，失败时绝不调用 start_pdd_view_intent。
|
||||
intent = self._adb_client.start_pdd_view_intent(serial, link.goods_id)
|
||||
self._wait_for_pdd_foreground(device)
|
||||
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
staging = target.parent / f".{target.name}.staging-{uuid4().hex}"
|
||||
staging.mkdir()
|
||||
screenshot_path = staging / "screenshot.png"
|
||||
try:
|
||||
_save_base64_screenshot(
|
||||
device.jsonrpc_call("takeScreenshot", SCREENSHOT_PARAMS, timeout=self._timeout_seconds),
|
||||
screenshot_path,
|
||||
)
|
||||
except (AdbTimeout, HTTPTimeoutError, TimeoutError):
|
||||
raise
|
||||
except Exception as error:
|
||||
raise ProductScreenshotCaptureError("商品打开后截图取证失败，未发布任何证据产物。") from error
|
||||
|
||||
try:
|
||||
hierarchy = device.jsonrpc_call(
|
||||
"dumpWindowHierarchy",
|
||||
HIERARCHY_PARAMS,
|
||||
timeout=self._timeout_seconds,
|
||||
)
|
||||
_validate_hierarchy(hierarchy)
|
||||
except (AdbTimeout, HTTPTimeoutError, TimeoutError):
|
||||
raise
|
||||
except Exception as error:
|
||||
raise ProductHierarchyCaptureError("商品打开后节点树取证失败，未发布任何证据产物。") from error
|
||||
hierarchy_path = staging / "hierarchy.xml"
|
||||
hierarchy_path.write_text(hierarchy, encoding="utf-8")
|
||||
|
||||
manifest_path = staging / "manifest.json"
|
||||
manifest_path.write_text(
|
||||
json.dumps(
|
||||
_manifest(inspection, serial, link, pdd_version, intent, screenshot_path, hierarchy_path),
|
||||
ensure_ascii=False,
|
||||
indent=2,
|
||||
sort_keys=True,
|
||||
)
|
||||
+ "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
os.replace(staging, target)
|
||||
except (ProductOpenError, DeviceConnectionError):
|
||||
_clean_staging(staging)
|
||||
raise
|
||||
except (AdbTimeout, HTTPTimeoutError, TimeoutError) as error:
|
||||
_clean_staging(staging)
|
||||
raise ProductOpenTimeoutError("商品打开后的只读取证超时，未发布任何证据产物。") from error
|
||||
except Exception as error:
|
||||
_clean_staging(staging)
|
||||
# 底层异常可能含 serial、路径或远端页面内容，不能直接向 CLI 或日志传播。
|
||||
raise ProductOpenError("商品打开或只读取证未完成，未发布任何证据产物。") from error
|
||||
|
||||
return ProductOpenResult(
|
||||
output_directory=target,
|
||||
manifest_path=target / "manifest.json",
|
||||
screenshot_path=target / "screenshot.png",
|
||||
hierarchy_path=target / "hierarchy.xml",
|
||||
)
|
||||
|
||||
def _wait_for_pdd_foreground(self, device: ProductOpenUiDevice) -> None:
|
||||
"""只轮询当前 package，直到 deadline；Activity 和节点树均不参与本判据。"""
|
||||
|
||||
deadline = self._monotonic_clock() + self._timeout_seconds
|
||||
while True:
|
||||
if _is_pdd_foreground(device.app_current()):
|
||||
return
|
||||
remaining = deadline - self._monotonic_clock()
|
||||
if remaining <= 0:
|
||||
raise ProductPackageMismatchError(
|
||||
"商品链接打开后未在限定时间内进入拼多多，已停止后续取证。"
|
||||
)
|
||||
# 每个失败观察后都等待正的、受 deadline 约束的时长，避免 busy-loop。
|
||||
self._sleep_function(min(self._foreground_poll_interval_seconds, remaining))
|
||||
|
||||
|
||||
def _validate_new_target(target: Path) -> None:
|
||||
if target.exists():
|
||||
raise ProductOpenError("输出目录已存在；为防止混入旧证据，拒绝覆盖。")
|
||||
if not target.name:
|
||||
raise ProductOpenError("输出目录必须是明确的新目录。")
|
||||
|
||||
|
||||
def _clean_staging(staging: Path | None) -> None:
|
||||
if staging is not None and staging.exists():
|
||||
# staging 仅在本次调用中创建，删除前不解析或扩展任何调用方提供的路径。
|
||||
shutil.rmtree(staging)
|
||||
|
||||
|
||||
def _require_expected_version(app_info: dict[str, Any]) -> str:
|
||||
if not isinstance(app_info, dict):
|
||||
raise ProductVersionMismatchError("拼多多版本与已取证版本不一致，已停止打开商品链接。")
|
||||
version = app_info.get("versionName") or app_info.get("version_name")
|
||||
if not isinstance(version, str) or version != EXPECTED_PDD_VERSION:
|
||||
raise ProductVersionMismatchError("拼多多版本与已取证版本不一致，已停止打开商品链接。")
|
||||
return version
|
||||
|
||||
|
||||
def _is_positive_finite(value: object) -> bool:
|
||||
return isinstance(value, (int, float)) and not isinstance(value, bool) and value > 0 and isfinite(value)
|
||||
|
||||
|
||||
def _is_pdd_foreground(current: object) -> bool:
|
||||
return isinstance(current, dict) and current.get("package") == PDD_PACKAGE
|
||||
|
||||
|
||||
def _manifest(
|
||||
inspection: DeviceInspection,
|
||||
serial: str,
|
||||
link: ProductUrl,
|
||||
pdd_version: str,
|
||||
intent: IntentLaunchSummary,
|
||||
screenshot_path: Path,
|
||||
hierarchy_path: Path,
|
||||
) -> dict[str, Any]:
|
||||
"""只写审计摘要；原始 serial、Activity、ADB 输出和页面正文均不进入 manifest。"""
|
||||
|
||||
return {
|
||||
"schema_version": 1,
|
||||
"captured_at": datetime.now(UTC).isoformat(),
|
||||
"product": {"goods_id": link.goods_id, "canonical_url": link.canonical_url},
|
||||
"channel": "wifi" if ":" in serial else "usb",
|
||||
"serial_sha256": sha256(serial.encode("utf-8")).hexdigest(),
|
||||
"device": {
|
||||
"model": inspection.model,
|
||||
"android_version": inspection.android_version,
|
||||
"pdd_package": PDD_PACKAGE,
|
||||
"pdd_version": pdd_version,
|
||||
},
|
||||
"intent": {"status": intent.status, "returncode": intent.returncode},
|
||||
"current_package": PDD_PACKAGE,
|
||||
"artifacts": [
|
||||
{"path": screenshot_path.name, "sha256": _sha256_file(screenshot_path)},
|
||||
{"path": hierarchy_path.name, "sha256": _sha256_file(hierarchy_path)},
|
||||
],
|
||||
}
|
||||
@@ -0,0 +1,61 @@
|
||||
"""唯一允许交给 Android Intent 的商品链接。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from urllib.parse import parse_qsl, urlsplit
|
||||
|
||||
|
||||
_SCHEME = "https"
|
||||
_HOST = "mobile.yangkeduo.com"
|
||||
_PATH = "/goods.html"
|
||||
|
||||
|
||||
class ProductUrlError(ValueError):
|
||||
"""输入不是可安全重建的 canonical 商品链接。"""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ProductUrl:
|
||||
"""经验证的商品标识及由它重建的 canonical URL。"""
|
||||
|
||||
goods_id: str
|
||||
canonical_url: str
|
||||
|
||||
|
||||
def parse_product_url(value: str) -> ProductUrl:
|
||||
"""只接受一个 ASCII 数字 ``goods_id`` 的拼多多商品直链。
|
||||
|
||||
解析结果绝不原样透传：Intent 使用的 URL 必须从 ``goods_id`` 重新构建，以排除
|
||||
短链、额外参数、userinfo、fragment 和 URL 解析器的边缘表示。
|
||||
"""
|
||||
|
||||
if not isinstance(value, str):
|
||||
raise ProductUrlError("商品链接必须是字符串。")
|
||||
try:
|
||||
parsed = urlsplit(value)
|
||||
port = parsed.port
|
||||
query_pairs = parse_qsl(parsed.query, keep_blank_values=True, strict_parsing=True)
|
||||
except ValueError as error:
|
||||
raise ProductUrlError("商品链接格式无效。") from error
|
||||
|
||||
if (
|
||||
parsed.scheme != _SCHEME
|
||||
or parsed.hostname != _HOST
|
||||
or parsed.username is not None
|
||||
or parsed.password is not None
|
||||
or port is not None
|
||||
or parsed.path != _PATH
|
||||
or parsed.fragment
|
||||
):
|
||||
raise ProductUrlError("商品链接不是允许的拼多多商品直链。")
|
||||
if len(query_pairs) != 1 or query_pairs[0][0] != "goods_id":
|
||||
raise ProductUrlError("商品链接必须且只能包含一个 goods_id 参数。")
|
||||
|
||||
goods_id = query_pairs[0][1]
|
||||
if not goods_id or any(character < "0" or character > "9" for character in goods_id):
|
||||
raise ProductUrlError("goods_id 必须是纯数字。")
|
||||
canonical_url = f"{_SCHEME}://{_HOST}{_PATH}?goods_id={goods_id}"
|
||||
if value != canonical_url:
|
||||
raise ProductUrlError("商品链接必须使用唯一 canonical 表示。")
|
||||
return ProductUrl(goods_id=goods_id, canonical_url=canonical_url)
|
||||
@@ -16,11 +16,14 @@ from cmbuyer_client.device.adb import (
|
||||
AdbClient,
|
||||
CommandResult,
|
||||
DeviceIdentityUnconfirmedError,
|
||||
DeviceCommandError,
|
||||
DeviceCommandTimeoutError,
|
||||
DeviceNotFoundError,
|
||||
DeviceOfflineError,
|
||||
DeviceStateError,
|
||||
DeviceUnauthorizedError,
|
||||
DuplicatePhysicalDeviceError,
|
||||
IntentLaunchUnconfirmedError,
|
||||
SerialRequiredError,
|
||||
)
|
||||
|
||||
@@ -176,7 +179,95 @@ class AdbClientTests(unittest.TestCase):
|
||||
def run(self, arguments: Sequence[str], timeout_seconds: float) -> CommandResult:
|
||||
raise subprocess.TimeoutExpired(arguments, timeout_seconds)
|
||||
|
||||
from cmbuyer_client.device.adb import DeviceCommandTimeoutError
|
||||
|
||||
with self.assertRaises(DeviceCommandTimeoutError):
|
||||
AdbClient(TimeoutRunner()).inspect(USB_SERIAL)
|
||||
|
||||
def test_product_intent_is_fixed_to_action_view_and_pdd_package(self) -> None:
|
||||
class IntentRunner:
|
||||
def __init__(self) -> None:
|
||||
self.calls: list[tuple[str, ...]] = []
|
||||
|
||||
def run(self, arguments: Sequence[str], timeout_seconds: float) -> CommandResult:
|
||||
self.calls.append(tuple(arguments))
|
||||
return CommandResult(stdout="Status: ok\n")
|
||||
|
||||
runner = IntentRunner()
|
||||
summary = AdbClient(runner).start_pdd_view_intent(
|
||||
USB_SERIAL,
|
||||
"123",
|
||||
)
|
||||
|
||||
self.assertEqual(summary.status, "ok")
|
||||
self.assertEqual(
|
||||
runner.calls,
|
||||
[
|
||||
(
|
||||
"-s",
|
||||
USB_SERIAL,
|
||||
"shell",
|
||||
"am",
|
||||
"start",
|
||||
"-W",
|
||||
"-a",
|
||||
"android.intent.action.VIEW",
|
||||
"-d",
|
||||
"https://mobile.yangkeduo.com/goods.html?goods_id=123",
|
||||
"-p",
|
||||
"com.xunmeng.pinduoduo",
|
||||
)
|
||||
],
|
||||
)
|
||||
|
||||
def test_product_intent_without_explicit_success_is_rejected(self) -> None:
|
||||
class UnknownIntentRunner:
|
||||
def run(self, arguments: Sequence[str], timeout_seconds: float) -> CommandResult:
|
||||
return CommandResult(stdout="Starting: Intent { ... }\n")
|
||||
|
||||
with self.assertRaises(IntentLaunchUnconfirmedError):
|
||||
AdbClient(UnknownIntentRunner()).start_pdd_view_intent(
|
||||
USB_SERIAL,
|
||||
"123",
|
||||
)
|
||||
|
||||
def test_product_intent_rejects_invalid_goods_id_before_runner(self) -> None:
|
||||
class RecordingRunner:
|
||||
def __init__(self) -> None:
|
||||
self.calls: list[tuple[str, ...]] = []
|
||||
|
||||
def run(self, arguments: Sequence[str], timeout_seconds: float) -> CommandResult:
|
||||
self.calls.append(tuple(arguments))
|
||||
return CommandResult(stdout="Status: ok\n")
|
||||
|
||||
invalid_values: tuple[object, ...] = (
|
||||
"",
|
||||
"12a",
|
||||
"１２３",
|
||||
" 123",
|
||||
"123 ",
|
||||
"https://mobile.yangkeduo.com/goods.html?goods_id=123",
|
||||
"am start -W -d anything",
|
||||
123,
|
||||
None,
|
||||
)
|
||||
for value in invalid_values:
|
||||
with self.subTest(value=repr(value)):
|
||||
runner = RecordingRunner()
|
||||
with self.assertRaises(ValueError):
|
||||
AdbClient(runner).start_pdd_view_intent(USB_SERIAL, value) # type: ignore[arg-type]
|
||||
self.assertEqual(runner.calls, [])
|
||||
|
||||
def test_product_intent_nonzero_and_timeout_remain_distinct(self) -> None:
|
||||
class FailedIntentRunner:
|
||||
def run(self, arguments: Sequence[str], timeout_seconds: float) -> CommandResult:
|
||||
return CommandResult(stdout="sensitive command output", returncode=1)
|
||||
|
||||
class TimeoutIntentRunner:
|
||||
def run(self, arguments: Sequence[str], timeout_seconds: float) -> CommandResult:
|
||||
raise subprocess.TimeoutExpired(arguments, timeout_seconds)
|
||||
|
||||
with self.assertRaises(DeviceCommandError) as command_error:
|
||||
AdbClient(FailedIntentRunner()).start_pdd_view_intent(USB_SERIAL, "123")
|
||||
self.assertNotIn("sensitive command output", str(command_error.exception))
|
||||
|
||||
with self.assertRaises(DeviceCommandTimeoutError):
|
||||
AdbClient(TimeoutIntentRunner()).start_pdd_view_intent(USB_SERIAL, "123")
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
"""拼多多受限打开模块的离线测试。"""
|
||||
@@ -0,0 +1,266 @@
|
||||
"""商品打开围栏的离线测试；所有设备和命令均为 fake。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
from io import BytesIO
|
||||
from pathlib import Path
|
||||
import sys
|
||||
from tempfile import TemporaryDirectory
|
||||
import unittest
|
||||
|
||||
from PIL import Image
|
||||
|
||||
|
||||
CLIENT_ROOT = Path(__file__).resolve().parents[2]
|
||||
sys.path.insert(0, str(CLIENT_ROOT / "src"))
|
||||
|
||||
from cmbuyer_client.device.adb import AdbDevice, DeviceInspection, IntentLaunchSummary
|
||||
from cmbuyer_client.pdd.product_open import (
|
||||
ProductOpenCapturer,
|
||||
ProductOpenTimeoutError,
|
||||
ProductOpenUiDevice,
|
||||
ProductHierarchyCaptureError,
|
||||
ProductPackageMismatchError,
|
||||
ProductScreenshotCaptureError,
|
||||
ProductVersionMismatchError,
|
||||
)
|
||||
from cmbuyer_client.pdd.product_url import ProductUrl, ProductUrlError
|
||||
|
||||
|
||||
SERIAL = "192.168.0.173:5555"
|
||||
URL = "https://mobile.yangkeduo.com/goods.html?goods_id=123"
|
||||
HIERARCHY = "<?xml version='1.0' encoding='UTF-8'?><hierarchy rotation='0'><node /></hierarchy>"
|
||||
|
||||
|
||||
def _png_base64() -> str:
|
||||
image_data = BytesIO()
|
||||
Image.new("RGB", (1, 1), color="white").save(image_data, format="PNG")
|
||||
return base64.b64encode(image_data.getvalue()).decode("ascii")
|
||||
|
||||
|
||||
class FakeAdbClient:
|
||||
def __init__(self) -> None:
|
||||
self.calls: list[tuple[str, str | None]] = []
|
||||
self.inspection = DeviceInspection(
|
||||
device=AdbDevice(serial=SERIAL, state="device", model="PKG110"),
|
||||
model="PKG110",
|
||||
android_version="16",
|
||||
)
|
||||
|
||||
def inspect(self, serial: str) -> DeviceInspection:
|
||||
self.calls.append(("inspect", serial))
|
||||
return self.inspection
|
||||
|
||||
def start_pdd_view_intent(self, serial: str, goods_id: str) -> IntentLaunchSummary:
|
||||
self.calls.append(("intent", goods_id))
|
||||
return IntentLaunchSummary(status="ok", returncode=0)
|
||||
|
||||
|
||||
class FakeUiDevice:
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
version: str = "8.17.0",
|
||||
current_package: str = "com.xunmeng.pinduoduo",
|
||||
current_packages: list[str] | None = None,
|
||||
hierarchy: str = HIERARCHY,
|
||||
timeout_on_screenshot: bool = False,
|
||||
) -> None:
|
||||
self.version = version
|
||||
self.current_package = current_package
|
||||
self.current_packages = list(current_packages) if current_packages is not None else None
|
||||
self.hierarchy = hierarchy
|
||||
self.timeout_on_screenshot = timeout_on_screenshot
|
||||
self.calls: list[str] = []
|
||||
|
||||
def app_info(self, package_name: str) -> dict[str, str]:
|
||||
self.calls.append("app_info")
|
||||
return {"versionName": self.version}
|
||||
|
||||
def app_current(self) -> dict[str, str]:
|
||||
self.calls.append("app_current")
|
||||
if self.current_packages:
|
||||
package = self.current_packages.pop(0)
|
||||
self.current_package = package
|
||||
return {"package": self.current_package, "activity": "sensitive.activity.name"}
|
||||
|
||||
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
|
||||
self.calls.append(method)
|
||||
if method == "takeScreenshot":
|
||||
if self.timeout_on_screenshot:
|
||||
raise TimeoutError("raw remote detail")
|
||||
return _png_base64()
|
||||
if method == "dumpWindowHierarchy":
|
||||
return self.hierarchy
|
||||
raise AssertionError(f"unexpected RPC {method}")
|
||||
|
||||
|
||||
class ProductOpenTests(unittest.TestCase):
|
||||
def _capturer(self, adb: FakeAdbClient, device: FakeUiDevice, **kwargs: object) -> ProductOpenCapturer:
|
||||
return ProductOpenCapturer(adb, lambda serial: device, timeout_seconds=2, **kwargs)
|
||||
|
||||
def test_success_uses_canonical_url_and_redacted_atomic_manifest(self) -> None:
|
||||
adb = FakeAdbClient()
|
||||
device = FakeUiDevice()
|
||||
with TemporaryDirectory() as temporary:
|
||||
target = Path(temporary) / "evidence"
|
||||
result = self._capturer(adb, device).open_and_capture(SERIAL, URL, target)
|
||||
manifest = result.manifest_path.read_text(encoding="utf-8")
|
||||
|
||||
self.assertTrue(result.screenshot_path.exists())
|
||||
self.assertTrue(result.hierarchy_path.exists())
|
||||
self.assertEqual(adb.calls, [("inspect", SERIAL), ("intent", "123")])
|
||||
self.assertEqual(device.calls, ["app_info", "app_current", "takeScreenshot", "dumpWindowHierarchy"])
|
||||
self.assertIn('"goods_id": "123"', manifest)
|
||||
self.assertIn('"canonical_url": "https://mobile.yangkeduo.com/goods.html?goods_id=123"', manifest)
|
||||
self.assertNotIn(SERIAL, manifest)
|
||||
self.assertNotIn("sensitive.activity.name", manifest)
|
||||
self.assertNotIn(HIERARCHY, manifest)
|
||||
|
||||
def test_version_mismatch_halts_before_intent(self) -> None:
|
||||
adb = FakeAdbClient()
|
||||
for version in ("8.17.1", " 8.17.0 "):
|
||||
with self.subTest(version=version), TemporaryDirectory() as temporary:
|
||||
target = Path(temporary) / "evidence"
|
||||
with self.assertRaises(ProductVersionMismatchError):
|
||||
self._capturer(adb, FakeUiDevice(version=version)).open_and_capture(SERIAL, URL, target)
|
||||
|
||||
self.assertEqual(adb.calls[-1:], [("inspect", SERIAL)])
|
||||
self.assertFalse(target.exists())
|
||||
|
||||
def test_public_entry_rejects_caller_constructed_url_value_object(self) -> None:
|
||||
adb = FakeAdbClient()
|
||||
with TemporaryDirectory() as temporary:
|
||||
with self.assertRaises(ProductUrlError):
|
||||
self._capturer(adb, FakeUiDevice()).open_and_capture(
|
||||
SERIAL,
|
||||
ProductUrl(goods_id="123", canonical_url="https://example.invalid/"), # type: ignore[arg-type]
|
||||
Path(temporary) / "evidence",
|
||||
)
|
||||
|
||||
self.assertEqual(adb.calls, [])
|
||||
|
||||
def test_foreground_package_mismatch_halts_before_capture(self) -> None:
|
||||
adb = FakeAdbClient()
|
||||
device = FakeUiDevice(current_package="com.example.other")
|
||||
clock = FakeClock()
|
||||
with TemporaryDirectory() as temporary:
|
||||
target = Path(temporary) / "evidence"
|
||||
with self.assertRaises(ProductPackageMismatchError):
|
||||
self._capturer(
|
||||
adb,
|
||||
device,
|
||||
foreground_poll_interval_seconds=0.5,
|
||||
monotonic_clock=clock.monotonic,
|
||||
sleep_function=clock.sleep,
|
||||
).open_and_capture(SERIAL, URL, target)
|
||||
|
||||
self.assertEqual(adb.calls, [("inspect", SERIAL), ("intent", "123")])
|
||||
self.assertEqual(device.calls, ["app_info", "app_current", "app_current", "app_current", "app_current", "app_current"])
|
||||
self.assertEqual(clock.sleeps, [0.5, 0.5, 0.5, 0.5])
|
||||
self.assertFalse(target.exists())
|
||||
|
||||
def test_foreground_package_poll_waits_for_pdd_before_reading_evidence(self) -> None:
|
||||
adb = FakeAdbClient()
|
||||
device = FakeUiDevice(current_packages=["com.example.other", "com.xunmeng.pinduoduo"])
|
||||
clock = FakeClock()
|
||||
with TemporaryDirectory() as temporary:
|
||||
target = Path(temporary) / "evidence"
|
||||
result = self._capturer(
|
||||
adb,
|
||||
device,
|
||||
foreground_poll_interval_seconds=0.25,
|
||||
monotonic_clock=clock.monotonic,
|
||||
sleep_function=clock.sleep,
|
||||
).open_and_capture(SERIAL, URL, target)
|
||||
|
||||
self.assertTrue(result.manifest_path.exists())
|
||||
self.assertEqual(clock.sleeps, [0.25])
|
||||
self.assertEqual(
|
||||
device.calls,
|
||||
["app_info", "app_current", "app_current", "takeScreenshot", "dumpWindowHierarchy"],
|
||||
)
|
||||
|
||||
def test_foreground_package_poll_stops_at_deadline_without_evidence(self) -> None:
|
||||
adb = FakeAdbClient()
|
||||
device = FakeUiDevice(current_packages=["com.example.other", "", "com.example.other"])
|
||||
clock = FakeClock()
|
||||
with TemporaryDirectory() as temporary:
|
||||
target = Path(temporary) / "evidence"
|
||||
with self.assertRaises(ProductPackageMismatchError):
|
||||
self._capturer(
|
||||
adb,
|
||||
device,
|
||||
foreground_poll_interval_seconds=0.8,
|
||||
monotonic_clock=clock.monotonic,
|
||||
sleep_function=clock.sleep,
|
||||
).open_and_capture(SERIAL, URL, target)
|
||||
|
||||
self.assertEqual(len(clock.sleeps), 3)
|
||||
for actual, expected in zip(clock.sleeps, (0.8, 0.8, 0.4), strict=True):
|
||||
self.assertAlmostEqual(actual, expected)
|
||||
self.assertEqual(device.calls, ["app_info", "app_current", "app_current", "app_current", "app_current"])
|
||||
self.assertFalse(target.exists())
|
||||
self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), [])
|
||||
|
||||
def test_foreground_poll_interval_must_be_positive_and_finite(self) -> None:
|
||||
for interval in (0, -0.1, float("inf"), float("nan"), True):
|
||||
with self.subTest(interval=interval):
|
||||
with self.assertRaises(ValueError):
|
||||
ProductOpenCapturer(
|
||||
FakeAdbClient(),
|
||||
lambda serial: FakeUiDevice(),
|
||||
timeout_seconds=2,
|
||||
foreground_poll_interval_seconds=interval, # type: ignore[arg-type]
|
||||
)
|
||||
|
||||
def test_timeout_and_invalid_hierarchy_leave_no_partial_evidence(self) -> None:
|
||||
scenarios = (
|
||||
(FakeUiDevice(timeout_on_screenshot=True), ProductOpenTimeoutError),
|
||||
(FakeUiDevice(hierarchy="<not-hierarchy />"), ProductHierarchyCaptureError),
|
||||
)
|
||||
for device, error_type in scenarios:
|
||||
with self.subTest(error_type=error_type.__name__), TemporaryDirectory() as temporary:
|
||||
target = Path(temporary) / "evidence"
|
||||
with self.assertRaises(error_type):
|
||||
self._capturer(FakeAdbClient(), device).open_and_capture(SERIAL, URL, target)
|
||||
|
||||
self.assertFalse(target.exists())
|
||||
self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), [])
|
||||
|
||||
def test_invalid_screenshot_is_a_distinct_redacted_failure(self) -> None:
|
||||
class InvalidScreenshotDevice(FakeUiDevice):
|
||||
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
|
||||
if method == "takeScreenshot":
|
||||
self.calls.append(method)
|
||||
return "not valid base64!"
|
||||
return super().jsonrpc_call(method, params, timeout)
|
||||
|
||||
with TemporaryDirectory() as temporary:
|
||||
target = Path(temporary) / "evidence"
|
||||
with self.assertRaises(ProductScreenshotCaptureError) as raised:
|
||||
self._capturer(FakeAdbClient(), InvalidScreenshotDevice()).open_and_capture(SERIAL, URL, target)
|
||||
|
||||
self.assertNotIn("base64", str(raised.exception).lower())
|
||||
self.assertFalse(target.exists())
|
||||
self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), [])
|
||||
|
||||
def test_read_only_protocol_has_no_ui_operation_methods(self) -> None:
|
||||
forbidden = {"click", "swipe", "send_keys", "set_text", "press", "long_click"}
|
||||
|
||||
self.assertTrue(forbidden.isdisjoint(ProductOpenUiDevice.__dict__))
|
||||
self.assertEqual(base64.b64decode(_png_base64())[:8], b"\x89PNG\r\n\x1a\n")
|
||||
|
||||
|
||||
class FakeClock:
|
||||
def __init__(self) -> None:
|
||||
self.value = 0.0
|
||||
self.sleeps: list[float] = []
|
||||
|
||||
def monotonic(self) -> float:
|
||||
return self.value
|
||||
|
||||
def sleep(self, seconds: float) -> None:
|
||||
self.sleeps.append(seconds)
|
||||
self.value += seconds
|
||||
@@ -0,0 +1,49 @@
|
||||
"""canonical 商品 URL 的离线解析测试。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
|
||||
CLIENT_ROOT = Path(__file__).resolve().parents[2]
|
||||
sys.path.insert(0, str(CLIENT_ROOT / "src"))
|
||||
|
||||
from cmbuyer_client.pdd.product_url import ProductUrlError, parse_product_url
|
||||
|
||||
|
||||
class ProductUrlTests(unittest.TestCase):
|
||||
def test_rebuilds_url_from_goods_id(self) -> None:
|
||||
link = parse_product_url("https://mobile.yangkeduo.com/goods.html?goods_id=00123")
|
||||
|
||||
self.assertEqual(link.goods_id, "00123")
|
||||
self.assertEqual(
|
||||
link.canonical_url,
|
||||
"https://mobile.yangkeduo.com/goods.html?goods_id=00123",
|
||||
)
|
||||
|
||||
def test_rejects_noncanonical_and_ambiguous_urls(self) -> None:
|
||||
rejected = (
|
||||
"http://mobile.yangkeduo.com/goods.html?goods_id=123",
|
||||
"https://other.example/goods.html?goods_id=123",
|
||||
"https://mobile.yangkeduo.com/other.html?goods_id=123",
|
||||
"https://user@mobile.yangkeduo.com/goods.html?goods_id=123",
|
||||
"https://mobile.yangkeduo.com:8443/goods.html?goods_id=123",
|
||||
"https://mobile.yangkeduo.com:443/goods.html?goods_id=123",
|
||||
"https://mobile.yangkeduo.com/goods.html?goods_id=123#fragment",
|
||||
"https://mobile.yangkeduo.com/goods.html",
|
||||
"https://mobile.yangkeduo.com/goods.html?goods_id=123&goods_id=456",
|
||||
"https://mobile.yangkeduo.com/goods.html?goods_id=123&source=share",
|
||||
"https://mobile.yangkeduo.com/goods.html?goods_id=12a",
|
||||
"https://mobile.yangkeduo.com/goods.html?goods_id=%EF%BC%91%EF%BC%92%EF%BC%93",
|
||||
"https://mobile.yangkeduo.com/goods.html?goods_id=",
|
||||
" https://mobile.yangkeduo.com/goods.html?goods_id=123",
|
||||
"https://MOBILE.YANGKEDUO.COM/goods.html?goods_id=123",
|
||||
"https://mobile.yangkeduo.com/goods.html?goods_id=%31%32%33",
|
||||
)
|
||||
|
||||
for value in rejected:
|
||||
with self.subTest(value=value):
|
||||
with self.assertRaises(ProductUrlError):
|
||||
parse_product_url(value)
|
||||
@@ -103,6 +103,28 @@ D:\Portable\adb\adb.exe devices -l
|
||||
和哈希，不得把原始证据提交 Git。USB 与 WiFi 已分别由人完成取证和隐私检查，设备型号、Android、
|
||||
拼多多版本、产物路径及 SHA-256 已记录到 T-101；上述命令保留用于可审计的复现与排障。
|
||||
|
||||
### T-102 按链接打开商品取证（2026-08-04 已完成人工真机验收）
|
||||
|
||||
`client/scripts/capture_product_open.py` 只接受
|
||||
`https://mobile.yangkeduo.com/goods.html?goods_id=<纯数字>` 的唯一 canonical 表示。解析后由
|
||||
`goods_id` 再次重建 URL，并以参数数组执行显式限定 `com.xunmeng.pinduoduo` 的 Android `VIEW`
|
||||
intent；没有任意 URL、任意 shell 或其他 App 控件操作入口。运行拼多多版本必须精确等于 T-101 已
|
||||
取证的 `8.17.0`，版本失配会在 intent 前停止。
|
||||
|
||||
```powershell
|
||||
# 仓库根目录；USB 或 WiFi 每次只保留一个通道在线，再手工复制该次 serial
|
||||
D:\Portable\adb\adb.exe devices -l
|
||||
.\client\.venv\Scripts\python.exe client\scripts\capture_product_open.py --serial <SERIAL> --url "https://mobile.yangkeduo.com/goods.html?goods_id=<GOODS_ID>" --output-dir "$env:LOCALAPPDATA\cmbuyer\artifacts\T-102\product-open-<GOODS_ID>" --timeout 10 --adb D:\Portable\adb\adb.exe
|
||||
```
|
||||
|
||||
脚本在 intent 成功后，以 `--timeout` 为明确上限只读轮询前台 package；只有观察到拼多多才采集截图与
|
||||
完整 XML，超时仍 fail closed。这一等待只解决 App 异步切换，不根据 Activity、节点文本或旧项目常量
|
||||
声称已到详情页。成功目录以原子方式发布，manifest 仅记录 `goods_id`、canonical URL、
|
||||
设备/App 非敏感元数据、受限命令摘要、文件路径和 SHA-256，不含原始 serial、Activity 或页面正文。
|
||||
必须由人本地确认截图对应目标商品并检查截图/XML 无地址、手机号、支付信息或其他无关隐私；原始
|
||||
证据不得提交 Git。T-102 已由人确认 goods_id `958756616606` 对应目标商品并完成截图/XML 隐私检查，
|
||||
设备、版本、证据路径与 SHA-256 已记录到任务执行记录。
|
||||
|
||||
Windows 的标准入口是仓库根 `./init.ps1`。它要求 Go、两端目录及其哨兵文件存在；已有合规
|
||||
`client/.venv` 时，所有采购工具检查与 validator 都使用该解释器。只有 venv 不存在时，才从 `py -0p`
|
||||
枚举的版本中确定性选择最高的 Python 3.11+ 创建它；没有合规版本时明确失败，绝不回退默认 `python`。
|
||||
|
||||
+27
-11
@@ -11,16 +11,16 @@
|
||||
|
||||
## 当前快照
|
||||
|
||||
- 日期：2026-08-03
|
||||
- 阶段：**Phase 0 · 地基（两端骨架、核心数据模型与统一入口已完成，尚无真机采购业务代码）**
|
||||
- 日期：2026-08-04
|
||||
- 阶段：**Phase 1 · 真机可行性（T-101、T-102 已完成人工真机验收，下一步 T-103）**
|
||||
- MVP 形态：手工填链接建单 → 批量开始试选 → 定时轮询 → **第一趟试选** → 人工确认 → **第二趟下单** → 待付款
|
||||
- 技术栈：已定。采购服务（`admin/`）使用 Go 1.23+ / gin / SQLite；采购工具（`client/`）
|
||||
使用 Python 3.11+ / uiautomator2 / PySide6。
|
||||
详见 [`03-tech-stack.md`](03-tech-stack.md)
|
||||
- 生产代码：`admin/` 已有最小 Go 服务、健康检查、核心领域模型、SQLite 迁移与任务状态机；
|
||||
`client/` 已有 Python 包、PySide6 最小入口、运行目录与日志脱敏策略，以及显式 serial 的 ADB
|
||||
连接边界与本地基线取证 CLI；尚无真机采购流程
|
||||
- 测试：采购服务已覆盖健康检查、核心模型、迁移与状态机等离线包级测试；采购工具 30 项离线单元测试
|
||||
连接边界、本地基线取证 CLI 和受限商品链接打开取证 CLI；尚无规格选择、价格读取或下单流程
|
||||
- 测试：采购服务已覆盖健康检查、核心模型、迁移与状态机等离线包级测试；采购工具 46 项离线单元测试
|
||||
（全部 mock，不连接真机）
|
||||
- 数据：SQLite 核心表与迁移已落成；无业务实例数据
|
||||
- 标准启动路径：Windows PowerShell 运行 `./init.ps1`，Unix shell 运行 `./init.sh`。Windows 入口
|
||||
@@ -28,19 +28,19 @@
|
||||
并且不覆盖低版本环境；成功后打印真实启动命令。
|
||||
- 标准验证路径：`./init.ps1` 已实际跑通 admin 的 mod download / test / vet / build、client 的
|
||||
editable install / 包导入 / unittest / compileall，以及仓库上下文校验。可单独运行两端命令诊断。
|
||||
- 当前 blocker：无外部 blocker。T-101 已完成人工 USB/WiFi 双通道验收；下一步落成并领取 T-102，
|
||||
验证按链接打开商品详情页。桌面 GUI 与后续真机采购流程尚未验收。
|
||||
- 当前 blocker：无外部 blocker。T-102 已证明 canonical 链接可在拼多多 8.17.0 打开到目标商品并完成
|
||||
人工隐私验收；下一步落成 T-103 的规格面板证据与试选验证。T-103 通过前 Phase 2 仍不能抢跑。
|
||||
|
||||
## 当前目录要点
|
||||
|
||||
| 路径 | 状态 | 说明 |
|
||||
| --- | --- | --- |
|
||||
| `docs/` | 已有 | 项目规范化文档，本次已完整生成 |
|
||||
| `docs/tasks/` | 已有（T-001～T-004、T-005～T-009、T-101） | T-001～T-004、T-101 已完成；下一任务为 T-102 |
|
||||
| `docs/tasks/` | 已有（T-001～T-004、T-005～T-009、T-101～T-102） | T-001～T-004、T-101～T-102 已完成；下一任务为 T-103 |
|
||||
| `docs/design/` | 已有（6 个原型） | web 登录 / 建单 / 工作台 / 详情，desk 采购执行 / 配置；均已人工确认 |
|
||||
| `scripts/` | 已有 | 上下文门禁、Vikunja 单向导出与 MCP 启动包装 |
|
||||
| `admin/` | 已初始化 | Go 1.23+ / gin / SQLite，含核心模型、迁移与状态机；无真机采购执行 |
|
||||
| `client/` | 已初始化 | Python 3.11+ 包、依赖源、PySide6 最小入口、显式 serial 的设备基线取证、离线测试与 wheel 元数据检查；无采购流程 |
|
||||
| `client/` | 已初始化 | Python 3.11+ 包、依赖源、PySide6 最小入口、显式 serial 的基线/商品打开取证、离线测试与 wheel 元数据检查；无规格选择、价格读取或下单流程 |
|
||||
| `init.ps1` / `init.sh` | 已完成 | 统一安装与离线验证入口；PowerShell 优先复用合规 venv，缺失时自动选择最高的 Python 3.11+，Unix 缺工具链明确失败 |
|
||||
|
||||
## 任务状态
|
||||
@@ -51,9 +51,10 @@
|
||||
源码目录契约）、T-008（Vikunja 任务权威与单向导出）、T-009（MVP 关键路径与并行波次），
|
||||
以及 T-001（采购服务 Go 骨架）。
|
||||
- 已完成：T-002（采购工具 Python 骨架）、T-003（双端统一初始化与验证入口）、
|
||||
T-004（核心数据模型）、T-101（真机环境盘点与 USB/WiFi 双通道人工验收）。下一步推进
|
||||
T-102 → T-103。
|
||||
- T-103 是当前最高优先级和 MVP 生死线。通过前不开发依赖真机可读字段的 Phase 2 生产页面。
|
||||
T-004（核心数据模型）、T-101（真机环境盘点与 USB/WiFi 双通道人工验收）、T-102（canonical
|
||||
链接打开与目标商品/隐私人工验收）。下一步推进 T-103。
|
||||
- T-103 是当前最高优先级和 MVP 生死线。通过前不开发依赖真机可读字段的
|
||||
Phase 2 生产页面。
|
||||
- 已确认原型继续只作信息架构依据；原型假数据不调用真实接口、不驱动真机。真机结论改变
|
||||
可读字段时必须先回修原型与交互清单。
|
||||
|
||||
@@ -117,6 +118,21 @@ D:\Portable\adb\adb.exe devices -l
|
||||
公开 JSON-RPC 调用，uiautomator2 初始化仍有上游固定启动上限。截图/XML 只保留在本地，执行记录只写
|
||||
路径和 SHA-256，原始证据不得提交 Git。
|
||||
|
||||
T-102 的人工真机验收命令（只接受唯一 canonical 链接；USB 或 WiFi 每次只保留一个通道在线）：
|
||||
|
||||
```powershell
|
||||
# 仓库根目录；从输出中手工复制本次在线 serial
|
||||
D:\Portable\adb\adb.exe devices -l
|
||||
.\client\.venv\Scripts\python.exe client\scripts\capture_product_open.py --serial <SERIAL> --url "https://mobile.yangkeduo.com/goods.html?goods_id=<GOODS_ID>" --output-dir "$env:LOCALAPPDATA\cmbuyer\artifacts\T-102\product-open-<GOODS_ID>" --timeout 10 --adb D:\Portable\adb\adb.exe
|
||||
```
|
||||
|
||||
脚本只允许 Android `VIEW` intent，并把 package 固定为 `com.xunmeng.pinduoduo`；intent 后会在
|
||||
`--timeout` 的有限窗口内只读轮询前台 package，解决 App 异步切换造成的一次性误判，超时仍会停止。
|
||||
它不点击、滑动、输入或判断商品页节点，也不打开规格、读取价格、进入下单或支付。运行拼多多版本必须精确为
|
||||
`8.17.0`，否则在 intent 前停止。成功后由人本地查看截图/XML，确认页面确为该 `goods_id` 对应商品并
|
||||
检查无地址、手机号、支付信息或其他无关隐私；只回报 manifest 路径及截图/XML SHA-256，原始证据
|
||||
不得提交 Git。T-102 已由人确认 goods_id `958756616606` 的目标商品及截图/XML 隐私，并完成验收。
|
||||
|
||||
## 关键背景
|
||||
|
||||
本项目是 `cmroubao`（Go 后端 + Android AccessibilityService）与 `cmpdd`
|
||||
|
||||
@@ -0,0 +1,99 @@
|
||||
---
|
||||
id: T-102
|
||||
title: 验证按链接打开商品详情页
|
||||
phase: 1
|
||||
deps: [T-101]
|
||||
status: DONE
|
||||
created: 2026-08-04
|
||||
vikunja_task_id: 22
|
||||
context_ref: 393f26d
|
||||
work_branch: task/t-102-open-product
|
||||
needs_device: true
|
||||
needs_human_review: true
|
||||
write_paths:
|
||||
- docs/tasks/T-102.md
|
||||
- client/src/cmbuyer_client/pdd/**
|
||||
- client/src/cmbuyer_client/device/**
|
||||
- client/tests/pdd/**
|
||||
- client/tests/device/**
|
||||
- client/scripts/capture_product_open.py
|
||||
- docs/03-tech-stack.md
|
||||
- docs/04-architecture.md
|
||||
- docs/current-state.md
|
||||
---
|
||||
|
||||
<!-- BEGIN VIKUNJA EXPORT id=22 synced=2026-08-04T01:50:40Z sha256=ee2405865a36aa01a3503e6d24d62a3dddd0f19b451845e1e2c3937734c31442 -->
|
||||
## 问题 / 背景
|
||||
|
||||
T-101 已证明同一台 PKG110（Android 16、拼多多 8.17.0）可通过 USB/WiFi 显式 serial 完成只读截图与完整 XML 取证。MVP 下一风险是任务自带的合法拼多多链接能否由 Android intent 打开到对应商品详情页。此前项目和旧文档不能证明当前 App 的行为；若先写详情页判据，会把未经本项目真机验证的假设带入 T-103。
|
||||
|
||||
## 关联需求与交互
|
||||
|
||||
- 功能：F-006 的第一步 `open_product(url)`；仅覆盖链接打开，不覆盖规格面板、规格选择或价格。
|
||||
- 用户故事 / 交互：采购工具执行链接任务前的只读真机 spike；无生产 GUI。
|
||||
- 架构 / API：`docs/04-architecture.md` 第三节 A 路径；`docs/api.md` 的 canonical `product_url` / `goods_id` 示例；T-103 前置。
|
||||
|
||||
## 方案
|
||||
|
||||
1. 只接受 HTTPS canonical 链接 `https://mobile.yangkeduo.com/goods.html?goods_id=<纯数字>`；拒绝 userinfo、非默认端口、fragment、重复/缺失/非数字 goods_id、其他 host/path 和额外业务参数。解析后由代码按 goods_id 重建 canonical URL，绝不把任意输入拼入 shell。
|
||||
2. 复用 T-101 的显式 serial、同机多通道 fail-closed、no-reconnect、类型化超时和原子证据发布边界；运行拼多多版本必须精确等于本项目已取证的 8.17.0，否则在打开链接前停止。
|
||||
3. 通过参数数组执行只含 Android `VIEW` intent 语义的 `adb shell am start -W`，显式限定包 `com.xunmeng.pinduoduo`；不使用 shell 字符串、不点击/滑动/输入任何控件。Android 命令失败、超时、未解析 intent、未停留在拼多多包、截图/XML 失败分别给出脱敏错误。
|
||||
4. 启动后只读取 current app/package、截图和 `dumpWindowHierarchy(compressed=False)`，manifest 记录 goods_id、canonical URL、设备/App 元数据、命令结果摘要、路径与 SHA-256，不记录页面全文或原始 serial。原始截图/XML 仅存 `%LOCALAPPDATA%`，不提交 Git。
|
||||
5. 离线 mock 测试覆盖 URL 正反例、参数数组与超时、版本失配在 intent 前停止、包不匹配、无 UI 点击 API、证据原子性和异常脱敏。代码完成后由人使用一个明确 goods_id 在仅单通道在线的真机执行，人工确认截图确为对应商品详情页并检查 XML/截图无敏感信息；再把设备、Android、拼多多版本、goods_id、路径、SHA-256 与结论写入执行记录。
|
||||
|
||||
## 验收要点
|
||||
|
||||
- 合法 canonical 链接可在拼多多 8.17.0 上打开；人工确认到达与 goods_id 对应的商品详情页。
|
||||
- 非法链接、设备状态、App 版本失配、intent 失败/超时、落在非拼多多包、证据失败均有可区分且不泄露页面内容的错误。
|
||||
- 取证记录包含 PKG110、Android 16、连接方式、拼多多 8.17.0、goods_id、canonical URL、截图/XML 本地路径与 SHA-256。
|
||||
- client 全部单测、compileall、wheel metadata、上下文校验及完整 `init.ps1` 通过。
|
||||
- `needs_device=true`：agent 只能完成离线实现并保持 DOING；只有人完成真机链接打开、页面对应性与隐私检查后才能 DONE。
|
||||
|
||||
## 执行记录
|
||||
|
||||
### 2026-08-04T01:09:47Z · ila
|
||||
|
||||
2026-08-04：已认领 T-102，工作分支 `task/t-102-open-product`，从任务定义提交 `4281b06` 开始实现。离线阶段只实现严格 canonical URL 校验、Android VIEW intent、版本/设备 fail-closed 与原子证据采集；不写商品详情页节点判据，不操作任何 App 控件。`needs_device=true`，离线实现完成后仍保持 DOING，等待人工提供明确商品链接并完成真机页面对应性和隐私验收。
|
||||
|
||||
### 2026-08-04T01:27:08Z · ila
|
||||
|
||||
2026-08-04：T-102 离线实现已由子 agent 完成并经主 agent 独立审查、两轮退回修正后通过，代码提交 `7040bb6`。实现包含：唯一 canonical 商品 URL 校验与重建、ADB 内层纯数字 goods_id 二次围栏、固定 package 的 Android VIEW intent、拼多多 8.17.0 intent 前版本围栏、前台 package 核验，以及截图/XML 原子取证与脱敏 manifest；无点击、滑动、输入、规格、价格、下单或支付能力，也未写商品详情页节点/Activity 判据。主审验证：43 项 client 离线测试、compileall、完整 init.ps1（含 admin test/vet/build）、wheel metadata、agent-context 与 diff-check 均通过；当前 wheel SHA-256 为 c675cd527849149ec25f213a0d67da31d0bb497cd8c5ed255d71154c103659ed（审计临时产物已清理）。未连接或操作真机，T-102 继续保持 DOING，等待人工使用明确 canonical 商品链接执行取证，并确认页面对应性与隐私。
|
||||
|
||||
### 2026-08-04T01:32:31Z · ila
|
||||
|
||||
2026-08-04 人工真机首轮：使用 WiFi serial `192.168.0.173:5555` 与 canonical 商品链接 `https://mobile.yangkeduo.com/goods.html?goods_id=958756616606` 执行 T-102 脚本。脚本通过 URL/intent 前置围栏，但在 intent 后返回“商品链接打开后前台应用不是拼多多”，按 fail-closed 规则停止，未发布完整取证。人在 Chrome 中可打开该 URL，只能证明网页链接有效，不能证明拼多多 App 已接管。T-102 保持 DOING；下一步由人只读取当前前台 package，区分瞬时切换/检查过早与实际落到 Chrome、系统解析器或其他包，在得到事实前不移除 package 围栏、不增加页面判据或 UI 兜底。
|
||||
|
||||
### 2026-08-04T01:34:39Z · ila
|
||||
|
||||
2026-08-04 人工追加诊断：T-102 首轮返回前台包不符后，人在未切换 App 的情况下立即执行只读 `dumpsys window`，得到 `mCurrentFocus` 与 `mFocusedApp` 均为 `com.xunmeng.pinduoduo/com.xunmeng.pinduoduo.activity.NewPageActivity`。该事实证明失败返回后前台已稳定到拼多多，支持“一次性 app_current 检查过早/存在异步切换窗口”的诊断；Activity 名只记录为诊断证据，不作为商品详情页判据。修复保持 package 围栏：改为有限时长、只读 package 的轮询，超时仍 fail closed，期间不点击、滑动、输入或读取页面节点。
|
||||
|
||||
### 2026-08-04T01:40:42Z · ila
|
||||
|
||||
2026-08-04：已按人工 foreground 证据完成最小修复，代码提交 `cb646b4`。保留 PDD package 围栏，将 intent 后的一次性判断改为以 CLI `--timeout` 为 deadline、默认 0.2 秒间隔的只读 package 轮询；只有观察到 `com.xunmeng.pinduoduo` 才采集截图/XML，超时仍 fail closed。轮询不读取 Activity/节点，不点击、滑动或输入。主 agent 已独立验证 46 项 client 测试、compileall、完整 init.ps1、wheel metadata、agent-context、diff-check；wheel SHA-256 `80fbc73785bb12b40856c6b1ef503bdb5cbc350fadd2b23cc6cc58358499bfe1`，临时产物已清理。未由 agent 连接真机，T-102 保持 DOING，等待人以 goods_id `958756616606` 重跑原命令并检查页面对应性与隐私。
|
||||
|
||||
### 2026-08-04T01:46:35Z · ila
|
||||
|
||||
2026-08-04 人工真机重跑已成功：人先退出到拼多多首页，再用 WiFi serial `192.168.0.173:5555`、goods_id `958756616606` 执行同一命令，证据原子发布到 `C:\Users\ila20\AppData\Local\cmbuyer\artifacts\T-102\product-open-958756616606`。manifest 记录 PKG110 / Android 16 / PDD 8.17.0 / current package `com.xunmeng.pinduoduo` / intent status ok；截图 SHA-256 `be9ea1f53b13fec82fd716875567870132ef0a1480cfbec5ad603436d774d81c`，XML SHA-256 `1449e78a154fddf7108e63a14a2e85bf74f162f4689c1fe95f2cb67fdb7b4fc5`。主 agent 只读取 manifest 并重新计算哈希，二者均匹配；未打开原始截图/XML。仍等待人明确确认截图对应目标商品且截图/XML 无地址、手机号、支付信息或其他无关隐私，确认前 T-102 保持 DOING。
|
||||
|
||||
### 2026-08-04T01:49:55Z · ila
|
||||
|
||||
2026-08-04 人工最终验收：人已查看本地 screenshot.png，确认对应 goods_id 958756616606；并确认截图与 hierarchy.xml 不含地址、手机号、支付信息或其他无关隐私。T-102 的 canonical 链接打开、PDD 8.17.0 前台 package、证据原子发布与人工页面对应性验收全部通过，允许关闭。
|
||||
<!-- END VIKUNJA EXPORT -->
|
||||
|
||||
## 边界
|
||||
|
||||
- 本任务只验证由显式 canonical 链接启动拼多多并采集本地证据；不编写或声称商品详情页节点判据，
|
||||
页面是否为对应商品必须由人查看本项目新产物确认。
|
||||
- 只允许 Android `VIEW` intent；不点击、滑动、长按、输入任何 App 控件，不打开规格面板，不选择
|
||||
颜色或尺码，不读取价格，不进入购买或订单页面。
|
||||
- 不引用或实现 `go_to_order_confirm()`、`submit_order()`、提交订单、付款、支付、授权或采购服务接口；
|
||||
不把前序项目的 URL、activity、节点文本或页面结论当作事实。
|
||||
- 输入 URL 必须先严格解析并按纯数字 `goods_id` 重建，再以参数数组传给 ADB；禁止 shell 拼接、
|
||||
任意 scheme/host/path、短链跳转、额外参数或自动猜测链接。
|
||||
- 运行拼多多版本与本项目已取证版本不一致时必须在 intent 前停止；证据失败、当前包不符或结果无法
|
||||
由人确认时均不得声称成功，不增加自动点击或其他兜底路径。
|
||||
- 截图和完整 XML 只保存在明确的本地 T-102 目录，人工检查后只把路径、SHA-256 与非敏感元数据
|
||||
写入执行记录；不得提交原始证据，不记录地址、手机号、支付信息或无关页面正文。
|
||||
- 不修改 `admin/`、业务 API、数据模型、生产 GUI 或 T-103 规格面板逻辑。
|
||||
- `needs_device: true`：agent 完成离线实现后保持 `DOING`；只有人使用明确 goods_id 完成真机打开、
|
||||
确认对应商品详情页并完成隐私检查后才能标 `DONE`。
|
||||
Reference in New Issue
Block a user