"""PDD 页面分类和商品深链跳转状态。 本模块只处理脱敏的页面结构，不依赖 Qt、SQLite 或具体业务服务。采集和采购必须 共用这里的分类，避免一边把首页当商品页、另一边只能等待超时。 """ from __future__ import annotations import hashlib import re import xml.etree.ElementTree as ET from dataclasses import dataclass from .pdd_device_service import PDD_PACKAGE_NAME PAGE_EXTERNAL = "external" PAGE_LOADING = "loading" PAGE_GOODS = "goods" PAGE_HOME = "home" PAGE_NETWORK_ERROR = "network_error" PAGE_LOGIN_REQUIRED = "login_required" PAGE_CAPTCHA = "captcha" PAGE_RISK_CONTROL = "risk_control" PAGE_PAYMENT = "payment" PAGE_ORDER_CONFIRMATION = "order_confirmation" PAGE_UNKNOWN = "unknown" ACTION_WAIT = "wait" ACTION_READY = "ready" ACTION_REOPEN = "reopen" ACTION_UNAVAILABLE = "unavailable" ACTION_NETWORK_ERROR = "network_error" _READY_MARKERS = ("发起拼单", "立即购买", "单独购买", "免拼购买", "快要抢光") _LOGIN_MARKERS = ("手机号登录", "登录后继续", "验证码登录", "账号登录") _CAPTCHA_MARKERS = ("请完成验证", "拖动滑块", "安全验证", "点击图中") _RISK_MARKERS = ("操作频繁", "异常请求", "风险提示", "账号异常") _PAYMENT_MARKERS = ("输入支付密码", "立即支付", "支付成功", "支付失败") _FINAL_SUBMIT_MARKERS = ("提交订单", "现在买，仅", "确认购买") _NETWORK_MARKERS = ( "网络不给力", "网络异常", "加载失败", "连接失败", "服务异常", "请检查网络", ) _LOADING_MARKERS = ("加载中", "正在加载", "努力加载") _HOME_NAVIGATION = frozenset({"首页", "聊天", "个人中心"}) _BOUNDS_PATTERN = re.compile(r"^\[(\d+),(\d+)\]\[(\d+),(\d+)\]$") _VOLATILE_NUMBER = re.compile(r"\d+(?:\.\d+)?") @dataclass(frozen=True) class PddPageObservation: """一次页面读取的分类和仅供当前任务比较的脱敏签名。""" kind: str signature: str pdd_hierarchy: bool @dataclass(frozen=True) class GoodsOpenDecision: """商品深链观察器给调用方的下一步动作。""" action: str attempt: int def classify_pdd_page( root: ET.Element, current_package: str, ) -> PddPageObservation: """使用包名和组合控件特征分类当前页面。""" nodes = list(root.iter("node")) labels = [_node_label(node) for node in nodes] labels = [label for label in labels if label] combined = " ".join(labels) pdd_node_count = sum( 1 for node in nodes if node.get("package") == PDD_PACKAGE_NAME ) pdd_hierarchy = pdd_node_count >= 3 is_pdd = current_package == PDD_PACKAGE_NAME or pdd_hierarchy if any(marker in combined for marker in _CAPTCHA_MARKERS): kind = PAGE_CAPTCHA elif any(marker in combined for marker in _LOGIN_MARKERS): kind = PAGE_LOGIN_REQUIRED elif any(marker in combined for marker in _RISK_MARKERS): kind = PAGE_RISK_CONTROL elif any(marker in combined for marker in _PAYMENT_MARKERS): kind = PAGE_PAYMENT elif any(marker in combined for marker in _FINAL_SUBMIT_MARKERS): kind = PAGE_ORDER_CONFIRMATION elif is_pdd and any(marker in combined for marker in _NETWORK_MARKERS): kind = PAGE_NETWORK_ERROR elif is_pdd and any(marker in combined for marker in _LOADING_MARKERS): kind = PAGE_LOADING elif is_pdd and any(marker in combined for marker in _READY_MARKERS): kind = PAGE_GOODS elif is_pdd and _is_loaded_home(nodes, labels): kind = PAGE_HOME elif not is_pdd: kind = PAGE_EXTERNAL else: kind = PAGE_UNKNOWN return PddPageObservation( kind=kind, signature=_page_signature(nodes, kind), pdd_hierarchy=pdd_hierarchy, ) class GoodsOpenTracker: """确认商品深链已离开旧页，并控制首页场景只重开一次。""" def __init__( self, before_open: PddPageObservation | None, opened_at: float, *, stable_home_reads: int = 3, home_grace_seconds: float = 2.0, ) -> None: if stable_home_reads < 2: raise ValueError("首页稳定读取次数至少为 2") if home_grace_seconds < 0: raise ValueError("首页跳转宽限时间不能为负数") self._before_open = before_open self._opened_at = float(opened_at) self._stable_home_reads = stable_home_reads self._home_grace_seconds = float(home_grace_seconds) self._attempt = 1 self._home_reads = 0 self._stale_goods_seen = False @property def attempt(self) -> int: return self._attempt @property def stale_goods_seen(self) -> bool: return self._stale_goods_seen def reopened(self, opened_at: float) -> None: """调用方重新打开同一 URL 后开始第二次、也是最后一次观察。""" if self._attempt != 1: raise RuntimeError("商品链接最多只允许受控重开一次") self._attempt = 2 self._opened_at = float(opened_at) self._home_reads = 0 def observe( self, observation: PddPageObservation, observed_at: float, ) -> GoodsOpenDecision: """读取一个新页面分类，并返回等待、成功、重开或失败。""" if observation.kind == PAGE_GOODS: self._home_reads = 0 before = self._before_open if ( before is not None and before.kind == PAGE_GOODS and before.signature == observation.signature ): self._stale_goods_seen = True return GoodsOpenDecision(ACTION_WAIT, self._attempt) return GoodsOpenDecision(ACTION_READY, self._attempt) if observation.kind == PAGE_NETWORK_ERROR: self._home_reads = 0 return GoodsOpenDecision(ACTION_NETWORK_ERROR, self._attempt) if observation.kind != PAGE_HOME: self._home_reads = 0 return GoodsOpenDecision(ACTION_WAIT, self._attempt) self._home_reads += 1 stable = self._home_reads >= self._stable_home_reads grace_elapsed = ( float(observed_at) - self._opened_at >= self._home_grace_seconds ) if not stable or not grace_elapsed: return GoodsOpenDecision(ACTION_WAIT, self._attempt) if self._attempt == 1: return GoodsOpenDecision(ACTION_REOPEN, self._attempt) return GoodsOpenDecision(ACTION_UNAVAILABLE, self._attempt) def _is_loaded_home(nodes: list[ET.Element], labels: list[str]) -> bool: exact_labels = set(labels) if not _HOME_NAVIGATION.issubset(exact_labels): return False home_selected = any( _node_label(node) == "首页" and node.get("selected") == "true" for node in nodes ) has_clickable_home = any( node.get("content-desc", "").strip() == "首页" and node.get("clickable") == "true" for node in nodes ) return home_selected and has_clickable_home def _page_signature(nodes: list[ET.Element], kind: str) -> str: """对页面上半部结构做哈希；不把原始文字写入日志或跨任务保存。""" screen_bottom = 0 parsed_bounds: dict[int, tuple[int, int, int, int]] = {} for node in nodes: match = _BOUNDS_PATTERN.fullmatch(node.get("bounds", "")) if match is None: continue bounds = tuple(map(int, match.groups())) parsed_bounds[id(node)] = bounds screen_bottom = max(screen_bottom, bounds[3]) parts = [kind] for node in nodes: label = _node_label(node) if not label or node.get("visible-to-user") == "false": continue bounds = parsed_bounds.get(id(node)) if bounds is not None and screen_bottom and bounds[1] > screen_bottom * 0.72: continue normalized = _VOLATILE_NUMBER.sub("#", " ".join(label.split())) parts.append( "|".join( ( node.get("class", ""), normalized[:120], node.get("selected", ""), node.get("checked", ""), ) ) ) raw = "\n".join(parts).encode("utf-8", errors="replace") return hashlib.sha256(raw).hexdigest()[:24] def _node_label(node: ET.Element) -> str: values = [] for key in ("text", "content-desc"): value = node.get(key, "").strip() if value and value not in values: values.append(value) return " ".join(values)