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

fix: 商品页就绪后立即进入规格面板 (#36)

Browse Source
This commit is contained in:
chengma
2026-08-07 18:13:10 +08:00
parent 09b27cec42
commit e65cc2afda
6 changed files with 114 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/collect_task_service.py
+1
View File
@@ -206,6 +206,7 @@ class CollectTaskService:
return {
"PDD_DATA_SKU_NAME_TRUNCATED": "SKU_NAME_TRUNCATED",
"PDD_PAGE_SPEC_ENTRY_MISSING": "SKU_PANEL_NOT_FOUND",
"PDD_PAGE_SPEC_PANEL_TIMEOUT": "SKU_PANEL_NOT_FOUND",
"PDD_DATA_TITLE_MISSING": "TITLE_TOO_SHORT",
"PDD_PAGE_OVERALL_TIMEOUT": "COLLECT_TIMEOUT",
"DEVICE_OFFLINE": "DEVICE_DISCONNECTED",
client/src/pdd_collect_service.py
+35 -73
View File
@@ -241,22 +241,6 @@ def _all_labels(root: ET.Element) -> list[str]:
return [label for node in root.iter("node") if (label := _node_label(node))]
def _screen_bounds(root: ET.Element) -> Optional[Bounds]:
bounds_list = [
bounds
for node in root.iter("node")
if (bounds := _parse_bounds(node.get("bounds", ""))) is not None
]
if not bounds_list:
return None
return (
min(item[0] for item in bounds_list),
min(item[1] for item in bounds_list),
max(item[2] for item in bounds_list),
max(item[3] for item in bounds_list),
)
def parse_quantity(raw: Optional[str]) -> QuantityMetric:
"""把“已拼1.2万+”等文字转成整数，同时保留原文。"""
@@ -600,21 +584,6 @@ def _validate_goods_url(goods_url: str, task_goods_id: str) -> str:
return value
def _combine_goods_snapshots(snapshots: Sequence[GoodsSnapshot]) -> GoodsSnapshot:
return GoodsSnapshot(
next((item.title for item in snapshots if item.title), None),
next((item.shop_name for item in snapshots if item.shop_name), None),
next(
(item.sales for item in snapshots if item.sales.raw),
QuantityMetric(None, None, False),
),
next(
(item.reviews for item in snapshots if item.reviews.raw),
QuantityMetric(None, None, False),
),
)
class PddCollectService:
"""打开商品页并采集结构化商品与 SKU 数据。"""
@@ -629,8 +598,8 @@ class PddCollectService:
now: Callable[[], datetime] = lambda: datetime.now(timezone.utc),
cancelled: Callable[[], bool] = lambda: False,
page_timeout: float = 30.0,
spec_panel_timeout: float = 10.0,
overall_timeout: float = 600.0,
max_page_swipes: int = 12,
max_spec_swipes: int = 12,
max_sku_count: int = 200,
artifact_directory: Optional[Path] = None,
@@ -643,9 +612,9 @@ class PddCollectService:
self._now = now
self._cancelled = cancelled
self._page_timeout = page_timeout
self._spec_panel_timeout = spec_panel_timeout
self._overall_timeout = overall_timeout
self._overall_deadline: Optional[float] = None
self._max_page_swipes = max_page_swipes
self._max_spec_swipes = max_spec_swipes
self._max_sku_count = max_sku_count
self._artifact_directory = artifact_directory
@@ -672,10 +641,13 @@ class PddCollectService:
raise PddCollectError("PDD_DATA_TITLE_MISSING", "商品页没有可识别的标题")
if not goods.sales.raw:
raise PddCollectError("PDD_DATA_SALES_MISSING", "商品页没有采集到已拼数量")
if not goods.reviews.raw:
raise PddCollectError("PDD_DATA_REVIEWS_MISSING", "商品页没有采集到评价数量")
if not goods.shop_name and self._last_goods_xml:
artifact = self._save_xml("shop-not-found", self._last_goods_xml)
if (
(not goods.shop_name or not goods.reviews.raw)
and self._last_goods_xml
):
artifact = self._save_xml(
"goods-metadata-incomplete", self._last_goods_xml
)
if artifact:
self._artifacts.append(artifact)
@@ -687,7 +659,7 @@ class PddCollectService:
"商品页没有找到可靠的规格入口",
)
device.click(*coordinate)
self._sleep(0.5)
self._wait_spec_panel(device)
snapshots = self._discover_dimensions(device)
dimensions = merge_dimensions(snapshots)
@@ -791,46 +763,36 @@ class PddCollectService:
raise PddCollectError("PDD_PAGE_TIMEOUT", "等待 PDD 商品详情页加载超时")
def _collect_goods_details(self, device: Any) -> GoodsSnapshot:
snapshots: list[GoodsSnapshot] = []
previous_signature: Optional[tuple[str, ...]] = None
unchanged = 0
for _ in range(self._max_page_swipes + 1):
"""只读商品首页当前视口，不在打开规格面板前滚动详情页。"""
self._check_cancelled()
xml_data = device.dump_hierarchy()
self._last_goods_xml = str(xml_data)
self._goods_screens_checked += 1
return parse_goods_page(xml_data)
def _wait_spec_panel(self, device: Any) -> SpecSnapshot:
"""等待点击后的规格面板真正出现，不能只依赖固定延时。"""
deadline = self._monotonic() + self._spec_panel_timeout
while self._monotonic() < deadline:
self._check_cancelled()
xml_data = device.dump_hierarchy()
self._last_goods_xml = str(xml_data)
self._goods_screens_checked += 1
root = _parse_xml(xml_data)
labels = tuple(_all_labels(root))
snapshots.append(parse_goods_page(xml_data))
combined = _combine_goods_snapshots(snapshots)
is_complete = (
combined.title
and combined.sales.raw
and combined.reviews.raw
)
if is_complete:
break
if labels == previous_signature:
unchanged += 1
try:
snapshot = parse_spec_panel(xml_data)
except PddCollectError as exc:
if exc.code != "PDD_DATA_SPEC_INCOMPLETE":
raise
else:
unchanged = 0
if unchanged >= 1:
break
previous_signature = labels
screen = _screen_bounds(root)
if screen is None:
break
left, top, right, bottom = screen
x = (left + right) // 2
device.swipe(
x,
top + int((bottom - top) * 0.75),
x,
top + int((bottom - top) * 0.30),
duration=0.35,
)
self._sleep(0.35)
return _combine_goods_snapshots(snapshots)
if snapshot.dimensions:
return snapshot
self._sleep(0.25)
raise PddCollectError(
"PDD_PAGE_SPEC_PANEL_TIMEOUT",
"点击规格入口后，等待规格面板加载超时",
)
def _save_xml(self, label: str, xml_data: str) -> Optional[Mapping[str, Any]]:
"""保存本地诊断 XML；测试未提供目录时不写文件。"""
client/test/test_pdd_collect_service.py
+63 -4
View File
@@ -33,6 +33,7 @@ class FakeCollectDevice:
self.panel_open = False
self.clicks = []
self.swipes = []
self.swipe_panel_states = []
def app_current(self):
return {"package": "com.xunmeng.pinduoduo"}
@@ -56,6 +57,7 @@ class FakeCollectDevice:
def swipe(self, *args, **kwargs):
self.swipes.append((args, kwargs))
self.swipe_panel_states.append(self.panel_open)
class LoadingDevice(FakeCollectDevice):
@@ -85,6 +87,11 @@ class FocusMismatchDevice(FakeCollectDevice):
self.app_started = True
class PanelDoesNotOpenDevice(FakeCollectDevice):
def click(self, x, y):
self.clicks.append((x, y))
def keep_only_one_sku(xml_data: str) -> str:
"""从脱敏固件中删除蓝色和 L，只保留一个组合。"""
@@ -176,7 +183,6 @@ class PddCollectParserTest(unittest.TestCase):
"client-001",
sleeper=lambda _seconds: None,
now=lambda: datetime(2026, 8, 7, 8, 0, tzinfo=timezone.utc),
max_page_swipes=1,
max_spec_swipes=1,
)
@@ -213,7 +219,6 @@ class PddCollectParserTest(unittest.TestCase):
"USB-001",
"client-001",
sleeper=lambda _seconds: None,
max_page_swipes=0,
max_spec_swipes=0,
)
@@ -231,7 +236,6 @@ class PddCollectParserTest(unittest.TestCase):
"USB-001",
"client-001",
sleeper=lambda _seconds: None,
max_page_swipes=0,
max_spec_swipes=0,
)
@@ -375,7 +379,6 @@ class PddCollectParserTest(unittest.TestCase):
"USB-001",
"client-001",
sleeper=lambda _seconds: None,
max_page_swipes=0,
)
result = service.collect(
@@ -383,6 +386,62 @@ class PddCollectParserTest(unittest.TestCase):
)
self.assertIsNone(result.shop_name)
def test_missing_reviews_does_not_scroll_before_opening_spec_panel(self):
home_without_reviews = self.home_xml.replace(
'<node class="android.widget.TextView" text="2356条评价" '
'bounds="[330,950][600,1010]" visible-to-user="true" '
'enabled="true" />',
"",
)
device = FakeCollectDevice(
home_without_reviews, keep_only_one_sku(self.spec_xml)
)
service = PddCollectService(
PddDeviceService(lambda _serial: device),
"USB-001",
"client-001",
sleeper=lambda _seconds: None,
max_spec_swipes=0,
)
result = service.collect(
FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
)
self.assertIsNone(result.reviews.raw)
self.assertTrue(device.clicks)
self.assertTrue(all(device.swipe_panel_states))
def test_spec_panel_timeout_has_stable_code_and_xml_evidence(self):
with tempfile.TemporaryDirectory() as directory:
device = PanelDoesNotOpenDevice(self.home_xml, self.spec_xml)
tick = {"value": -0.5}
def clock():
tick["value"] += 0.5
return tick["value"]
service = PddCollectService(
PddDeviceService(lambda _serial: device),
"USB-001",
"client-001",
sleeper=lambda _seconds: None,
monotonic=clock,
spec_panel_timeout=1.0,
artifact_directory=Path(directory),
)
with self.assertRaises(PddCollectError) as raised:
service.collect(
FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
)
self.assertEqual(
raised.exception.code, "PDD_PAGE_SPEC_PANEL_TIMEOUT"
)
artifacts = raised.exception.diagnostics["artifacts"]
self.assertTrue(Path(artifacts[0]["path"]).is_file())
if __name__ == "__main__":
unittest.main()
client/test/test_pdd_real_xml_fixtures.py
+9
View File
@@ -9,6 +9,7 @@ from src.pdd_collect_service import (
parse_spec_panel,
)
from src.pdd_device_service import PddDeviceService
from src.util.get_size_panle_coord import get_size_panel_coord
REAL_XML = Path(__file__).parents[1] / "image_xml"
@@ -74,6 +75,14 @@ class PddRealXmlFixtureTest(unittest.TestCase):
self.assertEqual(result.price_cent, 470)
self.assertEqual(result.list_price_cent, 1990)
def test_real_home_xml_has_immediate_spec_entry(self):
xml_data = (REAL_XML / "737116531267_home.xml").read_text(
encoding="utf-8"
)
coordinate = get_size_panel_coord(xml_data)
self.assertIsNotNone(coordinate)
self.assertGreater(coordinate[1], 2000)
if __name__ == "__main__":
unittest.main()
docs/client/01-requirements.md
+2 -2
View File
@@ -46,9 +46,9 @@ Client 顶级导航仅包含：
输入至少包含远程任务编号和商品链接。Client 应采集：
- 商品编号、商品链接和商品标题；
- 店铺名称；
- 店铺名称（无障碍树未提供时允许留空并保存诊断证据）；
- 已拼数量及原始显示文字；
- 评价数量及原始显示文字；
- 评价数量及原始显示文字（当前首页未提供时允许留空，不能为此反复滚动并阻塞规格采集）；
- 所有规格维度及其可见值；
- 每个规格组合对应的价格、币种和可用状态；
- 采集时间、设备及必要诊断产物引用。
docs/client/02-architecture.md
+4
View File
@@ -354,6 +354,10 @@ reconcile_purchase(task, run) -> PurchaseResult | ManualReview
PDD 页面可能出现登录失效、验证码、控件树不完整、A/B 页面、库存变化和价格变化。适配层必须返回结构化错误，不得把这些情况统一返回 `False`。
采集商品时按“首页就绪 → 读取当前首页摘要 → 点击规格入口 → 确认规格面板
出现 → 扫描颜色和尺码”的顺序执行。打开规格面板前不得为了寻找评价或店铺
连续滚动商品详情；这两个可选字段缺失时保存证据，但不阻塞核心 SKU 采集。
## 10. 关键架构决策
1. 使用 PyQt5、Qt Widgets 和 PyQt-Fluent-Widgets，不混用其他 Qt 绑定。