chengma/cmbuyer
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(client): productionize proven sku reveal flow

Browse Source
This commit is contained in:
QiuSW
2026-08-05 16:55:10 +08:00
parent fec21ae857
commit 60cf05b08b
5 changed files with 895 additions and 487 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/cmbuyer_client/pdd/sku_reveal_spike.py
+13 -373
View File
@@ -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")
client/src/cmbuyer_client/pdd/sku_selection.py
+361 -62
View File
@@ -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]"),
client/src/cmbuyer_client/pdd/sku_selection_runner.py
+141 -9
View File
@@ -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(),
client/tests/pdd/test_sku_reveal_spike.py
+23 -13
View File
@@ -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 = "<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))
client/tests/pdd/test_sku_selection.py
+357 -30
View File
@@ -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 = "<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):