fix: 阻止售罄页误入规格采集 (#159)

2026-08-11 15:49:12 +08:00
parent df1287f8eb
commit a875fbc981
3 changed files with 123 additions and 11 deletions
@@ -48,6 +48,13 @@ _DIMENSION_NAMES = ("颜色分类", "颜色", "尺码", "尺寸", "规格", "型
 _LOGIN_MARKERS = ("手机号登录", "登录后继续", "验证码登录", "账号登录")
 _CAPTCHA_MARKERS = ("请完成验证", "拖动滑块", "安全验证", "点击图中")
 _READY_MARKERS = ("发起拼单", "立即购买", "单独购买", "免拼购买", "快要抢光")
+_OUT_OF_STOCK_PAGE_MARKERS = (
+    "商品已售罄，推荐以下相似商品",
+    "商品已售罄，看看相似商品",
+    "该商品已售罄",
+    "该商品暂时缺货",
+    "商品暂时缺货",
+)
 _SHOP_NAME_EXCLUDES = frozenset({"进店", "关注", "店铺", "收藏", "客服"})
 _SHOP_ROW_TOLERANCE = 40

@@ -509,18 +516,19 @@ def _is_spec_panel_open(xml_data: str | bytes) -> bool:
    parents = {child: parent for parent in root.iter() for child in parent}
    if _find_non_scrollable_spec_panel(root, parents) is not None:
        return True
-    has_scrollable_region = any(
-        node.get("scrollable") == "true"
-        and _parse_bounds(node.get("bounds", "")) is not None
+    scrollable_regions = [
+        node
        for node in root.iter("node")
-    )
+        if node.get("scrollable") == "true"
+        and _parse_bounds(node.get("bounds", "")) is not None
+    ]
+    has_scrollable_region = bool(scrollable_regions)
    if not has_scrollable_region:
        return False

    compact_labels = [label.replace(" ", "") for label in labels]
    has_selection_summary = any(
-        label.startswith(("请选择", "已选"))
-        and any(word in label for word in ("颜色", "尺码", "规格", "款式"))
+        label.startswith(("请选择", "請選擇", "已选", "已選"))
        for label in compact_labels
    )
    has_submit_hint = any(
@@ -528,7 +536,15 @@ def _is_spec_panel_open(xml_data: str | bytes) -> bool:
        and any(word in label for word in ("选择", "颜色", "尺码", "规格"))
        for label in compact_labels
    )
-    return has_selection_summary or has_submit_hint
+    has_dimension_heading = any(
+        _is_dimension_heading(_preferred_node_label(node).strip())
+        and any(
+            region in _ancestors(node, parents)
+            for region in scrollable_regions
+        )
+        for node in root.iter("node")
+    )
+    return has_selection_summary or has_submit_hint or has_dimension_heading


 def _top_level_clickable_options(
@@ -600,6 +616,10 @@ def parse_spec_panel(xml_data: str | bytes) -> SpecSnapshot:
        if node.get("scrollable") == "true" and _parse_bounds(node.get("bounds", ""))
    ]
    non_scrollable_outer = _find_non_scrollable_spec_panel(root, parents)
+    if non_scrollable_outer is None and not _is_spec_panel_open(xml_data):
+        raise PddCollectError(
+            "PDD_DATA_SPEC_INCOMPLETE", "规格面板没有可识别的规格区域"
+        )
    panel_scrollables = []
    if non_scrollable_outer is not None:
        panel_scrollables = [
@@ -646,10 +666,12 @@ def parse_spec_panel(xml_data: str | bytes) -> SpecSnapshot:
        bounds = _parse_bounds(nested.get("bounds", ""))
        if bounds is None:
            continue
-        name = "颜色分类"
        preceding = [item for item in heading_nodes if item[0] <= bounds[1] + 30]
-        if preceding:
-            name = preceding[-1][1]
+        if not preceding:
+            # 普通推荐列表也是 RecyclerView。没有真实规格标题时，不能
+            # 默认把其中的商品卡片解释成颜色。
+            continue
+        name = preceding[-1][1]
        options = _top_level_clickable_options(nested, parents)
        if options:
            groups.append((name, options))
@@ -713,6 +735,17 @@ def _ancestors(


 def _raise_special_page(labels: Sequence[str]) -> None:
+    compact_labels = [label.replace(" ", "") for label in labels]
+    if any(
+        marker in label
+        for label in compact_labels
+        for marker in _OUT_OF_STOCK_PAGE_MARKERS
+    ):
+        raise PddCollectError(
+            "PDD_GOODS_UNAVAILABLE",
+            "商品已售罄，PDD 没有提供可采集的规格",
+            {"page_reason": "out_of_stock"},
+        )
    combined = " ".join(labels)
    if any(marker in combined for marker in _CAPTCHA_MARKERS):
        raise PddCollectError("PDD_PAGE_CAPTCHA", "PDD 出现安全验证，需要人工处理")
@@ -1134,7 +1167,7 @@ class PddCollectService:
                if exc.code != "PDD_DATA_SPEC_INCOMPLETE":
                    raise
            else:
-                if snapshot.dimensions or _is_spec_panel_open(xml_data):
+                if _is_spec_panel_open(xml_data):
                    return snapshot
            self._sleep(0.25)
        raise PddCollectError(
@@ -0,0 +1,27 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<hierarchy rotation="0">
+  <node class="android.widget.FrameLayout" package="com.xunmeng.pinduoduo"
+        bounds="[0,0][1080,2340]" visible-to-user="true" enabled="true">
+    <node class="androidx.recyclerview.widget.RecyclerView"
+          package="com.xunmeng.pinduoduo" scrollable="true"
+          bounds="[0,0][1080,2214]" visible-to-user="true" enabled="true">
+      <node class="android.widget.TextView" package="com.xunmeng.pinduoduo"
+            text="商品已售罄，推荐以下相似商品"
+            bounds="[225,332][855,393]" visible-to-user="true" enabled="true" />
+      <node class="android.view.ViewGroup" package="com.xunmeng.pinduoduo"
+            clickable="true" bounds="[0,927][535,1690]"
+            visible-to-user="true" enabled="true">
+        <node class="android.widget.TextView" package="com.xunmeng.pinduoduo"
+              text="测试推荐商品甲" bounds="[24,1482][511,1531]"
+              visible-to-user="true" enabled="true" />
+      </node>
+      <node class="android.view.ViewGroup" package="com.xunmeng.pinduoduo"
+            clickable="true" bounds="[545,927][1080,1690]"
+            visible-to-user="true" enabled="true">
+        <node class="android.widget.TextView" package="com.xunmeng.pinduoduo"
+              text="测试推荐商品乙" bounds="[569,1482][1056,1531]"
+              visible-to-user="true" enabled="true" />
+      </node>
+    </node>
+  </node>
+</hierarchy>
@@ -344,6 +344,9 @@ class PddCollectParserTest(unittest.TestCase):
        cls.non_scrollable_spec_xml = (
            FIXTURES / "pdd_spec_panel_non_scrollable.xml"
        ).read_text(encoding="utf-8")
+        cls.sold_out_xml = (
+            FIXTURES / "pdd_sold_out_recommendations.xml"
+        ).read_text(encoding="utf-8")

    def test_quantity_keeps_raw_value_and_approximate_flag(self):
        result = parse_quantity("已拼1.2万+件")
@@ -483,6 +486,55 @@ class PddCollectParserTest(unittest.TestCase):
    def test_goods_page_is_not_non_scrollable_spec_panel(self):
        self.assertFalse(_is_spec_panel_open(self.home_xml))

+    def test_sold_out_recommendations_are_not_spec_options(self):
+        with self.assertRaises(PddCollectError) as raised:
+            parse_spec_panel(self.sold_out_xml)
+
+        self.assertEqual(raised.exception.code, "PDD_GOODS_UNAVAILABLE")
+        self.assertEqual(
+            raised.exception.diagnostics, {"page_reason": "out_of_stock"}
+        )
+
+        page_without_marker = self.sold_out_xml.replace(
+            "商品已售罄，推荐以下相似商品", "为你推荐以下相似商品"
+        )
+        with self.assertRaises(PddCollectError) as incomplete:
+            parse_spec_panel(page_without_marker)
+        self.assertEqual(incomplete.exception.code, "PDD_DATA_SPEC_INCOMPLETE")
+
+    def test_blank_pdd_tree_is_not_a_spec_panel(self):
+        xml_data = """<hierarchy>
+          <node class="android.widget.FrameLayout"
+                package="com.xunmeng.pinduoduo"
+                bounds="[0,0][1080,2340]" />
+        </hierarchy>"""
+
+        self.assertFalse(_is_spec_panel_open(xml_data))
+        with self.assertRaises(PddCollectError) as raised:
+            parse_spec_panel(xml_data)
+        self.assertEqual(raised.exception.code, "PDD_DATA_SPEC_INCOMPLETE")
+
+    def test_collect_stops_on_sold_out_page_without_clicking_recommendations(self):
+        device = FakeCollectDevice(self.home_xml, self.sold_out_xml)
+        service = PddCollectService(
+            PddDeviceService(lambda _serial: device),
+            "USB-001",
+            "client-001",
+            sleeper=lambda _seconds: None,
+        )
+
+        with self.assertRaises(PddCollectError) as raised:
+            service.collect(
+                FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
+            )
+
+        self.assertEqual(raised.exception.code, "PDD_GOODS_UNAVAILABLE")
+        self.assertEqual(
+            raised.exception.diagnostics["page_reason"], "out_of_stock"
+        )
+        # 只点击商品页规格入口，不点击任何推荐商品。
+        self.assertEqual(len(device.clicks), 1)
+
    def test_full_window_with_scattered_labels_is_not_spec_panel(self):
        xml_data = """<hierarchy>
          <node class="android.widget.FrameLayout" bounds="[0,0][1080,2400]">