fix: 安全处理采购数量输入法 (#117)

client/src/pdd_u2_purchase_adapter.py

@@ -61,6 +61,13 @@ _PAYMENT_MARKERS = ("输入支付密码", "立即支付", "支付成功", "支
 _FINAL_SUBMIT_MARKERS = ("提交订单", "现在买，仅", "确认购买")
 _OUT_OF_STOCK_MARKERS = ("已售罄", "暂时缺货", "库存不足", "该商品已售罄")
 _SUPPORTED_OPTION_KEYS = frozenset({"color", "size"})
+_MAX_QUANTITY_BUTTON_CLICKS = 5
+_KEYBOARD_STATUS_MARKERS = (
+    "minputshown",
+    "misinputviewshown",
+    "minputviewshown",
+    "inputshown",
+)
 _DIAGNOSTIC_MARKERS = (
     "增加数量",
     "减少数量",
@@ -196,6 +203,52 @@ def _quantity(root: ET.Element) -> int:
     return values[0] if len(values) == 1 else 0
 
 
+def _quantity_button_targets(root: ET.Element, description: str) -> list[Bounds]:
+    """返回唯一、可点击的数量加减按钮坐标候选。"""
+
+    targets = set()
+    for node in root.iter("node"):
+        labels = {
+            node.get("text", "").strip(),
+            node.get("content-desc", "").strip(),
+        }
+        bounds = _parse_bounds(node.get("bounds", ""))
+        if description not in labels or bounds is None:
+            continue
+        if node.get("clickable") != "true":
+            continue
+        if node.get("enabled") == "false" or node.get("visible-to-user") == "false":
+            continue
+        targets.add(bounds)
+    return sorted(targets)
+
+
+def _shell_output(response: Any) -> str:
+    """兼容 uiautomator2 ShellResponse 和测试中的普通字符串。"""
+
+    output = getattr(response, "output", response)
+    if isinstance(output, bytes):
+        return output.decode("utf-8", errors="replace")
+    return str(output or "")
+
+
+def _keyboard_is_shown(device: Any) -> Optional[bool]:
+    """读取安卓输入法状态；无法可靠判断时返回 None。"""
+
+    try:
+        response = device.shell("dumpsys input_method")
+    except (AttributeError, OSError, RuntimeError):
+        return None
+    normalized = re.sub(r"\s+", "", _shell_output(response)).lower()
+    found_false = False
+    for marker in _KEYBOARD_STATUS_MARKERS:
+        if f"{marker}=true" in normalized:
+            return True
+        if f"{marker}=false" in normalized:
+            found_false = True
+    return False if found_false else None
+
+
 def _price_cent(root: ET.Element) -> int:
     """优先读取页面上半部的当前单价，不把划线价当成当前价。"""
 
@@ -439,26 +492,150 @@ class U2PddPurchaseAdapter(PddPurchaseAdapter):
             )
         self._check_cancelled("purchase_set_quantity")
         try:
-            editor = device(className="android.widget.EditText")
-            if int(getattr(editor, "count", 0)) != 1:
+            root = _parse_xml(self._dump_hierarchy())
+            current_quantity = _quantity(root)
+            if current_quantity <= 0:
                 raise PddPurchaseError(
                     "PURCHASE_QUANTITY_CONTROL_MISSING",
                     "页面没有唯一可靠的采购数量输入框",
                     step="purchase_set_quantity",
                 )
-            editor.set_text(str(quantity))
-            self._sleep(0.2)
-            if _quantity(_parse_xml(self._dump_hierarchy())) != quantity:
-                raise PddPurchaseError(
-                    "PURCHASE_QUANTITY_MISMATCH",
-                    "PDD 页面没有保存目标采购数量",
-                    step="purchase_set_quantity",
-                )
+            if current_quantity == quantity:
+                return
+
+            difference = quantity - current_quantity
+            if abs(difference) <= _MAX_QUANTITY_BUTTON_CLICKS:
+                description = "增加数量" if difference > 0 else "减少数量"
+                targets = _quantity_button_targets(root, description)
+                if len(targets) == 1:
+                    self._set_quantity_with_buttons(
+                        current_quantity, quantity, description
+                    )
+                    return
+
+            self._set_quantity_with_editor(quantity)
         except PddPurchaseError:
             raise
         except Exception as exc:
             self._raise_device_or_page_error(exc, "purchase_set_quantity")
 
+    def _set_quantity_with_buttons(
+        self, current_quantity: int, target_quantity: int, description: str
+    ) -> None:
+        """逐次点击加减按钮，每次都用最新控件树确认数量。"""
+
+        device = self._require_device()
+        step = 1 if target_quantity > current_quantity else -1
+        expected = current_quantity
+        while expected != target_quantity:
+            self._check_cancelled("purchase_set_quantity")
+            root = _parse_xml(self._dump_hierarchy())
+            targets = _quantity_button_targets(root, description)
+            if len(targets) != 1:
+                raise PddPurchaseError(
+                    "PURCHASE_QUANTITY_CONTROL_MISSING",
+                    f"页面没有唯一可靠的“{description}”按钮",
+                    step="purchase_set_quantity",
+                )
+            left, top, right, bottom = targets[0]
+            device.click((left + right) // 2, (top + bottom) // 2)
+            expected += step
+            self._sleep(0.2)
+            actual = _quantity(_parse_xml(self._dump_hierarchy()))
+            if actual != expected:
+                raise PddPurchaseError(
+                    "PURCHASE_QUANTITY_MISMATCH",
+                    f"调整采购数量后期望为 {expected}，页面实际为 {actual or '未知'}",
+                    step="purchase_set_quantity",
+                )
+
+    def _set_quantity_with_editor(self, quantity: int) -> None:
+        """输入框兜底；只在确认输入法显示后按一次返回键。"""
+
+        device = self._require_device()
+        editor = device(className="android.widget.EditText")
+        if int(getattr(editor, "count", 0)) != 1:
+            raise PddPurchaseError(
+                "PURCHASE_QUANTITY_CONTROL_MISSING",
+                "页面没有唯一可靠的采购数量输入框",
+                step="purchase_set_quantity",
+            )
+        editor.set_text(str(quantity))
+        self._sleep(0.2)
+        if _quantity(_parse_xml(self._dump_hierarchy())) != quantity:
+            raise PddPurchaseError(
+                "PURCHASE_QUANTITY_MISMATCH",
+                "PDD 页面没有保存目标采购数量",
+                step="purchase_set_quantity",
+            )
+
+        keyboard_shown = _keyboard_is_shown(device)
+        if keyboard_shown is None:
+            raise PddPurchaseError(
+                "PURCHASE_KEYBOARD_DISMISS_FAILED",
+                "无法确认安卓输入法状态，已停止采购以避免误按返回键",
+                step="purchase_set_quantity",
+                retryable=True,
+            )
+        if keyboard_shown:
+            device.press("back")
+            for _attempt in range(25):
+                self._check_cancelled("purchase_set_quantity")
+                self._sleep(0.2)
+                keyboard_shown = _keyboard_is_shown(device)
+                if keyboard_shown is False:
+                    break
+            if keyboard_shown is not False:
+                raise PddPurchaseError(
+                    "PURCHASE_KEYBOARD_DISMISS_FAILED",
+                    "安卓输入法没有在规定时间内关闭，已停止采购",
+                    step="purchase_set_quantity",
+                    retryable=True,
+                )
+
+        self._validate_panel_after_keyboard(quantity)
+
+    def _validate_panel_after_keyboard(self, quantity: int) -> None:
+        """输入框操作后重新确认规格面板和最终提交前状态。"""
+
+        root = _parse_xml(self._dump_hierarchy())
+        if _page_kind(root, "") != "order_confirmation":
+            raise PddPurchaseError(
+                "PURCHASE_PANEL_LOST_AFTER_KEYBOARD",
+                "关闭输入法后规格面板已经消失，已停止采购",
+                step="purchase_set_quantity",
+                retryable=True,
+            )
+        if _quantity(root) != quantity:
+            raise PddPurchaseError(
+                "PURCHASE_QUANTITY_MISMATCH",
+                "关闭输入法后采购数量发生变化",
+                step="purchase_set_quantity",
+            )
+        missing_options = [
+            value
+            for value in self._requested_options.values()
+            if not _selected(root, value)
+        ]
+        if missing_options:
+            raise PddPurchaseError(
+                "PURCHASE_OPTIONS_MISMATCH",
+                "关闭输入法后已选颜色或尺码发生变化",
+                step="purchase_set_quantity",
+            )
+        if _price_cent(root) <= 0:
+            raise PddPurchaseError(
+                "PURCHASE_PRICE_MISSING",
+                "关闭输入法后没有识别到有效价格",
+                step="purchase_set_quantity",
+            )
+        if len(_final_submit_targets(root)) != 1:
+            raise PddPurchaseError(
+                "PURCHASE_SUBMIT_TARGET_AMBIGUOUS",
+                "关闭输入法后没有唯一可靠的下单按钮",
+                step="purchase_set_quantity",
+            )
+
     def enter_confirmation(self) -> None:
         """只确认已到最终提交前页面，不点击底部提交按钮。"""