diff --git a/client/image_xml/737116531267_home.xml b/client/image_xml/737116531267_home.xml
new file mode 100644
index 0000000..01d874e
--- /dev/null
+++ b/client/image_xml/737116531267_home.xml
@@ -0,0 +1,423 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/client/image_xml/737116531267_size.xml b/client/image_xml/737116531267_size.xml
new file mode 100644
index 0000000..6d16930
--- /dev/null
+++ b/client/image_xml/737116531267_size.xml
@@ -0,0 +1,348 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/client/src/collect_task_service.py b/client/src/collect_task_service.py
new file mode 100644
index 0000000..6987de9
--- /dev/null
+++ b/client/src/collect_task_service.py
@@ -0,0 +1,229 @@
+"""一条采集任务的应用层流程：领取、采集、落库、提交。"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Callable, Optional
+
+from .admin_gateway import (
+ AdminGateway,
+ AdminGatewayError,
+ AndroidDeviceInfo,
+ ClaimCapabilities,
+ ClientInfo,
+)
+from .pdd_collect_service import PddCollectError, PddCollectService
+from .pdd_device_service import PddDeviceService
+from .db import data_dir
+from .task_models import OutboxEventRecord, OutboxEventType, TaskStatus, TaskType
+from .task_models import NewClaimedTask
+from .task_repository import DuplicateTaskError, TaskRepository
+
+
+@dataclass(frozen=True)
+class CollectTaskOutcome:
+ """工作线程返回给界面的简短结果。"""
+
+ kind: str
+ message: str
+ task_id: str = ""
+
+
+CollectServiceFactory = Callable[
+ [str, str, Callable[[], bool]], PddCollectService
+]
+
+
+class CollectTaskService:
+ """一次调用只处理一条本地工作或一个待提交事件。"""
+
+ def __init__(
+ self,
+ gateway: AdminGateway,
+ repository: TaskRepository,
+ client: ClientInfo,
+ device_address: str,
+ *,
+ cancelled: Callable[[], bool] = lambda: False,
+ collect_service_factory: Optional[CollectServiceFactory] = None,
+ ) -> None:
+ self._gateway = gateway
+ self._repository = repository
+ self._client = client
+ self._device_address = device_address
+ self._cancelled = cancelled
+ self._factory = collect_service_factory or self._default_factory
+
+ def execute_one(self) -> CollectTaskOutcome:
+ """先补交已有结果，再执行最早的本地任务，最后才领取新任务。"""
+
+ pending = self._repository.next_pending_outbox()
+ if pending is not None:
+ return self._submit(pending)
+
+ if not self._device_address.strip():
+ raise ValueError("请先在设置页选择并保存 Android 设备")
+
+ task = self._repository.next_collect_task()
+ if task is None:
+ remote = self._gateway.claim_next(
+ self._client,
+ ClaimCapabilities(
+ device=AndroidDeviceInfo(self._device_address),
+ supported_types=(TaskType.COLLECT,),
+ purchase_mode="dry_run",
+ schema_versions=(1,),
+ ),
+ )
+ if remote is None:
+ return CollectTaskOutcome("no_task", "暂无可领取的采集任务")
+ try:
+ self._repository.add_claimed_task(
+ NewClaimedTask(
+ remote_task_id=remote.task_id,
+ task_type=remote.task_type,
+ goods_url=str(remote.payload.get("goods_url") or ""),
+ goods_id=(
+ str(remote.payload["goods_id"])
+ if remote.payload.get("goods_id") is not None
+ else None
+ ),
+ priority=remote.priority,
+ version=remote.version,
+ admin_payload={
+ "id": remote.task_id,
+ "type": remote.task_type.value,
+ "version": remote.version,
+ "priority": remote.priority,
+ "payload": dict(remote.payload),
+ "created_at": remote.created_at,
+ "updated_at": remote.updated_at,
+ },
+ )
+ )
+ except DuplicateTaskError:
+ pass
+ except Exception as exc:
+ raise RuntimeError(
+ f"任务 {remote.task_id} 已领取，但本地保存失败：{exc}"
+ ) from exc
+ task = self._repository.get_task(remote.task_id)
+ if task is None:
+ raise RuntimeError(f"任务 {remote.task_id} 未能保存到本地")
+
+ if self._cancelled():
+ return CollectTaskOutcome("cancelled", "本次采集已取消", task.remote_task_id)
+
+ started = self._repository.start_collect_run(
+ task.remote_task_id, self._device_address
+ )
+ collector = self._factory(
+ self._device_address, self._client.client_id, self._cancelled
+ )
+ try:
+ result = collector.collect(started.task)
+ event = self._repository.save_collect_result(
+ task.remote_task_id, started.attempt_id, result.to_pdd_data()
+ )
+ except PddCollectError as exc:
+ status, retryable = self._classify_error(exc.code)
+ report_code = self._report_error_code(exc.code)
+ event = self._repository.save_collect_failure(
+ task.remote_task_id,
+ started.attempt_id,
+ status,
+ report_code,
+ exc.message,
+ retryable,
+ exc.diagnostics,
+ )
+ return self._submit(event)
+
+ def _submit(self, event: OutboxEventRecord) -> CollectTaskOutcome:
+ task_id = self._repository.outbox_task_id(event.id)
+ self._repository.mark_outbox_sending(event.id)
+ try:
+ if event.event_type is OutboxEventType.TASK_FAILURE:
+ receipt = self._gateway.submit_failure(
+ task_id, event.idempotency_key, event.payload_json
+ )
+ else:
+ receipt = self._gateway.submit_result(
+ task_id, event.idempotency_key, event.payload_json
+ )
+ if not receipt.accepted:
+ raise AdminGatewayError(
+ "ADMIN_RESULT_NOT_ACCEPTED", "Admin 未确认接收结果", False
+ )
+ except AdminGatewayError as exc:
+ message = str(exc)
+ if exc.retryable:
+ self._repository.mark_outbox_retry(event.id, message)
+ return CollectTaskOutcome(
+ "result_pending",
+ f"任务 {task_id} 数据已保存在本地，等待重新提交 Admin：{message}",
+ task_id,
+ )
+ self._repository.mark_outbox_failed(event.id, message)
+ return CollectTaskOutcome(
+ "manual_review",
+ f"任务 {task_id} 提交被 Admin 拒绝：{message}",
+ task_id,
+ )
+
+ self._repository.mark_outbox_sent(event.id)
+ if event.event_type is OutboxEventType.TASK_FAILURE:
+ return CollectTaskOutcome(
+ "failed",
+ f"任务 {task_id} 采集未完成，失败信息已提交 Admin",
+ task_id,
+ )
+ return CollectTaskOutcome(
+ "succeeded", f"任务 {task_id} 采集完成并已提交 Admin", task_id
+ )
+
+ @staticmethod
+ def _classify_error(code: str) -> tuple[TaskStatus, bool]:
+ if code == "PDD_CANCELLED":
+ return TaskStatus.CANCELLED, False
+ if code in {
+ "PDD_PAGE_LOGIN_REQUIRED",
+ "PDD_PAGE_CAPTCHA",
+ "PDD_DATA_SPEC_INCOMPLETE",
+ "PDD_DATA_TITLE_MISSING",
+ "PDD_DATA_PRICE_MISSING",
+ "PDD_DATA_SKU_NAME_TRUNCATED",
+ }:
+ return TaskStatus.MANUAL_REVIEW, False
+ if code.startswith("PDD_DATA_GOODS_"):
+ return TaskStatus.FAILED, False
+ return TaskStatus.RETRY_WAIT, True
+
+ @staticmethod
+ def _report_error_code(code: str) -> str:
+ """把内部解析代码映射为 Admin 可枚举的稳定代码。"""
+
+ return {
+ "PDD_DATA_SKU_NAME_TRUNCATED": "SKU_NAME_TRUNCATED",
+ "PDD_PAGE_SPEC_ENTRY_MISSING": "SKU_PANEL_NOT_FOUND",
+ "PDD_DATA_TITLE_MISSING": "TITLE_TOO_SHORT",
+ "PDD_PAGE_OVERALL_TIMEOUT": "COLLECT_TIMEOUT",
+ "DEVICE_OFFLINE": "DEVICE_DISCONNECTED",
+ "DEVICE_CONNECT_FAILED": "DEVICE_DISCONNECTED",
+ "DEVICE_DISCONNECTED": "DEVICE_DISCONNECTED",
+ }.get(code, code)
+
+ @staticmethod
+ def _default_factory(
+ device_address: str,
+ client_id: str,
+ cancelled: Callable[[], bool],
+ ) -> PddCollectService:
+ return PddCollectService(
+ PddDeviceService(),
+ device_address,
+ client_id,
+ cancelled=cancelled,
+ overall_timeout=600.0,
+ artifact_directory=data_dir() / "artifacts",
+ )
diff --git a/client/src/http_admin_gateway.py b/client/src/http_admin_gateway.py
index 3ba01e4..122d502 100644
--- a/client/src/http_admin_gateway.py
+++ b/client/src/http_admin_gateway.py
@@ -3,20 +3,20 @@
import json
import socket
from http.client import RemoteDisconnected
-from typing import Callable, Mapping, Optional
+from typing import Any, Callable, Mapping, Optional
from urllib.error import HTTPError, URLError
from urllib.parse import urlparse
from urllib.request import ProxyHandler, Request, build_opener
from uuid import uuid4
from .admin_gateway import (
+ AdminGateway,
AdminTask,
AdminGatewayError,
ClaimCapabilities,
ClientInfo,
- ClientRegistrationGateway,
RegistrationReceipt,
- TaskClaimGateway,
+ SubmissionReceipt,
)
from .task_models import TaskType
@@ -24,8 +24,8 @@ from .task_models import TaskType
DEFAULT_ADMIN_BASE_URL = "http://127.0.0.1:8080"
-class HttpAdminGateway(ClientRegistrationGateway, TaskClaimGateway):
- """通过 HTTP 登记 Client 和领取采集任务；令牌只保存在内存。"""
+class HttpAdminGateway(AdminGateway):
+ """通过 HTTP 登记、领取和提交任务；令牌只保存在内存。"""
def __init__(
self,
@@ -33,6 +33,7 @@ class HttpAdminGateway(ClientRegistrationGateway, TaskClaimGateway):
token: str = "",
timeout_seconds: float = 3.0,
opener: Optional[Callable] = None,
+ client_id: str = "",
):
normalized_url = base_url.strip().rstrip("/")
parsed = urlparse(normalized_url)
@@ -44,6 +45,7 @@ class HttpAdminGateway(ClientRegistrationGateway, TaskClaimGateway):
self._base_url = normalized_url
self._token = token.strip()
self._timeout_seconds = timeout_seconds
+ self._client_id = client_id.strip()
# Admin 通常运行在本机或局域网。明确禁用环境代理，避免
# HTTP_PROXY 把 127.0.0.1 请求错误转发到代理服务器。
self._opener = opener or build_opener(ProxyHandler({})).open
@@ -54,6 +56,7 @@ class HttpAdminGateway(ClientRegistrationGateway, TaskClaimGateway):
"""调用独立登记接口，不领取或修改任务。"""
request_id = str(uuid4())
+ self._client_id = client.client_id.strip()
payload = {
"client": {"name": client.name.strip()},
"supported_types": [
@@ -146,6 +149,7 @@ class HttpAdminGateway(ClientRegistrationGateway, TaskClaimGateway):
"""领取一个采集任务；Admin 返回 204 时返回 ``None``。"""
request_id = str(uuid4())
+ self._client_id = client.client_id.strip()
payload = {
"client": {"name": client.name.strip()},
# #30 只允许领取采集任务。采购能力必须由安全门禁工单开启。
@@ -219,6 +223,104 @@ class HttpAdminGateway(ClientRegistrationGateway, TaskClaimGateway):
)
return self._parse_claim_response(body, request_id)
+ def submit_result(
+ self,
+ task_id: str,
+ idempotency_key: str,
+ result: Mapping[str, Any],
+ ) -> SubmissionReceipt:
+ """幂等提交成功结果。"""
+
+ return self._submit(task_id, idempotency_key, result, "result")
+
+ def submit_failure(
+ self,
+ task_id: str,
+ idempotency_key: str,
+ failure: Mapping[str, Any],
+ ) -> SubmissionReceipt:
+ """幂等提交失败或人工处理结果。"""
+
+ return self._submit(task_id, idempotency_key, failure, "failure")
+
+ def _submit(
+ self,
+ task_id: str,
+ idempotency_key: str,
+ payload: Mapping[str, Any],
+ endpoint: str,
+ ) -> SubmissionReceipt:
+ if not task_id.strip():
+ raise ValueError("task_id 不能为空")
+ if not idempotency_key.strip():
+ raise ValueError("Idempotency-Key 不能为空")
+ if not self._client_id:
+ raise AdminGatewayError(
+ "CLIENT_ID_MISSING",
+ "提交结果前必须先配置 Client 设备号",
+ False,
+ )
+ request_id = str(uuid4())
+ headers = {
+ "Content-Type": "application/json; charset=utf-8",
+ "Accept": "application/json",
+ "Idempotency-Key": idempotency_key.strip(),
+ "X-Request-Id": request_id,
+ "X-Client-Id": self._client_id,
+ }
+ if self._token:
+ headers["Authorization"] = f"Bearer {self._token}"
+ request = Request(
+ f"{self._base_url}/api/v1/client/tasks/{task_id.strip()}/{endpoint}",
+ data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
+ headers=headers,
+ method="POST",
+ )
+ try:
+ with self._opener(request, timeout=self._timeout_seconds) as response:
+ status = getattr(response, "status", None)
+ if status is None:
+ status = response.getcode()
+ body = response.read()
+ except HTTPError as exc:
+ self._raise_http_error(exc, request_id, "提交任务")
+ except (
+ URLError,
+ RemoteDisconnected,
+ ConnectionError,
+ socket.timeout,
+ TimeoutError,
+ ) as exc:
+ reason = getattr(exc, "reason", exc)
+ code = "ADMIN_TIMEOUT" if isinstance(reason, (socket.timeout, TimeoutError)) else "ADMIN_UNAVAILABLE"
+ message = "Admin 提交请求超时，请稍后重试" if code == "ADMIN_TIMEOUT" else "无法连接 Admin，结果已保存在本地"
+ raise AdminGatewayError(code, message, True, request_id) from exc
+ if status not in (200, 201):
+ raise AdminGatewayError(
+ "ADMIN_UNEXPECTED_RESPONSE",
+ f"Admin 返回了未预期的状态码 {status}",
+ status >= 500,
+ request_id,
+ )
+ data = self._decode_json(body, request_id)
+ accepted = data.get("accepted")
+ result_id = data.get("result_id")
+ accepted_at = data.get("accepted_at")
+ if (
+ accepted is not True
+ or not isinstance(result_id, str)
+ or not result_id.strip()
+ or not isinstance(accepted_at, str)
+ or not accepted_at.strip()
+ ):
+ raise AdminGatewayError(
+ "ADMIN_INVALID_RESPONSE",
+ "Admin 提交响应字段不完整",
+ False,
+ request_id,
+ )
+ return SubmissionReceipt(True, result_id, accepted_at)
+
@classmethod
def _parse_claim_response(
cls,
diff --git a/client/src/pdd_collect_service.py b/client/src/pdd_collect_service.py
index 3123e4e..35ffecd 100644
--- a/client/src/pdd_collect_service.py
+++ b/client/src/pdd_collect_service.py
@@ -14,6 +14,7 @@ import xml.etree.ElementTree as ET
from dataclasses import dataclass
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
+from pathlib import Path
from typing import Any, Callable, Iterable, Mapping, Optional, Sequence
from urllib.parse import parse_qs, urlparse
@@ -31,13 +32,32 @@ _CAPTCHA_MARKERS = ("请完成验证", "拖动滑块", "安全验证", "点击
_READY_MARKERS = ("发起拼单", "立即购买", "单独购买", "免拼购买", "快要抢光")
+def _is_device_disconnect(error: BaseException) -> bool:
+ details = str(error).lower()
+ return any(
+ marker in details
+ for marker in (
+ "device not found",
+ "device offline",
+ "disconnected",
+ "closed transport",
+ )
+ )
+
+
class PddCollectError(RuntimeError):
"""采集失败，并携带稳定错误码。"""
- def __init__(self, code: str, message: str) -> None:
+ def __init__(
+ self,
+ code: str,
+ message: str,
+ diagnostics: Optional[Mapping[str, Any]] = None,
+ ) -> None:
super().__init__(message)
self.code = code
self.message = message
+ self.diagnostics = dict(diagnostics or {})
@dataclass(frozen=True)
@@ -85,11 +105,15 @@ class SkuResult:
price_cent: Optional[int]
available: bool
raw_price: Optional[str]
+ price_observed_at: Mapping[str, str]
+ list_price_cent: Optional[int] = None
def to_dict(self) -> dict[str, Any]:
return {
"options": dict(self.options),
"price_cent": self.price_cent,
+ "list_price_cent": self.list_price_cent,
+ "price_observed_at": dict(self.price_observed_at),
"currency": "CNY",
"available": self.available,
"raw_price": self.raw_price,
@@ -110,6 +134,7 @@ class SpecSnapshot:
selected_text: Optional[str]
price_cent: Optional[int]
raw_price: Optional[str]
+ list_price_cent: Optional[int]
@dataclass(frozen=True)
@@ -127,16 +152,16 @@ class CollectResult:
captured_at: str
client_id: str
device_address: str
+ artifacts: tuple[Mapping[str, Any], ...] = ()
def to_pdd_data(self) -> dict[str, Any]:
return {
"schema_version": 1,
- "goods": {
- "goods_id": self.goods_id,
- "url": self.goods_url,
- "title": self.title,
- },
- "shop": {"name": self.shop_name},
+ "goods_id": self.goods_id,
+ "goods_url": self.goods_url,
+ "title": self.title,
+ "shop_name": self.shop_name,
+ "price_granularity": "color",
"metrics": {
"sales": self.sales.to_dict(),
"reviews": self.reviews.to_dict(),
@@ -150,7 +175,7 @@ class CollectResult:
"device_address": self.device_address,
"pdd_package": PDD_PACKAGE_NAME,
},
- "artifacts": [],
+ "artifacts": [dict(item) for item in self.artifacts],
}
@@ -182,6 +207,14 @@ def _node_label(node: ET.Element) -> str:
)
+def _preferred_node_label(node: ET.Element) -> str:
+ """规格名被 text 截断时，优先采用完整的无障碍描述。"""
+
+ text = (node.get("text") or "").strip()
+ description = (node.get("content-desc") or "").strip()
+ return description or text
+
+
def _own_or_descendant_label(node: ET.Element) -> str:
own = _node_label(node)
if own:
@@ -193,6 +226,17 @@ def _own_or_descendant_label(node: ET.Element) -> str:
return ""
+def _preferred_or_descendant_label(node: ET.Element) -> str:
+ own = _preferred_node_label(node)
+ if own:
+ return own
+ for child in node.iter("node"):
+ label = _preferred_node_label(child)
+ if label and label != "打开大图":
+ return label
+ return ""
+
+
def _all_labels(root: ET.Element) -> list[str]:
return [label for node in root.iter("node") if (label := _node_label(node))]
@@ -250,9 +294,21 @@ def parse_goods_page(xml_data: str | bytes) -> GoodsSnapshot:
title = candidate
break
if title is None:
+ line_parts: dict[int, list[tuple[int, str]]] = {}
+ for node in root.iter("node"):
+ label = _preferred_node_label(node)
+ bounds = _parse_bounds(node.get("bounds", ""))
+ if not label or bounds is None:
+ continue
+ line_key = bounds[1] // 24
+ line_parts.setdefault(line_key, []).append((bounds[0], label))
+ joined_lines = [
+ "".join(text for _, text in sorted(parts))
+ for parts in line_parts.values()
+ ]
candidates = [
label
- for label in labels
+ for label in labels + joined_lines
if len(label) >= 12
and not any(word in label for word in ("通知", "支付", "已拼", "评价"))
]
@@ -347,7 +403,7 @@ def _top_level_clickable_options(
def _price_from_nodes(
nodes: Iterable[ET.Element],
spec_top: int,
-) -> tuple[Optional[int], Optional[str]]:
+) -> tuple[Optional[int], Optional[str], Optional[int]]:
candidates: list[tuple[int, int, str]] = []
for node in nodes:
label = _node_label(node)
@@ -366,9 +422,12 @@ def _price_from_nodes(
continue
candidates.append((bounds[1], cents, match.group(0).replace(" ", "")))
if not candidates:
- return None, None
- _, cents, raw = min(candidates, key=lambda item: (item[0], item[1]))
- return cents, raw
+ return None, None, None
+ first_top = min(item[0] for item in candidates)
+ first_row = [item for item in candidates if abs(item[0] - first_top) <= 12]
+ _, cents, raw = min(first_row, key=lambda item: item[1])
+ list_prices = [item[1] for item in first_row if item[1] > cents]
+ return cents, raw, max(list_prices, default=None)
def parse_spec_panel(xml_data: str | bytes) -> SpecSnapshot:
@@ -446,7 +505,12 @@ def parse_spec_panel(xml_data: str | bytes) -> SpecSnapshot:
values: list[DimensionValue] = []
seen: set[str] = set()
for node in option_nodes:
- text = _own_or_descendant_label(node).strip()
+ text = _preferred_or_descendant_label(node).strip()
+ if text.endswith(("…", "...")):
+ raise PddCollectError(
+ "PDD_DATA_SKU_NAME_TRUNCATED",
+ f"规格名称被截断，无法安全采集：{text}",
+ )
if not text or text in seen:
continue
seen.add(text)
@@ -458,8 +522,12 @@ def parse_spec_panel(xml_data: str | bytes) -> SpecSnapshot:
dimensions.append(SpecDimension(key, name, tuple(values)))
selected_text = next((label for label in labels if label.startswith("已选")), None)
- price_cent, raw_price = _price_from_nodes(root.iter("node"), outer_bounds[1])
- return SpecSnapshot(tuple(dimensions), selected_text, price_cent, raw_price)
+ price_cent, raw_price, list_price_cent = _price_from_nodes(
+ root.iter("node"), outer_bounds[1]
+ )
+ return SpecSnapshot(
+ tuple(dimensions), selected_text, price_cent, raw_price, list_price_cent
+ )
def _ancestors(
@@ -561,9 +629,11 @@ class PddCollectService:
now: Callable[[], datetime] = lambda: datetime.now(timezone.utc),
cancelled: Callable[[], bool] = lambda: False,
page_timeout: float = 30.0,
+ overall_timeout: float = 600.0,
max_page_swipes: int = 12,
max_spec_swipes: int = 12,
max_sku_count: int = 200,
+ artifact_directory: Optional[Path] = None,
) -> None:
self._device_service = device_service
self._device_address = device_address
@@ -573,9 +643,15 @@ class PddCollectService:
self._now = now
self._cancelled = cancelled
self._page_timeout = page_timeout
+ self._overall_timeout = overall_timeout
+ self._overall_deadline: Optional[float] = None
self._max_page_swipes = max_page_swipes
self._max_spec_swipes = max_spec_swipes
self._max_sku_count = max_sku_count
+ self._artifact_directory = artifact_directory
+ self._last_goods_xml: Optional[str] = None
+ self._goods_screens_checked = 0
+ self._artifacts: list[Mapping[str, Any]] = []
def collect(self, task: Any) -> CollectResult:
"""执行采集；``task`` 至少提供 ``goods_url`` 和可选 ``goods_id``。"""
@@ -585,6 +661,7 @@ class PddCollectService:
raise PddCollectError("PDD_DATA_GOODS_URL_MISSING", "采集任务缺少商品链接")
goods_id = str(getattr(task, "goods_id", "") or "").strip()
goods_id = _validate_goods_url(goods_url, goods_id)
+ self._overall_deadline = self._monotonic() + self._overall_timeout
self._check_cancelled()
try:
@@ -593,12 +670,14 @@ class PddCollectService:
goods = self._collect_goods_details(device)
if not goods.title:
raise PddCollectError("PDD_DATA_TITLE_MISSING", "商品页没有可识别的标题")
- if not goods.shop_name:
- raise PddCollectError("PDD_DATA_SHOP_MISSING", "商品页没有采集到店铺名称")
if not goods.sales.raw:
raise PddCollectError("PDD_DATA_SALES_MISSING", "商品页没有采集到已拼数量")
if not goods.reviews.raw:
raise PddCollectError("PDD_DATA_REVIEWS_MISSING", "商品页没有采集到评价数量")
+ if not goods.shop_name and self._last_goods_xml:
+ artifact = self._save_xml("shop-not-found", self._last_goods_xml)
+ if artifact:
+ self._artifacts.append(artifact)
home_xml = device.dump_hierarchy()
coordinate = get_size_panel_coord(home_xml)
@@ -623,11 +702,23 @@ class PddCollectService:
)
if not skus or not has_available_price:
raise PddCollectError("PDD_DATA_PRICE_MISSING", "没有采集到可用 SKU 的价格")
- except PddCollectError:
+ except PddCollectError as exc:
+ if not exc.diagnostics and self._last_goods_xml:
+ artifact = self._save_xml("collect-failed", self._last_goods_xml)
+ if artifact:
+ exc.diagnostics = {
+ "artifacts": [artifact],
+ "goods_screens_checked": self._goods_screens_checked,
+ }
raise
except PddDeviceError as exc:
raise PddCollectError(exc.code, exc.message) from exc
except Exception as exc:
+ if _is_device_disconnect(exc):
+ raise PddCollectError(
+ "DEVICE_DISCONNECTED",
+ f"Android 设备在采集过程中断开：{exc}",
+ ) from exc
raise PddCollectError("PDD_PAGE_UNKNOWN", f"PDD 采集过程中发生未知错误：{exc}") from exc
return CollectResult(
@@ -642,11 +733,17 @@ class PddCollectService:
captured_at=self._now().astimezone(timezone.utc).isoformat().replace("+00:00", "Z"),
client_id=self._client_id,
device_address=self._device_address,
+ artifacts=tuple(self._artifacts),
)
def _check_cancelled(self) -> None:
if self._cancelled():
raise PddCollectError("PDD_CANCELLED", "采集任务已安全取消")
+ if (
+ self._overall_deadline is not None
+ and self._monotonic() >= self._overall_deadline
+ ):
+ raise PddCollectError("PDD_PAGE_OVERALL_TIMEOUT", "PDD 采集超过 10 分钟")
def _open_goods(self, device: Any, goods_url: str) -> None:
try:
@@ -659,6 +756,11 @@ class PddCollectService:
except PddCollectError:
raise
except Exception as exc:
+ if _is_device_disconnect(exc):
+ raise PddCollectError(
+ "DEVICE_DISCONNECTED",
+ f"Android 设备在打开商品页时断开：{exc}",
+ ) from exc
raise PddCollectError("DEVICE_APP_START_FAILED", f"无法打开 PDD 商品链接：{exc}") from exc
deadline = self._monotonic() + self._page_timeout
@@ -686,13 +788,14 @@ class PddCollectService:
for _ in range(self._max_page_swipes + 1):
self._check_cancelled()
xml_data = device.dump_hierarchy()
+ self._last_goods_xml = str(xml_data)
+ self._goods_screens_checked += 1
root = _parse_xml(xml_data)
labels = tuple(_all_labels(root))
snapshots.append(parse_goods_page(xml_data))
combined = _combine_goods_snapshots(snapshots)
is_complete = (
combined.title
- and combined.shop_name
and combined.sales.raw
and combined.reviews.raw
)
@@ -720,6 +823,27 @@ class PddCollectService:
self._sleep(0.35)
return _combine_goods_snapshots(snapshots)
+ def _save_xml(self, label: str, xml_data: str) -> Optional[Mapping[str, Any]]:
+ """保存本地诊断 XML；测试未提供目录时不写文件。"""
+
+ if self._artifact_directory is None:
+ return None
+ try:
+ digest = hashlib.sha256(xml_data.encode("utf-8")).hexdigest()
+ directory = self._artifact_directory / self._client_id
+ directory.mkdir(parents=True, exist_ok=True)
+ path = directory / f"{label}-{digest[:12]}.xml"
+ if not path.exists():
+ path.write_text(xml_data, encoding="utf-8")
+ return {
+ "kind": "accessibility_xml",
+ "path": str(path.resolve()),
+ "sha256": digest,
+ "screens_checked": self._goods_screens_checked,
+ }
+ except OSError:
+ return None
+
def _discover_dimensions(self, device: Any) -> list[SpecSnapshot]:
snapshots: list[SpecSnapshot] = []
@@ -768,28 +892,66 @@ class PddCollectService:
f"规格组合共 {len(combinations)} 个，超过安全上限 {self._max_sku_count}",
)
- results: list[SkuResult] = []
- for values in combinations:
+ color_index = next(
+ (index for index, item in enumerate(dimensions) if item.key == "color"),
+ None,
+ )
+ if color_index is None:
+ raise PddCollectError(
+ "PDD_DATA_SPEC_INCOMPLETE", "规格面板没有可识别的颜色分类"
+ )
+
+ samples: dict[
+ str,
+ tuple[Optional[int], Optional[str], Optional[int], dict[str, str]],
+ ] = {}
+ color_dimension = dimensions[color_index]
+ for color in color_dimension.values:
self._check_cancelled()
- options = {dimension.key: value.text for dimension, value in zip(dimensions, values)}
- pairs = list(zip(dimensions, values))
- selected = self._select_combination(device, pairs)
- if not selected and len(pairs) > 1:
- # 某个选项可能只是在当前搭配下禁用。反向选择一次，可以先改变
- # 依赖维度，再重新判断目标组合是否真的缺货。
- selected = self._select_combination(device, list(reversed(pairs)))
+ if not color.available:
+ samples[color.text] = (None, None, None, {"color": color.text})
+ continue
+ selected = self._select_option(device, color.text, color_dimension.key)
if not selected:
- results.append(SkuResult(options, None, False, None))
+ samples[color.text] = (None, None, None, {"color": color.text})
continue
snapshot = parse_spec_panel(device.dump_hierarchy())
summary = snapshot.selected_text or ""
- confirmed = all(value.text in summary for value in values)
+ observed: dict[str, str] = {}
+ for dimension in dimensions:
+ match = next(
+ (value.text for value in dimension.values if value.text in summary),
+ None,
+ )
+ if match:
+ observed[dimension.key] = match
+ confirmed = (
+ observed.get(color_dimension.key) == color.text
+ and len(observed) == len(dimensions)
+ )
+ samples[color.text] = (
+ snapshot.price_cent if confirmed else None,
+ snapshot.raw_price if confirmed else None,
+ snapshot.list_price_cent if confirmed else None,
+ observed,
+ )
+
+ results: list[SkuResult] = []
+ for values in combinations:
+ options = {
+ dimension.key: value.text
+ for dimension, value in zip(dimensions, values)
+ }
+ price, raw_price, list_price, observed = samples[values[color_index].text]
+ available = all(value.available for value in values) and price is not None
results.append(
SkuResult(
options,
- snapshot.price_cent if confirmed else None,
- confirmed and snapshot.price_cent is not None,
- snapshot.raw_price if confirmed else None,
+ price if available else None,
+ available,
+ raw_price if available else None,
+ observed,
+ list_price if available else None,
)
)
return tuple(results)
@@ -864,7 +1026,7 @@ class PddCollectService:
]
screen_right = max(right_edges, default=0)
for node in root.iter("node"):
- if _own_or_descendant_label(node).strip() != target:
+ if _preferred_or_descendant_label(node).strip() != target:
continue
bounds = _parse_bounds(node.get("bounds", ""))
if node.get("clickable") != "true" or bounds is None:
diff --git a/client/src/pdd_ui_event.py b/client/src/pdd_ui_event.py
index 1c31606..1899658 100644
--- a/client/src/pdd_ui_event.py
+++ b/client/src/pdd_ui_event.py
@@ -13,7 +13,7 @@
- 窗口关闭时要断开信号并置标志位，否则迟到的后台结果会访问
已经销毁的控件、直接崩溃。做法见同文档 §5.2。
- 数据库读写走 Repository，**不要在这里拼业务 SQL**。
-- “获取任务”会真的去操作手机、可能下单；“搜索”只读本地数据库。
+- “获取任务”会真的去操作手机采集商品；“搜索”只读本地数据库。
两者必须分开，不得共用入口。
- 普通成功不弹窗，更新界面即可；可恢复错误用 `InfoBar`
（模板见 `docs/client/05-ui-specification.md` §9.1）；
@@ -28,11 +28,10 @@ from qfluentwidgets import InfoBar, InfoBarPosition
from .admin_gateway import (
AdminGatewayError,
AdminTask,
- AndroidDeviceInfo,
- ClaimCapabilities,
ClientInfo,
- TaskClaimGateway,
+ AdminGateway,
)
+from .collect_task_service import CollectServiceFactory, CollectTaskService
from .current_client_service import CurrentClientService
from .http_admin_gateway import DEFAULT_ADMIN_BASE_URL, HttpAdminGateway
from .pdd_ui import PDDTaskPage, TaskRow
@@ -45,7 +44,7 @@ from .task_models import (
TaskSummary,
TaskType,
)
-from .task_repository import DuplicateTaskError, TaskRepository
+from .task_repository import TaskRepository
TASK_TYPE_BY_TEXT = {
@@ -116,21 +115,23 @@ def admin_task_to_new_claimed_task(task: AdminTask) -> NewClaimedTask:
class ClaimTaskWorker(QObject):
- """在后台领取至多一个采集任务，并先写入本地数据库。"""
+ """在后台补交或执行至多一条采集任务。"""
noTask = pyqtSignal()
taskSaved = pyqtSignal(str)
duplicateTask = pyqtSignal(str)
localSaveFailed = pyqtSignal(str, str)
failed = pyqtSignal(str)
+ outcome = pyqtSignal(str, str, str)
completed = pyqtSignal()
def __init__(
self,
- gateway: TaskClaimGateway,
+ gateway: AdminGateway,
task_repository: TaskRepository,
client_service: CurrentClientService,
android_device_service: SelectedAndroidDeviceService,
+ collect_service_factory: Optional[CollectServiceFactory] = None,
) -> None:
super().__init__()
self._gateway = gateway
@@ -138,6 +139,7 @@ class ClaimTaskWorker(QObject):
self._client_service = client_service
self._android_device_service = android_device_service
self._cancelled = False
+ self._collect_service_factory = collect_service_factory
def cancel(self) -> None:
"""阻止尚未开始的领取；已领取的任务仍必须保存到本地。"""
@@ -146,7 +148,6 @@ class ClaimTaskWorker(QObject):
@pyqtSlot()
def run(self) -> None:
- claimed_task: Optional[AdminTask] = None
try:
if self._cancelled:
return
@@ -155,42 +156,20 @@ class ClaimTaskWorker(QObject):
self.failed.emit("请先在设置页保存当前设备号和设备名")
return
android_serial = self._android_device_service.load()
- if not android_serial:
- self.failed.emit("请先在设置页选择并保存 Android 设备")
- return
- capabilities = ClaimCapabilities(
- device=AndroidDeviceInfo(android_serial),
- supported_types=(TaskType.COLLECT,),
- purchase_mode="dry_run",
- schema_versions=(1,),
- )
- claimed_task = self._gateway.claim_next(
+ result = CollectTaskService(
+ self._gateway,
+ self._task_repository,
ClientInfo(
client_settings.client_id,
client_settings.client_name,
),
- capabilities,
- )
- if claimed_task is None:
- if not self._cancelled:
- self.noTask.emit()
- return
-
- local_task = admin_task_to_new_claimed_task(claimed_task)
- try:
- self._task_repository.add_claimed_task(local_task)
- except DuplicateTaskError:
- if not self._cancelled:
- self.duplicateTask.emit(claimed_task.task_id)
- return
- except Exception as exc:
- if not self._cancelled:
- self.localSaveFailed.emit(claimed_task.task_id, str(exc))
- return
-
- if not self._cancelled:
- self.taskSaved.emit(claimed_task.task_id)
+ android_serial or "",
+ cancelled=lambda: self._cancelled,
+ collect_service_factory=self._collect_service_factory,
+ ).execute_one()
+ if not self._cancelled or result.kind == "cancelled":
+ self.outcome.emit(result.kind, result.message, result.task_id)
except AdminGatewayError as exc:
if not self._cancelled:
request_hint = (
@@ -199,10 +178,7 @@ class ClaimTaskWorker(QObject):
self.failed.emit(f"{exc}{request_hint}")
except Exception as exc:
if not self._cancelled:
- if claimed_task is not None:
- self.localSaveFailed.emit(claimed_task.task_id, str(exc))
- else:
- self.failed.emit(f"领取任务失败：{exc}")
+ self.failed.emit(f"执行采集任务失败：{exc}")
finally:
self.completed.emit()
@@ -215,8 +191,9 @@ class PDDTaskPageEvent(QObject):
page: PDDTaskPage,
repository: Optional[TaskRepository] = None,
parent=None,
- claim_gateway: Optional[TaskClaimGateway] = None,
+ claim_gateway: Optional[AdminGateway] = None,
settings_repository: Optional[SettingsRepository] = None,
+ collect_service_factory: Optional[CollectServiceFactory] = None,
):
super().__init__(parent or page)
self._page = page
@@ -226,6 +203,13 @@ class PDDTaskPageEvent(QObject):
self._claim_busy = False
self._claim_thread: Optional[QThread] = None
self._claim_worker: Optional[ClaimTaskWorker] = None
+ self._collect_service_factory = collect_service_factory
+
+ try:
+ self._repository.recover_interrupted_work()
+ except AttributeError:
+ # 测试用的只读 Repository 可以不实现恢复接口。
+ pass
settings = settings_repository or SettingsRepository()
self._client_service = CurrentClientService(settings)
@@ -242,6 +226,7 @@ class PDDTaskPageEvent(QObject):
self._claim_gateway = HttpAdminGateway(
base_url if isinstance(base_url, str) else "",
timeout_seconds=timeout_seconds,
+ client_id=self._client_service.load().client_id,
)
except (TypeError, ValueError) as exc:
self._claim_gateway_error = str(exc)
@@ -290,7 +275,7 @@ class PDDTaskPageEvent(QObject):
self._claim_busy = True
self._page.autoFetchButton.setEnabled(False)
- self._page.set_engine_status("正在领取一个采集任务，请稍候…")
+ self._page.set_engine_status("正在处理一条采集任务，请稍候…")
thread = QThread(self)
worker = ClaimTaskWorker(
@@ -298,6 +283,7 @@ class PDDTaskPageEvent(QObject):
self._repository,
self._client_service,
self._selected_android_device_service,
+ self._collect_service_factory,
)
worker.moveToThread(thread)
thread.started.connect(worker.run)
@@ -306,6 +292,7 @@ class PDDTaskPageEvent(QObject):
worker.duplicateTask.connect(self._on_duplicate_claimed_task)
worker.localSaveFailed.connect(self._on_claimed_task_save_failed)
worker.failed.connect(self._on_claim_failed)
+ worker.outcome.connect(self._on_collect_outcome)
worker.completed.connect(thread.quit)
worker.completed.connect(worker.deleteLater)
thread.finished.connect(thread.deleteLater)
@@ -352,6 +339,15 @@ class PDDTaskPageEvent(QObject):
self._page.set_engine_status(content)
self._show_claim_error("领取任务失败", content)
+ @pyqtSlot(str, str, str)
+ def _on_collect_outcome(self, kind: str, message: str, _task_id: str) -> None:
+ if self._closing:
+ return
+ self._page.set_engine_status(message)
+ self._reload()
+ if kind in {"result_pending", "manual_review", "failed"}:
+ self._show_claim_error("采集任务需要处理", message)
+
def _show_claim_error(self, title: str, content: str) -> None:
"""显示不会自动消失的可恢复错误，同时保留底部状态文字。"""
@@ -413,6 +409,7 @@ class PDDTaskPageEvent(QObject):
self._on_claimed_task_save_failed,
),
(worker.failed, self._on_claim_failed),
+ (worker.outcome, self._on_collect_outcome),
)
except RuntimeError:
signal_slots = ()
@@ -424,7 +421,10 @@ class PDDTaskPageEvent(QObject):
if thread is not None and thread.isRunning():
thread.quit()
- thread.wait(10_000)
+ # uiautomator2/ADB 的单次调用可能需要数秒才返回。先通过
+ # cancelled 标志让采集在下一个安全点退出，再等待工作线程收尾，
+ # 避免窗口销毁时出现 "QThread destroyed while running"。
+ thread.wait(60_000)
def summary_to_row(summary: TaskSummary) -> TaskRow:
diff --git a/client/src/task_models.py b/client/src/task_models.py
index f7ed055..ab55768 100644
--- a/client/src/task_models.py
+++ b/client/src/task_models.py
@@ -184,6 +184,15 @@ class OutboxEventRecord:
sent_at: Optional[str]
+@dataclass(frozen=True)
+class StartedTaskRun:
+ """已进入执行状态的一次采集尝试。"""
+
+ task: TaskDetail
+ attempt_id: str
+ attempt_no: int
+
+
@dataclass(frozen=True)
class AppSettingRecord:
"""一条非敏感应用设置。"""
diff --git a/client/src/task_repository.py b/client/src/task_repository.py
index 6c469db..c66ac6b 100644
--- a/client/src/task_repository.py
+++ b/client/src/task_repository.py
@@ -8,10 +8,16 @@ import sqlite3
from datetime import datetime, timezone
from pathlib import Path
from typing import Dict, List, Optional, Tuple, Union
+from uuid import uuid4
from .db import initialize_database, open_database
from .task_models import (
NewClaimedTask,
+ OutboxEventRecord,
+ OutboxEventType,
+ OutboxStatus,
+ RunStatus,
+ StartedTaskRun,
TaskDetail,
TaskFilters,
TaskStatus,
@@ -145,6 +151,353 @@ class TaskRepository:
connection.close()
return self._to_detail(row) if row is not None else None
+ def recover_interrupted_work(self) -> None:
+ """恢复上次异常退出留下的可重试状态。"""
+
+ now = utc_now_iso()
+ connection = open_database(self._db_path)
+ try:
+ with connection:
+ connection.execute(
+ "UPDATE outbox_events SET status = 'pending', updated_at = ?"
+ " WHERE status = 'sending'",
+ (now,),
+ )
+ connection.execute(
+ "UPDATE pdd_tasks SET status = 'retry_wait',"
+ " current_step = 'interrupted', retry_count = retry_count + 1,"
+ " last_error_code = 'CLIENT_INTERRUPTED',"
+ " last_error_message = '客户端上次执行期间退出', updated_at = ?"
+ " WHERE status = 'running' AND task_type = 'collect'",
+ (now,),
+ )
+ connection.execute(
+ "UPDATE task_runs SET run_status = 'failed',"
+ " error_code = 'CLIENT_INTERRUPTED',"
+ " error_message = '客户端上次执行期间退出',"
+ " finished_at = ?, updated_at = ?"
+ " WHERE run_status = 'running' AND irreversible_action_at IS NULL"
+ " AND task_id IN (SELECT id FROM pdd_tasks WHERE task_type = 'collect')",
+ (now, now),
+ )
+ finally:
+ connection.close()
+
+ def next_collect_task(self) -> Optional[TaskDetail]:
+ """返回最早的本地待执行采集任务。"""
+
+ connection = open_database(self._db_path)
+ try:
+ row = connection.execute(
+ "SELECT * FROM pdd_tasks"
+ " WHERE task_type = 'collect' AND status IN ('claimed', 'retry_wait')"
+ " ORDER BY received_at ASC, id ASC LIMIT 1"
+ ).fetchone()
+ finally:
+ connection.close()
+ return self._to_detail(row) if row is not None else None
+
+ def start_collect_run(
+ self, remote_task_id: str, device_address: str
+ ) -> StartedTaskRun:
+ """原子地把待执行任务改为执行中，并创建一次运行记录。"""
+
+ now = utc_now_iso()
+ attempt_id = str(uuid4())
+ connection = open_database(self._db_path)
+ try:
+ with connection:
+ row = connection.execute(
+ "SELECT * FROM pdd_tasks WHERE remote_task_id = ?",
+ (remote_task_id,),
+ ).fetchone()
+ if row is None:
+ raise ValueError(f"任务 {remote_task_id} 不存在")
+ if row["task_type"] != TaskType.COLLECT.value:
+ raise ValueError("当前只能执行采集任务")
+ if row["status"] not in {
+ TaskStatus.CLAIMED.value,
+ TaskStatus.RETRY_WAIT.value,
+ }:
+ raise ValueError(f"任务状态 {row['status']} 不能开始采集")
+ attempt_no = int(
+ connection.execute(
+ "SELECT COALESCE(MAX(attempt_no), 0) + 1"
+ " FROM task_runs WHERE task_id = ?",
+ (row["id"],),
+ ).fetchone()[0]
+ )
+ connection.execute(
+ "UPDATE pdd_tasks SET status = 'running',"
+ " current_step = 'collecting', started_at = COALESCE(started_at, ?),"
+ " last_error_code = NULL, last_error_message = NULL, updated_at = ?"
+ " WHERE id = ?",
+ (now, now, row["id"]),
+ )
+ connection.execute(
+ "INSERT INTO task_runs (task_id, attempt_id, attempt_no,"
+ " device_address, run_status, current_step, started_at,"
+ " created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+ (
+ row["id"], attempt_id, attempt_no, device_address,
+ RunStatus.RUNNING.value, "collecting", now, now, now,
+ ),
+ )
+ task = self.get_task(remote_task_id)
+ assert task is not None
+ return StartedTaskRun(task, attempt_id, attempt_no)
+ finally:
+ connection.close()
+
+ def save_collect_result(
+ self,
+ remote_task_id: str,
+ attempt_id: str,
+ pdd_data: Dict[str, object],
+ ) -> OutboxEventRecord:
+ """在一个事务中保存采集结果并创建待提交事件。"""
+
+ now = utc_now_iso()
+ connection = open_database(self._db_path)
+ try:
+ with connection:
+ task = connection.execute(
+ "SELECT id, version FROM pdd_tasks WHERE remote_task_id = ?",
+ (remote_task_id,),
+ ).fetchone()
+ if task is None:
+ raise ValueError(f"任务 {remote_task_id} 不存在")
+ payload = {
+ "task_version": task["version"],
+ "attempt_id": attempt_id,
+ "result_type": "collect",
+ "completed_at": now,
+ "pdd_data": pdd_data,
+ }
+ idempotency_key = f"{remote_task_id}:{attempt_id}:result-v1"
+ connection.execute(
+ "UPDATE pdd_tasks SET status = 'result_pending',"
+ " current_step = 'submit_result', pdd_data = ?, goods_id = ?,"
+ " title = ?, price_cent = ?, finished_at = ?, updated_at = ?"
+ " WHERE id = ?",
+ (
+ json.dumps(pdd_data, ensure_ascii=False),
+ pdd_data.get("goods_id"), pdd_data.get("title"),
+ self._summary_price(pdd_data), now, now, task["id"],
+ ),
+ )
+ connection.execute(
+ "UPDATE task_runs SET run_status = 'succeeded',"
+ " current_step = 'submit_result', finished_at = ?, updated_at = ?"
+ " WHERE attempt_id = ?",
+ (now, now, attempt_id),
+ )
+ cursor = connection.execute(
+ "INSERT INTO outbox_events (task_id, event_type, idempotency_key,"
+ " payload_json, status, created_at, updated_at)"
+ " VALUES (?, 'collect_result', ?, ?, 'pending', ?, ?)",
+ (
+ task["id"], idempotency_key,
+ json.dumps(payload, ensure_ascii=False), now, now,
+ ),
+ )
+ event_id = int(cursor.lastrowid)
+ event = self.get_outbox_event(event_id)
+ assert event is not None
+ return event
+ finally:
+ connection.close()
+
+ def save_collect_failure(
+ self,
+ remote_task_id: str,
+ attempt_id: str,
+ status: TaskStatus,
+ error_code: str,
+ error_message: str,
+ retryable: bool,
+ diagnostics: Optional[Dict[str, object]] = None,
+ ) -> OutboxEventRecord:
+ """保存结构化失败，并可靠排队提交 Admin。"""
+
+ if status not in {
+ TaskStatus.RETRY_WAIT, TaskStatus.MANUAL_REVIEW,
+ TaskStatus.FAILED, TaskStatus.CANCELLED,
+ }:
+ raise ValueError("失败状态无效")
+ now = utc_now_iso()
+ connection = open_database(self._db_path)
+ try:
+ with connection:
+ task = connection.execute(
+ "SELECT id, version FROM pdd_tasks WHERE remote_task_id = ?",
+ (remote_task_id,),
+ ).fetchone()
+ if task is None:
+ raise ValueError(f"任务 {remote_task_id} 不存在")
+ payload = {
+ "task_version": task["version"],
+ "attempt_id": attempt_id,
+ "status": status.value,
+ "error": {
+ "code": error_code,
+ "message": error_message,
+ "retryable": retryable,
+ "step": "collecting",
+ },
+ "diagnostics": diagnostics or {"artifacts": []},
+ "reported_at": now,
+ }
+ idempotency_key = f"{remote_task_id}:{attempt_id}:failure-v1"
+ connection.execute(
+ "UPDATE pdd_tasks SET status = ?, current_step = 'failed',"
+ " retry_count = retry_count + ?, last_error_code = ?,"
+ " last_error_message = ?, finished_at = ?, updated_at = ?"
+ " WHERE id = ?",
+ (
+ status.value, 1 if status is TaskStatus.RETRY_WAIT else 0,
+ error_code, error_message, now, now, task["id"],
+ ),
+ )
+ run_status = {
+ TaskStatus.CANCELLED: RunStatus.CANCELLED,
+ TaskStatus.MANUAL_REVIEW: RunStatus.MANUAL_REVIEW,
+ }.get(status, RunStatus.FAILED)
+ connection.execute(
+ "UPDATE task_runs SET run_status = ?, current_step = 'failed',"
+ " error_code = ?, error_message = ?, finished_at = ?, updated_at = ?"
+ " WHERE attempt_id = ?",
+ (run_status.value, error_code, error_message, now, now, attempt_id),
+ )
+ cursor = connection.execute(
+ "INSERT INTO outbox_events (task_id, event_type, idempotency_key,"
+ " payload_json, status, created_at, updated_at)"
+ " VALUES (?, 'task_failure', ?, ?, 'pending', ?, ?)",
+ (
+ task["id"], idempotency_key,
+ json.dumps(payload, ensure_ascii=False), now, now,
+ ),
+ )
+ event_id = int(cursor.lastrowid)
+ event = self.get_outbox_event(event_id)
+ assert event is not None
+ return event
+ finally:
+ connection.close()
+
+ def next_pending_outbox(self) -> Optional[OutboxEventRecord]:
+ connection = open_database(self._db_path)
+ try:
+ row = connection.execute(
+ "SELECT o.* FROM outbox_events o JOIN pdd_tasks t ON t.id = o.task_id"
+ " WHERE o.status = 'pending'"
+ " AND (o.next_retry_at IS NULL OR o.next_retry_at <= ?)"
+ " ORDER BY o.id ASC LIMIT 1",
+ (utc_now_iso(),),
+ ).fetchone()
+ finally:
+ connection.close()
+ return self._to_outbox(row) if row is not None else None
+
+ def get_outbox_event(self, event_id: int) -> Optional[OutboxEventRecord]:
+ connection = open_database(self._db_path)
+ try:
+ row = connection.execute(
+ "SELECT * FROM outbox_events WHERE id = ?", (event_id,)
+ ).fetchone()
+ finally:
+ connection.close()
+ return self._to_outbox(row) if row is not None else None
+
+ def outbox_task_id(self, event_id: int) -> str:
+ connection = open_database(self._db_path)
+ try:
+ row = connection.execute(
+ "SELECT t.remote_task_id FROM outbox_events o"
+ " JOIN pdd_tasks t ON t.id = o.task_id WHERE o.id = ?",
+ (event_id,),
+ ).fetchone()
+ finally:
+ connection.close()
+ if row is None:
+ raise ValueError(f"Outbox {event_id} 不存在")
+ return str(row[0])
+
+ def mark_outbox_sending(self, event_id: int) -> None:
+ self._update_outbox(event_id, "sending", None)
+
+ def mark_outbox_retry(self, event_id: int, message: str) -> None:
+ self._update_outbox(event_id, "pending", message, increment=True)
+
+ def mark_outbox_failed(self, event_id: int, message: str) -> None:
+ self._update_outbox(event_id, "failed", message, increment=True)
+
+ def mark_outbox_sent(self, event_id: int) -> None:
+ now = utc_now_iso()
+ connection = open_database(self._db_path)
+ try:
+ with connection:
+ row = connection.execute(
+ "SELECT task_id, event_type FROM outbox_events WHERE id = ?",
+ (event_id,),
+ ).fetchone()
+ if row is None:
+ raise ValueError(f"Outbox {event_id} 不存在")
+ connection.execute(
+ "UPDATE outbox_events SET status = 'sent', sent_at = ?,"
+ " updated_at = ?, attempt_count = attempt_count + 1"
+ " WHERE id = ?",
+ (now, now, event_id),
+ )
+ if row["event_type"] == OutboxEventType.COLLECT_RESULT.value:
+ connection.execute(
+ "UPDATE pdd_tasks SET status = 'succeeded',"
+ " current_step = 'completed', updated_at = ? WHERE id = ?",
+ (now, row["task_id"]),
+ )
+ finally:
+ connection.close()
+
+ def _update_outbox(
+ self, event_id: int, status: str, message: Optional[str], increment: bool = False
+ ) -> None:
+ connection = open_database(self._db_path)
+ try:
+ with connection:
+ cursor = connection.execute(
+ "UPDATE outbox_events SET status = ?, last_error = ?, updated_at = ?,"
+ f" attempt_count = attempt_count + {1 if increment else 0} WHERE id = ?",
+ (status, message, utc_now_iso(), event_id),
+ )
+ if cursor.rowcount != 1:
+ raise ValueError(f"Outbox {event_id} 不存在")
+ finally:
+ connection.close()
+
+ @staticmethod
+ def _summary_price(pdd_data: Dict[str, object]) -> Optional[int]:
+ skus = pdd_data.get("skus")
+ if not isinstance(skus, list):
+ return None
+ prices = [
+ item.get("price_cent") for item in skus
+ if isinstance(item, dict) and isinstance(item.get("price_cent"), int)
+ ]
+ return min(prices) if prices else None
+
+ @staticmethod
+ def _to_outbox(row: sqlite3.Row) -> OutboxEventRecord:
+ return OutboxEventRecord(
+ id=row["id"], task_id=row["task_id"],
+ event_type=OutboxEventType(row["event_type"]),
+ idempotency_key=row["idempotency_key"],
+ payload_json=TaskRepository._load_json_object(row["payload_json"]),
+ status=OutboxStatus(row["status"]), attempt_count=row["attempt_count"],
+ next_retry_at=row["next_retry_at"], last_error=row["last_error"],
+ created_at=row["created_at"], updated_at=row["updated_at"],
+ sent_at=row["sent_at"],
+ )
+
@staticmethod
def _validate_page(limit: int, offset: int) -> None:
if not 1 <= limit <= MAX_PAGE_SIZE:
diff --git a/client/test/test_collect_task_service.py b/client/test/test_collect_task_service.py
new file mode 100644
index 0000000..8355da0
--- /dev/null
+++ b/client/test/test_collect_task_service.py
@@ -0,0 +1,164 @@
+"""采集任务应用流程测试，不连接真机和网络。"""
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from src.admin_gateway import AdminTask, ClaimCapabilities, ClientInfo
+from src.collect_task_service import CollectTaskService
+from src.mock_admin_gateway import MockAdminGateway
+from src.pdd_collect_service import PddCollectError
+from src.task_models import NewClaimedTask, TaskStatus, TaskType
+from src.task_repository import TaskRepository
+
+
+class FakeResult:
+ def to_pdd_data(self):
+ return {
+ "schema_version": 1,
+ "goods_id": "737116531267",
+ "title": "测试商品",
+ "shop_name": "测试店铺",
+ "price_granularity": "color",
+ "dimensions": [
+ {"key": "color", "name": "颜色分类"},
+ {"key": "size", "name": "尺码"},
+ ],
+ "skus": [
+ {
+ "options": {"color": "黑色", "size": "M"},
+ "price_cent": 990,
+ "price_observed_at": {"color": "黑色", "size": "M"},
+ "available": True,
+ }
+ ],
+ }
+
+
+class FakeCollector:
+ def __init__(self, calls, error=None):
+ self.calls = calls
+ self.error = error
+
+ def collect(self, task):
+ self.calls.append(task.remote_task_id)
+ if self.error is not None:
+ raise self.error
+ return FakeResult()
+
+
+class CollectTaskServiceTest(unittest.TestCase):
+ def setUp(self):
+ self.temporary = tempfile.TemporaryDirectory()
+ self.repository = TaskRepository(Path(self.temporary.name) / "client.db")
+ self.gateway = MockAdminGateway()
+ self.client = ClientInfo("CLIENT-001", "测试电脑")
+ self.task = AdminTask(
+ task_id="COL-001",
+ task_type=TaskType.COLLECT,
+ version=1,
+ priority=0,
+ payload={
+ "goods_id": "737116531267",
+ "goods_url": "https://mobile.yangkeduo.com/goods.html?goods_id=737116531267",
+ },
+ created_at="2026-08-07T08:00:00Z",
+ updated_at="2026-08-07T08:00:00Z",
+ )
+ self.gateway.enqueue_task(self.task, self.client.client_id)
+
+ def tearDown(self):
+ self.temporary.cleanup()
+
+ def _service(self, calls, error=None):
+ return CollectTaskService(
+ self.gateway,
+ self.repository,
+ self.client,
+ "USB-001",
+ collect_service_factory=lambda *_args: FakeCollector(calls, error),
+ )
+
+ def test_claim_collect_persist_submit_completes_one_task(self):
+ calls = []
+
+ outcome = self._service(calls).execute_one()
+
+ self.assertEqual(outcome.kind, "succeeded")
+ self.assertEqual(calls, ["COL-001"])
+ detail = self.repository.get_task("COL-001")
+ self.assertEqual(detail.status, TaskStatus.SUCCEEDED)
+ self.assertEqual(detail.pdd_data["price_granularity"], "color")
+ self.assertEqual(self.gateway.submission_count, 1)
+
+ def test_submit_timeout_retries_stored_outbox_without_recollecting(self):
+ remote = self.gateway.claim_next(
+ self.client,
+ ClaimCapabilities(supported_types=(TaskType.COLLECT,)),
+ )
+ self.repository.add_claimed_task(
+ NewClaimedTask(
+ remote_task_id=remote.task_id,
+ task_type=remote.task_type,
+ goods_url=remote.payload["goods_url"],
+ goods_id=remote.payload["goods_id"],
+ version=remote.version,
+ admin_payload={"payload": dict(remote.payload)},
+ )
+ )
+ calls = []
+ self.gateway.timeout_next_call()
+
+ first = self._service(calls).execute_one()
+ second = self._service(calls).execute_one()
+
+ self.assertEqual(first.kind, "result_pending")
+ self.assertEqual(second.kind, "succeeded")
+ self.assertEqual(calls, ["COL-001"])
+ self.assertEqual(self.gateway.submission_count, 1)
+
+ def test_pending_outbox_can_submit_without_android_device(self):
+ remote = self.gateway.claim_next(
+ self.client,
+ ClaimCapabilities(supported_types=(TaskType.COLLECT,)),
+ )
+ self.repository.add_claimed_task(
+ NewClaimedTask(
+ remote_task_id=remote.task_id,
+ task_type=remote.task_type,
+ goods_url=remote.payload["goods_url"],
+ goods_id=remote.payload["goods_id"],
+ version=remote.version,
+ )
+ )
+ started = self.repository.start_collect_run(remote.task_id, "USB-001")
+ self.repository.save_collect_result(
+ remote.task_id, started.attempt_id, FakeResult().to_pdd_data()
+ )
+
+ outcome = CollectTaskService(
+ self.gateway, self.repository, self.client, ""
+ ).execute_one()
+
+ self.assertEqual(outcome.kind, "succeeded")
+ self.assertEqual(
+ self.repository.get_task(remote.task_id).status,
+ TaskStatus.SUCCEEDED,
+ )
+
+ def test_captcha_becomes_manual_review_and_is_reported(self):
+ calls = []
+ outcome = self._service(
+ calls, PddCollectError("PDD_PAGE_CAPTCHA", "需要验证")
+ ).execute_one()
+
+ self.assertEqual(outcome.kind, "failed")
+ self.assertEqual(
+ self.repository.get_task("COL-001").status,
+ TaskStatus.MANUAL_REVIEW,
+ )
+ self.assertEqual(self.gateway.submission_count, 1)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/client/test/test_http_admin_gateway.py b/client/test/test_http_admin_gateway.py
index 383d6e0..9ccc0d8 100644
--- a/client/test/test_http_admin_gateway.py
+++ b/client/test/test_http_admin_gateway.py
@@ -104,7 +104,7 @@ class HttpAdminGatewayTest(unittest.TestCase):
},
)
)
- gateway = HttpAdminGateway(opener=opener)
+ gateway = HttpAdminGateway(opener=opener, client_id="CLIENT-001")
gateway.register_client(
ClientInfo("CLIENT-001"), self._capabilities(False)
@@ -116,6 +116,71 @@ class HttpAdminGatewayTest(unittest.TestCase):
key.lower(): value for key, value in opener.request.header_items()
})
+ def test_submit_result_sends_idempotency_key_and_parses_receipt(self):
+ opener = RecordingOpener(
+ FakeResponse(
+ 200,
+ {
+ "accepted": True,
+ "result_id": "RESULT-001",
+ "accepted_at": "2026-08-07T08:00:01Z",
+ },
+ )
+ )
+ gateway = HttpAdminGateway(opener=opener, client_id="CLIENT-001")
+ payload = {
+ "task_version": 1,
+ "attempt_id": "ATTEMPT-001",
+ "result_type": "collect",
+ "completed_at": "2026-08-07T08:00:00Z",
+ "pdd_data": {"goods_id": "123"},
+ }
+
+ receipt = gateway.submit_result(
+ "COL-001", "COL-001:ATTEMPT-001:result-v1", payload
+ )
+
+ self.assertTrue(receipt.accepted)
+ self.assertEqual(
+ opener.request.full_url,
+ "http://127.0.0.1:8080/api/v1/client/tasks/COL-001/result",
+ )
+ headers = {
+ key.lower(): value for key, value in opener.request.header_items()
+ }
+ self.assertEqual(
+ headers["idempotency-key"], "COL-001:ATTEMPT-001:result-v1"
+ )
+ self.assertEqual(headers["x-client-id"], "CLIENT-001")
+ self.assertEqual(json.loads(opener.request.data), payload)
+
+ def test_submit_failure_uses_failure_endpoint(self):
+ opener = RecordingOpener(
+ FakeResponse(
+ 201,
+ {
+ "accepted": True,
+ "result_id": "FAILURE-001",
+ "accepted_at": "2026-08-07T08:00:01Z",
+ },
+ )
+ )
+ gateway = HttpAdminGateway(opener=opener, client_id="CLIENT-001")
+
+ gateway.submit_failure(
+ "COL-001",
+ "COL-001:ATTEMPT-001:failure-v1",
+ {
+ "task_version": 1,
+ "attempt_id": "ATTEMPT-001",
+ "status": "manual_review",
+ "error": {"code": "PDD_PAGE_CAPTCHA"},
+ "reported_at": "2026-08-07T08:00:00Z",
+ },
+ )
+
+ self.assertTrue(opener.request.full_url.endswith("/COL-001/failure"))
+
def test_claim_maps_real_admin_payload_and_only_reports_collect(self):
opener = RecordingOpener(
FakeResponse(
diff --git a/client/test/test_pdd_collect_service.py b/client/test/test_pdd_collect_service.py
index 26275d4..c55d1f5 100644
--- a/client/test/test_pdd_collect_service.py
+++ b/client/test/test_pdd_collect_service.py
@@ -62,6 +62,18 @@ class LoadingDevice(FakeCollectDevice):
return ''
+class DisconnectingDevice(FakeCollectDevice):
+ def __init__(self, home_xml, spec_xml):
+ super().__init__(home_xml, spec_xml)
+ self.current_calls = 0
+
+ def app_current(self):
+ self.current_calls += 1
+ if self.current_calls > 1:
+ raise RuntimeError("device not found")
+ return super().app_current()
+
+
def keep_only_one_sku(xml_data: str) -> str:
"""从脱敏固件中删除蓝色和 L，只保留一个组合。"""
@@ -114,6 +126,19 @@ class PddCollectParserTest(unittest.TestCase):
["M", "L"],
)
+ def test_truncated_spec_name_prefers_full_content_description(self):
+ xml_data = self.spec_xml.replace(
+ 'content-desc="红色" clickable="true"',
+ 'text="红…" content-desc="红色" clickable="true"',
+ )
+ result = parse_spec_panel(xml_data)
+ self.assertEqual(result.dimensions[0].values[0].text, "红色")
+
+ without_description = xml_data.replace(' content-desc="红色"', "")
+ with self.assertRaises(PddCollectError) as raised:
+ parse_spec_panel(without_description)
+ self.assertEqual(raised.exception.code, "PDD_DATA_SKU_NAME_TRUNCATED")
+
def test_login_and_invalid_xml_have_different_error_codes(self):
with self.assertRaises(PddCollectError) as login:
parse_goods_page('')
@@ -150,14 +175,46 @@ class PddCollectParserTest(unittest.TestCase):
data = result.to_pdd_data()
self.assertEqual(data["schema_version"], 1)
- self.assertEqual(data["goods"]["goods_id"], "123")
+ self.assertEqual(data["goods_id"], "123")
+ self.assertEqual(data["title"], "测试纯棉短袖商品")
+ self.assertEqual(data["shop_name"], "测试服饰旗舰店")
+ self.assertEqual(data["price_granularity"], "color")
self.assertEqual(data["metrics"]["sales"]["value"], 12000)
self.assertEqual(data["dimensions"][0]["key"], "color")
self.assertEqual(data["skus"][0]["options"], {"color": "红色", "size": "M"})
self.assertEqual(data["skus"][0]["price_cent"], 1000)
+ self.assertEqual(
+ data["skus"][0]["price_observed_at"],
+ {"color": "红色", "size": "M"},
+ )
self.assertEqual(data["source"]["device_address"], "USB-001")
self.assertIsNone(data["purchase"])
+ def test_price_is_sampled_once_per_available_color(self):
+ device = FakeCollectDevice(self.home_xml, self.spec_xml)
+ service = PddCollectService(
+ PddDeviceService(lambda _serial: device),
+ "USB-001",
+ "client-001",
+ sleeper=lambda _seconds: None,
+ max_page_swipes=0,
+ max_spec_swipes=0,
+ )
+
+ data = service.collect(
+ FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
+ ).to_pdd_data()
+
+ self.assertEqual(len(data["skus"]), 4)
+ red_skus = [
+ item for item in data["skus"] if item["options"]["color"] == "红色"
+ ]
+ self.assertEqual(
+ {item["price_observed_at"]["size"] for item in red_skus}, {"M"}
+ )
+ # 1 次打开规格面板，另有红色点 1 次；尺码只读不点击。
+ self.assertEqual(len(device.clicks), 2)
+
def test_missing_goods_id_is_structured_error(self):
service = PddCollectService(
PddDeviceService(lambda _serial: object()),
@@ -195,7 +252,7 @@ class PddCollectParserTest(unittest.TestCase):
def test_page_timeout_has_specific_error_code(self):
device = LoadingDevice(self.home_xml, self.spec_xml)
- ticks = iter((0.0, 2.0))
+ ticks = iter((0.0, 0.0, 0.0, 2.0))
service = PddCollectService(
PddDeviceService(lambda _serial: device),
"USB-001",
@@ -211,7 +268,21 @@ class PddCollectParserTest(unittest.TestCase):
)
self.assertEqual(raised.exception.code, "PDD_PAGE_TIMEOUT")
- def test_incomplete_goods_data_does_not_return_partial_success(self):
+ def test_runtime_device_disconnect_has_stable_error_code(self):
+ service = PddCollectService(
+ PddDeviceService(
+ lambda _serial: DisconnectingDevice(self.home_xml, self.spec_xml)
+ ),
+ "USB-001",
+ "client-001",
+ )
+ with self.assertRaises(PddCollectError) as raised:
+ service.collect(
+ FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
+ )
+ self.assertEqual(raised.exception.code, "DEVICE_DISCONNECTED")
+
+ def test_missing_shop_name_is_allowed(self):
incomplete_home = self.home_xml.replace(
'