fix: 蛇形遍历颜色并逐色采价 (#37)
This commit is contained in:
chengma
+414
-192
@@ -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]:
|
||||
|
||||
Reference in New Issue
Block a user