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

fix: 修复采购尺码候选扫描面板误判 (#272)

Browse Source
This commit is contained in:
chengma
2026-08-18 17:00:27 +08:00
parent 3558da66d5
commit 536f4bc72b
3 changed files with 96 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/pdd_collect_service.py
+56 -8
View File
@@ -1186,12 +1186,18 @@ class PddCollectService:
)
def collect_second_dimension_candidates(
self, device: Any
self, device: Any, *, initial_xml: Optional[str] = None
) -> Optional[SpecDimension]:
"""复用采集流程完整遍历当前颜色的第二规格，不点击任何选项。"""
"""完整遍历当前颜色的第二规格，不点击任何选项。
``initial_xml`` 是调用方刚确认的最新规格面板首帧，用于避免滚动
动画期间第一次重新读取控件树不完整而误判面板已经消失。
"""
return self._collect_size_dimension(
device, require_confirmed_edges=True
device,
require_confirmed_edges=True,
initial_xml=initial_xml,
)
def _check_cancelled(self) -> None:
@@ -1500,17 +1506,44 @@ class PddCollectService:
deadline = self._monotonic() + min(2.0, self._spec_panel_timeout)
xml_data = initial_xml
transient_reads = 0
last_page_kind = "unknown"
last_parse_error = ""
# 次数上限避免测试时钟或设备时钟异常导致无限循环。
for attempt in range(20):
self._check_cancelled()
if xml_data is None:
xml_data = self._dump_hierarchy(device)
try:
root = _parse_xml(xml_data)
observation = classify_pdd_page(root, "")
last_page_kind = observation.kind
if observation.kind == PAGE_CAPTCHA:
raise PddCollectError(
"PDD_PAGE_CAPTCHA", "PDD 出现安全验证，需要人工处理"
)
if observation.kind == PAGE_LOGIN_REQUIRED:
raise PddCollectError(
"PDD_PAGE_LOGIN_REQUIRED", "PDD 登录已失效，需要人工重新登录"
)
if observation.kind == PAGE_RISK_CONTROL:
raise PddCollectError(
"PDD_PAGE_RISK_CONTROL", "PDD 出现风控页面，已停止采购"
)
if observation.kind == PAGE_PAYMENT:
raise PddCollectError(
"PDD_PAGE_PAYMENT", "PDD 出现支付页面，已停止采购"
)
snapshot = parse_spec_panel(xml_data)
panel_open = _is_spec_panel_open(xml_data)
except PddCollectError as exc:
if exc.code != "PDD_DATA_SPEC_INCOMPLETE":
if exc.code in {
"PDD_PAGE_CAPTCHA",
"PDD_PAGE_LOGIN_REQUIRED",
"PDD_PAGE_RISK_CONTROL",
"PDD_PAGE_PAYMENT",
}:
raise
last_parse_error = exc.code
panel_open = False
snapshot = None
@@ -1525,10 +1558,21 @@ class PddCollectService:
self._sleep(0.1)
xml_data = None
diagnostics = {
"transient_spec_reads": transient_reads,
"last_page_kind": last_page_kind,
"last_parse_error": last_parse_error,
}
if last_page_kind == PAGE_ORDER_CONFIRMATION:
raise PddCollectError(
"PDD_DATA_SPEC_INCOMPLETE",
"规格确认页仍在，但规格区域持续无法解析",
diagnostics,
)
raise PddCollectError(
"PDD_PAGE_SPEC_PANEL_LOST",
"规格面板操作期间控件树持续为空或面板已经消失",
{"transient_spec_reads": transient_reads},
"规格面板已经消失或已离开确认页",
diagnostics,
)
def _save_xml(self, label: str, xml_data: str) -> Optional[Mapping[str, Any]]:
@@ -1965,12 +2009,13 @@ class PddCollectService:
device: Any,
*,
require_confirmed_edges: bool = False,
initial_xml: Optional[str] = None,
) -> Optional[SpecDimension]:
"""完整遍历第二规格并收集文字，全程不点击尺码或套餐。"""
sizes: dict[str, bool] = {}
size_name = "尺码"
xml_data: Optional[str] = None
xml_data: Optional[str] = initial_xml
size_found = False
previous_signature: Optional[
tuple[tuple[str, tuple[str, ...]], ...]
@@ -1981,7 +2026,9 @@ class PddCollectService:
# 保留原有行为，避免改变普通衣服“颜色 + 尺码”的稳定流程。
for swipe_count in range(self._max_spec_swipes + 1):
self._check_cancelled()
xml_data, snapshot = self._read_valid_spec_panel(device)
xml_data, snapshot = self._read_valid_spec_panel(
device, initial_xml=xml_data
)
for dimension in snapshot.dimensions:
if dimension.key not in ("color", "size"):
raise PddCollectError(
@@ -2022,6 +2069,7 @@ class PddCollectService:
device, region, horizontal=False, reverse=False
)
self._sleep(0.35)
xml_data = None
if not size_found or xml_data is None:
return None
client/src/pdd_u2_purchase_adapter.py
+21 -8
View File
@@ -1004,9 +1004,7 @@ class U2PddPurchaseAdapter(PddPurchaseAdapter):
self._panel_timeout = panel_timeout
self._select_color = select_color_fn
self._select_size = select_size_fn
self._size_candidate_collector = (
size_candidate_collector or self._collect_size_candidates
)
self._size_candidate_collector = size_candidate_collector
self._now = now
self._artifact_directory = artifact_directory
self._last_xml: Optional[str] = None
@@ -1232,8 +1230,10 @@ class U2PddPurchaseAdapter(PddPurchaseAdapter):
"""完整只读遍历后，仅为页面确实不存在的目标返回候选快照。"""
try:
self._restore_purchase_panel_color_region(failed_xml)
dimension = self._size_candidate_collector(self._require_device())
restored_xml = self._restore_purchase_panel_color_region(failed_xml)
dimension = self._read_size_candidates(
self._require_device(), restored_xml
)
except PddPurchaseError:
raise
except PddCollectError as exc:
@@ -1337,7 +1337,18 @@ class U2PddPurchaseAdapter(PddPurchaseAdapter):
},
)
def _collect_size_candidates(self, device: Any) -> Optional[SpecDimension]:
def _read_size_candidates(
self, device: Any, initial_xml: Optional[str] = None
) -> Optional[SpecDimension]:
"""读取第二规格候选；注入测试函数保持原有单参数接口。"""
if self._size_candidate_collector is not None:
return self._size_candidate_collector(device)
return self._collect_size_candidates(device, initial_xml=initial_xml)
def _collect_size_candidates(
self, device: Any, *, initial_xml: Optional[str] = None
) -> Optional[SpecDimension]:
"""使用采集模块同一套第二规格遍历，不建立新连接也不保存 XML。"""
service = PddCollectService(
@@ -1349,7 +1360,9 @@ class U2PddPurchaseAdapter(PddPurchaseAdapter):
cancelled=self._cancelled,
artifact_directory=None,
)
return service.collect_second_dimension_candidates(device)
return service.collect_second_dimension_candidates(
device, initial_xml=initial_xml
)
def apply_resolved_size(
self,
@@ -1386,7 +1399,7 @@ class U2PddPurchaseAdapter(PddPurchaseAdapter):
},
)
try:
dimension = self._size_candidate_collector(self._require_device())
dimension = self._read_size_candidates(self._require_device())
except PddCollectError as exc:
raise PddPurchaseError(
"PURCHASE_SPEC_CANDIDATES_CHANGED",
client/test/test_pdd_collect_service.py
+19
View File
@@ -2184,6 +2184,25 @@ class PddCollectParserTest(unittest.TestCase):
self.assertEqual("PDD_PAGE_CAPTCHA", raised.exception.code)
self.assertEqual([], clock.sleeps)
def test_valid_initial_spec_frame_is_used_before_reading_device_again(self):
device = FakeCollectDevice(self.home_xml, self.spec_xml)
device.panel_open = True
device.opened_url = "https://mobile.yangkeduo.com/goods.html?goods_id=123"
service = PddCollectService(
PddDeviceService(lambda _serial: device),
"USB-001",
"client-001",
sleeper=lambda _seconds: None,
)
xml_data, snapshot = service._read_valid_spec_panel(
device, initial_xml=self.spec_xml
)
self.assertEqual(self.spec_xml, xml_data)
self.assertTrue(snapshot.dimensions)
self.assertEqual(0, service._goods_screens_checked)
if __name__ == "__main__":
unittest.main()