diff --git a/client/src/pdd_collect_service.py b/client/src/pdd_collect_service.py
index 68e0acc..4c7d1a9 100644
--- a/client/src/pdd_collect_service.py
+++ b/client/src/pdd_collect_service.py
@@ -645,6 +645,132 @@ def _is_spec_panel_open(xml_data: str | bytes) -> bool:
return has_selection_summary or has_submit_hint or has_dimension_heading
+def _quick_confirmation_spec_coord(root: ET.Element) -> Optional[tuple[int, int]]:
+ """定位快速确认页中的当前主规格值，不接触底部购买按钮。"""
+
+ if classify_pdd_page(root, "").kind != PAGE_ORDER_CONFIRMATION:
+ return None
+
+ nodes = list(root.iter("node"))
+ if sum(node.get("package") == PDD_PACKAGE_NAME for node in nodes) < 3:
+ return None
+
+ bounded_nodes = [
+ (node, bounds)
+ for node in nodes
+ if (bounds := _parse_bounds(node.get("bounds", ""))) is not None
+ and node.get("visible-to-user", "true") != "false"
+ ]
+ screen_bottom = max((bounds[3] for _, bounds in bounded_nodes), default=0)
+ if not screen_bottom:
+ return None
+
+ labels = [_preferred_or_descendant_label(node).replace(" ", "") for node in nodes]
+ has_close = any(label in ("关闭", "關閉") for label in labels)
+ has_summary = any(
+ label.startswith(("已选", "已選")) for label in labels
+ )
+ has_payment_area = any(
+ word in label for label in labels for word in ("微信支付", "先用后付", "支付方式")
+ )
+ has_quick_buy_action = any(
+ "现在买" in _preferred_or_descendant_label(node).replace(" ", "")
+ and (bounds[1] + bounds[3]) // 2 >= screen_bottom * 0.75
+ for node, bounds in bounded_nodes
+ )
+ quantity_editors = [
+ node
+ for node in nodes
+ if node.get("class") == "android.widget.EditText"
+ and node.get("text", "").strip().isdigit()
+ and int(node.get("text", "0")) > 0
+ ]
+ decreases = [
+ node
+ for node in nodes
+ if _preferred_or_descendant_label(node).replace(" ", "") == "减少数量"
+ and node.get("clickable") == "true"
+ ]
+ increases = [
+ node
+ for node in nodes
+ if _preferred_or_descendant_label(node).replace(" ", "") == "增加数量"
+ and node.get("clickable") == "true"
+ ]
+ if not (
+ has_close
+ and has_summary
+ and has_payment_area
+ and has_quick_buy_action
+ and len(quantity_editors) == 1
+ and len(decreases) == 1
+ and len(increases) == 1
+ ):
+ return None
+
+ headings = []
+ for node, bounds in bounded_nodes:
+ label = _preferred_node_label(node).strip()
+ if (
+ label
+ and node.get("clickable") != "true"
+ and _is_dimension_heading(label)
+ ):
+ headings.append((bounds[1], label, bounds))
+ headings.sort(key=lambda item: item[0])
+
+ used_keys: set[str] = set()
+ primary_index: Optional[int] = None
+ for index, (_, label, _) in enumerate(headings):
+ key = _dimension_key(label, used_keys)
+ used_keys.add(key)
+ if key == "color":
+ primary_index = index
+ break
+ if primary_index is None:
+ return None
+
+ _, _, heading_bounds = headings[primary_index]
+ next_heading_top = (
+ headings[primary_index + 1][0]
+ if primary_index + 1 < len(headings)
+ else int(screen_bottom * 0.75)
+ )
+ candidates: list[tuple[int, int, Bounds]] = []
+ for node, bounds in bounded_nodes:
+ if node.get("clickable") != "true" or not _is_available(node):
+ continue
+ label = _preferred_or_descendant_label(node).strip()
+ compact = label.replace(" ", "")
+ if not label:
+ continue
+ if compact.startswith(("已选", "已選", "请选择", "請選擇")):
+ continue
+ if any(
+ word in compact
+ for word in (
+ "现在买",
+ "提交订单",
+ "确认购买",
+ "支付",
+ "增加数量",
+ "减少数量",
+ "关闭",
+ )
+ ):
+ continue
+ if bounds[1] < heading_bounds[3] or bounds[3] > next_heading_top:
+ continue
+ area = (bounds[2] - bounds[0]) * (bounds[3] - bounds[1])
+ candidates.append((bounds[1], area, bounds))
+
+ if not candidates:
+ return None
+ _, _, target = min(candidates, key=lambda item: (item[0], item[1]))
+ left, top, right, bottom = target
+ return (left + right) // 2, (top + bottom) // 2
+
+
def _top_level_clickable_options(
container: ET.Element,
parents: Mapping[ET.Element, ET.Element],
@@ -1282,6 +1408,7 @@ class PddCollectService:
deadline = self._monotonic() + self._spec_panel_timeout
saw_confirmation_page = False
+ quick_confirmation_recovery_attempted = False
incomplete_reads = 0
while self._monotonic() < deadline:
self._check_cancelled()
@@ -1302,12 +1429,27 @@ class PddCollectService:
return snapshot
self._last_invalid_spec_xml = xml_data
incomplete_reads += 1
+
+ if not quick_confirmation_recovery_attempted:
+ coordinate = _quick_confirmation_spec_coord(root)
+ if coordinate is not None:
+ # 这里只打开当前主规格，禁止触碰“现在买”等不可逆动作。
+ device.click(*coordinate)
+ quick_confirmation_recovery_attempted = True
self._sleep(0.25)
if saw_confirmation_page:
+ message = "规格面板已经打开，但页面结构无法识别"
+ if quick_confirmation_recovery_attempted:
+ message = "快速确认页无法进入可识别的规格选择区域"
raise PddCollectError(
"PDD_DATA_SPEC_INCOMPLETE",
- "规格面板已经打开，但页面结构无法识别",
- {"incomplete_spec_reads": incomplete_reads},
+ message,
+ {
+ "incomplete_spec_reads": incomplete_reads,
+ "quick_confirmation_recovery_attempted": (
+ quick_confirmation_recovery_attempted
+ ),
+ },
)
raise PddCollectError(
"PDD_PAGE_SPEC_PANEL_TIMEOUT",
diff --git a/client/test/fixtures/pdd_quick_confirmation.xml b/client/test/fixtures/pdd_quick_confirmation.xml
new file mode 100644
index 0000000..7a16964
--- /dev/null
+++ b/client/test/fixtures/pdd_quick_confirmation.xml
@@ -0,0 +1,48 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/client/test/test_pdd_collect_service.py b/client/test/test_pdd_collect_service.py
index 15dc060..376b3d4 100644
--- a/client/test/test_pdd_collect_service.py
+++ b/client/test/test_pdd_collect_service.py
@@ -158,6 +158,30 @@ class PanelDoesNotOpenDevice(FakeCollectDevice):
self.clicks.append((x, y))
+class QuickConfirmationDevice(FakeCollectDevice):
+ """点击当前款式后，从快速确认页进入正常规格面板。"""
+
+ def __init__(self, quick_confirmation_xml, spec_xml):
+ super().__init__(quick_confirmation_xml, spec_xml)
+ self.opened_url = "https://mobile.yangkeduo.com/goods.html?goods_id=123"
+ self.panel_open = False
+
+ def dump_hierarchy(self):
+ return self.spec_xml if self.panel_open else self.home_xml
+
+ def click(self, x, y):
+ self.clicks.append((x, y))
+ if 32 <= x <= 470 and 1240 <= y <= 1320:
+ self.panel_open = True
+
+
+class StuckQuickConfirmationDevice(QuickConfirmationDevice):
+ """快速确认页接受点击后仍不打开规格选择区域。"""
+
+ def click(self, x, y):
+ self.clicks.append((x, y))
+
+
TRANSIENT_EMPTY_SPEC_XML = """
@@ -612,6 +636,9 @@ class PddCollectParserTest(unittest.TestCase):
cls.submit_hint_spec_xml = (
FIXTURES / "pdd_spec_panel_submit_hint.xml"
).read_text(encoding="utf-8")
+ cls.quick_confirmation_xml = (
+ FIXTURES / "pdd_quick_confirmation.xml"
+ ).read_text(encoding="utf-8")
cls.sold_out_xml = (
FIXTURES / "pdd_sold_out_recommendations.xml"
).read_text(encoding="utf-8")
@@ -944,6 +971,87 @@ class PddCollectParserTest(unittest.TestCase):
self.assertEqual([item.key for item in snapshot.dimensions], ["color", "size"])
self.assertEqual(clock.sleeps, [])
+ self.assertEqual(device.clicks, [])
+
+ def test_quick_confirmation_is_not_a_complete_spec_panel(self):
+ self.assertFalse(_is_spec_panel_open(self.quick_confirmation_xml))
+ with self.assertRaises(PddCollectError) as raised:
+ parse_spec_panel(self.quick_confirmation_xml)
+ self.assertEqual(raised.exception.code, "PDD_DATA_SPEC_INCOMPLETE")
+
+ def test_wait_opens_spec_selector_from_quick_confirmation_once(self):
+ device = QuickConfirmationDevice(
+ self.quick_confirmation_xml,
+ self.non_scrollable_spec_xml,
+ )
+ clock = FakeClock()
+ service = PddCollectService(
+ PddDeviceService(lambda _serial: device),
+ "USB-001",
+ "client-001",
+ sleeper=clock.sleep,
+ monotonic=clock.monotonic,
+ spec_panel_timeout=1.0,
+ )
+
+ snapshot = service._wait_spec_panel(device)
+
+ self.assertEqual([item.key for item in snapshot.dimensions], ["color", "size"])
+ self.assertEqual(device.clicks, [(251, 1280)])
+ self.assertTrue(all(y < 2000 for _, y in device.clicks))
+
+ def test_quick_confirmation_recovery_is_attempted_only_once(self):
+ device = StuckQuickConfirmationDevice(
+ self.quick_confirmation_xml,
+ self.non_scrollable_spec_xml,
+ )
+ clock = FakeClock()
+ service = PddCollectService(
+ PddDeviceService(lambda _serial: device),
+ "USB-001",
+ "client-001",
+ sleeper=clock.sleep,
+ monotonic=clock.monotonic,
+ spec_panel_timeout=1.0,
+ )
+
+ with self.assertRaises(PddCollectError) as raised:
+ service._wait_spec_panel(device)
+
+ self.assertEqual(raised.exception.code, "PDD_DATA_SPEC_INCOMPLETE")
+ self.assertIn("快速确认页", raised.exception.message)
+ self.assertEqual(device.clicks, [(251, 1280)])
+ self.assertTrue(
+ raised.exception.diagnostics["quick_confirmation_recovery_attempted"]
+ )
+
+ def test_quick_confirmation_without_payment_evidence_is_not_clicked(self):
+ incomplete_xml = self.quick_confirmation_xml.replace(
+ "使用微信支付，可更换支付方式",
+ "普通页面说明",
+ )
+ device = StuckQuickConfirmationDevice(
+ incomplete_xml,
+ self.non_scrollable_spec_xml,
+ )
+ clock = FakeClock()
+ service = PddCollectService(
+ PddDeviceService(lambda _serial: device),
+ "USB-001",
+ "client-001",
+ sleeper=clock.sleep,
+ monotonic=clock.monotonic,
+ spec_panel_timeout=1.0,
+ )
+
+ with self.assertRaises(PddCollectError) as raised:
+ service._wait_spec_panel(device)
+
+ self.assertEqual(raised.exception.code, "PDD_DATA_SPEC_INCOMPLETE")
+ self.assertEqual(device.clicks, [])
+ self.assertFalse(
+ raised.exception.diagnostics["quick_confirmation_recovery_attempted"]
+ )
def test_goods_page_is_not_non_scrollable_spec_panel(self):
self.assertFalse(_is_spec_panel_open(self.home_xml))