From e1e534c5d3a3bfaae47af220b9e412a5d2db5beb Mon Sep 17 00:00:00 2001
From: QiuSW <105186638@qq.com>
Date: Wed, 5 Aug 2026 15:20:09 +0800
Subject: [PATCH] fix(client): accept exact current-only sku state
---
client/scripts/capture_sku_reveal_spike.py | 24 +-
.../cmbuyer_client/pdd/sku_reveal_spike.py | 67 ++++-
.../src/cmbuyer_client/pdd/sku_selection.py | 249 +++++++++++++++---
...lected_size_hidden_current_only_8_17_0.xml | 59 +++++
.../sku_panel_empty_current_only_8_17_0.xml | 59 +++++
client/tests/pdd/test_sku_reveal_spike.py | 187 ++++++++++++-
client/tests/pdd/test_sku_selection.py | 228 +++++++++++++++-
docs/tasks/T-103.md | 6 +-
8 files changed, 821 insertions(+), 58 deletions(-)
create mode 100644 client/tests/pdd/fixtures/sku_panel_color_selected_size_hidden_current_only_8_17_0.xml
create mode 100644 client/tests/pdd/fixtures/sku_panel_empty_current_only_8_17_0.xml
diff --git a/client/scripts/capture_sku_reveal_spike.py b/client/scripts/capture_sku_reveal_spike.py
index f853b0e..4eecaed 100644
--- a/client/scripts/capture_sku_reveal_spike.py
+++ b/client/scripts/capture_sku_reveal_spike.py
@@ -13,9 +13,17 @@ 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.sku_reveal_spike import SkuRevealSpikeCapturer, SkuRevealSpikeError
-from cmbuyer_client.pdd.sku_selection import EXPECTED_GOODS_ID, SkuSelectionError
-from cmbuyer_client.pdd.sku_selection_runner import SkuSelectionRunError, safe_failure_stage
+from cmbuyer_client.pdd.sku_reveal_spike import (
+ safe_reveal_failure_stage,
+ SkuRevealSpikeCapturer,
+ SkuRevealSpikeError,
+)
+from cmbuyer_client.pdd.sku_selection import (
+ EXPECTED_GOODS_ID,
+ _safe_sku_entry_failure_stage,
+ SkuSelectionError,
+)
+from cmbuyer_client.pdd.sku_selection_runner import SkuSelectionRunError
def parse_arguments(argv: list[str] | None = None) -> argparse.Namespace:
@@ -42,6 +50,14 @@ def validate_arguments(arguments: argparse.Namespace) -> None:
raise ValueError("--timeout 必须是大于 0 的有限数值。")
+def _safe_capture_failure_stage(error: BaseException) -> str:
+ # 入口与 reveal 各有不可伪造的正式 marker;入口优先,不能被外层 reveal 标注覆盖。
+ entry_stage = _safe_sku_entry_failure_stage(error)
+ if entry_stage is not None:
+ return entry_stage
+ return safe_reveal_failure_stage(error) or "unknown"
+
+
def main(argv: list[str] | None = None) -> int:
arguments = parse_arguments(argv)
try:
@@ -74,7 +90,7 @@ def main(argv: list[str] | None = None) -> int:
except (DeviceConnectionError, SkuSelectionError, SkuSelectionRunError, SkuRevealSpikeError) as error:
# 不回显页面正文、节点、serial、坐标、路径或第三方异常。
print(
- f"规格 reveal 取证失败:stage={safe_failure_stage(error)};已停止,未发布本地证据目录。",
+ f"规格 reveal 取证失败:stage={_safe_capture_failure_stage(error)};已停止,未发布本地证据目录。",
file=sys.stderr,
)
return 1
diff --git a/client/src/cmbuyer_client/pdd/sku_reveal_spike.py b/client/src/cmbuyer_client/pdd/sku_reveal_spike.py
index 589fca6..b71bd3e 100644
--- a/client/src/cmbuyer_client/pdd/sku_reveal_spike.py
+++ b/client/src/cmbuyer_client/pdd/sku_reveal_spike.py
@@ -66,12 +66,48 @@ _TARGET_URL = f"https://mobile.yangkeduo.com/goods.html?goods_id={EXPECTED_GOODS
_REVEAL_START = (360, 1900)
_REVEAL_END = (360, 1300)
_REVEAL_STEPS = 30
+_REVEAL_FAILURE_STAGES = frozenset(
+ (
+ "reveal_precondition",
+ "reveal_attempted",
+ "reveal_candidate",
+ "reveal_after",
+ "reveal_publish",
+ )
+)
+_REVEAL_FAILURE_MARKER = object()
class SkuRevealSpikeError(RuntimeError):
"""一次性 reveal 取证未形成可发布证据。"""
+def _annotate_reveal_failure(error: BaseException, stage: str) -> None:
+ """只记录本模块实际走到的固定阶段;入口阶段使用独立 marker,不会被覆盖。"""
+
+ if type(stage) is not str or stage not in _REVEAL_FAILURE_STAGES:
+ return
+ try:
+ setattr(error, "_cmbuyer_reveal_failure_stage", stage)
+ # marker 最后写入,任一 setter 失败都不能形成可信阶段。
+ setattr(error, "_cmbuyer_reveal_failure_marker", _REVEAL_FAILURE_MARKER)
+ except BaseException:
+ pass
+
+
+def safe_reveal_failure_stage(error: BaseException) -> str | None:
+ """读取可公开的 reveal 控制流阶段;伪造属性或 hostile getter 均失败闭合。"""
+
+ try:
+ marker = getattr(error, "_cmbuyer_reveal_failure_marker", None)
+ stage = getattr(error, "_cmbuyer_reveal_failure_stage", None)
+ if marker is not _REVEAL_FAILURE_MARKER or type(stage) is not str:
+ return None
+ return stage if stage in _REVEAL_FAILURE_STAGES else None
+ except BaseException:
+ return None
+
+
@dataclass(frozen=True)
class SkuRevealSpikeResult:
output_directory: Path
@@ -127,13 +163,15 @@ class SkuRevealSpikeCapturer:
goods_id: str,
output_directory: Path,
) -> SkuRevealSpikeResult:
- if type(goods_id) is not str or goods_id != EXPECTED_GOODS_ID:
- raise SkuRevealSpikeError("商品不是 T-103 已取证目标,已停止取证。")
- link = parse_product_url(_TARGET_URL)
- target = Path(output_directory)
- _validate_new_target(target)
+ stage = "reveal_precondition"
staging: Path | None = None
+ adapter: _RevealEvidenceAdapter | None = None
try:
+ if type(goods_id) is not str or goods_id != EXPECTED_GOODS_ID:
+ raise SkuRevealSpikeError("商品不是 T-103 已取证目标,已停止取证。")
+ link = parse_product_url(_TARGET_URL)
+ target = Path(output_directory)
+ _validate_new_target(target)
staging = _prepare_staging(target)
deadline = self._clock() + self._timeout_seconds
inspection = self._adb_client.inspect(serial)
@@ -180,12 +218,16 @@ class SkuRevealSpikeCapturer:
(staging / "before" / "hierarchy.xml").write_text(before_hierarchy, encoding="utf-8")
rpc_outcome = "completed"
+ # 此后即属于 attempted:adapter 会在 RPC 前封存唯一机会,结果不明也只能调和。
+ stage = "reveal_attempted"
try:
adapter.reveal_size_options_once()
except SkuSelectionRunError:
rpc_outcome = "ambiguous_reconciled"
+ stage = "reveal_candidate"
projection, after_hierarchy = self._wait_for_candidate(adapter, deadline)
+ stage = "reveal_after"
after_directory = staging / "after"
_capture_frame(adapter, after_directory, after_hierarchy)
reverified = adapter.dump_window_hierarchy()
@@ -193,6 +235,7 @@ class SkuRevealSpikeCapturer:
raise SkuRevealSpikeError("截图后候选状态漂移,未发布证据。")
(after_directory / "hierarchy.xml").write_text(reverified, encoding="utf-8")
+ stage = "reveal_publish"
manifest = _manifest(inspection, serial, rpc_outcome, staging)
manifest_path = staging / "manifest.json"
manifest_path.write_text(
@@ -201,12 +244,22 @@ class SkuRevealSpikeCapturer:
)
os.rename(staging, target)
staging = None
- except (DeviceConnectionError, SkuSelectionError, SkuSelectionRunError, SkuRevealSpikeError):
+ except (DeviceConnectionError, SkuSelectionError, SkuSelectionRunError, SkuRevealSpikeError) as error:
_clean_staging(staging)
+ failure_stage = stage
+ if stage == "reveal_attempted" and (adapter is None or not adapter.reveal_attempted):
+ # attempted 只能在 adapter 已于 RPC 前封存唯一机会后成立。
+ failure_stage = "reveal_precondition"
+ _annotate_reveal_failure(error, failure_stage)
raise
except Exception as error:
_clean_staging(staging)
- raise SkuRevealSpikeError("规格 reveal 取证未完成,未发布本地证据目录。") from error
+ mapped = SkuRevealSpikeError("规格 reveal 取证未完成,未发布本地证据目录。")
+ failure_stage = stage
+ if stage == "reveal_attempted" and (adapter is None or not adapter.reveal_attempted):
+ failure_stage = "reveal_precondition"
+ _annotate_reveal_failure(mapped, failure_stage)
+ raise mapped from error
return SkuRevealSpikeResult(target, target / "manifest.json")
diff --git a/client/src/cmbuyer_client/pdd/sku_selection.py b/client/src/cmbuyer_client/pdd/sku_selection.py
index 362e993..584c62e 100644
--- a/client/src/cmbuyer_client/pdd/sku_selection.py
+++ b/client/src/cmbuyer_client/pdd/sku_selection.py
@@ -3,6 +3,7 @@ from __future__ import annotations
from dataclasses import dataclass
from enum import Enum
+from functools import partial
import re
from time import monotonic, sleep
from typing import Any, Callable, Protocol
@@ -126,15 +127,32 @@ class _PanelProfile(Enum):
@dataclass(frozen=True)
-class _PanelSpec:
- profile: _PanelProfile
- header_bounds: str
- outer_bounds: str
- price_row_bounds: str
+class _DualPriceLayout:
+ row_bounds: str
current_text: str
current_bounds: str
original_text: str
original_bounds: str
+
+
+@dataclass(frozen=True)
+class _CurrentOnlyPriceLayout:
+ # 同一 App 版本会把原价整段移除;这是独立证据布局,不是“原价可选”。
+ frame_bounds: str
+ row_bounds: str
+ current_text: str
+ current_bounds: str
+
+
+_PriceLayout = _DualPriceLayout | _CurrentOnlyPriceLayout
+
+
+@dataclass(frozen=True)
+class _PanelSpec:
+ profile: _PanelProfile
+ header_bounds: str
+ outer_bounds: str
+ price_layout: _PriceLayout
summary_text: str
summary_bounds: str
color_label_bounds: str | None
@@ -149,8 +167,10 @@ _PANEL_SPECS = (
_PanelSpec(
_PanelProfile.PANEL_OPEN_EMPTY,
"[0,366][1080,1077]", "[0,1077][1080,2079]",
- "[396,575][912,647]", "限1件 ¥12.88 ", "[396,580][675,647]",
- "券前¥29.88", "[693,580][912,647]",
+ _DualPriceLayout(
+ "[396,575][912,647]", "限1件 ¥12.88 ", "[396,580][675,647]",
+ "券前¥29.88", "[693,580][912,647]",
+ ),
_EMPTY_SUMMARY, "[396,731][1053,793]", "[36,1106][192,1159]",
"[36,1188][1080,2046]", "[372,1188][684,1587]", False,
"[36,2069][114,2079]", None,
@@ -158,8 +178,32 @@ _PANEL_SPECS = (
_PanelSpec(
_PanelProfile.COLOR_SELECTED_SIZE_HIDDEN,
"[0,366][1080,1077]", "[0,1077][1080,2079]",
- "[396,575][912,647]", "限1件 ¥12.88 ", "[396,580][675,647]",
- "券前¥29.88", "[693,580][912,647]",
+ _DualPriceLayout(
+ "[396,575][912,647]", "限1件 ¥12.88 ", "[396,580][675,647]",
+ "券前¥29.88", "[693,580][912,647]",
+ ),
+ _COLOR_ONLY_SUMMARY, "[396,731][1053,793]", "[36,1106][192,1159]",
+ "[36,1188][1080,2046]", "[372,1188][684,1587]", True,
+ "[36,2069][114,2079]", None,
+ ),
+ _PanelSpec(
+ _PanelProfile.PANEL_OPEN_EMPTY,
+ "[0,366][1080,1077]", "[0,1077][1080,2079]",
+ _CurrentOnlyPriceLayout(
+ "[396,575][1053,647]", "[396,575][693,647]",
+ "限1件 ¥12.88 ", "[396,580][675,647]",
+ ),
+ _EMPTY_SUMMARY, "[396,731][1053,793]", "[36,1106][192,1159]",
+ "[36,1188][1080,2046]", "[372,1188][684,1587]", False,
+ "[36,2069][114,2079]", None,
+ ),
+ _PanelSpec(
+ _PanelProfile.COLOR_SELECTED_SIZE_HIDDEN,
+ "[0,366][1080,1077]", "[0,1077][1080,2079]",
+ _CurrentOnlyPriceLayout(
+ "[396,575][1053,647]", "[396,575][693,647]",
+ "限1件 ¥12.88 ", "[396,580][675,647]",
+ ),
_COLOR_ONLY_SUMMARY, "[396,731][1053,793]", "[36,1106][192,1159]",
"[36,1188][1080,2046]", "[372,1188][684,1587]", True,
"[36,2069][114,2079]", None,
@@ -167,8 +211,10 @@ _PANEL_SPECS = (
_PanelSpec(
_PanelProfile.SIZE_VISIBLE_NON_TARGET,
"[0,366][1080,1000]", "[0,1000][1080,2079]",
- "[396,498][895,570]", _ROLLED_CURRENT_PRICE, "[396,503][712,570]",
- _ROLLED_ORIGINAL_PRICE, "[730,503][895,570]",
+ _DualPriceLayout(
+ "[396,498][895,570]", _ROLLED_CURRENT_PRICE, "[396,503][712,570]",
+ _ROLLED_ORIGINAL_PRICE, "[730,503][895,570]",
+ ),
_S_SUMMARY, "[396,654][1053,716]", None,
"[36,1000][1080,1483]", "[372,1000][684,1024]", True,
"[36,1506][114,1552]", _S_SIZE_UI,
@@ -176,8 +222,10 @@ _PANEL_SPECS = (
_PanelSpec(
_PanelProfile.TARGETS_SELECTED,
"[0,366][1080,1000]", "[0,1000][1080,2079]",
- "[396,498][895,570]", _ROLLED_CURRENT_PRICE, "[396,503][712,570]",
- _ROLLED_ORIGINAL_PRICE, "[730,503][895,570]",
+ _DualPriceLayout(
+ "[396,498][895,570]", _ROLLED_CURRENT_PRICE, "[396,503][712,570]",
+ _ROLLED_ORIGINAL_PRICE, "[730,503][895,570]",
+ ),
_TARGET_SUMMARY, "[396,654][1053,716]", None,
"[36,1000][1080,1483]", "[372,1000][684,1024]", True,
"[36,1506][114,1552]", _TARGET_SIZE_UI,
@@ -235,15 +283,21 @@ class SkuSelectionFlow:
self._require_foreground()
before = self._read_hierarchy()
nodes = _parse_nodes(before)
- profile = _classify_panel(nodes)
+ spec = _classify_panel(nodes)
+ profile = spec.profile
if profile is _PanelProfile.PANEL_OPEN_EMPTY:
target = _target_color_action(nodes, selected=False)
_action_bounds(target.bounds)
_require_action_occupants(nodes, target)
- self._pending = (before, _require_color_only_panel)
+ color_postcondition = partial(
+ _require_profile_with_price_layout,
+ expected=_PanelProfile.COLOR_SELECTED_SIZE_HIDDEN,
+ expected_price_layout=spec.price_layout,
+ )
+ self._pending = (before, color_postcondition)
self._device.tap_sku_option(target.bounds)
- self._wait_after_action(before, _require_color_only_panel)
+ self._wait_after_action(before, color_postcondition)
# 当前证据只证明颜色选择;尺码仍在视口外。没有动作证据时必须在此停住,
# 不能把一次通用 swipe 或下一次规格点击伪装成已验证流程。
raise SkuSelectionError(_REVEAL_NOT_PROVEN)
@@ -383,14 +437,14 @@ def _parse_nodes(raw: str) -> list[_Node]:
return result
-def _classify_panel(nodes: list[_Node]) -> _PanelProfile:
- matches: list[_PanelProfile] = []
+def _classify_panel(nodes: list[_Node]) -> _PanelSpec:
+ matches: list[_PanelSpec] = []
for spec in _PANEL_SPECS:
try:
_match_panel_profile(nodes, spec)
except SkuSelectionError:
continue
- matches.append(spec.profile)
+ matches.append(spec)
if len(matches) != 1:
raise SkuSelectionError("规格面板不符合唯一完整取证 profile,已停止操作。")
return matches[0]
@@ -408,9 +462,23 @@ def _require_target_panel(nodes: list[_Node]) -> None:
_require_profile(nodes, _PanelProfile.TARGETS_SELECTED)
-def _require_profile(nodes: list[_Node], expected: _PanelProfile) -> None:
- if _classify_panel(nodes) is not expected:
+def _require_profile(nodes: list[_Node], expected: _PanelProfile) -> _PanelSpec:
+ spec = _classify_panel(nodes)
+ if spec.profile is not expected:
raise SkuSelectionError("规格面板动作后状态与已取证 profile 不一致,已停止操作。")
+ return spec
+
+
+def _require_profile_with_price_layout(
+ nodes: list[_Node],
+ *,
+ expected: _PanelProfile,
+ expected_price_layout: _PriceLayout,
+) -> None:
+ spec = _require_profile(nodes, expected)
+ # 同一次颜色点击不得跨价格证据 variant;完整 dataclass identity 同时绑定类型与全部 exact 字段。
+ if spec.price_layout != expected_price_layout:
+ raise SkuSelectionError("规格面板动作后价格布局发生跨变体漂移,已停止操作。")
def _match_panel_profile(nodes: list[_Node], spec: _PanelSpec) -> None:
@@ -427,24 +495,24 @@ def _match_panel_profile(nodes: list[_Node], spec: _PanelSpec) -> None:
[node for node in nodes if node.parent is surface and _exact_recycler(node, spec.outer_bounds)],
"规格面板维度容器不唯一。",
)
- price_row = _one(
- [node for node in nodes if _descendant(node, header) and _exact_inert(node, "android.widget.LinearLayout", spec.price_row_bounds)],
- "规格面板价格行不唯一。",
- )
- if len([child for child in price_row.element if child.tag == "node"]) != 2:
- raise SkuSelectionError("规格面板价格行子节点数量漂移。")
- current = _one(
- [node for node in nodes if node.parent is price_row and _exact_readonly_text(node, spec.current_text, spec.current_bounds)],
- "规格面板当前价角色不唯一。",
- )
- _one(
- [node for node in nodes if node.parent is price_row and _exact_readonly_text(node, spec.original_text, spec.original_bounds)],
- "规格面板原价角色不唯一。",
+ current, summary_anchor, summary_is_direct = _match_price_layout(
+ nodes,
+ header,
+ spec.price_layout,
)
if _clickable_before(current, surface):
raise SkuSelectionError("规格面板价格角色位于可点击内容祖先下。")
_one(
- [node for node in nodes if _descendant(node, header) and _exact_readonly_text(node, spec.summary_text, spec.summary_bounds)],
+ [
+ node
+ for node in nodes
+ if (
+ node.parent is summary_anchor
+ if summary_is_direct
+ else _descendant(node, summary_anchor)
+ )
+ and _exact_readonly_text(node, spec.summary_text, spec.summary_bounds)
+ ],
"规格面板摘要不唯一。",
)
@@ -507,11 +575,13 @@ def _match_panel_profile(nodes: list[_Node], spec: _PanelSpec) -> None:
def _unit_price(nodes: list[_Node]) -> str:
- _require_profile(nodes, _PanelProfile.TARGETS_SELECTED)
- spec = _spec_for(_PanelProfile.TARGETS_SELECTED)
+ spec = _classify_panel(nodes)
+ if spec.profile is not _PanelProfile.TARGETS_SELECTED:
+ raise SkuSelectionError("规格面板动作后状态与已取证 profile 不一致,已停止操作。")
+ current_bounds = spec.price_layout.current_bounds
candidates = [
node for node in nodes
- if _exact_readonly_text(node, _ROLLED_CURRENT_PRICE, spec.current_bounds)
+ if _exact_readonly_text(node, _ROLLED_CURRENT_PRICE, current_bounds)
]
current = _one(candidates, "规格面板现价不唯一,已停止读取。")
surface = _one([node for node in nodes if _exact_inert(node, "android.view.ViewGroup", _PANEL_SURFACE)], "规格面板内容面不唯一。")
@@ -521,16 +591,16 @@ def _unit_price(nodes: list[_Node]) -> str:
def _target_color_action(nodes: list[_Node], *, selected: bool) -> _Node:
- profile = _classify_panel(nodes)
+ spec = _classify_panel(nodes)
+ profile = spec.profile
expected_profile = _PanelProfile.COLOR_SELECTED_SIZE_HIDDEN if selected else _PanelProfile.PANEL_OPEN_EMPTY
if profile is not expected_profile:
raise SkuSelectionError("目标颜色 action 不属于预期 profile。")
- spec = _spec_for(profile)
return _one([node for node in nodes if _exact_color_action(node, spec.color_bounds, selected)], "目标颜色 action 不唯一。")
def _target_size_action(nodes: list[_Node], *, selected: bool) -> _Node:
- profile = _classify_panel(nodes)
+ profile = _classify_panel(nodes).profile
expected_profile = _PanelProfile.TARGETS_SELECTED if selected else _PanelProfile.SIZE_VISIBLE_NON_TARGET
if profile is not expected_profile:
raise SkuSelectionError("目标尺码 action 不属于预期 profile。")
@@ -541,7 +611,102 @@ def _target_size_action(nodes: list[_Node], *, selected: bool) -> _Node:
def _spec_for(profile: _PanelProfile) -> _PanelSpec:
- return next(spec for spec in _PANEL_SPECS if spec.profile is profile)
+ matches = [spec for spec in _PANEL_SPECS if spec.profile is profile]
+ if len(matches) != 1:
+ raise SkuSelectionError("该 profile 没有唯一静态 spec。")
+ return matches[0]
+
+
+def _match_price_layout(
+ nodes: list[_Node],
+ header: _Node,
+ layout: _PriceLayout,
+) -> tuple[_Node, _Node, bool]:
+ if isinstance(layout, _DualPriceLayout):
+ price_row = _one(
+ [
+ node
+ for node in nodes
+ if _descendant(node, header)
+ and _exact_inert(node, "android.widget.LinearLayout", layout.row_bounds)
+ ],
+ "规格面板双价格行不唯一。",
+ )
+ if len([child for child in price_row.element if child.tag == "node"]) != 2:
+ raise SkuSelectionError("规格面板双价格行子节点数量漂移。")
+ current = _one(
+ [
+ node
+ for node in nodes
+ if node.parent is price_row
+ and _exact_readonly_text(node, layout.current_text, layout.current_bounds)
+ ],
+ "规格面板双价格当前价角色不唯一。",
+ )
+ _one(
+ [
+ node
+ for node in nodes
+ if node.parent is price_row
+ and _exact_readonly_text(node, layout.original_text, layout.original_bounds)
+ ],
+ "规格面板双价格原价角色不唯一。",
+ )
+ # 历史已批准 raw 的双价格行和摘要位于 header 内的嵌套内容容器,
+ # 旧证据只允许后代关系,不能套用 current-only 的直接父链。
+ return current, header, False
+
+ frame = _one(
+ [
+ node
+ for node in nodes
+ if _descendant(node, header)
+ and _exact_inert(node, "android.widget.FrameLayout", layout.frame_bounds)
+ ],
+ "规格面板单价格外层不唯一。",
+ )
+ if len([child for child in frame.element if child.tag == "node"]) != 1:
+ raise SkuSelectionError("规格面板单价格外层子节点数量漂移。")
+ content = frame.parent
+ relative = content.parent if content is not None else None
+ content_frame = relative.parent if relative is not None else None
+ if (
+ content is None
+ or relative is None
+ or content_frame is None
+ or content_frame.parent is not header
+ or not _exact_inert(content, "android.view.ViewGroup", "[0,551][1080,940]")
+ or not _exact_inert(relative, "android.widget.RelativeLayout", "[0,551][1080,940]")
+ or not _exact_inert(content_frame, "android.widget.FrameLayout", "[0,551][1080,940]")
+ ):
+ raise SkuSelectionError("规格面板单价格外层父链漂移。")
+ price_row = _one(
+ [
+ node
+ for node in nodes
+ if node.parent is frame
+ and _exact_inert(node, "android.widget.LinearLayout", layout.row_bounds)
+ ],
+ "规格面板单价格行父子关系漂移。",
+ )
+ if len([child for child in price_row.element if child.tag == "node"]) != 1:
+ raise SkuSelectionError("规格面板单价格行子节点数量漂移。")
+ current = _one(
+ [
+ node
+ for node in nodes
+ if node.parent is price_row
+ and _exact_readonly_text(node, layout.current_text, layout.current_bounds)
+ ],
+ "规格面板单价格当前价角色不唯一。",
+ )
+ if any(
+ _descendant(node, header)
+ and node.text == "券前¥29.88"
+ for node in nodes
+ ):
+ raise SkuSelectionError("规格面板单价格变体出现原价角色。")
+ return current, content, True
def _exact_inert(node: _Node, class_name: str, bounds: str) -> bool:
diff --git a/client/tests/pdd/fixtures/sku_panel_color_selected_size_hidden_current_only_8_17_0.xml b/client/tests/pdd/fixtures/sku_panel_color_selected_size_hidden_current_only_8_17_0.xml
new file mode 100644
index 0000000..9a49a2a
--- /dev/null
+++ b/client/tests/pdd/fixtures/sku_panel_color_selected_size_hidden_current_only_8_17_0.xml
@@ -0,0 +1,59 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/client/tests/pdd/fixtures/sku_panel_empty_current_only_8_17_0.xml b/client/tests/pdd/fixtures/sku_panel_empty_current_only_8_17_0.xml
new file mode 100644
index 0000000..4a1660e
--- /dev/null
+++ b/client/tests/pdd/fixtures/sku_panel_empty_current_only_8_17_0.xml
@@ -0,0 +1,59 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/client/tests/pdd/test_sku_reveal_spike.py b/client/tests/pdd/test_sku_reveal_spike.py
index 99bc79f..9f662bf 100644
--- a/client/tests/pdd/test_sku_reveal_spike.py
+++ b/client/tests/pdd/test_sku_reveal_spike.py
@@ -14,8 +14,12 @@ from xml.etree import ElementTree
from PIL import Image
+import cmbuyer_client.pdd.sku_reveal_spike as reveal_module
from cmbuyer_client.device.adb import AdbDevice, DeviceInspection
from cmbuyer_client.pdd.sku_reveal_spike import (
+ _annotate_reveal_failure,
+ _RevealEvidenceAdapter,
+ safe_reveal_failure_stage,
SkuRevealSpikeCapturer,
SkuRevealSpikeError,
_require_safe_reveal_path,
@@ -31,6 +35,8 @@ _FIXTURES = Path(__file__).with_name("fixtures")
_PRODUCT = (_FIXTURES / "product_entry_8_17_0.xml").read_text(encoding="utf-8")
_EMPTY = (_FIXTURES / "sku_panel_empty_8_17_0.xml").read_text(encoding="utf-8")
_COLOR_ONLY = (_FIXTURES / "sku_panel_color_selected_size_hidden_8_17_0.xml").read_text(encoding="utf-8")
+_CURRENT_ONLY_EMPTY = (_FIXTURES / "sku_panel_empty_current_only_8_17_0.xml").read_text(encoding="utf-8")
+_CURRENT_ONLY_COLOR = (_FIXTURES / "sku_panel_color_selected_size_hidden_current_only_8_17_0.xml").read_text(encoding="utf-8")
_M_SELECTED = (_FIXTURES / "sku_panel_size_m_restored_8_17_0.xml").read_text(encoding="utf-8")
@@ -171,6 +177,8 @@ def _png() -> str:
class _FakeDevice:
def __init__(self) -> None:
self.hierarchy = ""
+ self.empty_hierarchy = _EMPTY
+ self.color_hierarchy = _COLOR_ONLY
self.after_hierarchy = _candidate_unselected()
self.calls: list[tuple[object, ...]] = []
self.swipe_error = False
@@ -195,10 +203,10 @@ class _FakeDevice:
return _png()
if method == "click":
if params == [865, 2218]:
- self.hierarchy = _EMPTY
+ self.hierarchy = self.empty_hierarchy
return ""
if params == [528, 1387]:
- self.hierarchy = _COLOR_ONLY
+ self.hierarchy = self.color_hierarchy
return ""
raise AssertionError(params)
if method == "swipe":
@@ -282,6 +290,50 @@ class SkuRevealSpikeTests(unittest.TestCase):
self.assertNotIn("start", manifest)
self.assertNotIn("end", manifest)
+ def test_current_only_empty_and_color_only_are_accepted_before_reveal(self) -> None:
+ device = _FakeDevice()
+ device.empty_hierarchy = _CURRENT_ONLY_EMPTY
+ device.color_hierarchy = _CURRENT_ONLY_COLOR
+ with TemporaryDirectory() as directory:
+ target = Path(directory) / "evidence"
+ result = self._capturer(device).capture(
+ "192.168.0.173:5555",
+ "937122477375",
+ target,
+ )
+
+ self.assertEqual(result.output_directory, target)
+ self.assertEqual(len(_swipes(device)), 1)
+ self.assertEqual(
+ (target / "before" / "hierarchy.xml").read_text(encoding="utf-8"),
+ _CURRENT_ONLY_COLOR,
+ )
+
+ def test_cross_price_variant_after_color_click_is_precondition_failure_without_reveal(self) -> None:
+ for empty_hierarchy, color_hierarchy in (
+ (_CURRENT_ONLY_EMPTY, _COLOR_ONLY),
+ (_EMPTY, _CURRENT_ONLY_COLOR),
+ ):
+ with self.subTest():
+ device = _FakeDevice()
+ device.empty_hierarchy = empty_hierarchy
+ device.color_hierarchy = color_hierarchy
+ with TemporaryDirectory() as directory:
+ target = Path(directory) / "evidence"
+ with self.assertRaises(SkuSelectionError) as raised:
+ self._capturer(device).capture(
+ "192.168.0.173:5555",
+ "937122477375",
+ target,
+ )
+ self.assertEqual(
+ safe_reveal_failure_stage(raised.exception),
+ "reveal_precondition",
+ )
+ self.assertEqual(_swipes(device), [])
+ self.assertFalse(target.exists())
+ self.assertEqual(list(Path(directory).glob(".*.staging-*")), [])
+
def test_ambiguous_rpc_is_read_only_reconciled_without_retry(self) -> None:
device = _FakeDevice()
device.swipe_error = True
@@ -367,7 +419,7 @@ class SkuRevealSpikeTests(unittest.TestCase):
device.jsonrpc_call = drift # type: ignore[method-assign]
with TemporaryDirectory() as directory:
target = Path(directory) / "evidence"
- with self.assertRaises(SkuSelectionError):
+ with self.assertRaises(SkuSelectionError) as raised:
self._capturer(device).capture(
"192.168.0.173:5555",
"937122477375",
@@ -375,6 +427,89 @@ class SkuRevealSpikeTests(unittest.TestCase):
)
self.assertFalse(target.exists())
self.assertEqual(_swipes(device), [])
+ self.assertEqual(safe_reveal_failure_stage(raised.exception), "reveal_precondition")
+
+ def test_reveal_failure_stages_bind_attempt_candidate_after_and_publish(self) -> None:
+ device = _FakeDevice()
+
+ def fail_before_seal(adapter: _RevealEvidenceAdapter) -> None:
+ raise SkuRevealSpikeError("private pre-seal detail")
+
+ with TemporaryDirectory() as directory:
+ target = Path(directory) / "pre-seal"
+ with (
+ patch.object(_RevealEvidenceAdapter, "reveal_size_options_once", fail_before_seal),
+ self.assertRaises(SkuRevealSpikeError) as pre_seal,
+ ):
+ self._capturer(device).capture("192.168.0.173:5555", "937122477375", target)
+ self.assertEqual(safe_reveal_failure_stage(pre_seal.exception), "reveal_precondition")
+ self.assertEqual(_swipes(device), [])
+ self.assertFalse(target.exists())
+ self.assertEqual(list(Path(directory).glob(".*.staging-*")), [])
+
+ # attempted 只可能在 adapter 已封存并发送唯一手势后报告。
+ device = _FakeDevice()
+ original_reveal = _RevealEvidenceAdapter.reveal_size_options_once
+
+ def fail_after_attempt(adapter: _RevealEvidenceAdapter) -> None:
+ original_reveal(adapter)
+ raise SkuRevealSpikeError("private attempted detail")
+
+ with TemporaryDirectory() as directory:
+ target = Path(directory) / "attempted"
+ with (
+ patch.object(_RevealEvidenceAdapter, "reveal_size_options_once", fail_after_attempt),
+ self.assertRaises(SkuRevealSpikeError) as attempted,
+ ):
+ self._capturer(device).capture("192.168.0.173:5555", "937122477375", target)
+ self.assertEqual(safe_reveal_failure_stage(attempted.exception), "reveal_attempted")
+ self.assertEqual(len(_swipes(device)), 1)
+ self.assertFalse(target.exists())
+ self.assertEqual(list(Path(directory).glob(".*.staging-*")), [])
+
+ device = _FakeDevice()
+ device.after_hierarchy = _COLOR_ONLY
+ with TemporaryDirectory() as directory:
+ target = Path(directory) / "candidate"
+ with self.assertRaises(SkuRevealSpikeError) as candidate:
+ self._capturer(device).capture("192.168.0.173:5555", "937122477375", target)
+ self.assertEqual(safe_reveal_failure_stage(candidate.exception), "reveal_candidate")
+ self.assertEqual(len(_swipes(device)), 1)
+ self.assertFalse(target.exists())
+ self.assertEqual(list(Path(directory).glob(".*.staging-*")), [])
+
+ device = _FakeDevice()
+ original_capture = reveal_module._capture_frame
+
+ def fail_after_capture(adapter: object, directory: Path, hierarchy: str) -> None:
+ if directory.name == "after":
+ raise OSError("private after path")
+ original_capture(adapter, directory, hierarchy)
+
+ with TemporaryDirectory() as directory:
+ target = Path(directory) / "after"
+ with (
+ patch.object(reveal_module, "_capture_frame", fail_after_capture),
+ self.assertRaises(SkuRevealSpikeError) as after,
+ ):
+ self._capturer(device).capture("192.168.0.173:5555", "937122477375", target)
+ self.assertEqual(safe_reveal_failure_stage(after.exception), "reveal_after")
+ self.assertEqual(len(_swipes(device)), 1)
+ self.assertFalse(target.exists())
+ self.assertEqual(list(Path(directory).glob(".*.staging-*")), [])
+
+ device = _FakeDevice()
+ with TemporaryDirectory() as directory:
+ target = Path(directory) / "publish"
+ with (
+ patch.object(reveal_module.os, "rename", side_effect=OSError("private publish path")),
+ self.assertRaises(SkuRevealSpikeError) as publish,
+ ):
+ self._capturer(device).capture("192.168.0.173:5555", "937122477375", target)
+ self.assertEqual(safe_reveal_failure_stage(publish.exception), "reveal_publish")
+ self.assertEqual(len(_swipes(device)), 1)
+ self.assertFalse(target.exists())
+ self.assertEqual(list(Path(directory).glob(".*.staging-*")), [])
def test_complete_reveal_segment_rejects_narrow_impostor_and_invalid_bounds(self) -> None:
for hierarchy in (
@@ -481,12 +616,58 @@ class SkuRevealSpikeCliTests(unittest.TestCase):
annotated = SkuSelectionError(secret)
_annotate_sku_entry_failure(annotated, "sku_entry_panel_verify")
+ _annotate_reveal_failure(annotated, "reveal_precondition")
self.assertIn("stage=sku_entry_panel_verify", run_with(annotated))
spoofed = SkuSelectionError(secret)
setattr(spoofed, "_cmbuyer_failure_stage", "sku_entry_panel_verify")
self.assertIn("stage=unknown", run_with(spoofed))
+ def test_cli_reports_only_formally_annotated_reveal_stage(self) -> None:
+ script = _load_reveal_script()
+ secret = "SERIAL=192.168.0.173:5555 private"
+
+ def run_with(error: BaseException) -> str:
+ class FailingCapturer:
+ def __init__(self, *args: object, **kwargs: object) -> None:
+ return None
+
+ def capture(self, *args: object, **kwargs: object) -> object:
+ raise error
+
+ stderr = io.StringIO()
+ with patch.object(script, "SkuRevealSpikeCapturer", FailingCapturer), redirect_stderr(stderr):
+ self.assertEqual(
+ script.main(
+ [
+ "--serial", "192.168.0.173:5555",
+ "--goods-id", "937122477375",
+ "--output-dir", "evidence",
+ ]
+ ),
+ 1,
+ )
+ output = stderr.getvalue()
+ self.assertNotIn("192.168.0.173:5555", output)
+ self.assertNotIn("private", output)
+ return output
+
+ annotated = SkuRevealSpikeError(secret)
+ _annotate_reveal_failure(annotated, "reveal_candidate")
+ self.assertIn("stage=reveal_candidate", run_with(annotated))
+
+ spoofed = SkuRevealSpikeError(secret)
+ setattr(spoofed, "_cmbuyer_reveal_failure_stage", "reveal_candidate")
+ self.assertIn("stage=unknown", run_with(spoofed))
+
+ class HostileGetterError(SkuRevealSpikeError):
+ def __getattribute__(self, name: str) -> object:
+ if name.startswith("_cmbuyer_reveal_"):
+ raise RuntimeError(secret)
+ return super().__getattribute__(name)
+
+ self.assertIn("stage=unknown", run_with(HostileGetterError(secret)))
+
def _load_reveal_script() -> object:
path = Path(__file__).resolve().parents[2] / "scripts" / "capture_sku_reveal_spike.py"
diff --git a/client/tests/pdd/test_sku_selection.py b/client/tests/pdd/test_sku_selection.py
index a9f43fa..586a60f 100644
--- a/client/tests/pdd/test_sku_selection.py
+++ b/client/tests/pdd/test_sku_selection.py
@@ -18,6 +18,7 @@ import cmbuyer_client.pdd.sku_selection_runner as runner_module
from cmbuyer_client.device.adb import AdbDevice, DeviceConnectionError, DeviceInspection
from cmbuyer_client.pdd import SkuSelectionError, SkuSelectionFlow, SkuSelectionRunner
from cmbuyer_client.pdd.sku_selection import (
+ _PanelProfile,
SkuPanelDevice,
_action_bounds,
_classify_panel,
@@ -36,6 +37,8 @@ from cmbuyer_client.pdd.sku_selection_runner import (
_FIXTURES = Path(__file__).with_name("fixtures")
_EMPTY_FIXTURE = _FIXTURES / "sku_panel_empty_8_17_0.xml"
_COLOR_FIXTURE = _FIXTURES / "sku_panel_color_selected_size_hidden_8_17_0.xml"
+_CURRENT_ONLY_EMPTY_FIXTURE = _FIXTURES / "sku_panel_empty_current_only_8_17_0.xml"
+_CURRENT_ONLY_COLOR_FIXTURE = _FIXTURES / "sku_panel_color_selected_size_hidden_current_only_8_17_0.xml"
_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
@@ -57,6 +60,7 @@ class _RawDevice:
def __init__(self, hierarchy: str = _PRODUCT_PAGE, screenshot: str | None = None) -> None:
self.hierarchy = hierarchy
self.panel_hierarchy = _EMPTY_FIXTURE.read_text(encoding="utf-8")
+ self.color_hierarchy = _COLOR_FIXTURE.read_text(encoding="utf-8")
self.version = "8.17.0"
self.package = "com.xunmeng.pinduoduo"
self.screenshot = _png_base64() if screenshot is None else screenshot
@@ -99,7 +103,7 @@ class _RawDevice:
self.hierarchy = (
_EMPTY_FIXTURE.read_text(encoding="utf-8")
if self.fail_color_readback
- else _COLOR_FIXTURE.read_text(encoding="utf-8")
+ else self.color_hierarchy
)
return
if (x, y) == (635, 1624):
@@ -177,6 +181,63 @@ def _mutate_unique_node(
return ElementTree.tostring(root, encoding="unicode")
+def _nested_dual_price_layout(
+ hierarchy: str,
+ *,
+ header_bounds: str,
+ row_bounds: str,
+ summary_bounds: str,
+ content_bounds: str,
+ price_frame_bounds: str,
+) -> str:
+ root = ElementTree.fromstring(hierarchy)
+ parents = {child: parent for parent in root.iter() for child in parent}
+ header = next(
+ node for node in root.iter("node")
+ if node.get("class") == "android.widget.LinearLayout"
+ and node.get("bounds") == header_bounds
+ )
+ row = next(
+ node for node in root.iter("node")
+ if node.get("class") == "android.widget.LinearLayout"
+ and node.get("bounds") == row_bounds
+ )
+ summary = next(
+ node for node in root.iter("node")
+ if node.get("class") == "android.widget.TextView"
+ and node.get("bounds") == summary_bounds
+ )
+ parents[row].remove(row)
+ parents[summary].remove(summary)
+
+ def inert(class_name: str, bounds: str) -> ElementTree.Element:
+ return ElementTree.Element(
+ "node",
+ {
+ "package": "com.xunmeng.pinduoduo",
+ "class": class_name,
+ "bounds": bounds,
+ "clickable": "false",
+ "enabled": "true",
+ "visible-to-user": "true",
+ "selected": "false",
+ "scrollable": "false",
+ },
+ )
+
+ content_frame = inert("android.widget.FrameLayout", content_bounds)
+ relative = inert("android.widget.RelativeLayout", content_bounds)
+ content = inert("android.view.ViewGroup", content_bounds)
+ price_frame = inert("android.widget.FrameLayout", price_frame_bounds)
+ header.append(content_frame)
+ content_frame.append(relative)
+ relative.append(content)
+ content.append(price_frame)
+ price_frame.append(row)
+ content.append(summary)
+ return ElementTree.tostring(root, encoding="unicode")
+
+
def _entry_chain(root: ElementTree.Element) -> list[ElementTree.Element]:
parents = {child: parent for parent in root.iter() for child in parent}
child = next(node for node in root.iter("node") if node.get("text") == "快要抢光 ¥ 12.88")
@@ -364,6 +425,169 @@ class SkuSelectionFlowTests(unittest.TestCase):
self.assertEqual(_tap_centers(restored), [(635, 1624)])
self.assertEqual(_actions(restored, "pressKey"), [("jsonrpc", "pressKey", ["back"], 10)])
+ def test_current_only_empty_to_color_only_uses_the_matching_exact_spec(self) -> None:
+ device = _RawDevice()
+ device.panel_hierarchy = _CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8")
+ device.color_hierarchy = _CURRENT_ONLY_COLOR_FIXTURE.read_text(encoding="utf-8")
+ flow = _flow(device)
+
+ flow.open_sku_panel(_TARGET_URL)
+ with self.assertRaisesRegex(SkuSelectionError, "受控显示动作尚未取证"):
+ flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
+
+ self.assertEqual(_tap_centers(device), [(865, 2218), (528, 1387)])
+ matched = _classify_panel(_parse_nodes(device.hierarchy))
+ self.assertIs(matched.profile, _PanelProfile.COLOR_SELECTED_SIZE_HIDDEN)
+ self.assertEqual(
+ matched.price_layout.__class__.__name__,
+ "_CurrentOnlyPriceLayout",
+ )
+
+ def test_dual_and_current_only_variants_each_match_one_complete_spec(self) -> None:
+ cases = (
+ (_EMPTY_FIXTURE, _PanelProfile.PANEL_OPEN_EMPTY, "_DualPriceLayout"),
+ (_COLOR_FIXTURE, _PanelProfile.COLOR_SELECTED_SIZE_HIDDEN, "_DualPriceLayout"),
+ (_CURRENT_ONLY_EMPTY_FIXTURE, _PanelProfile.PANEL_OPEN_EMPTY, "_CurrentOnlyPriceLayout"),
+ (_CURRENT_ONLY_COLOR_FIXTURE, _PanelProfile.COLOR_SELECTED_SIZE_HIDDEN, "_CurrentOnlyPriceLayout"),
+ )
+ for fixture, profile, layout_name in cases:
+ with self.subTest(fixture=fixture.name):
+ spec = _classify_panel(_parse_nodes(fixture.read_text(encoding="utf-8")))
+ self.assertIs(spec.profile, profile)
+ self.assertEqual(spec.price_layout.__class__.__name__, layout_name)
+
+ def test_dual_price_historical_nested_parent_chains_remain_accepted(self) -> None:
+ for fixture, expected_profile in (
+ (_EMPTY_FIXTURE, _PanelProfile.PANEL_OPEN_EMPTY),
+ (_COLOR_FIXTURE, _PanelProfile.COLOR_SELECTED_SIZE_HIDDEN),
+ ):
+ with self.subTest(fixture=fixture.name):
+ nested = _nested_dual_price_layout(
+ fixture.read_text(encoding="utf-8"),
+ header_bounds="[0,366][1080,1077]",
+ row_bounds="[396,575][912,647]",
+ summary_bounds="[396,731][1053,793]",
+ content_bounds="[0,551][1080,940]",
+ price_frame_bounds="[396,575][1053,647]",
+ )
+ spec = _classify_panel(_parse_nodes(nested))
+ self.assertIs(spec.profile, expected_profile)
+ self.assertEqual(spec.price_layout.__class__.__name__, "_DualPriceLayout")
+
+ nested_s = _nested_dual_price_layout(
+ _S_FIXTURE.read_text(encoding="utf-8"),
+ header_bounds="[0,366][1080,1000]",
+ row_bounds="[396,498][895,570]",
+ summary_bounds="[396,654][1053,716]",
+ content_bounds="[0,474][1080,863]",
+ price_frame_bounds="[396,498][1053,570]",
+ )
+ self.assertIs(
+ _classify_panel(_parse_nodes(nested_s)).profile,
+ _PanelProfile.SIZE_VISIBLE_NON_TARGET,
+ )
+
+ nested_m = _nested_dual_price_layout(
+ _M_FIXTURE.read_text(encoding="utf-8"),
+ header_bounds="[0,366][1080,1000]",
+ row_bounds="[396,498][895,570]",
+ summary_bounds="[396,654][1053,716]",
+ content_bounds="[0,474][1080,863]",
+ price_frame_bounds="[396,498][1053,570]",
+ )
+ self.assertEqual(_flow(_RawDevice(nested_m)).read_sku_unit_price(), "12.88")
+
+ def test_current_only_price_variant_rejects_cross_variant_duplicate_unknown_and_parent_drift(self) -> None:
+ base = _CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8")
+
+ def mutate(kind: str) -> str:
+ root = ElementTree.fromstring(base)
+ parents = {child: parent for parent in root.iter() for child in parent}
+ frame = next(
+ node for node in root.iter("node")
+ if node.get("class") == "android.widget.FrameLayout"
+ and node.get("bounds") == "[396,575][1053,647]"
+ )
+ row = next(
+ node for node in frame
+ if node.get("class") == "android.widget.LinearLayout"
+ and node.get("bounds") == "[396,575][693,647]"
+ )
+ current = next(node for node in row if node.get("text") == "限1件 ¥12.88 ")
+ if kind == "cross_variant":
+ ElementTree.SubElement(
+ row,
+ "node",
+ {
+ "text": "券前¥29.88",
+ "package": "com.xunmeng.pinduoduo",
+ "class": "android.widget.TextView",
+ "bounds": "[693,580][912,647]",
+ "clickable": "false",
+ "enabled": "true",
+ "visible-to-user": "true",
+ "selected": "false",
+ "scrollable": "false",
+ },
+ )
+ elif kind == "duplicate_frame":
+ parents[frame].append(ElementTree.fromstring(ElementTree.tostring(frame, encoding="unicode")))
+ elif kind == "duplicate_current":
+ row.append(ElementTree.fromstring(ElementTree.tostring(current, encoding="unicode")))
+ elif kind == "unknown_text":
+ current.set("text", "未知促销文字")
+ elif kind == "unknown_bounds":
+ frame.set("bounds", "[396,574][1053,647]")
+ elif kind == "unknown_child":
+ row.append(ElementTree.Element("node", dict(current.attrib) | {"text": "未知节点"}))
+ elif kind == "parent_drift":
+ frame.remove(row)
+ parents[frame].append(row)
+ elif kind == "summary_parent_drift":
+ summary = next(
+ node for node in root.iter("node")
+ if node.get("bounds") == "[396,731][1053,793]"
+ and node.get("class") == "android.widget.TextView"
+ )
+ parents[summary].remove(summary)
+ parents[parents[summary]].append(summary)
+ else:
+ raise AssertionError(kind)
+ return ElementTree.tostring(root, encoding="unicode")
+
+ for kind in (
+ "cross_variant",
+ "duplicate_frame",
+ "duplicate_current",
+ "unknown_text",
+ "unknown_bounds",
+ "unknown_child",
+ "parent_drift",
+ "summary_parent_drift",
+ ):
+ with self.subTest(kind=kind), self.assertRaises(SkuSelectionError):
+ _classify_panel(_parse_nodes(mutate(kind)))
+
+ def test_color_click_rejects_dual_current_only_cross_variant_in_both_directions(self) -> None:
+ cases = (
+ (_CURRENT_ONLY_EMPTY_FIXTURE, _COLOR_FIXTURE),
+ (_EMPTY_FIXTURE, _CURRENT_ONLY_COLOR_FIXTURE),
+ )
+ for empty_fixture, color_fixture in cases:
+ with self.subTest(empty=empty_fixture.name, color=color_fixture.name):
+ device = _RawDevice()
+ device.panel_hierarchy = empty_fixture.read_text(encoding="utf-8")
+ device.color_hierarchy = color_fixture.read_text(encoding="utf-8")
+ flow = _flow(device)
+ flow.open_sku_panel(_TARGET_URL)
+
+ with self.assertRaises(SkuSelectionError) as raised:
+ flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
+
+ self.assertNotIn("受控显示动作尚未取证", str(raised.exception))
+ self.assertEqual(_tap_centers(device), [(865, 2218), (528, 1387)])
+ self.assertEqual(_actions(device, "pressKey"), [])
+
def test_full_verified_entry_structure_taps_exact_text_child_once(self) -> None:
device = _RawDevice()
_flow(device).open_sku_panel(_TARGET_URL)
@@ -708,6 +932,8 @@ class SkuSelectionFlowTests(unittest.TestCase):
for fixture in (
_EMPTY_FIXTURE,
_COLOR_FIXTURE,
+ _CURRENT_ONLY_EMPTY_FIXTURE,
+ _CURRENT_ONLY_COLOR_FIXTURE,
_S_FIXTURE,
_M_FIXTURE,
_ENTRY_FIXTURE,
diff --git a/docs/tasks/T-103.md b/docs/tasks/T-103.md
index ab248a4..f22af9f 100644
--- a/docs/tasks/T-103.md
+++ b/docs/tasks/T-103.md
@@ -25,7 +25,7 @@ write_paths:
- docs/current-state.md
---
-
+
## 问题 / 背景
T-102 已证明 canonical 链接可进入目标商品。T-103 在 PKG110 / Android 16 / 拼多多 8.17.0、goods_id `937122477375` 上确认:详情页底部购买区第一行“快要抢光 + 金额”是唯一获准的受控规格入口;商品内容区同名小字和第二行“免拼购买”都不得点击。
@@ -290,6 +290,10 @@ T-103 sanitizer v2 坐标修正与主审:提交 44c027a 将 screenshot space
### 2026-08-05T06:50:08Z · ila
2026-08-05 T-103 同版本单价格节点变体真机证据与实施方案(先证据、后判据):第一次正式 reveal 在 `sku_entry_panel_verify` fail closed。项目所有者已确认 force-stop 后颜色/尺码恢复未选;root 只读检查证明 PDD 8.17.0/PKG110/Android16/1080×2376/前台正确、summary=`请选择: 颜色分类 尺码`、selected-count=0、目标颜色 bounds `[372,1188][684,1587]` selected=false。证据目录 `C:\Users\ila20\AppData\Local\cmbuyer\artifacts\T-103\sku-panel-empty-current-only-20260805-143741`,screenshot SHA-256 `7f3143d7fab42b41a088e27a4b2d9f8dbdf31b5ecc9ba02b148ca021304026f6`,XML SHA-256 `d3dbf217b4567069feccb30ab43124c5fa7da3734659ca5452b98c2e0a7ff158`。根因不是 selected,而是价格行由旧双节点变为 exact 单节点:current 仍为 `限1件 ¥12.88 ` `[396,580][675,647]`,其唯一 LinearLayout 为 `[396,575][693,647]` 且仅一 child,外层 FrameLayout `[396,575][1053,647]`;`券前¥29.88` count=0。项目所有者随后在同一 goods_id 937122477375 面板人工选择目标颜色并明确完成第二态确认;目录 `...\sku-panel-color-only-current-only-20260805-144550`,screenshot `931dc2d31f4e34872fdd97879abd076e414b6494569e8a587353266011610545`,XML `c686737942e5fdcebc76c2bc1d5cd9ae90a45f4a921297119ddaa4313758a034`;summary=`请选择: 尺码`,目标颜色 selected=true,未选尺码,current count=1/original count=0,固定 reveal 通道用现有代码只读验证 PASS,未发送 swipe。实施只允许新增两个 exact 单价格前置 spec(PANEL_OPEN_EMPTY / COLOR_SELECTED_SIZE_HIDDEN),旧双价格 spec 继续保留;不得把 original 泛化为 optional、不得接受任意 row/bounds/child count/price text,也不得修改 rolled candidate、数量、确认页、提交或支付。正式再跑前新增带正式 marker 的 reveal 白名单阶段码,至少能明确 `reveal_not_attempted` 与 `reveal_attempted/candidate/after/publish` 边界,异常正文和页面内容仍不外泄;补 cross-variant/duplicate/unknown/伪造 stage 失败测试。T-103 继续 Doing,先提交此证据计划,再实现并独立主审。
+
+### 2026-08-05T07:19:02Z · ila
+
+2026-08-05 第二态由用户人工确认:目标颜色“黑色 CHA(纯棉)”已选,尺码选项已显示且尚未选尺码。实现已按两组新 raw 证据收紧:新增 PANEL_OPEN_EMPTY / COLOR_SELECTED_SIZE_HIDDEN 的 exact current-only 价格布局;历史 dual 布局继续按已取证嵌套后代关系匹配;颜色点击后的后置条件绑定点击前完整 price_layout,禁止 dual 与 current-only 双向漂移。reveal 失败阶段新增不可伪造 marker,precondition 保证零 swipe,attempted 及以后仅在唯一 RPC 机会已封存后报告。两名独立 agent 审计均 PASS;根代理直接读取六组真实 raw 验证唯一命中:dual empty/color/S/M 与 current-only empty/color 均正确,M 读价 12.88。根代理完整门禁:client unittest 269/269 PASS,compileall PASS,validate_agent_context PASS,git diff --check PASS。未新增数量、确认页、围栏、提交订单或付款能力。T-103 仍保持 DOING,下一步仅允许强停重置后运行一次真机 reveal,并等待人工验收。
## 边界