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

fix: 安全兼容采购规格繁简匹配 (#251)

Browse Source
This commit is contained in:
chengma
2026-08-17 15:20:35 +08:00
parent 1927e3edb6
commit 134514e382
4 changed files with 373 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/pdd_u2_purchase_adapter.py
+55 -34
View File
@@ -53,6 +53,7 @@ from .util.select_color_size import (
color_selection_failure_reason,
select_color,
select_size,
size_selection_failure_reason,
)
@@ -1033,34 +1034,37 @@ class U2PddPurchaseAdapter(PddPurchaseAdapter):
color_selected = self._select_color(
device, panel_xml, color, action_delay=0.2
)
if not color_selected:
failed_xml = self._dump_hierarchy()
failure_reason = color_selection_failure_reason(
failed_xml, color
)
diagnostics: dict[str, Any] = {
"selection_failure": failure_reason,
}
artifact = self._save_last_xml(
"purchase-color-selection-mismatch"
)
if artifact is not None:
diagnostics["artifacts"] = [artifact]
messages = {
"target_not_visible": f"没有找到目标颜色：{color}",
"safe_target_missing": (
f"目标颜色没有完整可见的安全点击位置：{color}"
),
"selection_unconfirmed": (
f"点击颜色后页面没有确认已选中：{color}"
),
}
raise PddPurchaseError(
"PURCHASE_OPTIONS_MISMATCH",
messages[failure_reason],
step="purchase_select_options",
diagnostics=diagnostics,
)
if not color_selected:
failed_xml = self._dump_hierarchy()
failure_reason = color_selection_failure_reason(
failed_xml, color
)
diagnostics: dict[str, Any] = {
"selection_failure": failure_reason,
}
artifact = self._save_last_xml(
"purchase-color-selection-mismatch"
)
if artifact is not None:
diagnostics["artifacts"] = [artifact]
messages = {
"target_not_visible": f"没有找到目标颜色：{color}",
"target_ambiguous": (
f"目标颜色存在多个繁简等价候选，已停止采购：{color}"
),
"safe_target_missing": (
f"目标颜色没有完整可见的安全点击位置：{color}"
),
"selection_unconfirmed": (
f"点击颜色后页面没有确认已选中：{color}"
),
}
raise PddPurchaseError(
"PURCHASE_OPTIONS_MISMATCH",
messages[failure_reason],
step="purchase_select_options",
diagnostics=diagnostics,
)
size = checked.get("size")
if size:
latest_xml = self._dump_hierarchy()
@@ -1073,12 +1077,29 @@ class U2PddPurchaseAdapter(PddPurchaseAdapter):
size_selected = self._select_size(
device, latest_xml, size, action_delay=0.2
)
if not size_selected:
raise PddPurchaseError(
"PURCHASE_OPTIONS_MISMATCH",
f"没有精确选中尺码：{size}",
step="purchase_select_options",
)
if not size_selected:
failed_xml = self._dump_hierarchy()
failure_reason = size_selection_failure_reason(
failed_xml, size
)
messages = {
"target_not_visible": f"没有找到目标尺码：{size}",
"target_ambiguous": (
f"目标尺码存在多个繁简等价候选，已停止采购：{size}"
),
"safe_target_missing": (
f"目标尺码没有可靠的安全点击位置：{size}"
),
"selection_unconfirmed": (
f"点击尺码后页面没有确认已选中：{size}"
),
}
raise PddPurchaseError(
"PURCHASE_OPTIONS_MISMATCH",
messages[failure_reason],
step="purchase_select_options",
diagnostics={"selection_failure": failure_reason},
)
except PddPurchaseError:
raise
except Exception as exc:
client/src/util/select_color_size.py
+200 -19
View File
@@ -1,6 +1,11 @@
import ctypes
from functools import lru_cache
import re
import sys
import time
import unicodedata
import xml.etree.ElementTree as ET
from ctypes import wintypes
from typing import Any, Optional, Union
@@ -10,6 +15,7 @@ Bounds = tuple[int, int, int, int]
_BOUNDS_PATTERN = re.compile(r"^\[(\d+),(\d+)\]\[(\d+),(\d+)\]$")
_COLOR_HEADINGS = ("颜色分类", "颜色", "款式", "颜色款式", "花色")
_PACKAGE_HEADINGS = ("套餐",)
_LCMAP_SIMPLIFIED_CHINESE = 0x02000000
def _parse_xml(xml_data: XmlData) -> ET.Element:
@@ -41,21 +47,97 @@ def _node_labels(node: ET.Element) -> tuple[str, str]:
return node.get("text", "").strip(), node.get("content-desc", "").strip()
@lru_cache(maxsize=512)
def normalize_spec_text(value: str) -> str:
"""生成采购规格比较键；Windows 下把繁体转换为简体。
原始规格文字仍用于点击、显示和保存。本函数只生成比较键；Windows API
不可用时安全退化为 Unicode 规范化，不做模糊匹配。
"""
normalized = unicodedata.normalize("NFKC", str(value or "")).strip()
if not normalized or sys.platform != "win32":
return normalized.casefold()
try:
function = ctypes.windll.kernel32.LCMapStringEx
function.argtypes = [
wintypes.LPCWSTR,
wintypes.DWORD,
wintypes.LPCWSTR,
ctypes.c_int,
wintypes.LPWSTR,
ctypes.c_int,
ctypes.c_void_p,
ctypes.c_void_p,
wintypes.LPARAM,
]
function.restype = ctypes.c_int
required = function(
"zh-CN",
_LCMAP_SIMPLIFIED_CHINESE,
normalized,
-1,
None,
0,
None,
None,
0,
)
if required <= 0:
return normalized.casefold()
buffer = ctypes.create_unicode_buffer(required)
written = function(
"zh-CN",
_LCMAP_SIMPLIFIED_CHINESE,
normalized,
-1,
buffer,
required,
None,
None,
0,
)
if written <= 0:
return normalized.casefold()
return buffer.value.casefold()
except (AttributeError, OSError, TypeError, ValueError):
return normalized.casefold()
def _label_matches(label: str, target: str) -> bool:
if label.casefold() == target.casefold():
return True
pattern = rf"(?:^|[\s,，:：;；]){re.escape(target)}(?:$|[\s,，:：;；])"
return re.search(pattern, label, flags=re.IGNORECASE) is not None
def _node_match_kind(node: ET.Element, target: str) -> int:
"""返回 2=原文精确匹配、1=繁简规范化匹配、0=不匹配。"""
checked_target = target.strip()
if not checked_target:
return 0
labels = _node_labels(node)
if any(_label_matches(label, checked_target) for label in labels):
return 2
normalized_target = normalize_spec_text(checked_target)
if not normalized_target:
return 0
for label in labels:
normalized_label = normalize_spec_text(label)
if normalized_label and _label_matches(normalized_label, normalized_target):
return 1
return 0
def _matches_target(node: ET.Element, target: str) -> bool:
target = target.strip()
if not target:
if not target.strip():
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
return _node_match_kind(node, target) > 0
def _parent_map(root: ET.Element) -> dict[ET.Element, ET.Element]:
@@ -328,17 +410,45 @@ def _safe_horizontal_target(bounds: Bounds, region: Bounds) -> bool:
)
def _target_click_bounds(
def _contains_bounds(outer: Bounds, inner: Bounds) -> bool:
return (
outer[0] <= inner[0]
and outer[1] <= inner[1]
and outer[2] >= inner[2]
and outer[3] >= inner[3]
)
def _collapse_nested_bounds(bounds: list[Bounds]) -> list[Bounds]:
"""把同一规格卡片的外层和文字点击区域合并为外层卡片。"""
unique = sorted(
set(bounds),
key=lambda item: (item[2] - item[0]) * (item[3] - item[1]),
reverse=True,
)
result: list[Bounds] = []
for candidate in unique:
if any(_contains_bounds(existing, candidate) for existing in result):
continue
result.append(candidate)
return result
def _target_click_candidates(
root: ET.Element,
target: str,
region: Bounds,
require_horizontal_safe: bool = False,
) -> Optional[Bounds]:
) -> list[tuple[int, int, Bounds]]:
parents = _parent_map(root)
candidates: list[tuple[int, Bounds]] = []
candidates: list[tuple[int, int, Bounds]] = []
for node in root.iter("node"):
if not _is_available(node) or not _matches_target(node, target):
if not _is_available(node):
continue
match_kind = _node_match_kind(node, target)
if match_kind == 0:
continue
bounds = _nearest_click_bounds(node, parents, region)
@@ -348,11 +458,47 @@ def _target_click_bounds(
continue
area = (bounds[2] - bounds[0]) * (bounds[3] - bounds[1])
candidates.append((area, bounds))
candidates.append((match_kind, area, bounds))
return candidates
def _normalized_target_is_ambiguous(
root: ET.Element,
target: str,
region: Bounds,
require_horizontal_safe: bool = False,
) -> bool:
candidates = _target_click_candidates(
root,
target,
region,
require_horizontal_safe=require_horizontal_safe,
)
if not candidates or any(kind == 2 for kind, _area, _bounds in candidates):
return False
return len(_collapse_nested_bounds([item[2] for item in candidates])) > 1
def _target_click_bounds(
root: ET.Element,
target: str,
region: Bounds,
require_horizontal_safe: bool = False,
) -> Optional[Bounds]:
candidates = _target_click_candidates(
root,
target,
region,
require_horizontal_safe=require_horizontal_safe,
)
if not candidates:
return None
return max(candidates, key=lambda item: item[0])[1]
best_kind = max(item[0] for item in candidates)
best = [item for item in candidates if item[0] == best_kind]
if best_kind == 1:
bounds = _collapse_nested_bounds([item[2] for item in best])
return bounds[0] if len(bounds) == 1 else None
return max(best, key=lambda item: item[1])[2]
def _target_is_selected(root: ET.Element, target: str) -> bool:
@@ -391,6 +537,13 @@ def color_selection_failure_reason(xml_data: XmlData, target: str) -> str:
if not matching_nodes:
return "target_not_visible"
region = _horizontal_color_region(root, target)
if region is not None and _normalized_target_is_ambiguous(
root,
target,
region,
require_horizontal_safe=True,
):
return "target_ambiguous"
if region is None or _target_click_bounds(
root,
target,
@@ -401,6 +554,25 @@ def color_selection_failure_reason(xml_data: XmlData, target: str) -> str:
return "selection_unconfirmed"
def size_selection_failure_reason(xml_data: XmlData, target: str) -> str:
"""说明尺码选择失败阶段，繁简等价候选不唯一时明确拒绝。"""
root = _parse_xml(xml_data)
matching_nodes = [
node
for node in root.iter("node")
if _is_available(node) and _matches_target(node, target)
]
if not matching_nodes:
return "target_not_visible"
region = _vertical_panel_region(root)
if region is not None and _normalized_target_is_ambiguous(root, target, region):
return "target_ambiguous"
if region is None or _target_click_bounds(root, target, region) is None:
return "safe_target_missing"
return "selection_unconfirmed"
def _visible_signature(
root: ET.Element,
region: Bounds,
@@ -426,6 +598,13 @@ def _click_target_and_verify(
action_delay: float,
require_horizontal_safe: bool = False,
) -> tuple[bool, ET.Element]:
if _normalized_target_is_ambiguous(
root,
target,
region,
require_horizontal_safe=require_horizontal_safe,
):
return False, root
if _target_is_selected(root, target):
return True, root
@@ -444,6 +623,8 @@ def _click_target_and_verify(
time.sleep(action_delay)
refreshed_root = _parse_xml(device.dump_hierarchy())
# 点击前已经完成唯一候选校验。点击后页面通常会额外出现“已选：目标”
# 摘要，不能把摘要和原卡片误判成两个可选候选。
return _target_is_selected(refreshed_root, target), refreshed_root
client/test/test_pdd_u2_purchase_adapter.py
+29
View File
@@ -1618,6 +1618,35 @@ class U2PddPurchaseAdapterTest(unittest.TestCase):
self.assertIn("purchase_select_size", operations)
adapter.close()
def test_failed_option_stage_is_recorded_as_failed(self):
cases = (
({"color": "不存在的颜色"}, "purchase_select_color"),
(
{"color": "黑色", "size": "不存在的尺码"},
"purchase_select_size",
),
)
for options, expected_operation in cases:
with self.subTest(operation=expected_operation):
device = FakeDevice()
adapter = self._adapter(device, [])
records = []
trace = TaskPerformanceTrace(sink=records.append)
trace.bind_task(f"PUR-{expected_operation}")
adapter.open_goods(GOODS_URL)
with trace.activate(), self.assertRaises(PddPurchaseError):
adapter.select_options(options)
matching = [
record
for record in records
if record["operation"] == expected_operation
]
self.assertEqual(len(matching), 1)
self.assertEqual(matching[0]["result"], "failed")
adapter.close()
def test_same_quantity_does_not_focus_editor(self):
device = FakeDevice()
adapter = self._adapter(device, [])
client/test/test_select_color_size.py
+89 -3
View File
@@ -6,7 +6,10 @@ import xml.etree.ElementTree as ET
from src.util.select_color_size import (
color_selection_failure_reason,
normalize_spec_text,
select_color,
select_size,
size_selection_failure_reason,
)
@@ -16,9 +19,15 @@ FIXTURES = Path(__file__).parent / "fixtures"
class StaticColorDevice:
"""模拟没有滚动属性的图片颜色规格面板。"""
def __init__(self, xml_data: str, confirm_selection: bool = True) -> None:
def __init__(
self,
xml_data: str,
confirm_selection: bool = True,
selected_text: str = "测试黑色",
) -> None:
self.xml_data = xml_data
self.confirm_selection = confirm_selection
self.selected_text = selected_text
self.clicks = []
self.swipes = []
@@ -31,10 +40,10 @@ class StaticColorDevice:
return
root = ET.fromstring(self.xml_data)
for node in root.iter("node"):
if node.get("text", "").strip() == "测试黑色":
if node.get("text", "").strip() == self.selected_text:
node.set("selected", "true")
if node.get("text", "").strip().startswith("请选择"):
node.set("text", "已选： 测试黑色")
node.set("text", f"已选： {self.selected_text}")
self.xml_data = ET.tostring(root, encoding="unicode")
def swipe(self, *args, **kwargs) -> None:
@@ -122,6 +131,83 @@ class StaticColorSelectionTest(unittest.TestCase):
self.assertEqual(device.clicks, [])
self.assertEqual(device.swipes, [])
def test_traditional_color_matches_unique_simplified_card(self):
simplified = "缥缈仙姬绾【黑色】"
traditional = "縹緲仙姬綰【黑色】"
panel_xml = self.panel_xml.replace("测试黑色", simplified)
device = StaticColorDevice(panel_xml, selected_text=simplified)
selected = select_color(
device,
panel_xml,
traditional,
action_delay=0.001,
)
self.assertEqual(normalize_spec_text(traditional), simplified.casefold())
self.assertTrue(selected)
self.assertEqual(device.clicks, [(194, 1389)])
def test_traditional_size_matches_unique_simplified_option(self):
simplified = "均码【40.0-60.0公斤】"
traditional = "均碼【40.0-60.0公斤】"
panel_xml = self.panel_xml.replace('text="M"', f'text="{simplified}"')
device = StaticColorDevice(panel_xml, selected_text=simplified)
selected = select_size(
device,
panel_xml,
traditional,
action_delay=0.001,
)
self.assertTrue(selected)
self.assertEqual(device.clicks, [(77, 1762)])
def test_normalized_duplicate_color_candidates_are_not_clicked(self):
simplified = "缥缈仙姬绾【黑色】"
traditional = "縹緲仙姬綰【黑色】"
panel_xml = self.panel_xml.replace("测试黑色", simplified).replace(
"测试红色", simplified
)
device = StaticColorDevice(panel_xml, selected_text=simplified)
selected = select_color(
device,
panel_xml,
traditional,
action_delay=0.001,
)
self.assertFalse(selected)
self.assertEqual(device.clicks, [])
self.assertEqual(
color_selection_failure_reason(panel_xml, traditional),
"target_ambiguous",
)
def test_normalized_duplicate_size_candidates_are_not_clicked(self):
simplified = "均码【40.0-60.0公斤】"
traditional = "均碼【40.0-60.0公斤】"
panel_xml = self.panel_xml.replace('text="M"', f'text="{simplified}"').replace(
'text="L"', f'text="{simplified}"'
)
device = StaticColorDevice(panel_xml, selected_text=simplified)
selected = select_size(
device,
panel_xml,
traditional,
action_delay=0.001,
)
self.assertFalse(selected)
self.assertEqual(device.clicks, [])
self.assertEqual(
size_selection_failure_reason(panel_xml, traditional),
"target_ambiguous",
)
if __name__ == "__main__":
unittest.main()