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

feat: 完整采集采购尺码候选 (#256)

Browse Source
This commit is contained in:
chengma
2026-08-17 17:04:43 +08:00
parent f187bfe0fe
commit 58dc9022eb
7 changed files with 458 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/pdd_collect_service.py
+29 -2
View File
@@ -1182,6 +1182,15 @@ class PddCollectService:
artifacts=tuple(self._artifacts),
)
def collect_second_dimension_candidates(
self, device: Any
) -> Optional[SpecDimension]:
"""复用采集流程完整遍历当前颜色的第二规格，不点击任何选项。"""
return self._collect_size_dimension(
device, require_confirmed_edges=True
)
def _check_cancelled(self) -> None:
if self._cancelled():
raise PddCollectError("PDD_CANCELLED", "采集任务已安全取消")
@@ -1926,7 +1935,12 @@ class PddCollectService:
break
return latest_xml
def _collect_size_dimension(self, device: Any) -> Optional[SpecDimension]:
def _collect_size_dimension(
self,
device: Any,
*,
require_confirmed_edges: bool = False,
) -> Optional[SpecDimension]:
"""完整遍历第二规格并收集文字，全程不点击尺码或套餐。"""
sizes: dict[str, bool] = {}
@@ -2008,7 +2022,10 @@ class PddCollectService:
continuation = self._build_second_dimension_context(
xml_data, initial_size_dimension
)
strict_edge_check = "套餐" in size_name and self._max_spec_swipes > 0
strict_edge_check = (
require_confirmed_edges
or ("套餐" in size_name and self._max_spec_swipes > 0)
)
if strict_edge_check and not start_confirmed:
raise PddCollectError(
"PDD_DATA_SPEC_INCOMPLETE",
@@ -2051,6 +2068,16 @@ class PddCollectService:
option_text = self._second_dimension_option_text(value.text)
if option_text is None:
continue
if (
require_confirmed_edges
and option_text in sizes
and sizes[option_text] != value.available
):
raise PddCollectError(
"PDD_DATA_SPEC_INCOMPLETE",
f"完整遍历时第二规格状态发生变化：{size_name}",
{"dimension_name": size_name},
)
sizes[option_text] = sizes.get(option_text, False) or value.available
signature = self._second_dimension_signature(xml_data, size_dimension)
client/src/pdd_purchase_adapter.py
+138 -1
View File
@@ -6,9 +6,128 @@
from __future__ import annotations
import hashlib
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Mapping
from typing import Mapping, Sequence
def _hash_frame(value: str) -> str:
"""按接口契约生成带 UTF-8 字节长度的哈希片段。"""
return f"{len(value.encode('utf-8'))}:{value}"
@dataclass(frozen=True)
class PurchaseSizeObservation:
"""完整遍历时看到的一条第二规格，包含不可购买项。"""
raw_text: str
available: bool
@dataclass(frozen=True)
class PurchaseSpecCandidate:
"""可以交给 Admin 解析的一条可购买尺码候选。"""
candidate_id: str
raw_text: str
options: Mapping[str, str]
def to_dict(self) -> dict[str, object]:
return {
"candidate_id": self.candidate_id,
"raw_text": self.raw_text,
"options": dict(self.options),
}
@dataclass(frozen=True)
class PurchaseSpecCandidateSnapshot:
"""当前商品、已选颜色下，一次完整且稳定的尺码候选观察。"""
goods_id: str
selected_color: str
target_size: str
dimension_name: str
observations: tuple[PurchaseSizeObservation, ...]
candidates: tuple[PurchaseSpecCandidate, ...]
candidate_snapshot_hash: str
observed_at: str
@classmethod
def build(
cls,
*,
goods_id: str,
selected_color: str,
target_size: str,
dimension_name: str,
observations: Sequence[PurchaseSizeObservation],
observed_at: str,
) -> "PurchaseSpecCandidateSnapshot":
"""按页面顺序生成候选短编号和 spec-resolution-v1 哈希。"""
checked_observations = tuple(observations)
candidates = tuple(
PurchaseSpecCandidate(
candidate_id=f"c{index}",
raw_text=item.raw_text,
options={"color": selected_color, "size": item.raw_text},
)
for index, item in enumerate(
(item for item in checked_observations if item.available),
start=1,
)
)
material = "".join(
(
_hash_frame("spec-resolution-v1"),
_hash_frame(goods_id),
_hash_frame(selected_color),
_hash_frame(str(len(candidates))),
*(
"".join(
(
_hash_frame(item.candidate_id),
_hash_frame(item.raw_text),
_hash_frame(str(item.options["color"])),
_hash_frame(str(item.options["size"])),
)
)
for item in candidates
),
)
)
return cls(
goods_id=goods_id,
selected_color=selected_color,
target_size=target_size,
dimension_name=dimension_name,
observations=checked_observations,
candidates=candidates,
candidate_snapshot_hash=hashlib.sha256(
material.encode("utf-8")
).hexdigest(),
observed_at=observed_at,
)
def to_dict(self) -> dict[str, object]:
"""返回确定顺序的内存结构；调用方不得把它直接写入日志。"""
return {
"goods_id": self.goods_id,
"selected_color": self.selected_color,
"target_size": self.target_size,
"dimension_name": self.dimension_name,
"observations": [
{"raw_text": item.raw_text, "available": item.available}
for item in self.observations
],
"candidates": [item.to_dict() for item in self.candidates],
"candidate_snapshot_hash": self.candidate_snapshot_hash,
"observed_at": self.observed_at,
}
@dataclass(frozen=True)
@@ -46,6 +165,24 @@ class PddPurchaseError(RuntimeError):
self.diagnostics = dict(diagnostics or {})
class PddPurchaseSpecResolutionRequired(PddPurchaseError):
"""本地无法精确匹配，但已得到完整可购买候选快照。"""
def __init__(self, snapshot: PurchaseSpecCandidateSnapshot) -> None:
super().__init__(
"PURCHASE_SPEC_RESOLUTION_REQUIRED",
f"目标尺码无法精确匹配，需要解析候选：{snapshot.target_size}",
step="purchase_select_options",
diagnostics={
"selection_failure": "target_not_visible",
"candidate_count": len(snapshot.candidates),
"candidate_snapshot_hash": snapshot.candidate_snapshot_hash,
"dimension_name": snapshot.dimension_name,
},
)
self.snapshot = snapshot
class PddPurchaseAdapter(ABC):
"""一台设备的一次采购演练会话；只能由创建它的工作线程使用。"""
client/src/pdd_u2_purchase_adapter.py
+122
View File
@@ -11,6 +11,7 @@ import re
import time
import xml.etree.ElementTree as ET
from contextlib import nullcontext
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
from pathlib import Path
from typing import Any, Callable, Mapping, Optional
@@ -24,6 +25,11 @@ from .pdd_device_service import (
current_thread_device_service,
)
from .performance_timing import current_performance_trace
from .pdd_collect_service import (
PddCollectError,
PddCollectService,
SpecDimension,
)
from .pdd_page_classifier import (
ACTION_NETWORK_ERROR,
ACTION_READY,
@@ -46,11 +52,15 @@ from .pdd_purchase_adapter import (
PddLivePurchaseAdapter,
PddPurchaseAdapter,
PddPurchaseError,
PddPurchaseSpecResolutionRequired,
PurchaseSizeObservation,
PurchaseSpecCandidateSnapshot,
PurchasePageState,
)
from .util.get_size_panle_coord import get_size_panel_coord
from .util.select_color_size import (
color_selection_failure_reason,
normalize_spec_text,
select_color,
select_size,
size_selection_failure_reason,
@@ -890,6 +900,10 @@ class U2PddPurchaseAdapter(PddPurchaseAdapter):
panel_timeout: float = 10.0,
select_color_fn: Callable[..., bool] = select_color,
select_size_fn: Callable[..., bool] = select_size,
size_candidate_collector: Optional[
Callable[[Any], Optional[SpecDimension]]
] = None,
now: Callable[[], datetime] = lambda: datetime.now(timezone.utc),
artifact_directory: Optional[Path] = None,
) -> None:
self._device_address = str(device_address or "").strip()
@@ -901,6 +915,10 @@ class U2PddPurchaseAdapter(PddPurchaseAdapter):
self._panel_timeout = panel_timeout
self._select_color = select_color_fn
self._select_size = select_size_fn
self._size_candidate_collector = (
size_candidate_collector or self._collect_size_candidates
)
self._now = now
self._artifact_directory = artifact_directory
self._last_xml: Optional[str] = None
self._hierarchy_reads = 0
@@ -1023,6 +1041,7 @@ class U2PddPurchaseAdapter(PddPurchaseAdapter):
panel_xml = self._wait_for_confirmation_panel()
color = checked.get("color")
selected_color_for_resolution = ""
if color:
panel_xml = self._restore_purchase_panel_color_region(panel_xml)
trace = current_performance_trace()
@@ -1065,6 +1084,7 @@ class U2PddPurchaseAdapter(PddPurchaseAdapter):
step="purchase_select_options",
diagnostics=diagnostics,
)
selected_color_for_resolution = color
size = checked.get("size")
if size:
latest_xml = self._dump_hierarchy()
@@ -1082,6 +1102,15 @@ class U2PddPurchaseAdapter(PddPurchaseAdapter):
failure_reason = size_selection_failure_reason(
failed_xml, size
)
if (
failure_reason == "target_not_visible"
and selected_color_for_resolution
):
self._raise_spec_resolution_required(
failed_xml,
selected_color_for_resolution,
size,
)
messages = {
"target_not_visible": f"没有找到目标尺码：{size}",
"target_ambiguous": (
@@ -1105,6 +1134,99 @@ class U2PddPurchaseAdapter(PddPurchaseAdapter):
except Exception as exc:
self._raise_device_or_page_error(exc, "purchase_select_options")
def _raise_spec_resolution_required(
self,
failed_xml: str,
selected_color: str,
target_size: str,
) -> None:
"""完整只读遍历后，仅为页面确实不存在的目标返回候选快照。"""
try:
self._restore_purchase_panel_color_region(failed_xml)
dimension = self._size_candidate_collector(self._require_device())
except PddPurchaseError:
raise
except PddCollectError as exc:
raise PddPurchaseError(
"PURCHASE_SIZE_CANDIDATE_SCAN_INCOMPLETE",
f"目标尺码未找到，且无法完整读取页面候选：{exc.message}",
step="purchase_select_options",
diagnostics={
"selection_failure": "target_not_visible",
"candidate_scan_error": exc.code,
},
) from exc
except Exception as exc:
self._raise_device_or_page_error(exc, "purchase_select_options")
if dimension is None or not dimension.values:
raise PddPurchaseError(
"PURCHASE_SIZE_CANDIDATE_SCAN_INCOMPLETE",
"目标尺码未找到，且页面没有可确认完整的第二规格候选",
step="purchase_select_options",
diagnostics={"selection_failure": "target_not_visible"},
)
available_values = tuple(
item for item in dimension.values if item.available
)
if not available_values:
raise PddPurchaseError(
"PURCHASE_OPTIONS_MISMATCH",
f"当前颜色没有可购买的尺码候选：{selected_color}",
step="purchase_select_options",
diagnostics={"selection_failure": "target_not_visible"},
)
# 目标若在完整列表中精确或繁简等价出现，说明前面的点击/确认失败，
# 不能把它伪装成需要远端解析的问题。
normalized_target = normalize_spec_text(target_size)
if any(
item.text == target_size
or normalize_spec_text(item.text) == normalized_target
for item in available_values
):
raise PddPurchaseError(
"PURCHASE_OPTIONS_MISMATCH",
f"页面存在目标尺码，但没有可靠确认选中：{target_size}",
step="purchase_select_options",
diagnostics={"selection_failure": "selection_unconfirmed"},
)
observed_at = (
self._now()
.astimezone(timezone.utc)
.isoformat()
.replace("+00:00", "Z")
)
snapshot = PurchaseSpecCandidateSnapshot.build(
goods_id=self._goods_id,
selected_color=selected_color,
target_size=target_size,
dimension_name=dimension.name,
observations=tuple(
PurchaseSizeObservation(item.text, item.available)
for item in dimension.values
),
observed_at=observed_at,
)
raise PddPurchaseSpecResolutionRequired(snapshot)
def _collect_size_candidates(self, device: Any) -> Optional[SpecDimension]:
"""使用采集模块同一套第二规格遍历，不建立新连接也不保存 XML。"""
service = PddCollectService(
self._device_service,
self._device_address,
"purchase-runtime",
sleeper=self._sleep,
monotonic=self._monotonic,
cancelled=self._cancelled,
artifact_directory=None,
)
return service.collect_second_dimension_candidates(device)
def set_quantity(self, quantity: int) -> None:
device = self._require_device()
if isinstance(quantity, bool) or not isinstance(quantity, int) or quantity <= 0: