diff --git a/client/buyer_main.py b/client/buyer_main.py new file mode 100644 index 0000000..fe84d78 --- /dev/null +++ b/client/buyer_main.py @@ -0,0 +1,3 @@ +import sys +from src.ui_main import ui_main +sys.exit(ui_main()) \ No newline at end of file diff --git a/client/src/demo1/auto_v1.py b/client/src/demo1/auto_v1.py new file mode 100644 index 0000000..990c65c --- /dev/null +++ b/client/src/demo1/auto_v1.py @@ -0,0 +1,142 @@ +"""实验脚本:手动跑通"打开商品页 → 进规格面板 → 选颜色尺码"的流程。 + +**这是实验脚本,正式流程不得导入本文件。** + +里面的东西都是为了手动调试方便才这么写的,一样都不能带进正式服务: + +- 硬编码的商品号、设备地址、商品标题类名; +- `print` 输出(正式代码用日志,见 `docs/client/06-quality-security.md` §6); +- 写到当前目录的 xml/png(正式代码写 `data/artifacts/`, + 见 `docs/client/03-data-model.md` §2.1); +- 出错就 `return`、不区分原因(正式代码要返回结构化错误, + 不能把所有失败都变成 `False` 或 `None`,见 `docs/client/02-architecture.md` §9)。 + +要把这里的逻辑正式化,迁到 PDD 适配层,接口见 `02-architecture.md` §9。 + +运行前提:手机已连接、已登录拼多多。本脚本**不会下单**—— +最后的 `device.click(*coord)` 是注释掉的,不要随手放开。 +""" + +import time + +import uiautomator2 as u2 +from util.loaded_pdd import wait_goods_page +from util.get_size_panle_coord import get_size_panel_coord +from util.select_color_size import select_color_size +from util.get_order_button import get_order_button + +package_name='com.xunmeng.pinduoduo' +goods_id="737116531267" +goods_url=f"https://mobile.yangkeduo.com/goods.html?goods_id={goods_id}" +goods_title_node_class="androidx.viewpager.widget.ViewPager" +goods_color="抹茶绿长款" +goods_size="L(80-115斤)" +def main(): + # 连接设备 + device = u2.connect("192.168.0.173:5555") + + #获取当前应用包名 + current_package_name=device.app_current()["package"] + + #如果不是拼多多则启动 + if current_package_name!=package_name: + device.app_start(package_name) + + # 等待拼多多启动 + device.app_wait(package_name) + + # 等待拼多多首页加载完成 + device.idle_delay=10 + device.implicitly_wait(5) + + # 打开商品链接 + device.open_url(goods_url) + if wait_goods_page(device): + print("商品页面加载完成") + else: + print("商品页面加载超时") + print(device.app_current()) + device.screenshot("load_timeout.png") + # 等待页面跳转完成 + # device.implicitly_wait(15) + + + # 获取只能直接获取 Android 无障碍控件树,用来给llm分析改点击哪个 + goods_home_xml = device.dump_hierarchy(pretty=True) + with open(f"{goods_id}_home.xml","w",encoding='utf-8') as f: + f.write(goods_home_xml) + + #判断商品标题node是否存在 + goods_title_node=device(className=goods_title_node_class) + if goods_title_node.exists==False: + print("在规格面板") + return + print("在商品详情页") + device.screenshot(f"{goods_id}_home.png") + + # 获取规格面板坐标 + coord=get_size_panel_coord(goods_home_xml) + if coord is None: + print("解析失败") + return + + # 点击进入规格面板 + device.click(*coord) + time.sleep(1) + + print("等待规格面板加载完成") + + color_size_xml = device.dump_hierarchy(pretty=True) + with open(f"{goods_id}_size.xml","w",encoding='utf-8') as f: + f.write(color_size_xml) + device.screenshot(f"{goods_id}_size.png") + + # #尝试滑倒面板下面选择尺码 + # size_panle_scroller = device(scrollable=True) + # if size_panle_scroller.exists==False: + # print("没有找到可滚动的规格面板") + # return + + # #尝试滑动到底部 + # size_panle_scroller.scroll.toEnd(max_swipes=20, steps=50) + # time.sleep(1) + # size_panle_xml = device.dump_hierarchy(pretty=True) + # with open(f"{goods_id}_size.xml","w",encoding='utf-8') as f: + # f.write(size_panle_xml) + # device.screenshot(f"{goods_id}_size.png") + + # #选中颜色分类 + # color_node=device(text=goods_color) + # if color_node.exists==False: + # print("颜色分类不存在") + # return + # color_node.click() + # time.sleep(1) + + # #选中尺码 + # size_node=device(text=goods_size) + # if size_node.exists==False: + # print("尺码不存在") + # return + # size_node.click() + # time.sleep(1) + + success = select_color_size( + device=device, + xml_data=color_size_xml, + target_color=goods_color, + target_size=goods_size, + ) + if success: + print("颜色和尺码选择完成") + else: + print("没有找到指定颜色或尺码") + + order_xml = device.dump_hierarchy(pretty=True) + coord=get_order_button(order_xml) + print(coord) + # device.click(*coord) +if __name__ == "__main__": + import sys,os + sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + main() diff --git a/client/src/util/get_order_button.py b/client/src/util/get_order_button.py new file mode 100644 index 0000000..971615e --- /dev/null +++ b/client/src/util/get_order_button.py @@ -0,0 +1,168 @@ +import re +import xml.etree.ElementTree as ET +from typing import Optional, Union + + +XmlData = Union[str, bytes] +Bounds = tuple[int, int, int, int] +Coord = tuple[int, int] + +_BOUNDS_PATTERN = re.compile(r"^\[(\d+),(\d+)\]\[(\d+),(\d+)\]$") +_CURRENCY_PATTERN = re.compile(r"[¥¥]") +_ORDER_WORDS = ( + "提交订单", + "现在买", + "立即购买", + "确认购买", + "去结算", + "确认订单", + "购买", + "下单", + "提交", + "确认", + "结算", +) + + +def _parse_bounds(value: str) -> Optional[Bounds]: + match = _BOUNDS_PATTERN.fullmatch(value.strip()) + if not match: + return None + + left, top, right, bottom = map(int, match.groups()) + if right <= left or bottom <= top: + return None + return left, top, right, bottom + + +def _is_available(node: ET.Element) -> bool: + return ( + node.get("visible-to-user", "true") == "true" + and node.get("enabled", "true") == "true" + ) + + +def _node_label(node: ET.Element) -> str: + return f'{node.get("text", "")} {node.get("content-desc", "")}'.strip() + + +def _subtree_label(node: ET.Element) -> str: + return " ".join(_node_label(item) for item in node.iter("node")) + + +def _screen_size(nodes: list[ET.Element]) -> tuple[int, int]: + bounds = [ + parsed + for node in nodes + if (parsed := _parse_bounds(node.get("bounds", ""))) is not None + ] + if not bounds: + return 0, 0 + return max(item[2] for item in bounds), max(item[3] for item in bounds) + + +def _nearest_order_ancestor( + node: ET.Element, + parents: dict[ET.Element, ET.Element], + screen_width: int, + screen_height: int, +) -> Optional[ET.Element]: + current: Optional[ET.Element] = node + + while current is not None: + bounds = _parse_bounds(current.get("bounds", "")) + if ( + bounds is not None + and current.get("clickable") == "true" + and _is_available(current) + ): + width = bounds[2] - bounds[0] + height = bounds[3] - bounds[1] + center_y = (bounds[1] + bounds[3]) / 2 + + # 下单按钮应位于屏幕底部,且不能是覆盖大半屏幕的可点击根节点。 + if ( + center_y >= screen_height * 0.80 + and bounds[3] >= screen_height * 0.88 + and width >= screen_width * 0.15 + and height <= screen_height * 0.30 + ): + return current + current = parents.get(current) + return None + + +def get_order_button_coord(xml_data: XmlData) -> Optional[Coord]: + """从规格面板控件树中获取可靠的下单按钮中心坐标。 + + ``xml_data`` 应为 ``device.dump_hierarchy()`` 返回的 XML。函数只解析 + 坐标,不执行点击;找不到底部价格、可点击祖先或下单语义时返回 None。 + """ + try: + root = ET.fromstring(xml_data) + except (ET.ParseError, TypeError) as exc: + raise ValueError("xml_data 不是有效的无障碍控件树 XML") from exc + + nodes = list(root.iter("node")) + screen_width, screen_height = _screen_size(nodes) + if screen_width <= 0 or screen_height <= 0: + return None + + parents = { + child: parent + for parent in root.iter() + for child in parent + if child.tag == "node" + } + + # 多个价格文字可能属于同一个下单按钮,使用节点身份去重。 + candidates: dict[ET.Element, float] = {} + for node in nodes: + if not _is_available(node): + continue + + label = _node_label(node) + if not _CURRENCY_PATTERN.search(label): + continue + + bounds = _parse_bounds(node.get("bounds", "")) + if bounds is None: + continue + + center_y = (bounds[1] + bounds[3]) / 2 + if center_y < screen_height * 0.80: + continue + + button = _nearest_order_ancestor( + node, + parents, + screen_width, + screen_height, + ) + if button is None: + continue + + button_label = _subtree_label(button) + matched_words = [word for word in _ORDER_WORDS if word in button_label] + if not matched_words: + continue + + button_bounds = _parse_bounds(button.get("bounds", "")) + if button_bounds is None: + continue + + button_center_y = (button_bounds[1] + button_bounds[3]) / 2 + strongest_word = max(len(word) for word in matched_words) + score = strongest_word * 10 + button_center_y / screen_height + candidates[button] = max(score, candidates.get(button, float("-inf"))) + + if not candidates: + return None + + button = max(candidates, key=candidates.get) + left, top, right, bottom = _parse_bounds(button.get("bounds", "")) # type: ignore[misc] + return (left + right) // 2, (top + bottom) // 2 + + +# 简短别名,推荐使用语义更明确的 get_order_button_coord。 +get_order_button = get_order_button_coord diff --git a/client/src/util/get_size_panle_coord.py b/client/src/util/get_size_panle_coord.py new file mode 100644 index 0000000..1bc363a --- /dev/null +++ b/client/src/util/get_size_panle_coord.py @@ -0,0 +1,136 @@ +import re +import xml.etree.ElementTree as ET +from typing import Optional, Union + + +Coord = tuple[int, int] +XmlData = Union[str, bytes] + +_BOUNDS_PATTERN = re.compile( + r"^\[(?P\d+),(?P\d+)\]" + r"\[(?P\d+),(?P\d+)\]$" +) +_PRICE_PATTERN = re.compile(r"[¥¥](?:\s*\d)?") +_BUY_WORDS = ("购买", "拼单", "下单", "立即", "单独", "免拼") + + +def _parse_bounds(bounds: str) -> Optional[tuple[int, int, int, int]]: + match = _BOUNDS_PATTERN.fullmatch(bounds.strip()) + if not match: + return None + + left, top, right, bottom = ( + int(match.group(name)) + for name in ("left", "top", "right", "bottom") + ) + if right <= left or bottom <= top: + return None + return left, top, right, bottom + + +def _is_available(node: ET.Element) -> bool: + return ( + node.get("visible-to-user", "true") == "true" + and node.get("enabled", "true") == "true" + ) + + +def _node_label(node: ET.Element) -> str: + return f'{node.get("text", "")} {node.get("content-desc", "")}'.strip() + + +def _subtree_label(node: ET.Element) -> str: + return " ".join(_node_label(item) for item in node.iter("node")) + + +def _nearest_clickable_parent( + node: ET.Element, + parents: dict[ET.Element, ET.Element], +) -> Optional[ET.Element]: + current: Optional[ET.Element] = node + while current is not None: + if current.get("clickable") == "true" and _is_available(current): + return current + current = parents.get(current) + return None + + +def get_size_panel_coord(xml_data: XmlData) -> Optional[Coord]: + """从拼多多商品详情页控件树中寻找打开规格面板的购买位置。 + + 返回最可信购买控件的中心坐标 ``(x, y)``;没有找到可靠目标时 + 返回 ``None``。传入内容应为 ``device.dump_hierarchy()`` 返回的 XML。 + """ + try: + root = ET.fromstring(xml_data) + except (ET.ParseError, TypeError) as exc: + raise ValueError("xml_data 不是有效的无障碍控件树 XML") from exc + + nodes = list(root.iter("node")) + bounds_by_node = { + node: bounds + for node in nodes + if (bounds := _parse_bounds(node.get("bounds", ""))) is not None + } + if not bounds_by_node: + return None + + screen_width = max(bounds[2] for bounds in bounds_by_node.values()) + screen_height = max(bounds[3] for bounds in bounds_by_node.values()) + if screen_width <= 0 or screen_height <= 0: + return None + + parents = { + child: parent + for parent in root.iter() + for child in parent + if child.tag == "node" + } + + candidates: list[tuple[float, ET.Element]] = [] + for node, (left, top, right, bottom) in bounds_by_node.items(): + if not _is_available(node): + continue + + label = _node_label(node) + if not _PRICE_PATTERN.search(label): + continue + + center_x = (left + right) / 2 + center_y = (top + bottom) / 2 + x_ratio = center_x / screen_width + y_ratio = center_y / screen_height + + # 购买入口应在商品页右侧底部操作栏,先排除正文价格及弹窗价格。 + if x_ratio < 0.40 or y_ratio < 0.80: + continue + + clickable = _nearest_clickable_parent(node, parents) + if clickable is None or clickable not in bounds_by_node: + continue + + clickable_label = _subtree_label(clickable) + has_buy_semantics = any(word in clickable_label for word in _BUY_WORDS) + if not has_buy_semantics: + continue + + # 语义可靠性优先,其次选择更靠近右下角且包含数字的价格节点。 + has_price_number = bool(re.search(r"[¥¥]\s*\d", label)) + score = ( + 10.0 + + (3.0 if has_price_number else 0.0) + + y_ratio * 2.0 + + x_ratio + ) + candidates.append((score, clickable)) + + if not candidates: + return None + + _, target = max(candidates, key=lambda item: item[0]) + left, top, right, bottom = bounds_by_node[target] + return (left + right) // 2, (top + bottom) // 2 + + +# 兼容文件名中的 panle 拼写,推荐新代码使用 get_size_panel_coord。 +get_size_panle_coord = get_size_panel_coord diff --git a/client/src/util/loaded_pdd.py b/client/src/util/loaded_pdd.py new file mode 100644 index 0000000..eccabdd --- /dev/null +++ b/client/src/util/loaded_pdd.py @@ -0,0 +1,30 @@ +import time +def wait_goods_page(device, timeout=30): + end_time = time.monotonic() + timeout + + while time.monotonic() < end_time: + current = device.app_current() + + if current.get("package") != "com.xunmeng.pinduoduo": + time.sleep(1) + continue + + ready = any([ + bool(device(textContains="发起拼单").exists), + bool(device(textContains="立即购买").exists), + bool(device(textContains="单独购买").exists), + bool(device(textContains="快要抢光").exists), + bool(device(textContains="免拼购买").exists), + ]) + + loading = any([ + bool(device(textContains="加载中").exists), + bool(device(textContains="正在加载").exists), + ]) + + if ready and not loading: + return True + + time.sleep(1) + + return False diff --git a/client/src/util/select_color_size.py b/client/src/util/select_color_size.py new file mode 100644 index 0000000..0921f88 --- /dev/null +++ b/client/src/util/select_color_size.py @@ -0,0 +1,607 @@ +import re +import time +import xml.etree.ElementTree as ET +from typing import Any, Optional, Union + + +XmlData = Union[str, bytes] +Bounds = tuple[int, int, int, int] + +_BOUNDS_PATTERN = re.compile(r"^\[(\d+),(\d+)\]\[(\d+),(\d+)\]$") +_COLOR_HEADINGS = ("颜色分类", "颜色", "款式", "颜色款式") + + +def _parse_xml(xml_data: XmlData) -> ET.Element: + try: + return ET.fromstring(xml_data) + except (ET.ParseError, TypeError) as exc: + raise ValueError("xml_data 不是有效的无障碍控件树 XML") from exc + + +def _parse_bounds(value: str) -> Optional[Bounds]: + match = _BOUNDS_PATTERN.fullmatch(value.strip()) + if not match: + return None + + left, top, right, bottom = map(int, match.groups()) + if right <= left or bottom <= top: + return None + return left, top, right, bottom + + +def _is_available(node: ET.Element) -> bool: + return ( + node.get("visible-to-user", "true") == "true" + and node.get("enabled", "true") == "true" + ) + + +def _node_labels(node: ET.Element) -> tuple[str, str]: + return node.get("text", "").strip(), node.get("content-desc", "").strip() + + +def _matches_target(node: ET.Element, target: str) -> bool: + target = target.strip() + if not target: + return False + + for label in _node_labels(node): + if label.casefold() == target.casefold(): + return True + + # 兼容“颜色 黑色,已选中”或“尺码: XL”等无障碍描述, + # 同时避免 XL 错误匹配到 2XL。 + pattern = rf"(?:^|[\s,,::;;]){re.escape(target)}(?:$|[\s,,::;;])" + if re.search(pattern, label, flags=re.IGNORECASE): + return True + return False + + +def _parent_map(root: ET.Element) -> dict[ET.Element, ET.Element]: + return { + child: parent + for parent in root.iter() + for child in parent + if child.tag == "node" + } + + +def _center_in(bounds: Bounds, region: Bounds) -> bool: + x = (bounds[0] + bounds[2]) // 2 + y = (bounds[1] + bounds[3]) // 2 + return region[0] <= x <= region[2] and region[1] <= y <= region[3] + + +def _intersection_area(first: Bounds, second: Bounds) -> int: + width = max(0, min(first[2], second[2]) - max(first[0], second[0])) + height = max(0, min(first[3], second[3]) - max(first[1], second[1])) + return width * height + + +def _screen_bounds(root: ET.Element) -> Optional[Bounds]: + bounds = [ + item + for node in root.iter("node") + if (item := _parse_bounds(node.get("bounds", ""))) is not None + ] + if not bounds: + return None + return 0, 0, max(item[2] for item in bounds), max(item[3] for item in bounds) + + +def _heading_bounds(root: ET.Element) -> Optional[Bounds]: + fuzzy_match: Optional[Bounds] = None + for node in root.iter("node"): + text = node.get("text", "").strip() + bounds = _parse_bounds(node.get("bounds", "")) + if bounds is None or not _is_available(node): + continue + + # “请选择:颜色分类 尺码”等摘要不能当成颜色列表标题。 + if text == "颜色分类": + return bounds + if fuzzy_match is None and any(heading in text for heading in _COLOR_HEADINGS): + fuzzy_match = bounds + return fuzzy_match + + +def _target_scrollable_ancestor( + root: ET.Element, + target: str, +) -> Optional[Bounds]: + """目标已在控件树时,直接使用它最近的可滚动祖先。""" + parents = _parent_map(root) + for node in root.iter("node"): + if not _matches_target(node, target): + continue + + current = parents.get(node) + while current is not None: + if current.get("scrollable") == "true" and _is_available(current): + bounds = _parse_bounds(current.get("bounds", "")) + if bounds is not None: + return bounds + current = parents.get(current) + return None + + +def _horizontal_color_region( + root: ET.Element, + target: Optional[str] = None, +) -> Optional[Bounds]: + if target: + target_region = _target_scrollable_ancestor(root, target) + if target_region is not None: + return target_region + + heading = _heading_bounds(root) + candidates: list[tuple[float, Bounds]] = [] + screen = _screen_bounds(root) + + for node in root.iter("node"): + if node.get("scrollable") != "true" or not _is_available(node): + continue + + bounds = _parse_bounds(node.get("bounds", "")) + if bounds is None: + continue + + width = bounds[2] - bounds[0] + height = bounds[3] - bounds[1] + + if heading is not None and screen is not None: + distance_below_heading = bounds[1] - heading[3] + if ( + -10 <= distance_below_heading <= int(screen[3] * 0.20) + and width >= screen[2] * 0.60 + ): + # 图片颜色可能是两行网格,高度较大,不能依赖宽高比。 + score = 100_000.0 - distance_below_heading - width * height / 1_000_000 + candidates.append((score, bounds)) + continue + + if width >= height * 1.4: + candidates.append((width / height, bounds)) + + if candidates: + return max(candidates, key=lambda item: item[0])[1] + + # 部分自绘横向列表不会暴露 scrollable,使用颜色标题下方区域兜底。 + if heading is None or screen is None: + return None + + top = heading[3] + bottom = min(screen[3], top + max(160, int(screen[3] * 0.14))) + if bottom <= top: + return None + return int(screen[2] * 0.04), top, int(screen[2] * 0.96), bottom + + +def _vertical_panel_region(root: ET.Element) -> Optional[Bounds]: + heading = _heading_bounds(root) + vertical: list[tuple[float, Bounds]] = [] + + for node in root.iter("node"): + if node.get("scrollable") != "true" or not _is_available(node): + continue + + bounds = _parse_bounds(node.get("bounds", "")) + if bounds is None: + continue + + width = bounds[2] - bounds[0] + height = bounds[3] - bounds[1] + if height < width * 0.60: + continue + + if heading is not None and _center_in(heading, bounds): + # 优先颜色标题所在的最内层纵向滚动容器。 + score = 10_000_000 - width * height + else: + # 弹出的规格面板通常比背后的商品页起点更靠下。 + score = bounds[1] * 1000 + width * height / 1_000_000 + vertical.append((score, bounds)) + + if vertical: + return max(vertical, key=lambda item: item[0])[1] + + screen = _screen_bounds(root) + if screen is None: + return None + return ( + int(screen[2] * 0.06), + int(screen[3] * 0.22), + int(screen[2] * 0.94), + int(screen[3] * 0.82), + ) + + +def _nearest_click_bounds( + node: ET.Element, + parents: dict[ET.Element, ET.Element], + region: Bounds, +) -> Optional[Bounds]: + node_bounds = _parse_bounds(node.get("bounds", "")) + if node_bounds is None or not _center_in(node_bounds, region): + return None + + node_area = max( + 1, + (node_bounds[2] - node_bounds[0]) * (node_bounds[3] - node_bounds[1]), + ) + current: Optional[ET.Element] = node + clickable_bounds: list[tuple[int, Bounds]] = [] + + while current is not None: + bounds = _parse_bounds(current.get("bounds", "")) + if bounds is None or not _center_in(bounds, region): + break + + area = (bounds[2] - bounds[0]) * (bounds[3] - bounds[1]) + if ( + current.get("clickable") == "true" + and _is_available(current) + and area <= node_area * 15 + ): + clickable_bounds.append((area, bounds)) + current = parents.get(current) + + if clickable_bounds: + # 规格文字和外层卡片可能都 clickable,优先点击完整卡片。 + return max(clickable_bounds, key=lambda item: item[0])[1] + + # 自绘规格经常没有 clickable 属性,点击文字节点中心作为兜底。 + return node_bounds + + +def _safe_horizontal_target(bounds: Bounds, region: Bounds) -> bool: + """排除只在横向列表左右边缘露出一小部分的规格卡片。""" + region_width = region[2] - region[0] + target_width = bounds[2] - bounds[0] + center_x = (bounds[0] + bounds[2]) // 2 + safe_margin = max(30, int(region_width * 0.08)) + min_width = max(48, int(region_width * 0.08)) + return ( + target_width >= min_width + and region[0] + safe_margin <= center_x <= region[2] - safe_margin + and _intersection_area(bounds, region) + == (bounds[2] - bounds[0]) * (bounds[3] - bounds[1]) + ) + + +def _target_click_bounds( + root: ET.Element, + target: str, + region: Bounds, + require_horizontal_safe: bool = False, +) -> Optional[Bounds]: + parents = _parent_map(root) + candidates: list[tuple[int, Bounds]] = [] + + for node in root.iter("node"): + if not _is_available(node) or not _matches_target(node, target): + continue + + bounds = _nearest_click_bounds(node, parents, region) + if bounds is None: + continue + if require_horizontal_safe and not _safe_horizontal_target(bounds, region): + continue + + area = (bounds[2] - bounds[0]) * (bounds[3] - bounds[1]) + candidates.append((area, bounds)) + + if not candidates: + return None + return max(candidates, key=lambda item: item[0])[1] + + +def _target_is_selected(root: ET.Element, target: str) -> bool: + """检查目标节点、祖先或“已选”摘要是否确认了选中状态。""" + parents = _parent_map(root) + + for node in root.iter("node"): + if not _matches_target(node, target): + continue + + text, description = _node_labels(node) + label = f"{text} {description}" + if "已选" in label or "选中" in label: + return True + + current: Optional[ET.Element] = node + while current is not None: + if ( + current.get("selected") == "true" + or current.get("checked") == "true" + ): + return True + current = parents.get(current) + return False + + +def _visible_signature( + root: ET.Element, + region: Bounds, +) -> tuple[tuple[str, str, Bounds], ...]: + """记录当前区域可见文本和坐标,用来判断是否到达滚动边界。""" + result: list[tuple[str, str, Bounds]] = [] + for node in root.iter("node"): + bounds = _parse_bounds(node.get("bounds", "")) + if bounds is None or not _is_available(node) or not _center_in(bounds, region): + continue + + text, description = _node_labels(node) + if text or description: + result.append((text, description, bounds)) + return tuple(result) + + +def _click_target_and_verify( + device: Any, + root: ET.Element, + target: str, + region: Bounds, + action_delay: float, + require_horizontal_safe: bool = False, +) -> tuple[bool, ET.Element]: + if _target_is_selected(root, target): + return True, root + + bounds = _target_click_bounds( + root, + target, + region, + require_horizontal_safe=require_horizontal_safe, + ) + if bounds is None: + return False, root + + x = (bounds[0] + bounds[2]) // 2 + y = (bounds[1] + bounds[3]) // 2 + device.click(x, y) + time.sleep(action_delay) + + refreshed_root = _parse_xml(device.dump_hierarchy()) + return _target_is_selected(refreshed_root, target), refreshed_root + + +def _color_swipe_y(root: ET.Element, region: Bounds) -> int: + """选择最靠近容器中部的实际颜色行中心,避免在两行间隙滑动。""" + region_center_y = (region[1] + region[3]) // 2 + + for node in root.iter("node"): + if ( + node.get("scrollable") != "true" + or _parse_bounds(node.get("bounds", "")) != region + ): + continue + + row_centers: list[int] = [] + for child in node: + bounds = _parse_bounds(child.get("bounds", "")) + if bounds is None or not _is_available(child): + continue + if child.get("clickable") == "true" or child.get("content-desc", ""): + row_centers.append((bounds[1] + bounds[3]) // 2) + + if row_centers: + return min(row_centers, key=lambda y: abs(y - region_center_y)) + + return region_center_y + + +def _swipe_color_row( + device: Any, + root: ET.Element, + region: Bounds, + finger_right: bool, +) -> None: + left, top, right, bottom = region + width = right - left + y = _color_swipe_y(root, region) + # 每次移动约 40% 宽度,避免一次跨过只短暂出现的颜色卡片。 + start_x = left + int(width * (0.30 if finger_right else 0.70)) + end_x = left + int(width * (0.70 if finger_right else 0.30)) + device.swipe(start_x, y, end_x, y, duration=0.4) + + +def _swipe_panel_down(device: Any, region: Bounds) -> None: + left, top, right, bottom = region + height = bottom - top + x = (left + right) // 2 + # 手指向上滑,规格内容向下滚动。 + device.swipe( + x, + top + int(height * 0.82), + x, + top + int(height * 0.25), + duration=0.35, + ) + + +def select_color( + device: Any, + xml_data: XmlData, + target_color: str, + max_swipes: int = 15, + action_delay: float = 0.5, +) -> bool: + """边横向滑动边刷新 XML,目标颜色充分可见后点击并验证选中。""" + if not target_color.strip(): + raise ValueError("target_color 不能为空") + + root = _parse_xml(xml_data) + region = _horizontal_color_region(root, target_color) + if region is None: + return False + + selected, root = _click_target_and_verify( + device, + root, + target_color, + region, + action_delay, + require_horizontal_safe=True, + ) + if selected: + return True + region = _horizontal_color_region(root, target_color) or region + + # 当前横向位置未知:先用手指向右滑到列表左端。每次刷新 XML 时, + # _click_target_and_verify 都会遍历整个 RecyclerView 内的所有行。 + previous_signature = _visible_signature(root, region) + unchanged_count = 0 + for _ in range(max_swipes): + _swipe_color_row(device, root, region, finger_right=True) + time.sleep(action_delay) + root = _parse_xml(device.dump_hierarchy()) + region = _horizontal_color_region(root, target_color) or region + + selected, root = _click_target_and_verify( + device, + root, + target_color, + region, + action_delay, + require_horizontal_safe=True, + ) + if selected: + return True + region = _horizontal_color_region(root, target_color) or region + + signature = _visible_signature(root, region) + if signature == previous_signature: + unchanged_count += 1 + else: + unchanged_count = 0 + previous_signature = signature + if unchanged_count >= 2: + break + + # 从左端用手指向左慢慢滑到右端;每个视口同时检查全部颜色行。 + previous_signature = _visible_signature(root, region) + unchanged_count = 0 + for _ in range(max_swipes): + _swipe_color_row(device, root, region, finger_right=False) + time.sleep(action_delay) + root = _parse_xml(device.dump_hierarchy()) + region = _horizontal_color_region(root, target_color) or region + + selected, root = _click_target_and_verify( + device, + root, + target_color, + region, + action_delay, + require_horizontal_safe=True, + ) + if selected: + return True + region = _horizontal_color_region(root, target_color) or region + + signature = _visible_signature(root, region) + if signature == previous_signature: + unchanged_count += 1 + else: + unchanged_count = 0 + previous_signature = signature + if unchanged_count >= 2: + break + + return False + + +def select_size( + device: Any, + xml_data: XmlData, + target_size: str, + max_swipes: int = 15, + action_delay: float = 0.5, +) -> bool: + """边向下滚动规格面板边刷新 XML,点击目标尺码并验证选中。""" + if not target_size.strip(): + raise ValueError("target_size 不能为空") + + root = _parse_xml(xml_data) + region = _vertical_panel_region(root) + if region is None: + return False + + selected, root = _click_target_and_verify( + device, + root, + target_size, + region, + action_delay, + ) + if selected: + return True + region = _vertical_panel_region(root) or region + + previous_signature = _visible_signature(root, region) + unchanged_count = 0 + for _ in range(max_swipes): + _swipe_panel_down(device, region) + time.sleep(action_delay) + root = _parse_xml(device.dump_hierarchy()) + region = _vertical_panel_region(root) or region + + selected, root = _click_target_and_verify( + device, + root, + target_size, + region, + action_delay, + ) + if selected: + return True + region = _vertical_panel_region(root) or region + + signature = _visible_signature(root, region) + if signature == previous_signature: + unchanged_count += 1 + else: + unchanged_count = 0 + previous_signature = signature + if unchanged_count >= 2: + break + + return False + + +def select_color_size( + device: Any, + xml_data: XmlData, + target_color: str, + target_size: str, + max_horizontal_swipes: int = 15, + max_vertical_swipes: int = 15, + action_delay: float = 0.5, +) -> bool: + """在已打开的拼多多规格面板中选择目标颜色和尺码。 + + 颜色列表先归位到左端,再单向扫描同一 RecyclerView 中的所有行。 + 颜色和尺码区域每滑动一次都会重新调用 ``dump_hierarchy``,不会假设 + 最后一次 XML 包含所有历史节点。颜色点击后会丢弃旧 XML,使用最新 + 规格状态继续查找尺码。两项都确认选中时返回 True。 + """ + color_selected = select_color( + device, + xml_data, + target_color, + max_swipes=max_horizontal_swipes, + action_delay=action_delay, + ) + if not color_selected: + return False + + # 颜色可能改变尺码库存和布局,必须获取新的控件树。 + current_xml = device.dump_hierarchy() + return select_size( + device, + current_xml, + target_size, + max_swipes=max_vertical_swipes, + action_delay=action_delay, + ) diff --git a/run_buy.bat b/run_buy.bat new file mode 100644 index 0000000..1dcc53d --- /dev/null +++ b/run_buy.bat @@ -0,0 +1,2 @@ +python ./client/buyer_main.py +pause \ No newline at end of file