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

fix: 售罄临时页仅下拉刷新一次 (#265)

Browse Source
This commit is contained in:
chengma
2026-08-18 11:58:23 +08:00
parent fc13b2c7a3
commit e7ac5a470f
8 changed files with 304 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/test/test_pdd_collect_service.py
+76
View File
@@ -93,6 +93,29 @@ class FakeCollectDevice:
self.swipe_panel_states.append(self.panel_open)
class SoldOutRefreshCollectDevice(FakeCollectDevice):
"""模拟深链先显示售罄，向下拉一次后恢复或继续售罄。"""
def __init__(self, sold_out_xml, recovered_xml, *, persistent=False):
super().__init__(recovered_xml, "")
self.sold_out_xml = sold_out_xml
self.recovered_xml = recovered_xml
self.persistent = persistent
self.refreshed = False
def dump_hierarchy(self):
if self.opened_url is None:
return '<hierarchy><node package="com.xunmeng.pinduoduo" /></hierarchy>'
if not self.refreshed or self.persistent:
return self.sold_out_xml
return self.recovered_xml
def swipe(self, x1, y1, x2, y2, duration=0.0):
self.swipes.append(((x1, y1, x2, y2), {"duration": duration}))
if y2 > y1 and x1 == x2:
self.refreshed = True
class GoodsMetadataScrollDevice(FakeCollectDevice):
"""模拟评价和店铺分别出现在商品页后续屏幕。"""
@@ -1764,6 +1787,59 @@ class PddCollectParserTest(unittest.TestCase):
)
self.assertEqual(raised.exception.code, "PDD_PAGE_TIMEOUT")
def test_sold_out_page_pulls_down_once_then_recovers_goods(self):
device = SoldOutRefreshCollectDevice(
self.sold_out_xml,
self.home_xml,
)
service = PddCollectService(
PddDeviceService(lambda _serial: device),
"USB-001",
"client-001",
sleeper=lambda _seconds: None,
)
service._open_goods(
device,
"https://mobile.yangkeduo.com/goods.html?goods_id=123",
{"package": "com.xunmeng.pinduoduo"},
)
self.assertEqual(len(device.swipes), 1)
(x1, y1, x2, y2), options = device.swipes[0]
self.assertEqual(x1, x2)
self.assertGreater(y2, y1)
self.assertEqual(options["duration"], 0.5)
self.assertEqual(device.clicks, [])
def test_sold_out_after_one_refresh_is_unavailable(self):
device = SoldOutRefreshCollectDevice(
self.sold_out_xml,
self.home_xml,
persistent=True,
)
service = PddCollectService(
PddDeviceService(lambda _serial: device),
"USB-001",
"client-001",
sleeper=lambda _seconds: None,
)
with self.assertRaises(PddCollectError) as raised:
service._open_goods(
device,
"https://mobile.yangkeduo.com/goods.html?goods_id=123",
{"package": "com.xunmeng.pinduoduo"},
)
self.assertEqual(raised.exception.code, "PDD_GOODS_UNAVAILABLE")
self.assertEqual(
raised.exception.diagnostics,
{"page_reason": "out_of_stock", "refresh_attempts": 1},
)
self.assertEqual(len(device.swipes), 1)
self.assertEqual(device.clicks, [])
def test_stable_home_reopens_once_then_stops_before_spec_click(self):
home = (FIXTURES / "pdd_home_page.xml").read_text(encoding="utf-8")
device = FakeCollectDevice(home, self.spec_xml)
client/test/test_pdd_page_classifier.py
+33
View File
@@ -6,7 +6,9 @@ import xml.etree.ElementTree as ET
from src.pdd_page_classifier import (
ACTION_NETWORK_ERROR,
ACTION_OUT_OF_STOCK,
ACTION_READY,
ACTION_REFRESH,
ACTION_REOPEN,
ACTION_UNAVAILABLE,
ACTION_WAIT,
@@ -17,6 +19,7 @@ from src.pdd_page_classifier import (
PAGE_LOADING,
PAGE_LOGIN_REQUIRED,
PAGE_NETWORK_ERROR,
PAGE_OUT_OF_STOCK,
PAGE_RISK_CONTROL,
GoodsOpenTracker,
classify_pdd_page,
@@ -54,6 +57,7 @@ class PddPageClassifierTest(unittest.TestCase):
"请完成验证": PAGE_CAPTCHA,
"操作频繁": PAGE_RISK_CONTROL,
"立即购买": PAGE_GOODS,
"商品已售罄，推荐以下相似商品": PAGE_OUT_OF_STOCK,
}
for label, expected in cases.items():
with self.subTest(label=label):
@@ -103,6 +107,35 @@ class PddPageClassifierTest(unittest.TestCase):
decision = tracker.observe(observation("网络不给力"), 3.0)
self.assertEqual(ACTION_NETWORK_ERROR, decision.action)
def test_sold_out_page_refreshes_once_then_can_recover(self) -> None:
tracker = GoodsOpenTracker(None, 0.0)
sold_out = observation("商品已售罄，推荐以下相似商品")
self.assertEqual(ACTION_REFRESH, tracker.observe(sold_out, 0.2).action)
tracker.refreshed_out_of_stock(0.3)
self.assertTrue(tracker.out_of_stock_refreshed)
self.assertEqual(
ACTION_READY,
tracker.observe(observation("目标商品 立即购买"), 0.8).action,
)
def test_sold_out_after_refresh_is_unavailable_without_second_refresh(self) -> None:
tracker = GoodsOpenTracker(None, 0.0)
sold_out = observation("该商品已售罄")
self.assertEqual(ACTION_REFRESH, tracker.observe(sold_out, 0.2).action)
tracker.refreshed_out_of_stock(0.3)
self.assertEqual(
ACTION_OUT_OF_STOCK,
tracker.observe(sold_out, 0.8).action,
)
with self.assertRaises(RuntimeError):
tracker.refreshed_out_of_stock(0.9)
def test_sku_sold_out_text_is_still_a_normal_goods_page(self) -> None:
result = observation("立即购买 尺码 L（售罄）")
self.assertEqual(PAGE_GOODS, result.kind)
if __name__ == "__main__":
unittest.main()
client/test/test_pdd_u2_purchase_adapter.py
+63
View File
@@ -234,6 +234,29 @@ class FakeDevice:
raise AssertionError(f"未预期的 selector: {selector}")
class SoldOutRefreshPurchaseDevice(FakeDevice):
"""模拟采购深链先显示售罄，下拉一次后恢复或继续售罄。"""
def __init__(self, sold_out_xml: str, *, persistent: bool = False) -> None:
super().__init__(sold_out_xml)
self.sold_out_xml = sold_out_xml
self.persistent = persistent
self.refreshed = False
self.swipes = []
def dump_hierarchy(self):
if not self.has_opened:
return '<hierarchy><node package="com.xunmeng.pinduoduo" /></hierarchy>'
if not self.refreshed or self.persistent:
return self.sold_out_xml
return home_xml()
def swipe(self, x1, y1, x2, y2, duration=0.0):
self.swipes.append((x1, y1, x2, y2, duration))
if y2 > y1 and x1 == x2:
self.refreshed = True
class VerticallyScrolledColorPanelDevice(FakeDevice):
"""模拟 PDD 记住规格面板纵向位置。"""
@@ -2143,6 +2166,46 @@ class U2PddPurchaseAdapterTest(unittest.TestCase):
self.assertEqual(raised.exception.code, "PDD_PAGE_CAPTCHA")
self.assertEqual(device.clicks, [])
def test_sold_out_page_pulls_down_once_then_recovers_goods(self):
sold_out_xml = (
FIXTURES / "pdd_sold_out_recommendations.xml"
).read_text(encoding="utf-8")
device = SoldOutRefreshPurchaseDevice(sold_out_xml)
adapter = self._adapter(device, [])
adapter.open_goods(GOODS_URL)
self.assertEqual(len(device.swipes), 1)
x1, y1, x2, y2, duration = device.swipes[0]
self.assertEqual(x1, x2)
self.assertGreater(y2, y1)
self.assertEqual(duration, 0.5)
self.assertEqual(device.clicks, [])
adapter.close()
def test_sold_out_after_one_refresh_is_unavailable(self):
sold_out_xml = (
FIXTURES / "pdd_sold_out_recommendations.xml"
).read_text(encoding="utf-8")
device = SoldOutRefreshPurchaseDevice(
sold_out_xml,
persistent=True,
)
adapter = self._adapter(device, [])
with self.assertRaises(PddPurchaseError) as raised:
adapter.open_goods(GOODS_URL)
self.assertEqual(raised.exception.code, "PDD_GOODS_UNAVAILABLE")
self.assertFalse(raised.exception.retryable)
self.assertEqual(
raised.exception.diagnostics,
{"page_reason": "out_of_stock", "refresh_attempts": 1},
)
self.assertEqual(len(device.swipes), 1)
self.assertEqual(device.clicks, [])
adapter.close()
def test_stable_home_reopens_once_then_stops_before_click(self):
device = FakeDevice(
(FIXTURES / "pdd_home_page.xml").read_text(encoding="utf-8")