chore: 提交现有客户端自动化代码
This commit is contained in:
@@ -0,0 +1,3 @@
|
||||
import sys
|
||||
from src.ui_main import ui_main
|
||||
sys.exit(ui_main())
|
||||
@@ -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()
|
||||
@@ -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
|
||||
@@ -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<left>\d+),(?P<top>\d+)\]"
|
||||
r"\[(?P<right>\d+),(?P<bottom>\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
|
||||
@@ -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
|
||||
@@ -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,
|
||||
)
|
||||
@@ -0,0 +1,2 @@
|
||||
python ./client/buyer_main.py
|
||||
pause
|
||||
Reference in New Issue
Block a user