diff --git a/client/src/pdd_collect_service.py b/client/src/pdd_collect_service.py
index 4a16d0d..67d567c 100644
--- a/client/src/pdd_collect_service.py
+++ b/client/src/pdd_collect_service.py
@@ -7,7 +7,6 @@
from __future__ import annotations
import hashlib
-import itertools
import re
import time
import xml.etree.ElementTree as ET
@@ -137,6 +136,24 @@ class SpecSnapshot:
list_price_cent: Optional[int]
+@dataclass(frozen=True)
+class VisibleSpecOption:
+ """当前控件树中完整可点击的规格节点。"""
+
+ text: str
+ available: bool
+ bounds: Bounds
+
+
+@dataclass(frozen=True)
+class ColorPriceSample:
+ """点击一个颜色后稳定读取到的颜色级价格。"""
+
+ price_cent: Optional[int]
+ raw_price: Optional[str]
+ list_price_cent: Optional[int]
+
+
@dataclass(frozen=True)
class CollectResult:
"""与 ``pdd_data`` v1 对应的采集结果。"""
@@ -526,31 +543,6 @@ def _ancestors(
return result
-def merge_dimensions(snapshots: Iterable[SpecSnapshot]) -> tuple[SpecDimension, ...]:
- """合并滚动过程中多棵控件树看到的规格值。"""
-
- order: list[str] = []
- names: dict[str, str] = {}
- values: dict[str, dict[str, bool]] = {}
- for snapshot in snapshots:
- for dimension in snapshot.dimensions:
- if dimension.key not in values:
- order.append(dimension.key)
- names[dimension.key] = dimension.name
- values[dimension.key] = {}
- for value in dimension.values:
- was_available = values[dimension.key].get(value.text, False)
- values[dimension.key][value.text] = was_available or value.available
- return tuple(
- SpecDimension(
- key,
- names[key],
- tuple(DimensionValue(text, available) for text, available in values[key].items()),
- )
- for key in order
- )
-
-
def _raise_special_page(labels: Sequence[str]) -> None:
combined = " ".join(labels)
if any(marker in combined for marker in _CAPTCHA_MARKERS):
@@ -651,7 +643,7 @@ class PddCollectService:
if artifact:
self._artifacts.append(artifact)
- home_xml = device.dump_hierarchy()
+ home_xml = self._dump_hierarchy(device)
coordinate = get_size_panel_coord(home_xml)
if coordinate is None:
raise PddCollectError(
@@ -661,14 +653,31 @@ class PddCollectService:
device.click(*coordinate)
self._wait_spec_panel(device)
- snapshots = self._discover_dimensions(device)
- dimensions = merge_dimensions(snapshots)
+ color_dimension, color_samples = self._collect_color_prices(device)
+ missing_prices = [
+ color.text
+ for color in color_dimension.values
+ if color.available
+ and color_samples[color.text].price_cent is None
+ ]
+ if missing_prices:
+ raise PddCollectError(
+ "PDD_DATA_PRICE_MISSING",
+ "以下可用颜色没有采集到稳定价格："
+ + "、".join(missing_prices),
+ )
+ size_dimension = self._collect_size_dimension(device)
+ dimensions = (color_dimension,)
+ if size_dimension is not None:
+ dimensions += (size_dimension,)
if not dimensions or any(not item.values for item in dimensions):
raise PddCollectError(
"PDD_DATA_SPEC_INCOMPLETE",
"规格面板没有采集到完整的规格维度",
)
- skus = self._collect_skus(device, dimensions)
+ skus = self._build_color_price_skus(
+ dimensions, color_samples
+ )
has_available_price = any(
item.price_cent is not None for item in skus if item.available
)
@@ -739,9 +748,7 @@ class PddCollectService:
while self._monotonic() < deadline:
self._check_cancelled()
current = device.app_current()
- xml_data = device.dump_hierarchy()
- self._last_goods_xml = str(xml_data)
- self._goods_screens_checked += 1
+ xml_data = self._dump_hierarchy(device)
root = _parse_xml(xml_data)
last_labels = _all_labels(root)
pdd_node_count = sum(
@@ -766,9 +773,7 @@ class PddCollectService:
"""只读商品首页当前视口，不在打开规格面板前滚动详情页。"""
self._check_cancelled()
- xml_data = device.dump_hierarchy()
- self._last_goods_xml = str(xml_data)
- self._goods_screens_checked += 1
+ xml_data = self._dump_hierarchy(device)
return parse_goods_page(xml_data)
def _wait_spec_panel(self, device: Any) -> SpecSnapshot:
@@ -777,9 +782,7 @@ class PddCollectService:
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
+ xml_data = self._dump_hierarchy(device)
try:
snapshot = parse_spec_panel(xml_data)
except PddCollectError as exc:
@@ -815,176 +818,395 @@ class PddCollectService:
except OSError:
return None
- def _discover_dimensions(self, device: Any) -> list[SpecSnapshot]:
- snapshots: list[SpecSnapshot] = []
+ def _dump_hierarchy(self, device: Any) -> str:
+ """读取并记住最新控件树，失败诊断必须指向最后一次操作。"""
- # 当前选中项可能让列表自动停在中间。因此横向和纵向都扫描两个方向，
- # 不能假设打开面板时正好位于列表起点。
- for horizontal, region_getter in (
- (True, self._horizontal_region),
- (False, self._vertical_region),
- ):
- for reverse in (False, True):
- previous_signature: Optional[
- tuple[tuple[str, tuple[str, ...]], ...]
- ] = None
- for _ in range(self._max_spec_swipes + 1):
- self._check_cancelled()
- xml_data = device.dump_hierarchy()
- snapshot = parse_spec_panel(xml_data)
- snapshots.append(snapshot)
- signature = tuple(
- (item.name, tuple(value.text for value in item.values))
- for item in snapshot.dimensions
- )
- root = _parse_xml(xml_data)
- region = region_getter(root)
- if region is None or signature == previous_signature:
- break
- previous_signature = signature
- self._swipe_region(
- device,
- region,
- horizontal=horizontal,
- reverse=reverse,
- )
- self._sleep(0.35)
- return snapshots
-
- def _collect_skus(
- self,
- device: Any,
- dimensions: tuple[SpecDimension, ...],
- ) -> tuple[SkuResult, ...]:
- combinations = list(itertools.product(*(item.values for item in dimensions)))
- if len(combinations) > self._max_sku_count:
- raise PddCollectError(
- "PDD_DATA_TOO_MANY_SKUS",
- f"规格组合共 {len(combinations)} 个，超过安全上限 {self._max_sku_count}",
- )
-
- color_index = next(
- (index for index, item in enumerate(dimensions) if item.key == "color"),
- None,
+ raw_xml = device.dump_hierarchy()
+ xml_data = (
+ raw_xml.decode("utf-8", errors="replace")
+ if isinstance(raw_xml, bytes)
+ else str(raw_xml)
)
- if color_index is None:
+ self._last_goods_xml = xml_data
+ self._goods_screens_checked += 1
+ return xml_data
+
+ def _collect_color_prices(
+ self, device: Any
+ ) -> tuple[SpecDimension, Mapping[str, ColorPriceSample]]:
+ """按行蛇形遍历颜色，并在每次点击后立即读取颜色级价格。"""
+
+ self._move_color_list_to_left(device)
+ first_xml = self._dump_hierarchy(device)
+ first_rows = self._visible_color_rows(first_xml)
+ if not first_rows:
raise PddCollectError(
"PDD_DATA_SPEC_INCOMPLETE", "规格面板没有可识别的颜色分类"
)
- samples: dict[
- str,
- tuple[Optional[int], Optional[str], Optional[int], dict[str, str]],
- ] = {}
- color_dimension = dimensions[color_index]
- for color in color_dimension.values:
- self._check_cancelled()
- if not color.available:
- samples[color.text] = (None, None, None, {"color": color.text})
- continue
- selected = self._select_option(device, color.text, color_dimension.key)
- if not selected:
- samples[color.text] = (None, None, None, {"color": color.text})
- continue
- snapshot = parse_spec_panel(device.dump_hierarchy())
- summary = snapshot.selected_text or ""
- observed: dict[str, str] = {}
- for dimension in dimensions:
- match = next(
- (value.text for value in dimension.values if value.text in summary),
- None,
+ row_count = len(first_rows)
+ visited: set[str] = set()
+ ordered_colors: list[DimensionValue] = []
+ samples: dict[str, ColorPriceSample] = {}
+
+ for row_index in range(row_count):
+ move_right = row_index % 2 == 0
+ current_row_colors: list[DimensionValue] = []
+ stable_edge_reads = 0
+ previous_signature: Optional[tuple[tuple[str, Bounds], ...]] = None
+
+ for swipe_count in range(self._max_spec_swipes + 1):
+ self._check_cancelled()
+ # 每次只处理一个节点；点击可能让列表自动移动，下一项必须
+ # 从最新 XML 重新计算，不能继续使用点击前的旧坐标。
+ while True:
+ xml_data = self._dump_hierarchy(device)
+ rows = self._visible_color_rows(xml_data)
+ if row_index >= len(rows):
+ raise PddCollectError(
+ "PDD_DATA_SPEC_INCOMPLETE",
+ f"颜色第 {row_index + 1} 行在滑动后消失",
+ )
+ current_row = sorted(
+ rows[row_index],
+ key=lambda item: item.bounds[0],
+ reverse=not move_right,
+ )
+ visible = next(
+ (item for item in current_row if item.text not in visited),
+ None,
+ )
+ if visible is None:
+ break
+ visited.add(visible.text)
+ color = DimensionValue(visible.text, visible.available)
+ current_row_colors.append(color)
+ if visible.available:
+ samples[visible.text] = self._click_and_sample_color(
+ device, visible.text
+ )
+ else:
+ samples[visible.text] = ColorPriceSample(
+ None, None, None
+ )
+
+ signature = tuple((item.text, item.bounds) for item in current_row)
+ if signature == previous_signature:
+ stable_edge_reads += 1
+ else:
+ stable_edge_reads = 0
+ previous_signature = signature
+ if stable_edge_reads >= 2 or swipe_count >= self._max_spec_swipes:
+ break
+
+ root = _parse_xml(self._dump_hierarchy(device))
+ region = self._horizontal_region(root)
+ if region is None:
+ break
+ self._swipe_region(
+ device,
+ region,
+ horizontal=True,
+ reverse=not move_right,
)
- if match:
- observed[dimension.key] = match
- confirmed = (
- observed.get(color_dimension.key) == color.text
- and len(observed) == len(dimensions)
+ self._sleep(0.35)
+
+ # 操作采用蛇形以减少无效滑动；输出仍恢复成页面自然的
+ # “每行从左到右”顺序，方便 Admin 下拉框稳定展示。
+ ordered_colors.extend(
+ current_row_colors if move_right else reversed(current_row_colors)
)
- samples[color.text] = (
- snapshot.price_cent if confirmed else None,
- snapshot.raw_price if confirmed else None,
- snapshot.list_price_cent if confirmed else None,
- observed,
+
+ if not ordered_colors:
+ raise PddCollectError(
+ "PDD_DATA_SPEC_INCOMPLETE", "规格面板没有采集到颜色"
+ )
+ return (
+ SpecDimension("color", "颜色分类", tuple(ordered_colors)),
+ samples,
+ )
+
+ def _move_color_list_to_left(self, device: Any) -> None:
+ """把颜色列表归位到左端；连续两次视口不变才认为到边。"""
+
+ previous_signature: Optional[tuple[tuple[str, Bounds], ...]] = None
+ stable_edge_reads = 0
+ for _ in range(self._max_spec_swipes):
+ self._check_cancelled()
+ xml_data = self._dump_hierarchy(device)
+ rows = self._visible_color_rows(xml_data)
+ signature = tuple(
+ (item.text, item.bounds) for row in rows for item in row
+ )
+ if signature == previous_signature:
+ stable_edge_reads += 1
+ else:
+ stable_edge_reads = 0
+ previous_signature = signature
+ if stable_edge_reads >= 2:
+ return
+
+ root = _parse_xml(xml_data)
+ region = self._horizontal_region(root)
+ if region is None:
+ return
+ self._swipe_region(
+ device, region, horizontal=True, reverse=True
+ )
+ self._sleep(0.35)
+
+ def _click_and_sample_color(
+ self, device: Any, target: str
+ ) -> ColorPriceSample:
+ """点击最新树中的颜色，等待选择和价格连续两次稳定。"""
+
+ xml_data = self._dump_hierarchy(device)
+ root = _parse_xml(xml_data)
+ latest_visible = next(
+ (
+ item
+ for row in self._visible_color_rows(xml_data)
+ for item in row
+ if item.text == target
+ ),
+ None,
+ )
+ node = self._find_option_node(root, target) if latest_visible else None
+ if node is None or not _is_available(node):
+ return ColorPriceSample(None, None, None)
+ bounds = _parse_bounds(node.get("bounds", ""))
+ assert bounds is not None
+ device.click(
+ (bounds[0] + bounds[2]) // 2,
+ (bounds[1] + bounds[3]) // 2,
+ )
+
+ previous_price: Optional[tuple[int, Optional[str], Optional[int]]] = None
+ stable_price_reads = 0
+ for _ in range(8):
+ self._check_cancelled()
+ self._sleep(0.2)
+ latest_xml = self._dump_hierarchy(device)
+ latest_root = _parse_xml(latest_xml)
+ snapshot = parse_spec_panel(latest_xml)
+ latest_node = self._find_option_node(latest_root, target)
+ selection_exposed = self._color_selection_state_exposed(latest_xml)
+ selected = bool(
+ latest_node is not None and self._node_is_selected(latest_node)
+ ) or bool(snapshot.selected_text and target in snapshot.selected_text)
+ # 老版本页面不暴露 selected/checked，也没有“已选”摘要时，
+ # 只能以完整可见可点击节点未报错的点击作为降级证据。
+ selection_confirmed = selected or not selection_exposed
+
+ if snapshot.price_cent is None or not selection_confirmed:
+ stable_price_reads = 0
+ previous_price = None
+ continue
+ current_price = (
+ snapshot.price_cent,
+ snapshot.raw_price,
+ snapshot.list_price_cent,
+ )
+ if current_price == previous_price:
+ stable_price_reads += 1
+ else:
+ stable_price_reads = 1
+ previous_price = current_price
+ if stable_price_reads >= 2:
+ return ColorPriceSample(
+ snapshot.price_cent,
+ snapshot.raw_price,
+ snapshot.list_price_cent,
+ )
+
+ return ColorPriceSample(None, None, None)
+
+ def _collect_size_dimension(self, device: Any) -> Optional[SpecDimension]:
+ """颜色采价完成后只滚动并收集尺码文字，不点击尺码。"""
+
+ sizes: dict[str, bool] = {}
+ previous_signature: Optional[
+ tuple[tuple[str, tuple[str, ...]], ...]
+ ] = None
+ stable_edge_reads = 0
+
+ for swipe_count in range(self._max_spec_swipes + 1):
+ self._check_cancelled()
+ xml_data = self._dump_hierarchy(device)
+ snapshot = parse_spec_panel(xml_data)
+ for dimension in snapshot.dimensions:
+ if dimension.key not in ("color", "size"):
+ raise PddCollectError(
+ "PDD_DATA_SPEC_UNSUPPORTED",
+ f"发现尚未支持的规格维度：{dimension.name}",
+ )
+ if dimension.key == "size":
+ for value in dimension.values:
+ sizes[value.text] = sizes.get(value.text, False) or value.available
+
+ signature = tuple(
+ (item.key, tuple(value.text for value in item.values))
+ for item in snapshot.dimensions
+ )
+ if signature == previous_signature:
+ stable_edge_reads += 1
+ else:
+ stable_edge_reads = 0
+ previous_signature = signature
+ if stable_edge_reads >= 2 or swipe_count >= self._max_spec_swipes:
+ break
+
+ root = _parse_xml(xml_data)
+ region = self._vertical_region(root)
+ if region is None:
+ break
+ self._swipe_region(
+ device, region, horizontal=False, reverse=False
+ )
+ self._sleep(0.35)
+
+ if not sizes:
+ return None
+ return SpecDimension(
+ "size",
+ "尺码",
+ tuple(DimensionValue(text, available) for text, available in sizes.items()),
+ )
+
+ def _build_color_price_skus(
+ self,
+ dimensions: tuple[SpecDimension, ...],
+ samples: Mapping[str, ColorPriceSample],
+ ) -> tuple[SkuResult, ...]:
+ """按颜色价格生成展示组合；采价证据只记录真实选中的颜色。"""
+
+ colors = dimensions[0].values
+ sizes = dimensions[1].values if len(dimensions) > 1 else (None,)
+ combination_count = len(colors) * len(sizes)
+ if combination_count > self._max_sku_count:
+ raise PddCollectError(
+ "PDD_DATA_TOO_MANY_SKUS",
+ f"规格组合共 {combination_count} 个，超过安全上限 {self._max_sku_count}",
)
results: list[SkuResult] = []
- for values in combinations:
- options = {
- dimension.key: value.text
- for dimension, value in zip(dimensions, values)
- }
- price, raw_price, list_price, observed = samples[values[color_index].text]
- available = all(value.available for value in values) and price is not None
- results.append(
- SkuResult(
- options,
- price if available else None,
- available,
- raw_price if available else None,
- observed,
- list_price if available else None,
+ for color in colors:
+ sample = samples[color.text]
+ for size in sizes:
+ options = {"color": color.text}
+ if size is not None:
+ options["size"] = size.text
+ available = (
+ color.available
+ and (size is None or size.available)
+ and sample.price_cent is not None
+ )
+ results.append(
+ SkuResult(
+ options,
+ sample.price_cent if available else None,
+ available,
+ sample.raw_price if available else None,
+ {"color": color.text}
+ if sample.price_cent is not None
+ else {},
+ sample.list_price_cent if available else None,
+ )
)
- )
return tuple(results)
- def _select_combination(
- self,
- device: Any,
- pairs: Sequence[tuple[SpecDimension, DimensionValue]],
- ) -> bool:
- for dimension, value in pairs:
- if not self._select_option(device, value.text, dimension.key):
- return False
- return True
+ def _visible_color_rows(
+ self, xml_data: str | bytes
+ ) -> list[list[VisibleSpecOption]]:
+ """把当前视口中完整显示的颜色节点按中心 y 聚类成行。"""
- def _select_option(self, device: Any, target: str, dimension_key: str) -> bool:
- prefer_horizontal = dimension_key == "color" or dimension_key.startswith(
- "color_"
+ root = _parse_xml(xml_data)
+ snapshot = parse_spec_panel(xml_data)
+ color_dimension = next(
+ (item for item in snapshot.dimensions if item.key == "color"),
+ None,
)
- for reverse in (False, True):
- previous_signature: Optional[tuple[str, ...]] = None
- for _ in range(self._max_spec_swipes + 1):
- self._check_cancelled()
- xml_data = device.dump_hierarchy()
- root = _parse_xml(xml_data)
- node = self._find_option_node(root, target)
- if node is not None:
- if not _is_available(node):
- return False
- bounds = _parse_bounds(node.get("bounds", ""))
- assert bounds is not None
- device.click(
- (bounds[0] + bounds[2]) // 2,
- (bounds[1] + bounds[3]) // 2,
- )
- self._sleep(0.25)
- labels = _all_labels(_parse_xml(device.dump_hierarchy()))
- return target in " ".join(labels)
+ if color_dimension is None:
+ return []
- signature = tuple(_all_labels(root))
- if signature == previous_signature:
- break
- previous_signature = signature
- horizontal = self._horizontal_region(root)
- vertical = self._vertical_region(root)
- if horizontal is not None and (prefer_horizontal or vertical is None):
- self._swipe_region(
- device,
- horizontal,
- horizontal=True,
- reverse=reverse,
- )
- elif vertical is not None:
- self._swipe_region(
- device,
- vertical,
- horizontal=False,
- reverse=reverse,
- )
- else:
- return False
- self._sleep(0.25)
- return False
+ candidates: list[VisibleSpecOption] = []
+ for value in color_dimension.values:
+ node = self._find_option_node(root, value.text)
+ if node is None:
+ continue
+ bounds = _parse_bounds(node.get("bounds", ""))
+ if bounds is None:
+ continue
+ candidates.append(
+ VisibleSpecOption(
+ value.text,
+ value.available,
+ bounds,
+ )
+ )
+ if not candidates:
+ return []
+
+ widest = max(item.bounds[2] - item.bounds[0] for item in candidates)
+ minimum_width = max(48, int(widest * 0.6))
+ complete = [
+ item
+ for item in candidates
+ if item.bounds[2] - item.bounds[0] >= minimum_width
+ ]
+ complete.sort(key=lambda item: ((item.bounds[1] + item.bounds[3]) // 2, item.bounds[0]))
+
+ rows: list[list[VisibleSpecOption]] = []
+ row_centers: list[int] = []
+ for item in complete:
+ center_y = (item.bounds[1] + item.bounds[3]) // 2
+ matching_index = next(
+ (
+ index
+ for index, row_center in enumerate(row_centers)
+ if abs(center_y - row_center) <= 80
+ ),
+ None,
+ )
+ if matching_index is None:
+ rows.append([item])
+ row_centers.append(center_y)
+ else:
+ rows[matching_index].append(item)
+ for row in rows:
+ row.sort(key=lambda item: item.bounds[0])
+ return rows
+
+ @staticmethod
+ def _node_selection_state_exposed(node: ET.Element) -> bool:
+ return any(
+ item.get("selected") is not None or item.get("checked") is not None
+ for item in node.iter()
+ )
+
+ @staticmethod
+ def _node_is_selected(node: ET.Element) -> bool:
+ return any(
+ item.get("selected") == "true" or item.get("checked") == "true"
+ for item in node.iter()
+ )
+
+ def _color_selection_state_exposed(self, xml_data: str | bytes) -> bool:
+ root = _parse_xml(xml_data)
+ snapshot = parse_spec_panel(xml_data)
+ if snapshot.selected_text:
+ return True
+ color_dimension = next(
+ (item for item in snapshot.dimensions if item.key == "color"),
+ None,
+ )
+ if color_dimension is None:
+ return False
+ return any(
+ node is not None and self._node_selection_state_exposed(node)
+ for value in color_dimension.values
+ if (node := self._find_option_node(root, value.text)) is not None
+ )
@staticmethod
def _find_option_node(root: ET.Element, target: str) -> Optional[ET.Element]:
diff --git a/client/test/test_pdd_collect_service.py b/client/test/test_pdd_collect_service.py
index bc2878c..369ac3a 100644
--- a/client/test/test_pdd_collect_service.py
+++ b/client/test/test_pdd_collect_service.py
@@ -92,6 +92,127 @@ class PanelDoesNotOpenDevice(FakeCollectDevice):
self.clicks.append((x, y))
+class SnakeColorDevice(FakeCollectDevice):
+ """模拟两行横向颜色网格和滚动后才出现的尺码区域。"""
+
+ prices = {
+ "A色": 1000,
+ "B色": 1000,
+ "C色": 1200,
+ "D色": 1300,
+ "E色": 1400,
+ "F色": 1500,
+ }
+
+ def __init__(self, home_xml):
+ super().__init__(home_xml, "")
+ self.page = 1
+ self.selected_color = "A色"
+ self.sizes_visible = False
+ self.clicked_colors = []
+
+ def dump_hierarchy(self):
+ if not self.panel_open:
+ return self.home_xml
+ return self._spec_xml()
+
+ def click(self, x, y):
+ self.clicks.append((x, y))
+ if not self.panel_open:
+ self.panel_open = True
+ return
+ for name, bounds in self._current_color_bounds().items():
+ left, top, right, bottom = bounds
+ if left <= x <= right and top <= y <= bottom:
+ self.selected_color = name
+ self.clicked_colors.append(name)
+ return
+
+ def swipe(self, x1, y1, x2, y2, duration=0.35):
+ self.swipes.append(((x1, y1, x2, y2), {"duration": duration}))
+ self.swipe_panel_states.append(self.panel_open)
+ if abs(x2 - x1) > abs(y2 - y1):
+ self.page = 1 if x1 > x2 else 0
+ else:
+ self.sizes_visible = True
+
+ def _current_color_bounds(self):
+ if self.page == 0:
+ return {
+ "A色": (36, 740, 340, 850),
+ "B色": (370, 740, 674, 850),
+ "C色": (1044, 740, 1080, 850),
+ "D色": (36, 900, 340, 1010),
+ "E色": (370, 900, 674, 1010),
+ "F色": (1044, 900, 1080, 1010),
+ }
+ return {
+ "A色": (0, 740, 20, 850),
+ "B色": (36, 740, 340, 850),
+ "C色": (370, 740, 674, 850),
+ "D色": (0, 900, 20, 1010),
+ "E色": (36, 900, 340, 1010),
+ "F色": (370, 900, 674, 1010),
+ }
+
+ def _spec_xml(self):
+ color_nodes = []
+ for name, bounds in self._current_color_bounds().items():
+ left, top, right, bottom = bounds
+ selected = "true" if name == self.selected_color else "false"
+ color_nodes.append(
+ f''
+ )
+ size_nodes = ""
+ if self.sizes_visible:
+ size_nodes = """
+
+
+
+ """
+ price = self.prices[self.selected_color] / 100
+ return f"""
+
+
+
+
+
+ {''.join(color_nodes)}
+
+ {size_nodes}
+
+
+ """
+
+
+class IgnoredColorClickDevice(SnakeColorDevice):
+ """页面暴露选中状态，但模拟一次没有生效的颜色点击。"""
+
+ def click(self, x, y):
+ if not self.panel_open:
+ super().click(x, y)
+ return
+ self.clicks.append((x, y))
+
+
def keep_only_one_sku(xml_data: str) -> str:
"""从脱敏固件中删除蓝色和 L，只保留一个组合。"""
@@ -202,7 +323,7 @@ class PddCollectParserTest(unittest.TestCase):
self.assertEqual(data["skus"][0]["price_cent"], 1000)
self.assertEqual(
data["skus"][0]["price_observed_at"],
- {"color": "红色", "size": "M"},
+ {"color": "红色"},
)
self.assertEqual(data["source"]["device_address"], "USB-001")
self.assertIsNone(data["purchase"])
@@ -248,11 +369,97 @@ class PddCollectParserTest(unittest.TestCase):
item for item in data["skus"] if item["options"]["color"] == "红色"
]
self.assertEqual(
- {item["price_observed_at"]["size"] for item in red_skus}, {"M"}
+ {item["price_observed_at"]["color"] for item in red_skus},
+ {"红色"},
+ )
+ self.assertTrue(
+ all(
+ set(item["price_observed_at"]) == {"color"}
+ for item in red_skus
+ )
)
# 1 次打开规格面板，另有红色点 1 次；尺码只读不点击。
self.assertEqual(len(device.clicks), 2)
+ def test_colors_are_sampled_in_snake_order_before_sizes_are_read(self):
+ device = SnakeColorDevice(self.home_xml)
+ service = PddCollectService(
+ PddDeviceService(lambda _serial: device),
+ "USB-001",
+ "client-001",
+ sleeper=lambda _seconds: None,
+ )
+
+ data = service.collect(
+ FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
+ ).to_pdd_data()
+
+ self.assertEqual(
+ device.clicked_colors,
+ ["A色", "B色", "C色", "F色", "E色", "D色"],
+ )
+ self.assertEqual(
+ [item["text"] for item in data["dimensions"][0]["values"]],
+ ["A色", "B色", "C色", "D色", "E色", "F色"],
+ )
+ self.assertEqual(
+ [item["text"] for item in data["dimensions"][1]["values"]],
+ ["M", "L"],
+ )
+ color_prices = {
+ item["options"]["color"]: item["price_cent"]
+ for item in data["skus"]
+ if item["options"]["size"] == "M"
+ }
+ self.assertEqual(color_prices["A色"], 1000)
+ self.assertEqual(color_prices["B色"], 1000)
+ self.assertEqual(color_prices["F色"], 1500)
+ self.assertTrue(all(
+ item["price_observed_at"] == {"color": item["options"]["color"]}
+ for item in data["skus"]
+ ))
+
+ def test_visible_click_is_fallback_when_page_exposes_no_selection_state(self):
+ xml_data = keep_only_one_sku(self.spec_xml)
+ root = ET.fromstring(xml_data)
+ for parent in root.iter():
+ for child in list(parent):
+ if (child.get("text") or "").startswith("已选"):
+ parent.remove(child)
+ xml_without_selection = ET.tostring(root, encoding="unicode")
+ device = FakeCollectDevice(self.home_xml, xml_without_selection)
+ service = PddCollectService(
+ PddDeviceService(lambda _serial: device),
+ "USB-001",
+ "client-001",
+ sleeper=lambda _seconds: None,
+ max_spec_swipes=0,
+ )
+
+ data = service.collect(
+ FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
+ ).to_pdd_data()
+
+ self.assertEqual(data["skus"][0]["price_cent"], 1000)
+ self.assertEqual(
+ data["skus"][0]["price_observed_at"], {"color": "红色"}
+ )
+
+ def test_visible_text_does_not_confirm_an_ignored_click(self):
+ device = IgnoredColorClickDevice(self.home_xml)
+ device.panel_open = True
+ device.page = 0
+ service = PddCollectService(
+ PddDeviceService(lambda _serial: device),
+ "USB-001",
+ "client-001",
+ sleeper=lambda _seconds: None,
+ )
+
+ sample = service._click_and_sample_color(device, "B色")
+
+ self.assertIsNone(sample.price_cent)
+
def test_missing_goods_id_is_structured_error(self):
service = PddCollectService(
PddDeviceService(lambda _serial: object()),
diff --git a/client/test/test_pdd_real_xml_fixtures.py b/client/test/test_pdd_real_xml_fixtures.py
index 93c1848..15d5e58 100644
--- a/client/test/test_pdd_real_xml_fixtures.py
+++ b/client/test/test_pdd_real_xml_fixtures.py
@@ -65,9 +65,10 @@ class PddRealXmlFixtureTest(unittest.TestCase):
)
def test_real_spec_panel_has_all_colors_and_two_prices(self):
- result = parse_spec_panel(
- (REAL_XML / "737116531267_size.xml").read_text(encoding="utf-8")
+ xml_data = (REAL_XML / "737116531267_size.xml").read_text(
+ encoding="utf-8"
)
+ result = parse_spec_panel(xml_data)
colors = next(item for item in result.dimensions if item.key == "color")
self.assertEqual(len(colors.values), 8)
@@ -75,6 +76,23 @@ class PddRealXmlFixtureTest(unittest.TestCase):
self.assertEqual(result.price_cent, 470)
self.assertEqual(result.list_price_cent, 1990)
+ service = PddCollectService(
+ PddDeviceService(lambda _serial: object()),
+ "USB-001",
+ "client-001",
+ max_spec_swipes=0,
+ )
+ rows = service._visible_color_rows(xml_data)
+ self.assertEqual(len(rows), 2)
+ self.assertEqual(
+ [item.text for item in rows[0]],
+ ["黑色中长款", "白色中长款", "卡其中长款"],
+ )
+ self.assertEqual(
+ [item.text for item in rows[1]],
+ ["黑色长款", "白色长款", "卡其长款"],
+ )
+
def test_real_home_xml_has_immediate_spec_entry(self):
xml_data = (REAL_XML / "737116531267_home.xml").read_text(
encoding="utf-8"
diff --git a/docs/client/01-requirements.md b/docs/client/01-requirements.md
index 5c23d0d..93d7b7f 100644
--- a/docs/client/01-requirements.md
+++ b/docs/client/01-requirements.md
@@ -50,10 +50,16 @@ Client 顶级导航仅包含：
- 已拼数量及原始显示文字；
- 评价数量及原始显示文字（当前首页未提供时允许留空，不能为此反复滚动并阻塞规格采集）；
- 所有规格维度及其可见值；
-- 每个规格组合对应的价格、币种和可用状态；
+- 每个颜色对应的价格、币种，以及当前控件树显示的规格可用状态；
- 采集时间、设备及必要诊断产物引用。
-颜色、尺码和价格不能只保存为互不关联的列表，必须以 SKU 规格组合保存价格关系。对于“1.2 万+”等近似数值，同时保存规范化数值、原始文字和近似标记。
+当前业务按颜色采价：进入规格面板后逐个选中颜色并立即读取稳定价格；尺码只
+收集文字和当前显示的可用状态，不逐个选择、不单独采价。提交时仍以 SKU 规格
+组合保存颜色与尺码关系，同一颜色的组合复用该颜色价格，
+`price_observed_at` 只记录实际选中的颜色，不得虚构已选尺码。以后需要按尺码
+区分价格或组合库存时，必须另行确认并改用 `price_granularity=sku`。
+
+对于“1.2 万+”等近似数值，同时保存规范化数值、原始文字和近似标记。
### 4.2 采购任务
diff --git a/docs/client/02-architecture.md b/docs/client/02-architecture.md
index 609c1e7..8c23be5 100644
--- a/docs/client/02-architecture.md
+++ b/docs/client/02-architecture.md
@@ -355,8 +355,14 @@ reconcile_purchase(task, run) -> PurchaseResult | ManualReview
PDD 页面可能出现登录失效、验证码、控件树不完整、A/B 页面、库存变化和价格变化。适配层必须返回结构化错误，不得把这些情况统一返回 `False`。
采集商品时按“首页就绪 → 读取当前首页摘要 → 点击规格入口 → 确认规格面板
-出现 → 扫描颜色和尺码”的顺序执行。打开规格面板前不得为了寻找评价或店铺
-连续滚动商品详情；这两个可选字段缺失时保存证据，但不阻塞核心 SKU 采集。
+出现 → 颜色列表归左 → 按行蛇形逐色点击并采价 → 向下滚动并只读尺码 →
+组装结果”的顺序执行。颜色点击可能改变列表位置，因此每次点击后必须重新读取
+控件树，不能复用旧坐标；左右边缘以连续两次没有新颜色且视口稳定为准。
+
+当前业务价格粒度固定为颜色。第一行从左到右、下一行从右到左交替遍历，操作
+顺序采用蛇形以减少滑动，输出维度仍恢复为页面每行从左到右的自然顺序。尺码
+只读取文字和当前可用状态，不点击、不参与采价。打开规格面板前不得为了寻找
+评价或店铺连续滚动商品详情；这两个可选字段缺失时保存证据，但不阻塞核心采集。
## 10. 关键架构决策
diff --git a/docs/client/03-data-model.md b/docs/client/03-data-model.md
index 81fc65a..98783c8 100644
--- a/docs/client/03-data-model.md
+++ b/docs/client/03-data-model.md
@@ -502,6 +502,7 @@ CREATE TABLE app_settings (
{
"options": {"color": "黑色", "size": "L"},
"price_cent": 3990,
+ "price_observed_at": {"color": "黑色"},
"currency": "CNY",
"available": true,
"raw_price": "¥39.90"
@@ -518,6 +519,12 @@ CREATE TABLE app_settings (
}
```
+当前采集任务使用 `price_granularity=color`。Client 逐个选中颜色读取价格，
+尺码只读取文字；因此同一颜色的多个尺码组合可以共享颜色价格，但
+`price_observed_at` 只能包含实际选中的颜色。它不能填写未点击、未确认的尺码。
+这里的 `available` 来自当前规格节点显示状态，不表示 Client 已逐个验证了每个
+颜色与尺码组合的实时库存。
+
### 8.2 采购结果
采购任务在同一结构中增加：
diff --git a/docs/client/04-admin-api-contract.md b/docs/client/04-admin-api-contract.md
index d073cf4..57d32d9 100644
--- a/docs/client/04-admin-api-contract.md
+++ b/docs/client/04-admin-api-contract.md
@@ -283,7 +283,7 @@ Idempotency-Key: task-id:attempt-id:result-v1
"options": {"color": "黑色", "size": "M"},
"price_cent": 470,
"list_price_cent": 1990,
- "price_observed_at": {"color": "黑色", "size": "M"},
+ "price_observed_at": {"color": "黑色"},
"available": true,
"raw_price": "折后¥4.7"
}]
@@ -309,7 +309,7 @@ Idempotency-Key: task-id:attempt-id:result-v1
- `[必须]` 相同键但不同内容返回 `409 IDEMPOTENCY_CONFLICT`。
- `[必须]` Client 只有收到 `accepted: true` 后才能把本地任务标为 `succeeded`。
- `[必须]` `price_cent` 是实际支付价的整数分；划线价可放在 `list_price_cent`。
-- `[必须]` `price_granularity` 只能是 `color` 或 `sku`。按颜色采样时填 `color`，并为每条 SKU 填写读取该价格时实际选中的 `price_observed_at`。
+- `[必须]` `price_granularity` 只能是 `color` 或 `sku`。当前 Client 按颜色采样时填 `color`；同一颜色的尺码组合可以共享颜色价格，但 `price_observed_at` 只填写实际选中的颜色，不得虚构尺码。以后逐个选择完整组合采价时才使用 `sku`。
- `[必须]` `shop_name` 采不到时允许省略或留空；Admin 不得因此拒绝老版本 Client，也不得用空值覆盖已保存的店铺名。
### 6.1 Admin 必须无条件接受