diff --git a/client/src/cmbuyer_client/pdd/sku_selection.py b/client/src/cmbuyer_client/pdd/sku_selection.py
index cb60f1a..cc651f2 100644
--- a/client/src/cmbuyer_client/pdd/sku_selection.py
+++ b/client/src/cmbuyer_client/pdd/sku_selection.py
@@ -28,6 +28,12 @@ _ENTRY_ACTION_DESC = "快要抢光¥12.88"
_ENTRY_ACTION_BOUNDS = "[446,2166][1080,2328]"
_ENTRY_SIBLING = "免拼购买"
_ENTRY_SIBLING_BOUNDS = "[688,2256][856,2305]"
+_PRODUCT_TITLE_DESC = "2026年新款高档重工潮流烫钻中长款T恤显瘦宽松上衣淡人穿搭"
+_PRODUCT_TITLE_FIRST_LINE = "2026年新款高档重工潮流烫钻中长款T恤显瘦宽松"
+_PRODUCT_TITLE_SECOND_LINE = "上衣淡人穿搭"
+_PRODUCT_TITLE_RESOURCE_ID = "com.xunmeng.pinduoduo:id/tv_title"
+_PRODUCT_TITLE_BOUNDS = "[36,1571][1044,1689]"
+_DANGEROUS_EXIT_ACTION_TERMS = ("提交订单", "确认订单", "立即支付", "去支付", "付款")
_FORBIDDEN_ENTRY_ACTION_DESC = (
"购买", "下单", "付款", "订单", "单独购买", "直接拼成", "提交订单", "支付",
"先用后付", "0元下单", "0 元下单",
@@ -423,16 +429,22 @@ class SkuSelectionFlow:
self._terminal = True
raise
deadline = self._clock() + self._entry_timeout
+ stable: tuple[object, ...] | None = None
try:
while True:
self._require_foreground()
raw = self._read_hierarchy()
- if raw != before:
- try:
- _classify_panel(_parse_nodes(raw))
- except SkuSelectionError:
+ try:
+ projection = _product_exit_projection(_parse_nodes(raw))
+ except SkuSelectionError:
+ stable = None
+ else:
+ # 用前台/版本检查夹住第二棵候选树；页面在 dump 后漂移时不能成功。
+ self._require_foreground()
+ if stable == projection:
self._terminal = True
return
+ stable = projection
remaining = deadline - self._clock()
if remaining <= 0:
raise SkuSelectionError("安全退出后未确认离开规格面板，未重试返回。")
@@ -1394,6 +1406,62 @@ def _entry_projection(node: _Node, nodes: list[_Node]) -> tuple[object, ...] | N
)
+def _product_exit_projection(nodes: list[_Node]) -> tuple[object, ...]:
+ """返回 T-104 人验同商品详情页的最小稳定投影；不读取页面价格。"""
+
+ if any(
+ _is_live_clickable(node)
+ and any(term in value for value in (node.text, node.desc) for term in _DANGEROUS_EXIT_ACTION_TERMS)
+ for node in nodes
+ ):
+ # T-106/T-107 前只把这些结构当保守拒绝，不宣称它们是完整确认/支付页分类器。
+ raise SkuSelectionError("退出后出现危险动作语义，不能确认安全退出。")
+
+ entries = _eligible_entries(nodes)
+ if len(entries) != 1:
+ raise SkuSelectionError("退出后商品规格入口不唯一。")
+ entry = _entry_projection(entries[0], nodes)
+ if entry is None:
+ raise SkuSelectionError("退出后商品规格入口结构失效。")
+
+ titles = [
+ node
+ for node in nodes
+ if _exact_common(
+ node,
+ "android.view.ViewGroup",
+ _PRODUCT_TITLE_BOUNDS,
+ clickable="true",
+ selected="false",
+ scrollable="false",
+ )
+ and node.element.get("resource-id") == _PRODUCT_TITLE_RESOURCE_ID
+ and node.element.get("long-clickable") == "true"
+ and not node.text
+ and node.desc == _PRODUCT_TITLE_DESC
+ ]
+ title = _one(titles, "退出后同商品标题正锚不唯一。")
+ children = _walk_direct_children(title)
+ expected_children = (
+ (_PRODUCT_TITLE_FIRST_LINE, "[36,1571][1023,1624]"),
+ (_PRODUCT_TITLE_SECOND_LINE, "[36,1636][306,1689]"),
+ )
+ if len(children) != len(expected_children) or any(
+ not _exact_readonly_text(child, text, bounds)
+ for child, (text, bounds) in zip(children, expected_children, strict=True)
+ ):
+ raise SkuSelectionError("退出后同商品标题子结构漂移。")
+
+ return (
+ "product_exit_8_17_0",
+ entry,
+ _entry_node_projection(title),
+ tuple(_entry_node_projection(child) for child in children),
+ title.element.get("resource-id", ""),
+ title.element.get("long-clickable", ""),
+ )
+
+
def _entry_node_projection(node: _Node) -> tuple[str, ...]:
return (
node.element.tag,
diff --git a/client/src/cmbuyer_client/pdd/sku_selection_runner.py b/client/src/cmbuyer_client/pdd/sku_selection_runner.py
index 365458d..54aeb91 100644
--- a/client/src/cmbuyer_client/pdd/sku_selection_runner.py
+++ b/client/src/cmbuyer_client/pdd/sku_selection_runner.py
@@ -148,6 +148,7 @@ class SkuSelectionRunResult:
screenshot_path: Path
manifest_path: Path
unit_price: str
+ captured_at: datetime
@dataclass(frozen=True)
@@ -589,7 +590,9 @@ class SkuSelectionRunner:
stage = "screenshot_capture"
screenshot_path = staging / "screenshot.png"
try:
- _save_base64_screenshot(adapter.capture_screenshot(), screenshot_path)
+ screenshot_payload = adapter.capture_screenshot()
+ captured_at = datetime.now(UTC)
+ _save_base64_screenshot(screenshot_payload, screenshot_path)
_require_screenshot_size(screenshot_path)
except SkuSelectionRunError:
raise
@@ -610,7 +613,22 @@ class SkuSelectionRunner:
stage = "publish"
manifest_path.write_text(
- json.dumps(_manifest(inspection, serial, link, screenshot_path, task_color, task_size, adapter), ensure_ascii=False, indent=2, sort_keys=True) + "\n",
+ json.dumps(
+ _manifest(
+ inspection,
+ serial,
+ link,
+ screenshot_path,
+ task_color,
+ task_size,
+ adapter,
+ captured_at,
+ ),
+ ensure_ascii=False,
+ indent=2,
+ sort_keys=True,
+ )
+ + "\n",
encoding="utf-8",
)
# Windows 的 rename 不替换既有目标；并发创建 target 时保留其内容并把本次运行判失败。
@@ -648,6 +666,7 @@ class SkuSelectionRunner:
screenshot_path=target / "screenshot.png",
manifest_path=target / "manifest.json",
unit_price=EXPECTED_UNIT_PRICE,
+ captured_at=captured_at,
)
@@ -819,13 +838,14 @@ def _manifest(
task_color: str,
task_size: str,
adapter: UiautomatorSkuPanelAdapter,
+ captured_at: datetime,
) -> dict[str, Any]:
"""仅写可审计摘要；原始 serial、节点树、页面文案和实际截图内容均不写入 manifest。"""
option_outcomes = dict(adapter.option_rpc_outcomes)
return {
"schema_version": 1,
- "captured_at": datetime.now(UTC).isoformat(),
+ "captured_at": captured_at.isoformat(),
"operation": "t103-sku-selection",
"product": {"goods_id": link.goods_id, "canonical_url": link.canonical_url},
"target_selection": {"color": task_color, "size": task_size},
@@ -860,8 +880,9 @@ def _manifest(
"rpc_outcome": adapter.back_rpc_outcome,
},
},
- "post_exit_status": "human_review_required",
- "page_identity": "human_review_required",
+ "post_exit_status": "same_product_verified",
+ "safe_exit": "completed",
+ "page_identity": "same_goods_evidence_bound",
"channel": "wifi" if ":" in serial else "usb",
"serial_sha256": sha256(serial.encode("utf-8")).hexdigest(),
"device": {
diff --git a/client/tests/pdd/fixtures/product_exit_8_17_0.xml b/client/tests/pdd/fixtures/product_exit_8_17_0.xml
new file mode 100644
index 0000000..d421456
--- /dev/null
+++ b/client/tests/pdd/fixtures/product_exit_8_17_0.xml
@@ -0,0 +1,12 @@
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/client/tests/pdd/test_sku_selection.py b/client/tests/pdd/test_sku_selection.py
index 9c738f8..893d6ce 100644
--- a/client/tests/pdd/test_sku_selection.py
+++ b/client/tests/pdd/test_sku_selection.py
@@ -25,6 +25,7 @@ from cmbuyer_client.pdd.sku_selection import (
_action_bounds,
_classify_panel,
_parse_nodes,
+ _product_exit_projection,
resolve_task_selection,
)
from cmbuyer_client.pdd.sku_selection_runner import (
@@ -48,10 +49,12 @@ _S_FIXTURE = _FIXTURES / "sku_panel_size_s_selected_8_17_0.xml"
_M_FIXTURE = _FIXTURES / "sku_panel_size_m_restored_8_17_0.xml"
_FIXTURE = _M_FIXTURE
_ENTRY_FIXTURE = Path(__file__).with_name("fixtures") / "product_entry_8_17_0.xml"
+_EXIT_FIXTURE = Path(__file__).with_name("fixtures") / "product_exit_8_17_0.xml"
_TARGET_URL = "https://mobile.yangkeduo.com/goods.html?goods_id=937122477375"
_TASK_COLOR = "黑色CHA(纯棉)"
_TASK_SIZE = "M(建议100-115)"
_PRODUCT_PAGE = _ENTRY_FIXTURE.read_text(encoding="utf-8")
+_PRODUCT_EXIT = _EXIT_FIXTURE.read_text(encoding="utf-8")
def _revealed_current_only() -> str:
@@ -96,7 +99,7 @@ class _RawDevice:
if method == "takeScreenshot":
return self.screenshot
if method == "pressKey":
- self.hierarchy = ""
+ self.hierarchy = _PRODUCT_EXIT
return ""
if method == "click":
if not isinstance(params, list) or len(params) != 2:
@@ -1156,6 +1159,7 @@ class SkuSelectionFlowTests(unittest.TestCase):
_S_FIXTURE,
_M_FIXTURE,
_ENTRY_FIXTURE,
+ _EXIT_FIXTURE,
):
content = fixture.read_text(encoding="utf-8")
with self.subTest(fixture=fixture.name):
@@ -1164,6 +1168,170 @@ class SkuSelectionFlowTests(unittest.TestCase):
self.assertNotIn(forbidden, content)
self.assertNotIn("提交订单", content)
+ def test_product_exit_fixture_binds_exact_same_product_and_entry(self) -> None:
+ projection = _product_exit_projection(_parse_nodes(_PRODUCT_EXIT))
+ self.assertEqual(projection[0], "product_exit_8_17_0")
+
+ root = ElementTree.fromstring(_PRODUCT_EXIT)
+ title = next(
+ node
+ for node in root.iter("node")
+ if node.get("resource-id") == "com.xunmeng.pinduoduo:id/tv_title"
+ )
+ root.append(ElementTree.fromstring(ElementTree.tostring(title, encoding="unicode")))
+ duplicate = ElementTree.tostring(root, encoding="unicode")
+ missing = _PRODUCT_EXIT.replace(
+ "com.xunmeng.pinduoduo:id/tv_title",
+ "com.xunmeng.pinduoduo:id/other",
+ 1,
+ )
+ drifted = _PRODUCT_EXIT.replace("[36,1571][1044,1689]", "[35,1571][1044,1689]", 1)
+ dangerous = _PRODUCT_EXIT.replace(
+ "",
+ '',
+ )
+ system_ui = _PRODUCT_EXIT.replace("com.xunmeng.pinduoduo", "com.android.systemui")
+ non_pdd_overlay = _PRODUCT_EXIT.replace(
+ "",
+ '',
+ )
+ lock_screen = (
+ ''
+ )
+
+ for name, hierarchy in (
+ ("empty", ""),
+ ("panel_still_open", _M_FIXTURE.read_text(encoding="utf-8")),
+ ("system_ui", system_ui),
+ ("non_pdd_overlay", non_pdd_overlay),
+ ("lock_screen", lock_screen),
+ ("other_pdd_page", _PRODUCT_PAGE),
+ ("anchor_missing", missing),
+ ("anchor_duplicate", duplicate),
+ ("anchor_drift", drifted),
+ ("dangerous_action", dangerous),
+ ):
+ with self.subTest(name=name), self.assertRaises(SkuSelectionError):
+ _product_exit_projection(_parse_nodes(hierarchy))
+
+ def test_safe_exit_requires_two_consecutive_evidence_projections(self) -> None:
+ device = _RawDevice(_M_FIXTURE.read_text(encoding="utf-8"))
+ flow = _flow(device)
+ flow.exit_sku_panel_safely()
+ self.assertEqual(len(_actions(device, "pressKey")), 1)
+ self.assertGreaterEqual(len(_actions(device, "dumpWindowHierarchy")), 3)
+ with self.assertRaises(SkuSelectionError):
+ flow.exit_sku_panel_safely()
+ self.assertEqual(len(_actions(device, "pressKey")), 1)
+
+ def test_safe_exit_single_hit_then_drift_resets_stability(self) -> None:
+ class SequenceAfterBack(_RawDevice):
+ def __init__(self) -> None:
+ super().__init__(_M_FIXTURE.read_text(encoding="utf-8"))
+ self.frames: list[str] = []
+
+ def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+ if method == "pressKey":
+ self.calls.append(("jsonrpc", method, params, timeout))
+ self.frames = [_PRODUCT_EXIT, "", _PRODUCT_EXIT, _PRODUCT_EXIT]
+ return ""
+ if method == "dumpWindowHierarchy" and self.frames:
+ self.hierarchy = self.frames.pop(0)
+ return super().jsonrpc_call(method, params, timeout)
+
+ device = SequenceAfterBack()
+ _flow(device, 0.08).exit_sku_panel_safely()
+ self.assertEqual(len(_actions(device, "pressKey")), 1)
+ self.assertEqual(device.frames, [])
+
+ def test_safe_exit_negative_pages_fail_closed_after_one_back(self) -> None:
+ root = ElementTree.fromstring(_PRODUCT_EXIT)
+ title = next(
+ node
+ for node in root.iter("node")
+ if node.get("resource-id") == "com.xunmeng.pinduoduo:id/tv_title"
+ )
+ root.append(ElementTree.fromstring(ElementTree.tostring(title, encoding="unicode")))
+ duplicate = ElementTree.tostring(root, encoding="unicode")
+ dangerous = _PRODUCT_EXIT.replace(
+ "",
+ '',
+ )
+
+ class FixedPostState(_RawDevice):
+ def __init__(self, post: str) -> None:
+ super().__init__(_M_FIXTURE.read_text(encoding="utf-8"))
+ self.post = post
+
+ def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+ if method == "pressKey":
+ self.calls.append(("jsonrpc", method, params, timeout))
+ self.hierarchy = self.post
+ return ""
+ return super().jsonrpc_call(method, params, timeout)
+
+ cases = (
+ ("empty", ""),
+ ("panel", _M_FIXTURE.read_text(encoding="utf-8")),
+ ("system_ui", _PRODUCT_EXIT.replace("com.xunmeng.pinduoduo", "com.android.systemui")),
+ (
+ "non_pdd_overlay",
+ _PRODUCT_EXIT.replace(
+ "",
+ '',
+ ),
+ ),
+ ("other_pdd", _PRODUCT_PAGE),
+ ("missing", _PRODUCT_EXIT.replace("com.xunmeng.pinduoduo:id/tv_title", "missing", 1)),
+ ("duplicate", duplicate),
+ ("drift", _PRODUCT_EXIT.replace("[36,1571][1044,1689]", "[35,1571][1044,1689]", 1)),
+ ("danger", dangerous),
+ )
+ for name, post in cases:
+ with self.subTest(name=name):
+ device = FixedPostState(post)
+ with self.assertRaises(SkuSelectionError):
+ _flow(device).exit_sku_panel_safely()
+ self.assertEqual(len(_actions(device, "pressKey")), 1)
+
+ version_drift = FixedPostState(_PRODUCT_EXIT)
+ original_app_info = version_drift.app_info
+
+ def drifted_app_info(package_name: str) -> dict[str, str]:
+ value = original_app_info(package_name)
+ if _actions(version_drift, "pressKey"):
+ value["versionName"] = "8.18.0"
+ return value
+
+ version_drift.app_info = drifted_app_info # type: ignore[method-assign]
+ with self.assertRaises(SkuSelectionError):
+ _flow(version_drift).exit_sku_panel_safely()
+ self.assertEqual(len(_actions(version_drift, "pressKey")), 1)
+
+ foreground_drift = FixedPostState(_PRODUCT_EXIT)
+ original_app_current = foreground_drift.app_current
+
+ def drifted_app_current() -> dict[str, str]:
+ value = original_app_current()
+ if _actions(foreground_drift, "pressKey"):
+ value["package"] = "external.app"
+ return value
+
+ foreground_drift.app_current = drifted_app_current # type: ignore[method-assign]
+ with self.assertRaises(SkuSelectionError):
+ _flow(foreground_drift).exit_sku_panel_safely()
+ self.assertEqual(len(_actions(foreground_drift, "pressKey")), 1)
+
class _CompletedFlow:
"""仅隔离 runner 文件发布测试，同时必须形成完整动作审计链。"""
@@ -1212,6 +1380,7 @@ class SkuSelectionRunnerTests(unittest.TestCase):
manifest = result.manifest_path.read_text(encoding="utf-8")
manifest_data = json.loads(manifest)
self.assertTrue(result.screenshot_path.is_file())
+ self.assertEqual(result.screenshot_path, target / "screenshot.png")
self.assertNotIn("device-1", manifest)
self.assertNotIn("hierarchy", manifest)
self.assertNotIn("已选", manifest)
@@ -1220,8 +1389,13 @@ class SkuSelectionRunnerTests(unittest.TestCase):
self.assertIn('"panel_status": "verified_before_back"', manifest)
self.assertIn('"back_attempts": 1', manifest)
self.assertIn('"back_rpc_outcome": "completed"', manifest)
- self.assertIn('"post_exit_status": "human_review_required"', manifest)
- self.assertNotIn('"safe_exit"', manifest)
+ self.assertIn('"post_exit_status": "same_product_verified"', manifest)
+ self.assertIn('"safe_exit": "completed"', manifest)
+ self.assertIn('"page_identity": "same_goods_evidence_bound"', manifest)
+ self.assertNotIn("human_review_required", manifest)
+ self.assertIsNotNone(result.captured_at.tzinfo)
+ self.assertEqual(result.captured_at.utcoffset().total_seconds(), 0)
+ self.assertEqual(manifest_data["captured_at"], result.captured_at.isoformat())
self.assertEqual(
manifest_data["actions"],
{