diff --git a/client/src/cmbuyer_client/pdd/sku_reveal_spike.py b/client/src/cmbuyer_client/pdd/sku_reveal_spike.py index d3c1d60..7af793c 100644 --- a/client/src/cmbuyer_client/pdd/sku_reveal_spike.py +++ b/client/src/cmbuyer_client/pdd/sku_reveal_spike.py @@ -1,4 +1,4 @@ -"""T-103 尺码显示动作的一次性真机取证；不属于生产采购 Flow。""" +"""T-103 尺码显示动作的一次性真机证据采集。""" from __future__ import annotations @@ -20,32 +20,10 @@ from ..device.baseline import PDD_PACKAGE, _save_base64_screenshot, _sha256_file from .product_url import parse_product_url from .sku_selection import ( EXPECTED_GOODS_ID, - _COLOR_ONLY_SUMMARY, - _PANEL_SURFACE, - _PanelProfile, - _REVEAL_NOT_PROVEN, - _S_SIZE_UI, - _TARGET_COLOR_UI, - _TARGET_SIZE_UI, - _action_bounds, - _clickable_before, - _descendant, - _exact_color_action, - _exact_inert, - _exact_live_layout, - _exact_readonly_text, - _exact_recycler, - _is_size_action_text, - _one, _parse_nodes, _require_color_only_panel, - _require_color_action_chain, - _require_color_subtree, - _require_exact_selected_set, - _require_panel_chain, - _require_size_wrapper, - _require_size_action_chain, - _spec_for, + _require_safe_reveal_path, + _revealed_unselected_projection, resolve_task_selection, SkuSelectionError, SkuSelectionFlow, @@ -63,9 +41,6 @@ from .sku_selection_runner import ( _TARGET_URL = f"https://mobile.yangkeduo.com/goods.html?goods_id={EXPECTED_GOODS_ID}" -_REVEAL_START = (360, 1900) -_REVEAL_END = (360, 1300) -_REVEAL_STEPS = 30 _REVEAL_FAILURE_STAGES = frozenset( ( "reveal_precondition", @@ -114,30 +89,6 @@ class SkuRevealSpikeResult: manifest_path: Path -class _RevealEvidenceAdapter(UiautomatorSkuPanelAdapter): - """只为 spike 增加一个无参数、固定 profile 的一次性手势。""" - - def __init__(self, device: Any, timeout_seconds: float) -> None: - super().__init__(device, timeout_seconds) - self._reveal_attempted = False - - @property - def reveal_attempted(self) -> bool: - return self._reveal_attempted - - def reveal_size_options_once(self) -> None: - if self._reveal_attempted: - raise SkuRevealSpikeError("规格显示动作已经尝试过，拒绝重试。") - # RPC 超时也可能表示手势已经送达，必须在调用前封存唯一机会。 - self._reveal_attempted = True - self._call( - "jsonrpc_call", - "swipe", - [*_REVEAL_START, *_REVEAL_END, _REVEAL_STEPS], - timeout=self._timeout_seconds, - ) - - class SkuRevealSpikeCapturer: """打开已取证面板、选一次目标颜色，再采集唯一 reveal 的前后证据。""" @@ -165,7 +116,7 @@ class SkuRevealSpikeCapturer: ) -> SkuRevealSpikeResult: stage = "reveal_precondition" staging: Path | None = None - adapter: _RevealEvidenceAdapter | None = None + adapter: UiautomatorSkuPanelAdapter | None = None try: if type(goods_id) is not str or goods_id != EXPECTED_GOODS_ID: raise SkuRevealSpikeError("商品不是 T-103 已取证目标，已停止取证。") @@ -176,7 +127,7 @@ class SkuRevealSpikeCapturer: deadline = self._clock() + self._timeout_seconds inspection = self._adb_client.inspect(serial) _require_expected_device(inspection) - adapter = _RevealEvidenceAdapter(self._connector(serial), self._timeout_seconds) + adapter = UiautomatorSkuPanelAdapter(self._connector(serial), self._timeout_seconds) _require_expected_version(adapter.app_info(PDD_PACKAGE)) if adapter.display_size() != EXPECTED_SCREEN_SIZE: raise SkuRevealSpikeError("设备不是已取证的竖屏坐标空间，已停止取证。") @@ -193,13 +144,10 @@ class SkuRevealSpikeCapturer: sleep_function=self._sleep, ) flow.open_sku_panel(link.canonical_url, pre_intent) - try: - flow.select_sku_options(resolve_task_selection("黑色CHA(纯棉)", "M(建议100-115)")) - except SkuSelectionError as error: - if error.args != (_REVEAL_NOT_PROVEN,): - raise - else: - raise SkuRevealSpikeError("规格流程未停在已取证的仅颜色状态。") + # spike 只复用生产 Flow 的私有前置准备，不调用完整选择流程，避免重复 reveal/M。 + flow._prepare_reveal_precondition( + resolve_task_selection("黑色CHA(纯棉)", "M(建议100-115)") + ) before_hierarchy = adapter.dump_window_hierarchy() before_nodes = _parse_nodes(before_hierarchy) @@ -231,7 +179,7 @@ class SkuRevealSpikeCapturer: after_directory = staging / "after" _capture_frame(adapter, after_directory, after_hierarchy) reverified = adapter.dump_window_hierarchy() - if _candidate_projection(_parse_nodes(reverified)) != projection: + if _revealed_unselected_projection(_parse_nodes(reverified)) != projection: raise SkuRevealSpikeError("截图后候选状态漂移，未发布证据。") (after_directory / "hierarchy.xml").write_text(reverified, encoding="utf-8") @@ -265,7 +213,7 @@ class SkuRevealSpikeCapturer: def _wait_for_candidate( self, - adapter: _RevealEvidenceAdapter, + adapter: UiautomatorSkuPanelAdapter, deadline: float, ) -> tuple[tuple[tuple[str, ...], ...], str]: stable: tuple[tuple[str, ...], ...] | None = None @@ -276,7 +224,7 @@ class SkuRevealSpikeCapturer: raise SkuRevealSpikeError("reveal 后拼多多不在前台，未发布证据。") hierarchy = adapter.dump_window_hierarchy() try: - projection = _candidate_projection(_parse_nodes(hierarchy)) + projection = _revealed_unselected_projection(_parse_nodes(hierarchy)) except SkuSelectionError: projection = None if projection is not None and projection == stable: @@ -288,315 +236,7 @@ class SkuRevealSpikeCapturer: self._sleep(min(0.2, remaining)) -def _require_safe_reveal_path(nodes: list[Any]) -> None: - # 当前真机证据证明 x=360 是两列规格卡之间的空隙；必须验证完整线段，离散采样会漏掉窄浮层。 - surface = _one( - [node for node in nodes if _exact_inert(node, "android.view.ViewGroup", _PANEL_SURFACE)], - "固定 reveal 通道无法绑定面板内容面。", - ) - _require_panel_chain(surface) - content = surface.parent - action_root = content.parent if content is not None else None - action_parent = action_root.parent if action_root is not None else None - if ( - content is None - or action_root is None - or action_parent is None - or not _exact_live_layout(action_root, "android.view.ViewGroup", "[0,366][1080,2328]") - or not _exact_live_layout(action_parent, "android.widget.LinearLayout", "[0,120][1080,2328]") - ): - raise SkuRevealSpikeError("固定 reveal 通道祖先身份漂移，未执行手势。") - allowed = {id(action_root.element), id(action_parent.element)} - occupants: set[int] = set() - for node in nodes: - if node.element.get("clickable") != "true": - continue - try: - left, top, right, bottom = _action_bounds(node.bounds) - except SkuSelectionError as error: - raise SkuRevealSpikeError("可点击节点坐标不可验证，未执行手势。") from error - if ( - left <= _REVEAL_START[0] < right - and top <= _REVEAL_START[1] - and bottom > _REVEAL_END[1] - ): - occupants.add(id(node.element)) - if occupants != allowed: - raise SkuRevealSpikeError("固定 reveal 通道被未取证可点击节点占用，未执行手势。") - if _REVEAL_START[1] >= 2079 or _REVEAL_END[1] >= 2079: - raise SkuRevealSpikeError("固定 reveal 通道越过规格内容区，未执行手势。") - - -def _candidate_projection(nodes: list[Any]) -> tuple[tuple[str, ...], ...]: - surface = _one( - [node for node in nodes if _exact_inert(node, "android.view.ViewGroup", _PANEL_SURFACE)], - "候选面板内容面不唯一。", - ) - _require_panel_chain(surface) - header = _one( - [node for node in nodes if node.parent is surface and _exact_inert(node, "android.widget.LinearLayout", "[0,366][1080,1000]")], - "候选面板头部不唯一。", - ) - if len([child for child in header.element if child.tag == "node"]) != 4: - raise SkuSelectionError("候选面板头部子节点数量漂移。") - outer = _one( - [node for node in nodes if node.parent is surface and _exact_recycler(node, "[0,1000][1080,2079]")], - "候选维度容器不唯一。", - ) - price_content, price_frame = _candidate_price_container(nodes, header) - price_nodes = _candidate_price_nodes(nodes, header, surface, price_frame) - summary = _one( - [ - node - for node in nodes - if node.parent is price_content - and _exact_readonly_text(node, _COLOR_ONLY_SUMMARY, "[396,654][1053,716]") - and len(node.element) == 0 - ], - "候选摘要不唯一。", - ) - color_region = _one( - [node for node in nodes if _descendant(node, outer) and _exact_recycler(node, "[36,1000][1080,1483]")], - "候选颜色容器不唯一。", - ) - color = _one( - [node for node in nodes if node.parent is color_region and _exact_color_action(node, "[372,1000][684,1024]", True)], - "候选目标颜色不唯一。", - ) - rolled_spec = _spec_for(_PanelProfile.TARGETS_SELECTED) - selected_color_nodes = _require_color_subtree(color, rolled_spec) - _require_color_action_chain(color, color_region, outer, surface, rolled_spec) - size_label = _one( - [node for node in nodes if _descendant(node, outer) and _exact_readonly_text(node, "尺码", "[36,1506][114,1552]")], - "候选尺码标题不唯一。", - ) - size_header = size_label.parent - if size_header is None or not _exact_live_layout(size_header, "android.widget.LinearLayout", "[36,1489][1044,1570]"): - raise SkuSelectionError("候选尺码标题父结构漂移。") - size_options = _one( - [node for node in nodes if _descendant(node, outer) and _exact_inert(node, "android.view.ViewGroup", "[36,1582][1044,1897]")], - "候选尺码 options 根不唯一。", - ) - actions = [node for node in nodes if _descendant(node, size_options) and _is_size_action_text(node)] - s_action = _one( - [node for node in actions if node.text == _S_SIZE_UI and node.bounds == "[36,1582][409,1667]"], - "候选 S action 不唯一。", - ) - m_action = _one( - [node for node in actions if node.text == _TARGET_SIZE_UI and node.bounds == "[439,1582][831,1667]"], - "候选 M action 不唯一。", - ) - _require_size_wrapper(s_action, size_options) - _require_size_wrapper(m_action, size_options) - _require_size_action_chain(s_action, size_options, outer, surface) - _require_size_action_chain(m_action, size_options, outer, surface) - if any(node.element.get("selected") == "true" for node in actions): - raise SkuSelectionError("reveal 候选态已有尺码被选中。") - _require_exact_selected_set(nodes, surface, selected_color_nodes) - dangerous = [ - node for node in nodes - if "提交订单" in node.text - and node.element.get("package") == PDD_PACKAGE - and node.element.get("class") == "android.widget.TextView" - ] - if len(dangerous) != 1 or _action_bounds(dangerous[0].bounds)[1] < 2079: - raise SkuSelectionError("提交硬拒绝区位置不唯一。") - return tuple( - ( - node.element.get("class", ""), - node.bounds, - node.text, - node.desc, - node.element.get("selected", ""), - node.element.get("clickable", ""), - ) - for node in ( - surface, - header, - outer, - *price_nodes, - summary, - color_region, - color, - size_label, - size_options, - s_action, - m_action, - dangerous[0], - ) - ) - - -def _candidate_price_nodes( - nodes: list[Any], - header: Any, - surface: Any, - price_frame: Any, -) -> tuple[Any, Any, Any]: - matches: list[tuple[Any, Any, Any]] = [] - for matcher in (_candidate_dual_price_nodes, _candidate_current_only_price_nodes): - try: - match = matcher(nodes, header, surface, price_frame) - except SkuSelectionError: - continue - matches.append(match) - if len(matches) != 1: - raise SkuSelectionError("候选价格布局不符合唯一完整取证变体。") - return matches[0] - - -def _candidate_dual_price_nodes( - nodes: list[Any], - header: Any, - surface: Any, - price_frame: Any, -) -> tuple[Any, Any, Any]: - price_row = _one( - [ - node - for node in nodes - if node.parent is price_frame - and _exact_inert(node, "android.widget.LinearLayout", "[396,498][895,570]") - ], - "候选双价格行不唯一。", - ) - 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, "快卖完 ¥12.88", "[396,503][712,570]") - and len(node.element) == 0 - ], - "候选双价格当前价不唯一。", - ) - original = _one( - [ - node - for node in nodes - if node.parent is price_row - and _exact_readonly_text(node, "¥29.88", "[730,503][895,570]") - and len(node.element) == 0 - ], - "候选双价格原价不唯一。", - ) - if _clickable_before(current, surface): - raise SkuSelectionError("候选双价格当前价位于可点击内容祖先下。") - if any( - _candidate_current_only_fragment(node) - for node in nodes - if _descendant(node, header) - ): - raise SkuSelectionError("候选双价格布局混入单价格残片。") - return price_row, current, original - - -def _candidate_current_only_price_nodes( - nodes: list[Any], - header: Any, - surface: Any, - price_frame: Any, -) -> tuple[Any, Any, Any]: - price_row = _one( - [ - node - for node in nodes - if node.parent is price_frame - and _exact_inert(node, "android.widget.LinearLayout", "[396,498][693,570]") - ], - "候选单价格行父子关系漂移。", - ) - 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, "限1件 ¥12.88 ", "[396,503][675,570]") - and len(node.element) == 0 - ], - "候选单价格当前价不唯一。", - ) - if any(node.text in {"¥29.88", "券前¥29.88"} for node in nodes): - raise SkuSelectionError("候选单价格变体出现原价角色。") - if any( - _candidate_dual_fragment(node) - for node in nodes - if _descendant(node, header) - ): - raise SkuSelectionError("候选单价格布局混入双价格残片。") - if _clickable_before(current, surface): - raise SkuSelectionError("候选单价格当前价位于可点击内容祖先下。") - return price_frame, price_row, current - - -def _candidate_price_container(nodes: list[Any], header: Any) -> tuple[Any, Any]: - content_frame = _one( - [ - node - for node in nodes - if node.parent is header - and _exact_inert(node, "android.widget.FrameLayout", "[0,474][1080,863]") - ], - "候选价格内容外框不唯一。", - ) - if len([child for child in content_frame.element if child.tag == "node"]) != 1: - raise SkuSelectionError("候选价格内容外框子节点数量漂移。") - relative = _one( - [ - node - for node in nodes - if node.parent is content_frame - and _exact_inert(node, "android.widget.RelativeLayout", "[0,474][1080,863]") - ], - "候选价格 RelativeLayout 父子关系漂移。", - ) - if len([child for child in relative.element if child.tag == "node"]) != 1: - raise SkuSelectionError("候选价格 RelativeLayout 子节点数量漂移。") - content = _one( - [ - node - for node in nodes - if node.parent is relative - and _exact_inert(node, "android.view.ViewGroup", "[0,474][1080,863]") - ], - "候选价格内容 ViewGroup 父子关系漂移。", - ) - if len([child for child in content.element if child.tag == "node"]) != 8: - raise SkuSelectionError("候选价格内容 ViewGroup 子节点数量漂移。") - price_frame = _one( - [ - node - for node in nodes - if node.parent is content - and _exact_inert(node, "android.widget.FrameLayout", "[396,498][1053,570]") - ], - "候选价格 FrameLayout 父子关系漂移。", - ) - if len([child for child in price_frame.element if child.tag == "node"]) != 1: - raise SkuSelectionError("候选价格 FrameLayout 子节点数量漂移。") - return content, price_frame - - -def _candidate_current_only_fragment(node: Any) -> bool: - return ( - _exact_inert(node, "android.widget.LinearLayout", "[396,498][693,570]") - or _exact_readonly_text(node, "限1件 ¥12.88 ", "[396,503][675,570]") - ) - - -def _candidate_dual_fragment(node: Any) -> bool: - return ( - _exact_inert(node, "android.widget.LinearLayout", "[396,498][895,570]") - or _exact_readonly_text(node, "快卖完 ¥12.88", "[396,503][712,570]") - or _exact_readonly_text(node, "¥29.88", "[730,503][895,570]") - ) - - -def _capture_frame(adapter: _RevealEvidenceAdapter, directory: Path, hierarchy: str) -> None: +def _capture_frame(adapter: UiautomatorSkuPanelAdapter, directory: Path, hierarchy: str) -> None: directory.mkdir() hierarchy_path = directory / "hierarchy.xml" hierarchy_path.write_text(hierarchy, encoding="utf-8") diff --git a/client/src/cmbuyer_client/pdd/sku_selection.py b/client/src/cmbuyer_client/pdd/sku_selection.py index 584c62e..2538aa7 100644 --- a/client/src/cmbuyer_client/pdd/sku_selection.py +++ b/client/src/cmbuyer_client/pdd/sku_selection.py @@ -18,6 +18,8 @@ EXPECTED_UNIT_PRICE = "12.88" # 任务值不是页面判据；右侧是 v5 取证的唯一 accessibility 文案（空格/全角括号均有意义）。 TASK_TO_UI_SELECTION = {("黑色CHA(纯棉)", "M(建议100-115)"): ("黑色 CHA (纯棉）", "M（建议100-115）")} _TARGET_COLOR_UI, _TARGET_SIZE_UI = next(iter(TASK_TO_UI_SELECTION.values())) +_TARGET_COLOR_UNROLLED_BOUNDS = "[372,1188][684,1587]" +_TARGET_SIZE_BOUNDS = "[439,1582][831,1667]" _ENTRY = "快要抢光 ¥ 12.88" _ENTRY_PROMOTION_LABEL = "快要抢光" _ENTRY_TEXT_BOUNDS = "[688,2184][1042,2253]" @@ -37,10 +39,11 @@ _COLOR_ONLY_SUMMARY = "请选择： 尺码" _S_SIZE_UI = "S（建议80-100）" _S_SUMMARY = f"已选： {_TARGET_COLOR_UI} {_S_SIZE_UI}" _TARGET_SUMMARY = f"已选： {_TARGET_COLOR_UI} {_TARGET_SIZE_UI}" -_REVEAL_NOT_PROVEN = "尺码仍在已取证视口外；受控显示动作尚未取证，已停止后续点击。" _BOUNDS = re.compile(r"^\[(\d+),(\d+)\]\[(\d+),(\d+)\]$") _ROLLED_CURRENT_PRICE = "快卖完 ¥12.88" _ROLLED_ORIGINAL_PRICE = "¥29.88" +_REVEALED_CURRENT_PRICE = "限1件 ¥12.88 " +_REVEAL_GESTURE = (360, 1900, 360, 1300, 30) _BAD_PRICE_ROLE = ("提交订单", "支付", "优惠", "券", "会员", "补贴", "区间", "实付", "到手", "原价", "划线价", "最低", "低至", "起价", "下单", "先用后付", "预估") @@ -91,6 +94,7 @@ class SkuPanelDevice(Protocol): def dump_window_hierarchy(self) -> str: ... def tap_sku_entry(self, bounds: str) -> None: ... def tap_sku_option(self, bounds: str) -> None: ... + def reveal_size_options_once(self) -> None: ... def leave_sku_panel(self) -> None: ... @@ -122,6 +126,7 @@ class _Node: class _PanelProfile(Enum): PANEL_OPEN_EMPTY = "panel_open_empty" COLOR_SELECTED_SIZE_HIDDEN = "color_selected_size_hidden" + SIZE_VISIBLE_UNSELECTED = "size_visible_unselected" SIZE_VISIBLE_NON_TARGET = "size_visible_non_target" TARGETS_SELECTED = "targets_selected" @@ -172,7 +177,7 @@ _PANEL_SPECS = ( "券前¥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,1188][1080,2046]", _TARGET_COLOR_UNROLLED_BOUNDS, False, "[36,2069][114,2079]", None, ), _PanelSpec( @@ -183,7 +188,7 @@ _PANEL_SPECS = ( "券前¥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,1188][1080,2046]", _TARGET_COLOR_UNROLLED_BOUNDS, True, "[36,2069][114,2079]", None, ), _PanelSpec( @@ -194,7 +199,7 @@ _PANEL_SPECS = ( "限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,1188][1080,2046]", _TARGET_COLOR_UNROLLED_BOUNDS, False, "[36,2069][114,2079]", None, ), _PanelSpec( @@ -205,7 +210,7 @@ _PANEL_SPECS = ( "限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,1188][1080,2046]", _TARGET_COLOR_UNROLLED_BOUNDS, True, "[36,2069][114,2079]", None, ), _PanelSpec( @@ -233,6 +238,19 @@ _PANEL_SPECS = ( ) +_REVEALED_UNSELECTED_SPEC = _PanelSpec( + _PanelProfile.SIZE_VISIBLE_UNSELECTED, + "[0,366][1080,1000]", "[0,1000][1080,2079]", + _CurrentOnlyPriceLayout( + "[396,498][1053,570]", "[396,498][693,570]", + _REVEALED_CURRENT_PRICE, "[396,503][675,570]", + ), + _COLOR_ONLY_SUMMARY, "[396,654][1053,716]", None, + "[36,1000][1080,1483]", "[372,1000][684,1024]", True, + "[36,1506][114,1552]", None, +) + + class SkuSelectionFlow: def __init__(self, device: SkuPanelDevice, entry_wait_timeout_seconds: float = 0.2, entry_poll_interval_seconds: float = 0.2, monotonic_clock: Callable[[], float] = monotonic, @@ -241,9 +259,12 @@ class SkuSelectionFlow: raise ValueError("入口等待参数无效。") self._device, self._entry_timeout, self._poll = device, entry_wait_timeout_seconds, entry_poll_interval_seconds self._clock, self._sleep = monotonic_clock, sleep_function - self._pending: tuple[str, Callable[[list[_Node]], Any]] | None = None + self._pending: tuple[str, Callable[[list[_Node]], Any], bool] | None = None + self._terminal = False + self._color_selected_by_flow = False def open_sku_panel(self, product_url: str, pre_intent_hierarchy: str | None = None) -> None: + self._require_active() try: if parse_product_url(product_url).goods_id != EXPECTED_GOODS_ID: raise SkuSelectionError("商品不是已取证目标，已停止操作。") @@ -265,90 +286,175 @@ class SkuSelectionFlow: raise try: - self._pending = (before, _require_empty_panel) + self._pending = (before, _require_empty_panel, False) self._device.tap_sku_entry(entry.bounds) except BaseException as error: + self._terminal = True _annotate_sku_entry_failure(error, "sku_entry_click") raise try: self._wait_after_action(before, _require_empty_panel) except BaseException as error: + self._terminal = True _annotate_sku_entry_failure(error, "sku_entry_panel_verify") raise def select_sku_options(self, selection: SkuSelection) -> None: + self._require_active() + if selection != SkuSelection(_TARGET_COLOR_UI, _TARGET_SIZE_UI): + raise SkuSelectionError("规格 UI 文案不是获准目标，已停止操作。") + nodes = self._verified_nodes() + initial_profile = _classify_panel(nodes).profile + if initial_profile is _PanelProfile.TARGETS_SELECTED: + return + if initial_profile is _PanelProfile.SIZE_VISIBLE_NON_TARGET: + self._require_foreground() + before = self._read_hierarchy() + nodes = _parse_nodes(before) + target = _target_size_action(nodes, selected=False) + _action_bounds(target.bounds) + _require_action_occupants(nodes, target) + self._perform_mutation( + before, + _require_target_panel, + partial(self._device.tap_sku_option, target.bounds), + ) + return + + self._prepare_reveal_precondition(selection) + self._require_foreground() + before = self._read_hierarchy() + nodes = _parse_nodes(before) + spec = _require_profile(nodes, _PanelProfile.COLOR_SELECTED_SIZE_HIDDEN) + if not isinstance(spec.price_layout, _CurrentOnlyPriceLayout): + # 真机动作证据只覆盖 current-only；dual 仅颜色态必须在 reveal 前零动作停止。 + raise SkuSelectionError("规格显示前置不是已取证单价格布局，已停止操作。") + _require_safe_reveal_path(nodes) + self._perform_mutation( + before, + _revealed_unselected_projection, + self._device.reveal_size_options_once, + require_stable=True, + ) + + self._require_foreground() + before = self._read_hierarchy() + nodes = _parse_nodes(before) + _revealed_unselected_projection(nodes) + target = _target_size_action(nodes, selected=False) + _action_bounds(target.bounds) + _require_action_occupants(nodes, target) + self._perform_mutation( + before, + _require_target_panel, + partial(self._device.tap_sku_option, target.bounds), + ) + + def _prepare_reveal_precondition(self, selection: SkuSelection) -> None: + """只把 current-only 初态推进到仅颜色态，供生产 Flow 与证据 spike 共用。""" + + self._require_active() if selection != SkuSelection(_TARGET_COLOR_UI, _TARGET_SIZE_UI): raise SkuSelectionError("规格 UI 文案不是获准目标，已停止操作。") self._require_foreground() before = self._read_hierarchy() nodes = _parse_nodes(before) 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) - 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, color_postcondition) - # 当前证据只证明颜色选择；尺码仍在视口外。没有动作证据时必须在此停住， - # 不能把一次通用 swipe 或下一次规格点击伪装成已验证流程。 - raise SkuSelectionError(_REVEAL_NOT_PROVEN) - if profile is _PanelProfile.COLOR_SELECTED_SIZE_HIDDEN: - raise SkuSelectionError(_REVEAL_NOT_PROVEN) - if profile is _PanelProfile.SIZE_VISIBLE_NON_TARGET: - target = _target_size_action(nodes, selected=False) - _action_bounds(target.bounds) - _require_action_occupants(nodes, target) - self._pending = (before, _require_target_panel) - self._device.tap_sku_option(target.bounds) - self._wait_after_action(before, _require_target_panel) + if spec.profile is _PanelProfile.COLOR_SELECTED_SIZE_HIDDEN: + if not isinstance(spec.price_layout, _CurrentOnlyPriceLayout): + raise SkuSelectionError("规格显示前置不是已取证单价格布局，已停止操作。") + if not self._color_selected_by_flow: + raise SkuSelectionError("目标颜色不是本次 Flow 精确选中，未执行 reveal。") return - if profile is _PanelProfile.TARGETS_SELECTED: - return - raise SkuSelectionError("规格面板状态不属于已取证 profile，已停止操作。") + if spec.profile is not _PanelProfile.PANEL_OPEN_EMPTY or not isinstance( + spec.price_layout, _CurrentOnlyPriceLayout + ): + raise SkuSelectionError("规格面板不属于获准的单价格初态，已停止操作。") + target = _target_color_action(nodes, selected=False) + _action_bounds(target.bounds) + _require_action_occupants(nodes, target) + color_postcondition = partial( + _require_profile_with_price_layout, + expected=_PanelProfile.COLOR_SELECTED_SIZE_HIDDEN, + expected_price_layout=spec.price_layout, + ) + self._perform_mutation( + before, + color_postcondition, + partial(self._device.tap_sku_option, target.bounds), + ) + self._color_selected_by_flow = True def read_sku_unit_price(self) -> str: + self._require_active() return _unit_price(self._verified_nodes()) def verify_target_selection_and_read_price(self, selection: SkuSelection) -> str: + self._require_active() if selection != SkuSelection(_TARGET_COLOR_UI, _TARGET_SIZE_UI): raise SkuSelectionError("规格 UI 文案不是获准目标，已停止读取。") nodes = self._verified_nodes() return _unit_price(nodes) def exit_sku_panel_safely(self) -> None: + self._require_active() self._require_foreground() before = self._read_hierarchy() - _classify_panel(_parse_nodes(before)) - self._device.leave_sku_panel() + # 截图重证与 Back 之间仍可能漂移；只有目标双规格和已验单价同时保持时才允许返回。 + _unit_price(_parse_nodes(before)) + try: + self._device.leave_sku_panel() + except BaseException: + self._terminal = True + raise deadline = self._clock() + self._entry_timeout - while True: - self._require_foreground() - raw = self._read_hierarchy() - if raw != before: - try: - _classify_panel(_parse_nodes(raw)) - except SkuSelectionError: - return - remaining = deadline - self._clock() - if remaining <= 0: - raise SkuSelectionError("安全退出后未确认离开规格面板，未重试返回。") - self._sleep(min(self._poll, remaining)) + try: + while True: + self._require_foreground() + raw = self._read_hierarchy() + if raw != before: + try: + _classify_panel(_parse_nodes(raw)) + except SkuSelectionError: + self._terminal = True + return + remaining = deadline - self._clock() + if remaining <= 0: + raise SkuSelectionError("安全退出后未确认离开规格面板，未重试返回。") + self._sleep(min(self._poll, remaining)) + except BaseException: + self._terminal = True + raise def reconcile_pending_action(self) -> None: """仅只读调和一次已发出但尚未得到后置条件确认的动作。""" if self._pending is None: return - before, condition = self._pending - self._wait_after_action(before, condition) + before, condition, require_stable = self._pending + self._wait_after_action(before, condition, require_stable=require_stable) + + def _perform_mutation( + self, + before: str, + condition: Callable[[list[_Node]], Any], + mutation: Callable[[], None], + *, + require_stable: bool = False, + ) -> list[_Node]: + # 后置条件连同唯一动作机会必须在 RPC 前封存；超时可能表示已送达，不能重发。 + self._pending = (before, condition, require_stable) + try: + mutation() + return self._wait_after_action(before, condition, require_stable=require_stable) + except BaseException: + # 结果不明时只允许 reconcile_pending_action 读回；即使调和成功，本 Flow 也不再继续。 + self._terminal = True + raise + + def _require_active(self) -> None: + if self._terminal: + raise SkuSelectionError("规格流程已进入不可重入终止态，已停止操作。") def _wait_for_entry(self, previous: str | None) -> tuple[_Node, str]: deadline, stable = self._clock() + self._entry_timeout, None @@ -380,19 +486,31 @@ class SkuSelectionFlow: raise SkuSelectionError("等待已取证规格入口超时，未执行点击。") self._sleep(min(self._poll, remaining)) - def _wait_after_action(self, previous: str, condition: Callable[[list[_Node]], Any]) -> list[_Node]: + def _wait_after_action( + self, + previous: str, + condition: Callable[[list[_Node]], Any], + *, + require_stable: bool = False, + ) -> list[_Node]: deadline = self._clock() + self._entry_timeout + stable: Any = None + has_stable = False while True: self._require_foreground() raw = self._read_hierarchy() if raw != previous: nodes = _parse_nodes(raw) try: - condition(nodes) - self._pending = None - return nodes + projection = condition(nodes) + if not require_stable or (has_stable and projection == stable): + self._pending = None + return nodes + stable = projection + has_stable = True except SkuSelectionError: - pass + stable = None + has_stable = False remaining = deadline - self._clock() if remaining <= 0: raise SkuSelectionError("动作后页面未在限定时间内满足已取证后置条件，未重试动作。") @@ -445,6 +563,12 @@ def _classify_panel(nodes: list[_Node]) -> _PanelSpec: except SkuSelectionError: continue matches.append(spec) + try: + _revealed_unselected_projection(nodes) + except SkuSelectionError: + pass + else: + matches.append(_REVEALED_UNSELECTED_SPEC) if len(matches) != 1: raise SkuSelectionError("规格面板不符合唯一完整取证 profile，已停止操作。") return matches[0] @@ -574,6 +698,177 @@ def _match_panel_profile(nodes: list[_Node], spec: _PanelSpec) -> None: _require_exact_selected_set(nodes, surface, [*selected_nodes, expected_action]) +def _require_safe_reveal_path(nodes: list[_Node]) -> None: + """证明固定手势整条线段仍是已取证的两列卡片间安全通道。""" + + surface = _one( + [node for node in nodes if _exact_inert(node, "android.view.ViewGroup", _PANEL_SURFACE)], + "固定 reveal 通道无法绑定面板内容面。", + ) + _require_panel_chain(surface) + content = surface.parent + action_root = content.parent if content is not None else None + action_parent = action_root.parent if action_root is not None else None + if ( + content is None + or action_root is None + or action_parent is None + or not _exact_live_layout(action_root, "android.view.ViewGroup", "[0,366][1080,2328]") + or not _exact_live_layout(action_parent, "android.widget.LinearLayout", "[0,120][1080,2328]") + ): + raise SkuSelectionError("固定 reveal 通道祖先身份漂移，未执行手势。") + start_x, start_y, end_x, end_y, _steps = _REVEAL_GESTURE + allowed = {id(action_root.element), id(action_parent.element)} + occupants: set[int] = set() + for node in nodes: + if node.element.get("clickable") != "true": + continue + try: + left, top, right, bottom = _action_bounds(node.bounds) + except SkuSelectionError as error: + raise SkuSelectionError("可点击节点坐标不可验证，未执行手势。") from error + if ( + left <= start_x < right + and top <= start_y + and bottom > end_y + ): + occupants.add(id(node.element)) + if occupants != allowed: + raise SkuSelectionError("固定 reveal 通道被未取证可点击节点占用，未执行手势。") + if start_y >= 2079 or end_y >= 2079: + raise SkuSelectionError("固定 reveal 通道越过规格内容区，未执行手势。") + + +def _revealed_unselected_projection(nodes: list[_Node]) -> tuple[tuple[str, ...], ...]: + """唯一生产判据：current-only reveal 后，目标颜色已选且 S/M 均未选。""" + + surface = _one( + [node for node in nodes if _exact_inert(node, "android.view.ViewGroup", _PANEL_SURFACE)], + "reveal 后面板内容面不唯一。", + ) + _require_panel_chain(surface) + header = _one( + [node for node in nodes if node.parent is surface and _exact_inert(node, "android.widget.LinearLayout", "[0,366][1080,1000]")], + "reveal 后面板头部不唯一。", + ) + if len([child for child in header.element if child.tag == "node"]) != 4: + raise SkuSelectionError("reveal 后面板头部子节点数量漂移。") + outer = _one( + [node for node in nodes if node.parent is surface and _exact_recycler(node, "[0,1000][1080,2079]")], + "reveal 后维度容器不唯一。", + ) + content_frame = _one( + [node for node in nodes if node.parent is header and _exact_inert(node, "android.widget.FrameLayout", "[0,474][1080,863]")], + "reveal 后价格内容外框不唯一。", + ) + if len([child for child in content_frame.element if child.tag == "node"]) != 1: + raise SkuSelectionError("reveal 后价格内容外框子节点数量漂移。") + relative = _one( + [node for node in nodes if node.parent is content_frame and _exact_inert(node, "android.widget.RelativeLayout", "[0,474][1080,863]")], + "reveal 后价格 RelativeLayout 父子关系漂移。", + ) + if len([child for child in relative.element if child.tag == "node"]) != 1: + raise SkuSelectionError("reveal 后价格 RelativeLayout 子节点数量漂移。") + content = _one( + [node for node in nodes if node.parent is relative and _exact_inert(node, "android.view.ViewGroup", "[0,474][1080,863]")], + "reveal 后价格内容 ViewGroup 父子关系漂移。", + ) + if len([child for child in content.element if child.tag == "node"]) != 8: + raise SkuSelectionError("reveal 后价格内容 ViewGroup 子节点数量漂移。") + price_frame = _one( + [node for node in nodes if node.parent is content and _exact_inert(node, "android.widget.FrameLayout", "[396,498][1053,570]")], + "reveal 后价格 FrameLayout 父子关系漂移。", + ) + if len([child for child in price_frame.element if child.tag == "node"]) != 1: + raise SkuSelectionError("reveal 后价格 FrameLayout 子节点数量漂移。") + price_row = _one( + [node for node in nodes if node.parent is price_frame and _exact_inert(node, "android.widget.LinearLayout", "[396,498][693,570]")], + "reveal 后单价格行父子关系漂移。", + ) + if len([child for child in price_row.element if child.tag == "node"]) != 1: + raise SkuSelectionError("reveal 后单价格行子节点数量漂移。") + current = _one( + [node for node in nodes if node.parent is price_row and _exact_readonly_text(node, _REVEALED_CURRENT_PRICE, "[396,503][675,570]") and len(node.element) == 0], + "reveal 后单价格当前价不唯一。", + ) + if _clickable_before(current, surface): + raise SkuSelectionError("reveal 后单价格当前价位于可点击内容祖先下。") + if any( + node.text in {_ROLLED_CURRENT_PRICE, _ROLLED_ORIGINAL_PRICE, "券前¥29.88"} + or (node.bounds == "[396,498][895,570]" and node.element.get("class") == "android.widget.LinearLayout") + for node in nodes + ): + raise SkuSelectionError("reveal 后单价格布局混入双价格角色。") + summary = _one( + [node for node in nodes if node.parent is content and _exact_readonly_text(node, _COLOR_ONLY_SUMMARY, "[396,654][1053,716]") and len(node.element) == 0], + "reveal 后摘要不唯一。", + ) + color_region = _one( + [node for node in nodes if _descendant(node, outer) and _exact_recycler(node, "[36,1000][1080,1483]")], + "reveal 后颜色容器不唯一。", + ) + color = _one( + [node for node in nodes if node.parent is color_region and _exact_color_action(node, "[372,1000][684,1024]", True)], + "reveal 后目标颜色不唯一。", + ) + selected_color_nodes = _require_color_subtree(color, _REVEALED_UNSELECTED_SPEC) + _require_color_action_chain(color, color_region, outer, surface, _REVEALED_UNSELECTED_SPEC) + size_label = _one( + [node for node in nodes if _descendant(node, outer) and _exact_readonly_text(node, _SIZE, "[36,1506][114,1552]")], + "reveal 后尺码标题不唯一。", + ) + size_header = size_label.parent + if size_header is None or not _exact_live_layout(size_header, "android.widget.LinearLayout", "[36,1489][1044,1570]"): + raise SkuSelectionError("reveal 后尺码标题父结构漂移。") + size_options = _one( + [node for node in nodes if _descendant(node, outer) and _exact_inert(node, "android.view.ViewGroup", "[36,1582][1044,1897]")], + "reveal 后尺码 options 根不唯一。", + ) + actions = [node for node in nodes if _descendant(node, size_options) and _is_size_action_text(node)] + s_action = _one( + [node for node in actions if node.text == _S_SIZE_UI and node.bounds == "[36,1582][409,1667]" and node.element.get("selected") == "false"], + "reveal 后 S action 不唯一。", + ) + m_action = _one( + [node for node in actions if node.text == _TARGET_SIZE_UI and node.bounds == _TARGET_SIZE_BOUNDS and node.element.get("selected") == "false"], + "reveal 后 M action 不唯一。", + ) + _require_size_wrapper(s_action, size_options) + _require_size_wrapper(m_action, size_options) + _require_size_action_chain(s_action, size_options, outer, surface) + _require_size_action_chain(m_action, size_options, outer, surface) + if any(node.element.get("selected") == "true" for node in actions): + raise SkuSelectionError("reveal 后已有尺码被选中。") + _require_exact_selected_set(nodes, surface, selected_color_nodes) + dangerous = [ + node + for node in nodes + if node.element.get("package") == PDD_PACKAGE + and node.element.get("class") == "android.widget.TextView" + and "提交订单" in node.text + ] + danger = _one(dangerous, "提交硬拒绝区位置不唯一。") + if ( + not _exact_readonly_text(danger, "选择尺码后，提交订单", "[285,2225][795,2284]") + or len(danger.element) != 0 + ): + raise SkuSelectionError("提交硬拒绝区结构漂移。") + return tuple( + ( + node.element.get("class", ""), + node.bounds, + node.text, + node.desc, + node.element.get("selected", ""), + node.element.get("clickable", ""), + ) + for node in ( + surface, header, outer, price_frame, price_row, current, summary, + color_region, color, size_label, size_options, s_action, m_action, danger, + ) + ) + + def _unit_price(nodes: list[_Node]) -> str: spec = _classify_panel(nodes) if spec.profile is not _PanelProfile.TARGETS_SELECTED: @@ -601,11 +896,15 @@ def _target_color_action(nodes: list[_Node], *, selected: bool) -> _Node: def _target_size_action(nodes: list[_Node], *, selected: bool) -> _Node: profile = _classify_panel(nodes).profile - expected_profile = _PanelProfile.TARGETS_SELECTED if selected else _PanelProfile.SIZE_VISIBLE_NON_TARGET - if profile is not expected_profile: + expected_profiles = ( + {_PanelProfile.TARGETS_SELECTED} + if selected + else {_PanelProfile.SIZE_VISIBLE_UNSELECTED, _PanelProfile.SIZE_VISIBLE_NON_TARGET} + ) + if profile not in expected_profiles: raise SkuSelectionError("目标尺码 action 不属于预期 profile。") return _one( - [node for node in nodes if _is_size_action_text(node) and node.text == _TARGET_SIZE_UI and node.bounds == "[439,1582][831,1667]" and node.element.get("selected") == str(selected).lower()], + [node for node in nodes if _is_size_action_text(node) and node.text == _TARGET_SIZE_UI and node.bounds == _TARGET_SIZE_BOUNDS and node.element.get("selected") == str(selected).lower()], "目标尺码 action 不唯一。", ) @@ -825,7 +1124,7 @@ def _require_color_subtree(color: _Node, spec: _PanelSpec) -> list[_Node]: selected = str(spec.color_selected).lower() children = [child for child in color.element if child.tag == "node"] expected_selected: list[_Node] = [color] if spec.color_selected else [] - if spec.color_bounds == "[372,1188][684,1587]": + if spec.color_bounds == _TARGET_COLOR_UNROLLED_BOUNDS: if len(children) != 4: raise SkuSelectionError("目标颜色完整卡片子树数量漂移。") child_nodes = [node for node in _walk_direct_children(color)] @@ -869,7 +1168,7 @@ def _require_color_action_chain( ) -> None: if color.parent is not color_region: raise SkuSelectionError("目标颜色 action 不属于已取证颜色容器。") - if spec.color_bounds == "[372,1188][684,1587]": + if spec.color_bounds == _TARGET_COLOR_UNROLLED_BOUNDS: expected = ( ("android.widget.FrameLayout", "[0,1188][1080,2046]"), ("android.widget.LinearLayout", "[0,1188][1080,2052]"), diff --git a/client/src/cmbuyer_client/pdd/sku_selection_runner.py b/client/src/cmbuyer_client/pdd/sku_selection_runner.py index 2509143..763d99c 100644 --- a/client/src/cmbuyer_client/pdd/sku_selection_runner.py +++ b/client/src/cmbuyer_client/pdd/sku_selection_runner.py @@ -31,6 +31,10 @@ from .sku_selection import ( SkuSelectionError, SkuSelectionFlow, _SKU_ENTRY_FAILURE_STAGES, + _REVEAL_GESTURE, + _ENTRY_TEXT_BOUNDS, + _TARGET_COLOR_UNROLLED_BOUNDS, + _TARGET_SIZE_BOUNDS, _action_bounds, _annotate_sku_entry_failure, _safe_sku_entry_failure_stage, @@ -135,9 +139,9 @@ class SkuSelectionRunResult: class UiautomatorSkuPanelAdapter(SkuPanelDevice): - """把 uiautomator2 缩为 T-103 所需的读取与三种命名操作。 + """把 uiautomator2 缩为 T-103 所需的读取与四种命名操作。 - ``tap_sku_entry``、``tap_sku_option`` 和 ``leave_sku_panel`` 是仅有的状态改变方法； + 四个命名方法是仅有的状态改变入口；reveal 的手势参数固定且不向 Flow 暴露； 坐标由 Flow 和本类双重检查后才计算中心点，每次调用只执行一次底层动作。 """ @@ -147,7 +151,14 @@ class UiautomatorSkuPanelAdapter(SkuPanelDevice): self._device = device self._timeout_seconds = timeout_seconds self._entry_was_tapped = False + self._entry_bounds_attempted: str | None = None + self._entry_rpc_outcome = "not_attempted" + self._option_bounds_attempted: set[str] = set() + self._option_rpc_outcomes: dict[str, str] = {} + self._reveal_attempted = False + self._reveal_rpc_outcome = "not_attempted" self._left_panel = False + self._back_rpc_outcome = "not_attempted" @property def entry_was_tapped(self) -> bool: @@ -159,6 +170,38 @@ class UiautomatorSkuPanelAdapter(SkuPanelDevice): def left_panel(self) -> bool: return self._left_panel + @property + def reveal_attempted(self) -> bool: + return self._reveal_attempted + + @property + def entry_rpc_outcome(self) -> str: + return self._entry_rpc_outcome + + @property + def entry_bounds_attempted(self) -> str | None: + return self._entry_bounds_attempted + + @property + def option_rpc_outcomes(self) -> tuple[tuple[str, str], ...]: + return tuple(sorted(self._option_rpc_outcomes.items())) + + @property + def reveal_rpc_outcome(self) -> str: + return self._reveal_rpc_outcome + + @property + def option_attempts(self) -> int: + return len(self._option_bounds_attempted) + + @property + def back_attempts(self) -> int: + return int(self._left_panel) + + @property + def back_rpc_outcome(self) -> str: + return self._back_rpc_outcome + def app_info(self, package_name: str) -> dict[str, Any]: value = self._call("app_info", package_name) if not isinstance(value, dict): @@ -178,19 +221,46 @@ class UiautomatorSkuPanelAdapter(SkuPanelDevice): return value def tap_sku_entry(self, bounds: str) -> None: + if self._entry_was_tapped: + raise SkuSelectionDeviceAdapterError("规格入口已经尝试过，拒绝重试。") # 超时也可能表示底层事件已经送达；必须先封存 attempt，后续绝不重试该入口。 self._entry_was_tapped = True + self._entry_bounds_attempted = bounds + self._entry_rpc_outcome = "ambiguous" self._tap_bounds_once(bounds) + self._entry_rpc_outcome = "completed" def tap_sku_option(self, bounds: str) -> None: + if bounds in self._option_bounds_attempted: + raise SkuSelectionDeviceAdapterError("同一规格选项已经尝试过，拒绝重试。") + # 规格 RPC 也可能送达后超时；按 exact bounds 封存本次唯一机会。 + self._option_bounds_attempted.add(bounds) + self._option_rpc_outcomes[bounds] = "ambiguous" self._tap_bounds_once(bounds) + self._option_rpc_outcomes[bounds] = "completed" + + def reveal_size_options_once(self) -> None: + if self._reveal_attempted: + raise SkuSelectionDeviceAdapterError("规格显示动作已经尝试过，拒绝重试。") + # 手势唯一机会在 RPC 前封存；生产 API 不接受方向、坐标或步数参数。 + self._reveal_attempted = True + self._reveal_rpc_outcome = "ambiguous" + self._call( + "jsonrpc_call", + "swipe", + list(_REVEAL_GESTURE), + timeout=self._timeout_seconds, + ) + self._reveal_rpc_outcome = "completed" def leave_sku_panel(self) -> None: if self._left_panel: raise SkuSelectionDeviceAdapterError("规格面板已经执行过返回，已停止操作。") # 底层调用即使报错也可能已把返回事件送达；先封存本次机会，finally 不得再次返回。 self._left_panel = True + self._back_rpc_outcome = "ambiguous" self._call("jsonrpc_call", "pressKey", ["back"], timeout=self._timeout_seconds) + self._back_rpc_outcome = "completed" def capture_screenshot(self) -> str: value = self._call("jsonrpc_call", "takeScreenshot", SCREENSHOT_PARAMS, timeout=self._timeout_seconds) @@ -312,9 +382,11 @@ class SkuSelectionRunner: stage = "safe_exit" flow.exit_sku_panel_safely() + _require_completed_action_audit(adapter) + stage = "publish" manifest_path.write_text( - json.dumps(_manifest(inspection, serial, link, screenshot_path, task_color, task_size), ensure_ascii=False, indent=2, sort_keys=True) + "\n", + json.dumps(_manifest(inspection, serial, link, screenshot_path, task_color, task_size, adapter), ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8", ) # Windows 的 rename 不替换既有目标；并发创建 target 时保留其内容并把本次运行判失败。 @@ -340,11 +412,10 @@ class SkuSelectionRunner: _annotate_mapped_failure(mapped, error, stage) raise mapped from error finally: - # 失败路径只能复用 Flow 的版本、前台和面板证明；证明不了便停止，绝不盲目返回。 - if flow is not None and adapter is not None and adapter.entry_was_tapped and not adapter.left_panel: + # 结果不明只允许读回 pending；失败路径绝不继续后续动作或自动 Back。 + if flow is not None: try: flow.reconcile_pending_action() - flow.exit_sku_panel_safely() except (SkuSelectionRunError, SkuSelectionError): pass @@ -411,9 +482,42 @@ def _require_screenshot_size(screenshot_path: Path) -> None: raise SkuSelectionScreenshotError("原始截图无效，未发布任何证据产物。") from error -def _manifest(inspection: DeviceInspection, serial: str, link: ProductUrl, screenshot_path: Path, task_color: str, task_size: str) -> dict[str, Any]: +def _require_completed_action_audit(adapter: UiautomatorSkuPanelAdapter) -> None: + """发布只接受完整正常链；测试替身或缺动作结果不能伪造真机闭环。""" + + if ( + not adapter.entry_was_tapped + or adapter.entry_bounds_attempted != _ENTRY_TEXT_BOUNDS + or adapter.entry_rpc_outcome != "completed" + or adapter.option_rpc_outcomes + != tuple( + sorted( + ( + (_TARGET_COLOR_UNROLLED_BOUNDS, "completed"), + (_TARGET_SIZE_BOUNDS, "completed"), + ) + ) + ) + or not adapter.reveal_attempted + or adapter.reveal_rpc_outcome != "completed" + or adapter.back_attempts != 1 + or adapter.back_rpc_outcome != "completed" + ): + raise SkuSelectionRunError("规格动作审计链不完整，未发布任何证据产物。") + + +def _manifest( + inspection: DeviceInspection, + serial: str, + link: ProductUrl, + screenshot_path: Path, + task_color: str, + task_size: str, + adapter: UiautomatorSkuPanelAdapter, +) -> dict[str, Any]: """仅写可审计摘要；原始 serial、节点树、页面文案和实际截图内容均不写入 manifest。""" + option_outcomes = dict(adapter.option_rpc_outcomes) return { "schema_version": 1, "captured_at": datetime.now(UTC).isoformat(), @@ -422,8 +526,36 @@ def _manifest(inspection: DeviceInspection, serial: str, link: ProductUrl, scree "target_selection": {"color": task_color, "size": task_size}, "unit_price": EXPECTED_UNIT_PRICE, "selection_status": "restored", - "panel_status": "verified", - "safe_exit": "completed", + "panel_status": "verified_before_back", + "back_attempts": adapter.back_attempts, + "back_rpc_outcome": adapter.back_rpc_outcome, + "actions": { + "sku_entry": { + "attempts": int(adapter.entry_was_tapped), + "rpc_outcome": adapter.entry_rpc_outcome, + }, + "target_color": { + "attempts": int(_TARGET_COLOR_UNROLLED_BOUNDS in option_outcomes), + "rpc_outcome": option_outcomes.get( + _TARGET_COLOR_UNROLLED_BOUNDS, "not_attempted" + ), + }, + "size_reveal": { + "attempts": int(adapter.reveal_attempted), + "rpc_outcome": adapter.reveal_rpc_outcome, + }, + "target_size": { + "attempts": int(_TARGET_SIZE_BOUNDS in option_outcomes), + "rpc_outcome": option_outcomes.get( + _TARGET_SIZE_BOUNDS, "not_attempted" + ), + }, + "back": { + "attempts": adapter.back_attempts, + "rpc_outcome": adapter.back_rpc_outcome, + }, + }, + "post_exit_status": "human_review_required", "page_identity": "human_review_required", "channel": "wifi" if ":" in serial else "usb", "serial_sha256": sha256(serial.encode("utf-8")).hexdigest(), diff --git a/client/tests/pdd/test_sku_reveal_spike.py b/client/tests/pdd/test_sku_reveal_spike.py index bb816de..1458f64 100644 --- a/client/tests/pdd/test_sku_reveal_spike.py +++ b/client/tests/pdd/test_sku_reveal_spike.py @@ -18,17 +18,19 @@ 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, - _candidate_projection, - _RevealEvidenceAdapter, safe_reveal_failure_stage, SkuRevealSpikeCapturer, SkuRevealSpikeError, - _require_safe_reveal_path, ) from cmbuyer_client.pdd.sku_selection import ( SkuSelectionError, _annotate_sku_entry_failure, _parse_nodes, + _require_safe_reveal_path, + _revealed_unselected_projection as _candidate_projection, +) +from cmbuyer_client.pdd.sku_selection_runner import ( + UiautomatorSkuPanelAdapter as _RevealEvidenceAdapter, ) @@ -350,7 +352,7 @@ def _candidate_cross_variant_residue(variant: str, kind: str) -> str: def _candidate_with_color_unselected() -> str: - root = ElementTree.fromstring(_candidate_unselected()) + root = ElementTree.fromstring(_candidate_current_only()) target = next( node for node in root.iter("node") @@ -364,7 +366,7 @@ def _candidate_with_color_unselected() -> str: def _candidate_with_submit_risk(kind: str) -> str: - root = ElementTree.fromstring(_candidate_unselected()) + root = ElementTree.fromstring(_candidate_current_only()) parents = {child: parent for parent in root.iter() for child in parent} submit = next( node @@ -379,6 +381,13 @@ def _candidate_with_submit_risk(kind: str) -> str: parent.append(ElementTree.fromstring(ElementTree.tostring(submit, encoding="unicode"))) elif kind == "moved_up": submit.set("bounds", "[285,1900][795,1959]") + elif kind in {"extra_altered", "extra_moved_up"}: + extra = ElementTree.fromstring(ElementTree.tostring(submit, encoding="unicode")) + if kind == "extra_altered": + extra.set("text", "异常提交订单文案") + else: + extra.set("bounds", "[285,1900][795,1959]") + parent.append(extra) else: raise AssertionError(kind) return ElementTree.tostring(root, encoding="unicode") @@ -416,9 +425,9 @@ 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.empty_hierarchy = _CURRENT_ONLY_EMPTY + self.color_hierarchy = _CURRENT_ONLY_COLOR + self.after_hierarchy = _candidate_current_only() self.calls: list[tuple[object, ...]] = [] self.swipe_error = False @@ -545,8 +554,9 @@ class SkuRevealSpikeTests(unittest.TestCase): self.assertTrue((target / "after" / "hierarchy.xml").is_file()) def test_candidate_price_variants_reject_hybrid_duplicate_unknown_and_parent_drift(self) -> None: - # 两个真实嵌套 variant 各自唯一命中；任意混搭、额外子节点或父子关系漂移均拒绝。 - _candidate_projection(_parse_nodes(_candidate_unselected())) + # 生产只放行 current-only；历史 dual 及任意混搭、额外节点、父链漂移均拒绝。 + with self.assertRaises(SkuSelectionError): + _candidate_projection(_parse_nodes(_candidate_unselected())) _candidate_projection(_parse_nodes(_candidate_current_only())) for variant in ("dual", "current_only"): kinds = ["hybrid", "duplicate", "unknown", "unknown_child", "parent_drift"] @@ -683,7 +693,7 @@ class SkuRevealSpikeTests(unittest.TestCase): self._assert_after_rejected_once(_candidate_with_color_unselected()) def test_submit_hard_reject_zone_risk_is_not_published_or_retried(self) -> None: - for kind in ("missing", "duplicate", "moved_up"): + for kind in ("missing", "duplicate", "moved_up", "extra_altered", "extra_moved_up"): with self.subTest(kind=kind): self._assert_after_rejected_once(_candidate_with_submit_risk(kind)) @@ -739,7 +749,7 @@ class SkuRevealSpikeTests(unittest.TestCase): device.jsonrpc_call = drift # type: ignore[method-assign] with TemporaryDirectory() as directory: target = Path(directory) / "evidence" - with self.assertRaises(SkuRevealSpikeError) as raised: + with self.assertRaises((SkuRevealSpikeError, SkuSelectionError)) as raised: self._capturer(device).capture( "192.168.0.173:5555", "937122477375", @@ -850,7 +860,7 @@ class SkuRevealSpikeTests(unittest.TestCase): "not-a-bound", ), ): - with self.subTest(), self.assertRaises(SkuRevealSpikeError): + with self.subTest(), self.assertRaises((SkuRevealSpikeError, SkuSelectionError)): _require_safe_reveal_path(_parse_nodes(hierarchy)) diff --git a/client/tests/pdd/test_sku_selection.py b/client/tests/pdd/test_sku_selection.py index 586a60f..e718ca3 100644 --- a/client/tests/pdd/test_sku_selection.py +++ b/client/tests/pdd/test_sku_selection.py @@ -3,8 +3,10 @@ from __future__ import annotations import ast import base64 from contextlib import redirect_stderr +from functools import lru_cache from io import BytesIO import importlib.util +import json from pathlib import Path from tempfile import TemporaryDirectory import unittest @@ -49,6 +51,14 @@ _TASK_SIZE = "M(建议100-115)" _PRODUCT_PAGE = _ENTRY_FIXTURE.read_text(encoding="utf-8") +def _revealed_current_only() -> str: + # 与 spike 测试共用由已验 M fixture 机械恢复出的 current-only 未选态；生产判据只在源码一处。 + from tests.pdd.test_sku_reveal_spike import _candidate_current_only + + return _candidate_current_only() + + +@lru_cache(maxsize=1) def _png_base64() -> str: image = Image.new("RGB", (1080, 2376), "white") raw = BytesIO() @@ -59,8 +69,9 @@ def _png_base64() -> str: 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.panel_hierarchy = _CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8") + self.color_hierarchy = _CURRENT_ONLY_COLOR_FIXTURE.read_text(encoding="utf-8") + self.revealed_hierarchy = _revealed_current_only() self.version = "8.17.0" self.package = "com.xunmeng.pinduoduo" self.screenshot = _png_base64() if screenshot is None else screenshot @@ -89,6 +100,11 @@ class _RawDevice: raise AssertionError(params) self._apply_tap(int(params[0]), int(params[1])) return "" + if method == "swipe": + if params != [360, 1900, 360, 1300, 30]: + raise AssertionError(params) + self.hierarchy = self.revealed_hierarchy + return "" raise AssertionError(method) def _apply_tap(self, x: int, y: int) -> None: @@ -405,25 +421,223 @@ class SkuSelectionFlowTests(unittest.TestCase): flow.open_sku_panel(_TARGET_URL) self.assertEqual(_actions(device, "click"), []) - def test_target_mapping_stops_at_unproven_reveal_then_s_to_m_is_exact(self) -> None: + def test_current_only_exact_transition_reveals_then_selects_m_once(self) -> None: device = _RawDevice() flow = _flow(device) flow.open_sku_panel(_TARGET_URL) - with self.assertRaisesRegex(SkuSelectionError, "受控显示动作尚未取证"): - flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE)) + flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE)) + self.assertEqual(flow.read_sku_unit_price(), "12.88") - self.assertEqual(_tap_centers(device), [(865, 2218), (528, 1387)]) + self.assertEqual(_tap_centers(device), [(865, 2218), (528, 1387), (635, 1624)]) + self.assertEqual(len(_actions(device, "swipe")), 1) self.assertEqual(_actions(device, "pressKey"), []) restored = _RawDevice(_S_FIXTURE.read_text(encoding="utf-8")) restored_flow = _flow(restored) restored_flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE)) self.assertEqual(restored_flow.read_sku_unit_price(), "12.88") - restored_flow.exit_sku_panel_safely() - self.assertEqual(_tap_centers(restored), [(635, 1624)]) - self.assertEqual(_actions(restored, "pressKey"), [("jsonrpc", "pressKey", ["back"], 10)]) + self.assertEqual(_actions(restored, "swipe"), []) + + def test_dual_color_only_is_zero_action_before_reveal(self) -> None: + for fixture in (_COLOR_FIXTURE, _CURRENT_ONLY_COLOR_FIXTURE): + with self.subTest(fixture=fixture.name): + device = _RawDevice(fixture.read_text(encoding="utf-8")) + with self.assertRaises(SkuSelectionError): + _flow(device).select_sku_options( + resolve_task_selection(_TASK_COLOR, _TASK_SIZE) + ) + self.assertEqual(_actions(device, "click"), []) + self.assertEqual(_actions(device, "swipe"), []) + + def test_revealed_current_only_is_a_unique_profile_but_not_a_reentry_shortcut(self) -> None: + hierarchy = _revealed_current_only() + spec = _classify_panel(_parse_nodes(hierarchy)) + self.assertIs(spec.profile, _PanelProfile.SIZE_VISIBLE_UNSELECTED) + self.assertEqual(spec.price_layout.__class__.__name__, "_CurrentOnlyPriceLayout") + + device = _RawDevice(hierarchy) + with self.assertRaises(SkuSelectionError): + _flow(device).select_sku_options( + resolve_task_selection(_TASK_COLOR, _TASK_SIZE) + ) + self.assertEqual(_actions(device, "click"), []) + self.assertEqual(_actions(device, "swipe"), []) + + def test_revealed_exact_m_contract_rejects_missing_duplicate_disabled_near_and_other_selected(self) -> None: + base = _revealed_current_only() + for kind in ("missing", "duplicate", "disabled", "near", "other_selected"): + root = ElementTree.fromstring(base) + parents = {child: parent for parent in root.iter() for child in parent} + m_action = next( + node for node in root.iter("node") + if node.get("text") == "M（建议100-115）" + ) + wrapper = parents[m_action] + if kind == "missing": + parents[wrapper].remove(wrapper) + elif kind == "duplicate": + parents[wrapper].append( + ElementTree.fromstring(ElementTree.tostring(wrapper, encoding="unicode")) + ) + elif kind == "disabled": + m_action.set("enabled", "false") + elif kind == "near": + m_action.set("text", "M（建议100-115）相近") + else: + s_action = next( + node for node in root.iter("node") + if node.get("text") == "S（建议80-100）" + ) + s_action.set("selected", "true") + with self.subTest(kind=kind), self.assertRaises(SkuSelectionError): + _classify_panel( + _parse_nodes(ElementTree.tostring(root, encoding="unicode")) + ) + + def test_reveal_requires_two_consecutive_stable_frames_and_terminal_blocks_reentry(self) -> None: + class UnstableRevealDevice(_RawDevice): + def __init__(self) -> None: + super().__init__(_CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8")) + self.after_reveal_reads = 0 + + def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str: + if method == "dumpWindowHierarchy" and _actions(self, "swipe"): + self.after_reveal_reads += 1 + candidate = _revealed_current_only() + self.hierarchy = ( + candidate + if self.after_reveal_reads % 2 + else candidate.replace("请选择： 尺码", "未知摘要") + ) + return super().jsonrpc_call(method, params, timeout) + + device = UnstableRevealDevice() + flow = _flow(device) + with self.assertRaises(SkuSelectionError): + flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE)) + action_count = len(_actions(device, "click")) + len(_actions(device, "swipe")) + self.assertEqual(len(_actions(device, "swipe")), 1) + self.assertEqual(_tap_centers(device), [(528, 1387)]) + with self.assertRaisesRegex(SkuSelectionError, "不可重入终止态"): + flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE)) + self.assertEqual( + len(_actions(device, "click")) + len(_actions(device, "swipe")), + action_count, + ) + + def test_reveal_timeout_is_read_only_reconciled_without_m_back_or_retry(self) -> None: + class RevealTimeoutDevice(_RawDevice): + def __init__(self, delivered: bool) -> None: + super().__init__(_CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8")) + self.delivered = delivered + + def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str: + if method == "swipe": + if self.delivered: + super().jsonrpc_call(method, params, timeout) + else: + self.calls.append(("jsonrpc", method, params, timeout)) + raise TimeoutError("private reveal timeout") + return super().jsonrpc_call(method, params, timeout) + + for delivered in (False, True): + with self.subTest(delivered=delivered): + device = RevealTimeoutDevice(delivered) + flow = _flow(device) + with self.assertRaises(SkuSelectionRunError): + flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE)) + try: + flow.reconcile_pending_action() + except SkuSelectionError: + pass + self.assertEqual(len(_actions(device, "swipe")), 1) + self.assertEqual(_tap_centers(device), [(528, 1387)]) + self.assertEqual(_actions(device, "pressKey"), []) + with self.assertRaisesRegex(SkuSelectionError, "不可重入终止态"): + flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE)) + self.assertEqual(len(_actions(device, "swipe")), 1) + + def test_m_timeout_is_read_only_reconciled_without_back_or_retry(self) -> None: + class MTimeoutDevice(_RawDevice): + def __init__(self, delivered: bool) -> None: + super().__init__(_CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8")) + self.delivered = delivered + + def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str: + if method == "click" and params == [635, 1624]: + if self.delivered: + super().jsonrpc_call(method, params, timeout) + else: + self.calls.append(("jsonrpc", method, params, timeout)) + raise TimeoutError("private M timeout") + return super().jsonrpc_call(method, params, timeout) + + for delivered in (False, True): + with self.subTest(delivered=delivered): + device = MTimeoutDevice(delivered) + flow = _flow(device) + with self.assertRaises(SkuSelectionRunError): + flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE)) + try: + flow.reconcile_pending_action() + except SkuSelectionError: + pass + self.assertEqual(len(_actions(device, "swipe")), 1) + self.assertEqual(_tap_centers(device), [(528, 1387), (635, 1624)]) + self.assertEqual(_actions(device, "pressKey"), []) + with self.assertRaisesRegex(SkuSelectionError, "不可重入终止态"): + flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE)) + self.assertEqual(_tap_centers(device), [(528, 1387), (635, 1624)]) + + def test_foreground_is_freshly_rechecked_before_s_reveal_and_m_mutations(self) -> None: + class ForegroundDriftDevice(_RawDevice): + def __init__(self, hierarchy: str, flip_at: int) -> None: + super().__init__(hierarchy) + self.flip_at = flip_at + self.current_reads = 0 + + def app_current(self) -> dict[str, str]: + current = super().app_current() + self.current_reads += 1 + if self.current_reads == self.flip_at: + return {"package": "other"} + return current + + cases = ( + ( + "s", + _S_FIXTURE.read_text(encoding="utf-8"), + 2, + [], + 0, + ), + ( + "reveal", + _CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8"), + 4, + [(528, 1387)], + 0, + ), + ( + "m", + _CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8"), + 7, + [(528, 1387)], + 1, + ), + ) + for name, hierarchy, flip_at, expected_taps, expected_swipes in cases: + with self.subTest(name=name): + device = ForegroundDriftDevice(hierarchy, flip_at) + with self.assertRaises(SkuSelectionError): + _flow(device).select_sku_options( + resolve_task_selection(_TASK_COLOR, _TASK_SIZE) + ) + self.assertEqual(_tap_centers(device), expected_taps) + self.assertEqual(len(_actions(device, "swipe")), expected_swipes) + self.assertEqual(_actions(device, "pressKey"), []) def test_current_only_empty_to_color_only_uses_the_matching_exact_spec(self) -> None: device = _RawDevice() @@ -432,15 +646,15 @@ class SkuSelectionFlowTests(unittest.TestCase): flow = _flow(device) flow.open_sku_panel(_TARGET_URL) - with self.assertRaisesRegex(SkuSelectionError, "受控显示动作尚未取证"): - flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE)) + flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE)) - self.assertEqual(_tap_centers(device), [(865, 2218), (528, 1387)]) + self.assertEqual(_tap_centers(device), [(865, 2218), (528, 1387), (635, 1624)]) + self.assertEqual(len(_actions(device, "swipe")), 1) matched = _classify_panel(_parse_nodes(device.hierarchy)) - self.assertIs(matched.profile, _PanelProfile.COLOR_SELECTED_SIZE_HIDDEN) + self.assertIs(matched.profile, _PanelProfile.TARGETS_SELECTED) self.assertEqual( matched.price_layout.__class__.__name__, - "_CurrentOnlyPriceLayout", + "_DualPriceLayout", ) def test_dual_and_current_only_variants_each_match_one_complete_spec(self) -> None: @@ -585,7 +799,9 @@ class SkuSelectionFlowTests(unittest.TestCase): 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)]) + expected = [(865, 2218), (528, 1387)] if empty_fixture is _CURRENT_ONLY_EMPTY_FIXTURE else [(865, 2218)] + self.assertEqual(_tap_centers(device), expected) + self.assertEqual(_actions(device, "swipe"), []) self.assertEqual(_actions(device, "pressKey"), []) def test_full_verified_entry_structure_taps_exact_text_child_once(self) -> None: @@ -851,10 +1067,8 @@ class SkuSelectionFlowTests(unittest.TestCase): flow.open_sku_panel(_TARGET_URL) device.select_alternates() flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE)) - self.assertEqual( - _tap_centers(device), - [(865, 2218), (635, 1624)], - ) + self.assertEqual(_tap_centers(device), [(865, 2218), (635, 1624)]) + self.assertEqual(_actions(device, "swipe"), []) def test_price_rejects_coupon_prefix_extra_amount_and_bottom_action(self) -> None: for replacement in ("券后 ¥12.88", "会员补贴 ¥12.88", "到手 ¥12.88", "实付 ¥12.88", "区间 ¥12.88", "原价 ¥12.88", "划线价 ¥12.88", "最低 ¥12.88", "低至 ¥12.88", "起价 ¥12.88", "快卖完 1 ¥12.88", "快卖完 ¥12.88 ¥11.88"): @@ -876,7 +1090,7 @@ class SkuSelectionFlowTests(unittest.TestCase): _flow(_RawDevice(clickable_parent)).read_sku_unit_price() def test_public_api_and_protocol_have_no_broad_or_order_operations(self) -> None: - forbidden = {"quantity", "confirm", "authorization", "fence", "submit", "payment", "click"} + forbidden = {"quantity", "confirm", "authorization", "fence", "submit", "payment", "click", "swipe", "scroll"} self.assertTrue(forbidden.isdisjoint(SkuSelectionFlow.__dict__)) self.assertTrue(forbidden.isdisjoint(SkuPanelDevice.__dict__)) self.assertTrue(forbidden.isdisjoint(pdd.__all__)) @@ -900,6 +1114,8 @@ class SkuSelectionFlowTests(unittest.TestCase): runner_tree = ast.parse(files[1].read_text(encoding="utf-8")) click_calls = [node for node in ast.walk(runner_tree) if isinstance(node, ast.Constant) and node.value == "click"] self.assertEqual(len(click_calls), 1) + swipe_calls = [node for node in ast.walk(runner_tree) if isinstance(node, ast.Constant) and node.value == "swipe"] + self.assertEqual(len(swipe_calls), 1) def test_entry_wait_rejects_unchanged_or_duplicate_page_without_click(self) -> None: now = [0.0] @@ -947,22 +1163,24 @@ class SkuSelectionFlowTests(unittest.TestCase): class _CompletedFlow: - """仅隔离 runner 文件发布测试；生产 Flow 在 reveal 取证前仍必须停止。""" + """仅隔离 runner 文件发布测试，同时必须形成完整动作审计链。""" def __init__(self, device: object, *args: object, **kwargs: object) -> None: self.device = device def open_sku_panel(self, product_url: str, pre_intent_hierarchy: str | None = None) -> None: - return None + self.device.tap_sku_entry("[688,2184][1042,2253]") def select_sku_options(self, selection: object) -> None: - return None + self.device.tap_sku_option("[372,1188][684,1587]") + self.device.reveal_size_options_once() + self.device.tap_sku_option("[439,1582][831,1667]") def verify_target_selection_and_read_price(self, selection: object) -> str: return "12.88" def exit_sku_panel_safely(self) -> None: - return None + self.device.leave_sku_panel() def reconcile_pending_action(self) -> None: return None @@ -989,15 +1207,55 @@ class SkuSelectionRunnerTests(unittest.TestCase): self.assertEqual(result.unit_price, "12.88") manifest = result.manifest_path.read_text(encoding="utf-8") + manifest_data = json.loads(manifest) self.assertTrue(result.screenshot_path.is_file()) self.assertNotIn("device-1", manifest) self.assertNotIn("hierarchy", manifest) self.assertNotIn("已选", manifest) self.assertIn('"unit_price": "12.88"', manifest) self.assertIn('"selection_status": "restored"', manifest) - self.assertIn('"panel_status": "verified"', manifest) - self.assertIn('"safe_exit": "completed"', manifest) + 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.assertEqual( + manifest_data["actions"], + { + "sku_entry": {"attempts": 1, "rpc_outcome": "completed"}, + "target_color": {"attempts": 1, "rpc_outcome": "completed"}, + "size_reveal": {"attempts": 1, "rpc_outcome": "completed"}, + "target_size": {"attempts": 1, "rpc_outcome": "completed"}, + "back": {"attempts": 1, "rpc_outcome": "completed"}, + }, + ) self.assertFalse((target / "hierarchy.xml").exists()) + self.assertEqual(len(_actions(device, "pressKey")), 1) + + def test_publish_rejects_flow_stub_without_exact_action_audit_chain(self) -> None: + class IncompleteFlow(_CompletedFlow): + def open_sku_panel(self, product_url: str, pre_intent_hierarchy: str | None = None) -> None: + return None + + def select_sku_options(self, selection: object) -> None: + return None + + def exit_sku_panel_safely(self) -> None: + return None + + device = _RawDevice() + with TemporaryDirectory() as temporary: + target = Path(temporary) / "result" + with ( + patch.object(runner_module, "SkuSelectionFlow", IncompleteFlow), + self.assertRaises(SkuSelectionRunError) as raised, + ): + self._runner(_FakeAdb(), device).run( + "device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target + ) + self.assertEqual(safe_failure_stage(raised.exception), "safe_exit") + self.assertFalse(target.exists()) + self.assertEqual(list(Path(temporary).glob(".result.staging-*")), []) self.assertEqual(_actions(device, "pressKey"), []) def test_failure_stage_is_fixed_control_flow_metadata_without_error_text(self) -> None: @@ -1223,6 +1481,34 @@ class SkuSelectionRunnerTests(unittest.TestCase): self.assertTrue(adapter.entry_was_tapped) self.assertEqual(_actions(adapter._device, "click"), [("jsonrpc", "click", [865, 2218], 10)]) + def test_adapter_seals_every_named_mutation_before_rpc(self) -> None: + device = _RawDevice(_PRODUCT_PAGE) + adapter = UiautomatorSkuPanelAdapter(device, 10) + adapter.tap_sku_entry("[688,2184][1042,2253]") + with self.assertRaises(SkuSelectionDeviceAdapterError): + adapter.tap_sku_entry("[688,2184][1042,2253]") + + adapter.tap_sku_option("[372,1188][684,1587]") + with self.assertRaises(SkuSelectionDeviceAdapterError): + adapter.tap_sku_option("[372,1188][684,1587]") + + adapter.reveal_size_options_once() + with self.assertRaises(SkuSelectionDeviceAdapterError): + adapter.reveal_size_options_once() + + adapter.leave_sku_panel() + with self.assertRaises(SkuSelectionDeviceAdapterError): + adapter.leave_sku_panel() + + self.assertEqual(len(_actions(device, "click")), 2) + self.assertEqual(len(_actions(device, "swipe")), 1) + self.assertEqual(len(_actions(device, "pressKey")), 1) + self.assertEqual(adapter.option_attempts, 1) + self.assertEqual(adapter.entry_rpc_outcome, "completed") + self.assertEqual(adapter.reveal_rpc_outcome, "completed") + self.assertEqual(adapter.back_attempts, 1) + self.assertEqual(adapter.back_rpc_outcome, "completed") + def test_entry_stability_interruptions_never_click(self) -> None: now = [0.0] class SequenceDevice(_RawDevice): @@ -1276,8 +1562,48 @@ class SkuSelectionRunnerTests(unittest.TestCase): with self.assertRaises(SkuSelectionError): _flow(device).exit_sku_panel_safely() self.assertEqual(_actions(device, "pressKey"), []) - def test_unproven_reveal_publishes_nothing_and_safely_exits_once(self) -> None: + def test_back_precondition_rejects_every_non_target_profile_after_screenshot_reverify(self) -> None: + drifts = ( + _CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8"), + _CURRENT_ONLY_COLOR_FIXTURE.read_text(encoding="utf-8"), + _S_FIXTURE.read_text(encoding="utf-8"), + _revealed_current_only(), + ) + + for drift in drifts: + class DriftBeforeBackDevice(_RawDevice): + def __init__(self) -> None: + super().__init__() + self.after_screenshot_reads = 0 + self.screenshot_seen = False + + def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str: + if method == "takeScreenshot": + value = super().jsonrpc_call(method, params, timeout) + self.screenshot_seen = True + return value + if method == "dumpWindowHierarchy" and self.screenshot_seen: + self.after_screenshot_reads += 1 + if self.after_screenshot_reads == 2: + self.hierarchy = drift + return super().jsonrpc_call(method, params, timeout) + + with self.subTest(profile=_classify_panel(_parse_nodes(drift)).profile), TemporaryDirectory() as temporary: + device = DriftBeforeBackDevice() + target = Path(temporary) / "result" + with self.assertRaises(SkuSelectionError) as raised: + self._runner(_FakeAdb(), device).run( + "device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target + ) + self.assertEqual(safe_failure_stage(raised.exception), "safe_exit") + self.assertEqual(_actions(device, "pressKey"), []) + self.assertFalse(target.exists()) + self.assertEqual(list(Path(temporary).glob(".result.staging-*")), []) + + def test_dual_precondition_publishes_nothing_without_reveal_or_back(self) -> None: device = _RawDevice() + device.panel_hierarchy = _EMPTY_FIXTURE.read_text(encoding="utf-8") + device.color_hierarchy = _COLOR_FIXTURE.read_text(encoding="utf-8") with TemporaryDirectory() as temporary: target = Path(temporary) / "out" with self.assertRaises(SkuSelectionError): @@ -1285,7 +1611,8 @@ class SkuSelectionRunnerTests(unittest.TestCase): self.assertFalse(target.exists()) self.assertFalse((target / "manifest.json").exists()) self.assertEqual(list(Path(temporary).glob(".out.staging-*")), []) - self.assertEqual(len(_actions(device, "pressKey")), 1) + self.assertEqual(_actions(device, "swipe"), []) + self.assertEqual(len(_actions(device, "pressKey")), 0) def test_option_timeout_reconciliation_controls_back_once(self) -> None: class OptionTimeoutDevice(_RawDevice): @@ -1300,7 +1627,7 @@ class SkuSelectionRunnerTests(unittest.TestCase): raise TimeoutError("uncertain option") return super().jsonrpc_call(method, params, timeout) - for delivered, expected_back in ((False, 0), (True, 1)): + for delivered, expected_back in ((False, 0), (True, 0)): with self.subTest(delivered=delivered), TemporaryDirectory() as temporary: device = OptionTimeoutDevice(delivered) adb = _FakeAdb(); device.hierarchy = "" @@ -1336,7 +1663,7 @@ class SkuSelectionRunnerTests(unittest.TestCase): with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionRunError): self._runner(_FakeAdb(), device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result") self.assertEqual(len(_actions(device, "click")), 1) - self.assertEqual(len(_actions(device, "pressKey")), 1) + self.assertEqual(len(_actions(device, "pressKey")), 0) class SkuSelectionCliTests(unittest.TestCase):