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

fix: 修复 PDD 页面误判与失败回执 (#35)

Browse Source
This commit is contained in:
chengma
2026-08-07 17:58:30 +08:00
parent 51c2292be1
commit c59b4b0d59
11 changed files with 267 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
admin/service/submit.go
+1
View File
@@ -210,6 +210,7 @@ func SubmitFailure(db *sql.DB, taskID, clientID, idemKey string, rawBody []byte)
}
return map[string]any{
"accepted": true,
"result_id": newID(),
"task_status": string(newStatus),
"accepted_at": model.NowISO(),
}, nil
admin/service/submit_test.go
+9 -1
View File
@@ -435,9 +435,17 @@ func TestSubmitFailure_状态映射(t *testing.T) {
body := `{"task_version":1,"attempt_id":"a-1","status":"` + tc.reported +
`","error":{"code":"PDD_PAGE_TIMEOUT","message":"页面超时"}}`
if _, err := SubmitFailure(db, "TASK-A", "client-001", "k", []byte(body)); err != nil {
response, err := SubmitFailure(db, "TASK-A", "client-001", "k", []byte(body))
if err != nil {
t.Fatalf("提交失败结果出错: %v", err)
}
var receipt map[string]any
if err := json.Unmarshal([]byte(response), &receipt); err != nil {
t.Fatalf("失败响应不是 JSON: %v", err)
}
if receipt["result_id"] == "" || receipt["result_id"] == nil {
t.Error("失败响应 result_id 不应为空")
}
if s := taskStatus(t, db, "TASK-A"); s != tc.want {
t.Errorf("%s 应映射为 %s，实际 %s", tc.reported, tc.want, s)
}
client/src/http_admin_gateway.py
+22 -3
View File
@@ -1,5 +1,6 @@
"""使用 Python 标准库调用 Admin 登记和任务领取接口。"""
"""使用 Python 标准库调用 Admin 登记、领取和提交接口。"""
import hashlib
import json
import socket
from http.client import RemoteDisconnected
@@ -306,6 +307,24 @@ class HttpAdminGateway(AdminGateway):
accepted = data.get("accepted")
result_id = data.get("result_id")
accepted_at = data.get("accepted_at")
legacy_failure = (
endpoint == "failure"
and accepted is True
and isinstance(accepted_at, str)
and bool(accepted_at.strip())
and isinstance(data.get("task_status"), str)
and bool(data["task_status"].strip())
and not (isinstance(result_id, str) and result_id.strip())
)
if legacy_failure:
digest = hashlib.sha256(
idempotency_key.encode("utf-8")
).hexdigest()[:16]
return SubmissionReceipt(
True,
f"legacy-failure-{digest}",
accepted_at,
)
if (
accepted is not True
or not isinstance(result_id, str)
@@ -315,8 +334,8 @@ class HttpAdminGateway(AdminGateway):
):
raise AdminGatewayError(
"ADMIN_INVALID_RESPONSE",
"Admin 提交响应字段不完整",
False,
"Admin 返回成功状态，但响应格式异常，数据可能已接收，将安全重试",
True,
request_id,
)
return SubmissionReceipt(True, result_id, accepted_at)
client/src/pdd_collect_service.py
+15 -6
View File
@@ -764,21 +764,30 @@ class PddCollectService:
raise PddCollectError("DEVICE_APP_START_FAILED", f"无法打开 PDD 商品链接：{exc}") from exc
deadline = self._monotonic() + self._page_timeout
last_labels: list[str] = []
while self._monotonic() < deadline:
self._check_cancelled()
current = device.app_current()
if current.get("package") == PDD_PACKAGE_NAME:
xml_data = device.dump_hierarchy()
root = _parse_xml(xml_data)
last_labels = _all_labels(root)
xml_data = device.dump_hierarchy()
self._last_goods_xml = str(xml_data)
self._goods_screens_checked += 1
root = _parse_xml(xml_data)
last_labels = _all_labels(root)
pdd_node_count = sum(
1
for node in root.iter("node")
if node.get("package") == PDD_PACKAGE_NAME
)
# 部分 OPPO/ColorOS 设备会一直把无线调试设置页报告为焦点，
# 即使屏幕和无障碍树已经是 PDD。此时以树中真实包名为准。
is_pdd_hierarchy = pdd_node_count >= 3
is_pdd_focused = current.get("package") == PDD_PACKAGE_NAME
if is_pdd_focused or is_pdd_hierarchy:
_raise_special_page(last_labels)
combined = " ".join(last_labels)
loading = "加载中" in combined or "正在加载" in combined
if not loading and any(marker in combined for marker in _READY_MARKERS):
return
self._sleep(0.5)
_raise_special_page(last_labels)
raise PddCollectError("PDD_PAGE_TIMEOUT", "等待 PDD 商品详情页加载超时")
def _collect_goods_details(self, device: Any) -> GoodsSnapshot:
client/src/task_repository.py
+9
View File
@@ -163,6 +163,15 @@ class TaskRepository:
" WHERE status = 'sending'",
(now,),
)
# #35 之前，Admin 的失败响应缺少 result_id。Admin 实际已经
# 接收，但旧 Client 把这类 2xx 响应误标为永久失败。恢复为
# pending 后使用原幂等键重试，不会重复写入业务结果。
connection.execute(
"UPDATE outbox_events SET status = 'pending', updated_at = ?"
" WHERE status = 'failed'"
" AND last_error = 'Admin 提交响应字段不完整'",
(now,),
)
connection.execute(
"UPDATE pdd_tasks SET status = 'retry_wait',"
" current_step = 'interrupted', retry_count = retry_count + 1,"
client/test/test_http_admin_gateway.py
+37
View File
@@ -181,6 +181,43 @@ class HttpAdminGatewayTest(unittest.TestCase):
self.assertTrue(opener.request.full_url.endswith("/COL-001/failure"))
def test_submit_failure_accepts_legacy_admin_receipt_without_result_id(self):
gateway = HttpAdminGateway(
opener=RecordingOpener(
FakeResponse(
200,
{
"accepted": True,
"task_status": "assigned",
"accepted_at": "2026-08-07T08:00:01Z",
},
)
),
client_id="CLIENT-001",
)
receipt = gateway.submit_failure(
"COL-001",
"COL-001:ATTEMPT-001:failure-v1",
{"status": "retry_wait"},
)
self.assertTrue(receipt.accepted)
self.assertTrue(receipt.result_id.startswith("legacy-failure-"))
def test_malformed_2xx_submission_is_retryable_and_not_called_rejected(self):
gateway = HttpAdminGateway(
opener=RecordingOpener(FakeResponse(200, {"accepted": True})),
client_id="CLIENT-001",
)
with self.assertRaises(AdminGatewayError) as raised:
gateway.submit_result("COL-001", "key-1", {})
self.assertEqual(raised.exception.code, "ADMIN_INVALID_RESPONSE")
self.assertTrue(raised.exception.retryable)
self.assertIn("可能已接收", str(raised.exception))
def test_claim_maps_real_admin_payload_and_only_reports_collect(self):
opener = RecordingOpener(
FakeResponse(
client/test/test_pdd_collect_service.py
+80
View File
@@ -3,6 +3,7 @@
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
import tempfile
import unittest
import xml.etree.ElementTree as ET
@@ -74,6 +75,16 @@ class DisconnectingDevice(FakeCollectDevice):
return super().app_current()
class FocusMismatchDevice(FakeCollectDevice):
"""系统焦点报告为设置页，但无障碍树实际属于 PDD。"""
def app_current(self):
return {"package": "com.oplus.wirelesssettings"}
def app_start(self, _package):
self.app_started = True
def keep_only_one_sku(xml_data: str) -> str:
"""从脱敏固件中删除蓝色和 L，只保留一个组合。"""
@@ -190,6 +201,29 @@ class PddCollectParserTest(unittest.TestCase):
self.assertEqual(data["source"]["device_address"], "USB-001")
self.assertIsNone(data["purchase"])
def test_pdd_hierarchy_is_ready_even_when_focused_package_is_settings(self):
pdd_home_xml = self.home_xml.replace(
"<node ", '<node package="com.xunmeng.pinduoduo" '
)
device = FocusMismatchDevice(
pdd_home_xml, keep_only_one_sku(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,
)
result = service.collect(
FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
)
self.assertEqual(result.goods_id, "123")
self.assertTrue(device.app_started)
def test_price_is_sampled_once_per_available_color(self):
device = FakeCollectDevice(self.home_xml, self.spec_xml)
service = PddCollectService(
@@ -268,6 +302,52 @@ class PddCollectParserTest(unittest.TestCase):
)
self.assertEqual(raised.exception.code, "PDD_PAGE_TIMEOUT")
def test_non_pdd_tree_with_purchase_words_is_not_ready(self):
device = FocusMismatchDevice(
'<hierarchy><node package="com.android.settings" text="免拼购买" '
'bounds="[0,0][100,100]" /></hierarchy>',
self.spec_xml,
)
ticks = iter((0.0, 0.0, 0.0, 2.0))
service = PddCollectService(
PddDeviceService(lambda _serial: device),
"USB-001",
"client-001",
sleeper=lambda _seconds: None,
monotonic=lambda: next(ticks),
page_timeout=1.0,
)
with self.assertRaises(PddCollectError) as raised:
service.collect(
FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
)
self.assertEqual(raised.exception.code, "PDD_PAGE_TIMEOUT")
def test_page_timeout_saves_last_xml_diagnostic(self):
with tempfile.TemporaryDirectory() as directory:
device = LoadingDevice(self.home_xml, self.spec_xml)
ticks = iter((0.0, 0.0, 0.0, 0.0, 0.0, 2.0))
service = PddCollectService(
PddDeviceService(lambda _serial: device),
"USB-001",
"client-001",
sleeper=lambda _seconds: None,
monotonic=lambda: next(ticks),
page_timeout=1.0,
artifact_directory=Path(directory),
)
with self.assertRaises(PddCollectError) as raised:
service.collect(
FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
)
artifacts = raised.exception.diagnostics["artifacts"]
self.assertEqual(len(artifacts), 1)
self.assertTrue(Path(artifacts[0]["path"]).is_file())
self.assertEqual(len(artifacts[0]["sha256"]), 64)
def test_runtime_device_disconnect_has_stable_error_code(self):
service = PddCollectService(
PddDeviceService(
client/test/test_pdd_real_xml_fixtures.py
+45 -1
View File
@@ -3,13 +3,57 @@
from pathlib import Path
import unittest
from src.pdd_collect_service import parse_goods_page, parse_spec_panel
from src.pdd_collect_service import (
PddCollectService,
parse_goods_page,
parse_spec_panel,
)
from src.pdd_device_service import PddDeviceService
REAL_XML = Path(__file__).parents[1] / "image_xml"
class SettingsFocusedRealPage:
"""复现真机：焦点 API 是无线设置，控件树实际是 PDD。"""
def __init__(self, xml_data):
self.xml_data = xml_data
def app_current(self):
return {"package": "com.oplus.wirelesssettings"}
def app_start(self, _package):
return None
def app_wait(self, _package, timeout=10):
return 1
def open_url(self, _url):
return None
def dump_hierarchy(self):
return self.xml_data
class PddRealXmlFixtureTest(unittest.TestCase):
def test_real_pdd_page_is_ready_when_focus_api_reports_settings(self):
xml_data = (REAL_XML / "737116531267_home.xml").read_text(
encoding="utf-8"
)
device = SettingsFocusedRealPage(xml_data)
service = PddCollectService(
PddDeviceService(lambda _serial: device),
"USB-001",
"client-001",
sleeper=lambda _seconds: None,
)
service._open_goods(
device,
"https://mobile.yangkeduo.com/goods.html?goods_id=737116531267",
)
def test_split_title_is_joined_from_real_goods_page(self):
result = parse_goods_page(
(REAL_XML / "737116531267_home.xml").read_text(encoding="utf-8")
client/test/test_task_repository.py
+20
View File
@@ -230,6 +230,26 @@ class TaskRepositoryTests(unittest.TestCase):
OutboxStatus.PENDING,
)
def test_recovery_retries_old_ambiguous_admin_response(self):
self.repository.add_claimed_task(self._task("TASK-AMBIGUOUS"))
started = self.repository.start_collect_run("TASK-AMBIGUOUS", "USB-001")
event = self.repository.save_collect_failure(
"TASK-AMBIGUOUS",
started.attempt_id,
TaskStatus.RETRY_WAIT,
"PDD_PAGE_TIMEOUT",
"页面超时",
True,
)
self.repository.mark_outbox_failed(event.id, "Admin 提交响应字段不完整")
self.repository.recover_interrupted_work()
self.assertEqual(
self.repository.get_outbox_event(event.id).status,
OutboxStatus.PENDING,
)
if __name__ == "__main__":
unittest.main()
docs/admin/04-client-api.md
+14
View File
@@ -217,6 +217,20 @@ Idempotency-Key: <task_id>:<attempt_id>:failure-v1
采集任务失败时，同步把 `pdd_products.collect_status` 置 `failed`，
错误信息写进 `collect_msg`、诊断产物位置写进 `artifact_ref`，界面上要看得见。
成功响应：
```json
{
"accepted": true,
"result_id": "result-uuid",
"task_status": "assigned",
"accepted_at": "2026-08-06T08:03:01Z"
}
```
`[必须]` `result_id` 非空，并与同一幂等键保存的响应一起返回。失败提交也使用
Client `SubmissionReceipt` 的统一确认字段，不能只返回 `task_status`。
`[必须]` §4.1 的无条件接受**同样适用于本接口**。
## 6. 幂等怎么做
docs/client/04-admin-api-contract.md
+15
View File
@@ -355,6 +355,21 @@ Idempotency-Key: task-id:attempt-id:failure-v1
`status` 只能是 `retry_wait`、`manual_review`、`failed` 或 `cancelled`。
成功响应与结果接口使用同一种确认结构，并额外返回任务状态：
```json
{
"accepted": true,
"result_id": "result-uuid",
"task_status": "assigned",
"accepted_at": "2026-08-06T08:03:01Z"
}
```
`[必须]` `result_id` 非空。旧 Admin 已经写入幂等表、但没有 `result_id` 的
历史失败响应，Client 可按 `accepted + task_status + accepted_at` 兼容一次，
避免把已经接收的数据误报为拒绝。
规则：
- `[必须]` §6.1 的无条件接受规则同样适用于本接口。