feat(client): add T-103 manual SKU evidence capture

client/scripts/capture_sku_panel_spike.py

@@ -0,0 +1,103 @@
+"""采集人工已停在规格面板的三种状态证据；不执行任何页面操作。"""
+
+from __future__ import annotations
+
+import argparse
+from math import isfinite
+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_url import ProductUrl, ProductUrlError, parse_product_url
+from cmbuyer_client.pdd.sku_panel_spike import (
+    HUMAN_DECLARED_STATES,
+    SkuPanelEvidenceCapturer,
+    SkuPanelEvidenceError,
+)
+
+
+def parse_arguments(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="采集人工已打开的拼多多规格面板状态证据。")
+    parser.add_argument("--serial", required=True, help="ADB device serial；禁止自动选择。")
+    product_source = parser.add_mutually_exclusive_group(required=True)
+    product_source.add_argument("--url", help="唯一 canonical goods.html?goods_id= 直链。")
+    product_source.add_argument("--goods-id", help="纯数字商品标识；仅用于记录证据归属。")
+    parser.add_argument("--state", required=True, choices=sorted(HUMAN_DECLARED_STATES), help="人工声明的面板状态。")
+    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 resolve_product_url(arguments: argparse.Namespace) -> ProductUrl:
+    if isinstance(arguments.url, str):
+        return parse_product_url(arguments.url)
+    if isinstance(arguments.goods_id, str):
+        # 仅使用严格 parser 重新验证并构建，不把输入交给 ADB 或页面。
+        return parse_product_url(f"https://mobile.yangkeduo.com/goods.html?goods_id={arguments.goods_id}")
+    raise ValueError("必须提供 --url 或 --goods-id。")
+
+
+def validate_arguments(arguments: argparse.Namespace) -> ProductUrl:
+    if not isinstance(arguments.serial, str) or not arguments.serial.strip():
+        raise ValueError("必须显式提供非空 --serial。")
+    if (
+        not isinstance(arguments.timeout, (int, float))
+        or isinstance(arguments.timeout, bool)
+        or arguments.timeout <= 0
+        or not isfinite(arguments.timeout)
+    ):
+        raise ValueError("--timeout 必须是大于 0 的有限数值。")
+    if arguments.state not in HUMAN_DECLARED_STATES:
+        raise ValueError("--state 必须是允许的人工声明状态。")
+    return resolve_product_url(arguments)
+
+
+def main(argv: list[str] | None = None) -> int:
+    arguments = parse_arguments(argv)
+    try:
+        link = validate_arguments(arguments)
+    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 = SkuPanelEvidenceCapturer(client, connector, timeout_seconds=arguments.timeout)
+    try:
+        result = capturer.capture(
+            arguments.serial,
+            link.canonical_url,
+            arguments.state,
+            arguments.output_dir,
+        )
+    except (DeviceConnectionError, SkuPanelEvidenceError) 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())

client/src/cmbuyer_client/pdd/sku_panel_spike.py

@@ -0,0 +1,260 @@
+"""人工停留在规格面板后的只读取证。
+
+本模块不识别规格面板，不打开商品链接，也不读取价格；三种面板状态完全由现场人员声明。
+"""
+
+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 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
+from ..device.baseline import (
+    HIERARCHY_PARAMS,
+    PDD_PACKAGE,
+    SCREENSHOT_PARAMS,
+    _save_base64_screenshot,
+    _sha256_file,
+    _validate_hierarchy,
+)
+from .product_open import EXPECTED_PDD_VERSION
+from .product_url import ProductUrl, parse_product_url
+
+
+HUMAN_DECLARED_STATES = frozenset(
+    {
+        "initial",
+        "one-dimension-selected",
+        "all-dimensions-selected",
+    }
+)
+
+
+class SkuPanelEvidenceError(RuntimeError):
+    """人工规格面板证据无法完整发布。"""
+
+
+class SkuPanelDeclaredStateError(SkuPanelEvidenceError):
+    """调用方没有提供允许的人工声明状态。"""
+
+
+class SkuPanelVersionMismatchError(SkuPanelEvidenceError):
+    """运行时拼多多版本不是已取证版本。"""
+
+
+class SkuPanelPackageMismatchError(SkuPanelEvidenceError):
+    """人工声明前台不是拼多多时仍试图留证。"""
+
+
+class SkuPanelEvidenceTimeoutError(SkuPanelEvidenceError):
+    """只读截图或节点树取证超时。"""
+
+
+class SkuPanelScreenshotError(SkuPanelEvidenceError):
+    """截图不能保存为严格有效的 PNG。"""
+
+
+class SkuPanelHierarchyError(SkuPanelEvidenceError):
+    """节点树不能保存为严格有效的 XML。"""
+
+
+class SkuPanelUiDevice(Protocol):
+    """人工面板证据所需的只读接口，故意没有任何页面操作方法。"""
+
+    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 SkuPanelEvidenceResult:
+    """已原子发布的本地证据目录。"""
+
+    output_directory: Path
+    manifest_path: Path
+    screenshot_path: Path
+    hierarchy_path: Path
+
+
+class SkuPanelEvidenceCapturer:
+    """把人工已停留的面板状态留证，不对页面作任何自动结论。
+
+    状态字段命名为 ``human_declared_state``，防止消费者把本模块误解为自动面板/规格/价格识别。
+    """
+
+    def __init__(
+        self,
+        adb_client: AdbClient,
+        connector: Callable[[str], SkuPanelUiDevice],
+        timeout_seconds: float,
+    ) -> None:
+        if not _is_positive_finite(timeout_seconds):
+            raise ValueError("timeout_seconds 必须是大于 0 的有限数值")
+        self._adb_client = adb_client
+        self._connector = connector
+        self._timeout_seconds = timeout_seconds
+
+    def capture(
+        self,
+        serial: str,
+        product_url: str,
+        human_declared_state: str,
+        output_directory: Path,
+    ) -> SkuPanelEvidenceResult:
+        """采集人工已准备的状态；不会打开链接、面板或执行任何 UI 操作。"""
+
+        link = parse_product_url(product_url)
+        state = _validate_human_declared_state(human_declared_state)
+        target = Path(output_directory)
+        _validate_new_target(target)
+
+        staging: Path | None = None
+        try:
+            # 沿用 T-101 的显式 serial、在线状态与重复物理设备 fail-closed 核验。
+            inspection = self._adb_client.inspect(serial)
+            device = self._connector(serial)
+            pdd_version = _require_expected_version(device.app_info(PDD_PACKAGE))
+            _require_pdd_foreground(device.app_current())
+
+            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 SkuPanelScreenshotError("规格面板截图取证失败，未发布任何证据产物。") 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 SkuPanelHierarchyError("规格面板节点树取证失败，未发布任何证据产物。") 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, state, pdd_version, screenshot_path, hierarchy_path),
+                    ensure_ascii=False,
+                    indent=2,
+                    sort_keys=True,
+                )
+                + "\n",
+                encoding="utf-8",
+            )
+            os.replace(staging, target)
+        except (SkuPanelEvidenceError, DeviceConnectionError):
+            _clean_staging(staging)
+            raise
+        except (AdbTimeout, HTTPTimeoutError, TimeoutError) as error:
+            _clean_staging(staging)
+            raise SkuPanelEvidenceTimeoutError("规格面板只读取证超时，未发布任何证据产物。") from error
+        except Exception as error:
+            _clean_staging(staging)
+            # 第三方异常可能含 serial、Activity 或页面正文，不能直接向 CLI/日志传播。
+            raise SkuPanelEvidenceError("规格面板只读取证未完成，未发布任何证据产物。") from error
+
+        return SkuPanelEvidenceResult(
+            output_directory=target,
+            manifest_path=target / "manifest.json",
+            screenshot_path=target / "screenshot.png",
+            hierarchy_path=target / "hierarchy.xml",
+        )
+
+
+def _is_positive_finite(value: object) -> bool:
+    return isinstance(value, (int, float)) and not isinstance(value, bool) and value > 0 and isfinite(value)
+
+
+def _validate_human_declared_state(value: object) -> str:
+    if not isinstance(value, str) or value not in HUMAN_DECLARED_STATES:
+        raise SkuPanelDeclaredStateError("必须提供允许的人工声明规格面板状态。")
+    return value
+
+
+def _validate_new_target(target: Path) -> None:
+    if target.exists():
+        raise SkuPanelEvidenceError("输出目录已存在；为防止混入旧证据，拒绝覆盖。")
+    if not target.name:
+        raise SkuPanelEvidenceError("输出目录必须是明确的新目录。")
+
+
+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: object) -> str:
+    if not isinstance(app_info, dict):
+        raise SkuPanelVersionMismatchError("拼多多版本与已取证版本不一致，已停止取证。")
+    version = app_info.get("versionName") or app_info.get("version_name")
+    if not isinstance(version, str) or version != EXPECTED_PDD_VERSION:
+        raise SkuPanelVersionMismatchError("拼多多版本与已取证版本不一致，已停止取证。")
+    return version
+
+
+def _require_pdd_foreground(current: object) -> None:
+    if not isinstance(current, dict) or current.get("package") != PDD_PACKAGE:
+        raise SkuPanelPackageMismatchError("当前前台应用不是拼多多，已停止取证。")
+
+
+def _manifest(
+    inspection: DeviceInspection,
+    serial: str,
+    link: ProductUrl,
+    human_declared_state: str,
+    pdd_version: str,
+    screenshot_path: Path,
+    hierarchy_path: Path,
+) -> dict[str, Any]:
+    """仅记录人工声明与非敏感审计摘要，不写入 Activity 或页面内容。"""
+
+    return {
+        "schema_version": 1,
+        "captured_at": datetime.now(UTC).isoformat(),
+        "product": {"goods_id": link.goods_id, "canonical_url": link.canonical_url},
+        "human_declared_state": human_declared_state,
+        "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,
+        },
+        "artifacts": [
+            {"path": screenshot_path.name, "sha256": _sha256_file(screenshot_path)},
+            {"path": hierarchy_path.name, "sha256": _sha256_file(hierarchy_path)},
+        ],
+    }

client/tests/pdd/test_sku_panel_spike.py

@@ -0,0 +1,258 @@
+"""人工声明规格面板状态的离线只读取证测试。"""
+
+from __future__ import annotations
+
+import argparse
+import base64
+from importlib.util import module_from_spec, spec_from_file_location
+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
+from cmbuyer_client.pdd.product_url import ProductUrlError
+from cmbuyer_client.pdd.sku_panel_spike import (
+    HUMAN_DECLARED_STATES,
+    SkuPanelDeclaredStateError,
+    SkuPanelEvidenceCapturer,
+    SkuPanelEvidenceError,
+    SkuPanelEvidenceTimeoutError,
+    SkuPanelHierarchyError,
+    SkuPanelPackageMismatchError,
+    SkuPanelScreenshotError,
+    SkuPanelUiDevice,
+    SkuPanelVersionMismatchError,
+)
+
+
+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><node text='sensitive page text' /></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[str] = []
+        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(serial)
+        return self.inspection
+
+
+class FakeUiDevice:
+    def __init__(
+        self,
+        *,
+        version: str = "8.17.0",
+        package: str = "com.xunmeng.pinduoduo",
+        screenshot: str | None = None,
+        hierarchy: str = HIERARCHY,
+    ) -> None:
+        self.version = version
+        self.package = package
+        self.screenshot = screenshot if screenshot is not None else _png_base64()
+        self.hierarchy = hierarchy
+        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")
+        return {"package": self.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":
+            return self.screenshot
+        if method == "dumpWindowHierarchy":
+            return self.hierarchy
+        raise AssertionError(f"unexpected RPC {method}")
+
+
+def _load_spike_script() -> object:
+    script_path = CLIENT_ROOT / "scripts" / "capture_sku_panel_spike.py"
+    spec = spec_from_file_location("capture_sku_panel_spike_for_test", script_path)
+    assert spec is not None and spec.loader is not None
+    module = module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+class SkuPanelEvidenceTests(unittest.TestCase):
+    def _capturer(self, adb: FakeAdbClient, device: FakeUiDevice) -> SkuPanelEvidenceCapturer:
+        return SkuPanelEvidenceCapturer(adb, lambda serial: device, timeout_seconds=2)
+
+    def test_all_human_declared_states_publish_redacted_manifest(self) -> None:
+        for state in sorted(HUMAN_DECLARED_STATES):
+            with self.subTest(state=state), TemporaryDirectory() as temporary:
+                adb = FakeAdbClient()
+                device = FakeUiDevice()
+                target = Path(temporary) / "evidence"
+                result = self._capturer(adb, device).capture(SERIAL, URL, state, target)
+                manifest = result.manifest_path.read_text(encoding="utf-8")
+
+                self.assertEqual(adb.calls, [SERIAL])
+                self.assertEqual(device.calls, ["app_info", "app_current", "takeScreenshot", "dumpWindowHierarchy"])
+                self.assertIn(f'"human_declared_state": "{state}"', manifest)
+                self.assertIn('"goods_id": "123"', manifest)
+                self.assertIn('"canonical_url": "https://mobile.yangkeduo.com/goods.html?goods_id=123"', manifest)
+                self.assertNotIn("detected_state", manifest)
+                self.assertNotIn(SERIAL, manifest)
+                self.assertNotIn("sensitive.activity.name", manifest)
+                self.assertNotIn(HIERARCHY, manifest)
+
+    def test_invalid_state_and_url_fail_before_device_access(self) -> None:
+        adb = FakeAdbClient()
+        with TemporaryDirectory() as temporary:
+            with self.assertRaises(SkuPanelDeclaredStateError):
+                self._capturer(adb, FakeUiDevice()).capture(SERIAL, URL, "guessed", Path(temporary) / "state")
+            with self.assertRaises(ProductUrlError):
+                self._capturer(adb, FakeUiDevice()).capture(
+                    SERIAL,
+                    "https://mobile.yangkeduo.com/goods.html?goods_id=12x",
+                    "initial",
+                    Path(temporary) / "url",
+                )
+
+        self.assertEqual(adb.calls, [])
+
+    def test_version_or_foreground_package_mismatch_stops_before_artifacts(self) -> None:
+        scenarios = (
+            (FakeUiDevice(version="8.17.1"), SkuPanelVersionMismatchError, ["app_info"]),
+            (FakeUiDevice(package="com.example.other"), SkuPanelPackageMismatchError, ["app_info", "app_current"]),
+        )
+        for device, error_type, expected_calls 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).capture(SERIAL, URL, "initial", target)
+
+                self.assertEqual(device.calls, expected_calls)
+                self.assertFalse(target.exists())
+
+    def test_screenshot_and_xml_failure_leave_no_partial_evidence(self) -> None:
+        scenarios = (
+            (FakeUiDevice(screenshot="not valid base64!"), SkuPanelScreenshotError),
+            (FakeUiDevice(hierarchy="<not-hierarchy />"), SkuPanelHierarchyError),
+        )
+        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).capture(SERIAL, URL, "initial", target)
+
+                self.assertFalse(target.exists())
+                self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), [])
+
+    def test_screenshot_timeout_is_redacted_and_leaves_no_partial_evidence(self) -> None:
+        class TimeoutScreenshotDevice(FakeUiDevice):
+            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                if method == "takeScreenshot":
+                    self.calls.append(method)
+                    raise TimeoutError("adb 192.168.0.173:5555 raw detail")
+                return super().jsonrpc_call(method, params, timeout)
+
+        with TemporaryDirectory() as temporary:
+            target = Path(temporary) / "evidence"
+            with self.assertRaises(SkuPanelEvidenceTimeoutError) as raised:
+                self._capturer(FakeAdbClient(), TimeoutScreenshotDevice()).capture(SERIAL, URL, "initial", target)
+
+            self.assertNotIn(SERIAL, str(raised.exception))
+            self.assertNotIn("adb", str(raised.exception).lower())
+            self.assertFalse(target.exists())
+            self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), [])
+
+    def test_existing_output_directory_is_not_overwritten_or_connected(self) -> None:
+        adb = FakeAdbClient()
+        device = FakeUiDevice()
+        with TemporaryDirectory() as temporary:
+            target = Path(temporary) / "evidence"
+            target.mkdir()
+            sentinel = target / "sentinel.txt"
+            sentinel.write_text("keep", encoding="utf-8")
+
+            with self.assertRaises(SkuPanelEvidenceError):
+                self._capturer(adb, device).capture(SERIAL, URL, "initial", target)
+
+            self.assertEqual(adb.calls, [])
+            self.assertEqual(device.calls, [])
+            self.assertEqual(sentinel.read_text(encoding="utf-8"), "keep")
+
+    def test_protocol_has_no_ui_or_purchase_operation_methods(self) -> None:
+        forbidden = {
+            "click",
+            "swipe",
+            "send_keys",
+            "set_text",
+            "press",
+            "open_product",
+            "open_sku_panel",
+            "set_quantity",
+            "go_to_order_confirm",
+            "submit_order",
+            "pay",
+        }
+        self.assertTrue(forbidden.isdisjoint(SkuPanelUiDevice.__dict__))
+        self.assertEqual({name for name in SkuPanelEvidenceCapturer.__dict__ if not name.startswith("_")}, {"capture"})
+
+
+class SkuPanelSpikeCliTests(unittest.TestCase):
+    def test_validate_arguments_rejects_invalid_serial_timeout_state_and_url(self) -> None:
+        script = _load_spike_script()
+        valid = {
+            "serial": SERIAL,
+            "url": URL,
+            "goods_id": None,
+            "state": "initial",
+            "output_dir": Path("evidence"),
+            "timeout": 10.0,
+            "adb": "adb",
+        }
+        invalid_values = (
+            ("serial", ""),
+            ("timeout", 0),
+            ("timeout", float("inf")),
+            ("state", "not-declared"),
+            ("url", "https://mobile.yangkeduo.com/goods.html?goods_id=bad"),
+        )
+        for field, value in invalid_values:
+            with self.subTest(field=field, value=value):
+                arguments = argparse.Namespace(**(valid | {field: value}))
+                with self.assertRaises((ValueError, ProductUrlError)):
+                    script.validate_arguments(arguments)  # type: ignore[attr-defined]
+
+    def test_goods_id_is_rebuilt_as_canonical_url(self) -> None:
+        script = _load_spike_script()
+        arguments = argparse.Namespace(
+            serial=SERIAL,
+            url=None,
+            goods_id="00123",
+            state="all-dimensions-selected",
+            output_dir=Path("evidence"),
+            timeout=10.0,
+            adb="adb",
+        )
+
+        link = script.validate_arguments(arguments)  # type: ignore[attr-defined]
+        self.assertEqual(link.canonical_url, "https://mobile.yangkeduo.com/goods.html?goods_id=00123")