diff --git a/client/src/admin_gateway.py b/client/src/admin_gateway.py index 76d978e..eb88073 100644 --- a/client/src/admin_gateway.py +++ b/client/src/admin_gateway.py @@ -47,10 +47,7 @@ class ClaimCapabilities: """Client 领取任务时声明的设备与执行能力。""" device: Optional[AndroidDeviceInfo] = None - supported_types: Tuple[TaskType, ...] = ( - TaskType.COLLECT, - TaskType.PURCHASE, - ) + supported_types: Tuple[TaskType, ...] = (TaskType.COLLECT,) purchase_mode: str = "dry_run" schema_versions: Tuple[int, ...] = (1,) diff --git a/client/src/http_admin_gateway.py b/client/src/http_admin_gateway.py index 4269cd7..0211d60 100644 --- a/client/src/http_admin_gateway.py +++ b/client/src/http_admin_gateway.py @@ -147,16 +147,17 @@ class HttpAdminGateway(AdminGateway): client: ClientInfo, capabilities: ClaimCapabilities, ) -> Optional[AdminTask]: - """领取一个采集任务；Admin 返回 204 时返回 ``None``。""" + """按调用方已安全确认的能力领取一个任务。""" request_id = str(uuid4()) self._client_id = client.client_id.strip() payload = { "client": {"name": client.name.strip()}, - # #30 只允许领取采集任务。采购能力必须由安全门禁工单开启。 - "supported_types": [TaskType.COLLECT.value], + "supported_types": [ + task_type.value for task_type in capabilities.supported_types + ], "capabilities": { - "purchase_mode": "dry_run", + "purchase_mode": capabilities.purchase_mode, "schema_versions": list(capabilities.schema_versions), }, } @@ -394,6 +395,8 @@ class HttpAdminGateway(AdminGateway): request_id, ) from exc + cls._validate_claim_payload(task_type, task_payload, request_id) + return AdminTask( task_id=task_id, task_type=task_type, @@ -404,6 +407,61 @@ class HttpAdminGateway(AdminGateway): updated_at=updated_at, ) + @staticmethod + def _validate_claim_payload( + task_type: TaskType, + payload: Mapping[str, object], + request_id: str, + ) -> None: + """在采购任务进入本地执行前校验全部安全字段。""" + + goods_url = payload.get("goods_url") + if not isinstance(goods_url, str) or not goods_url.strip(): + raise AdminGatewayError( + "ADMIN_INVALID_RESPONSE", + "Admin 任务 payload.goods_url 不能为空", + False, + request_id, + ) + if task_type is not TaskType.PURCHASE: + return + goods_id = payload.get("goods_id") + options = payload.get("options") + quantity = payload.get("quantity") + max_price_cent = payload.get("max_price_cent") + valid_options = ( + isinstance(options, Mapping) + and bool(options) + and all( + isinstance(key, str) + and bool(key.strip()) + and isinstance(value, str) + and bool(value.strip()) + for key, value in options.items() + ) + ) + valid = ( + isinstance(goods_id, str) + and bool(goods_id.strip()) + and valid_options + and isinstance(quantity, int) + and not isinstance(quantity, bool) + and quantity > 0 + and isinstance(max_price_cent, int) + and not isinstance(max_price_cent, bool) + and max_price_cent > 0 + ) + if not valid: + raise AdminGatewayError( + "ADMIN_INVALID_RESPONSE", + ( + "Admin 采购任务缺少有效的 goods_id、动态 options、" + "quantity 或 max_price_cent" + ), + False, + request_id, + ) + @staticmethod def _decode_json(body: bytes, request_id: str) -> dict: try: diff --git a/client/src/pdd_ui_event.py b/client/src/pdd_ui_event.py index f238f0f..fab9846 100644 --- a/client/src/pdd_ui_event.py +++ b/client/src/pdd_ui_event.py @@ -34,7 +34,6 @@ from qfluentwidgets import InfoBar, InfoBarPosition, MessageBox from .admin_gateway import ( AdminGatewayError, - AdminTask, ClientInfo, AdminGateway, ) @@ -42,16 +41,20 @@ 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 +from .purchase_task_service import PurchaseAdapterFactory from .selected_android_device_service import SelectedAndroidDeviceService from .settings_repository import SettingsRepository from .task_models import ( - NewClaimedTask, TaskFilters, TaskStatus, TaskSummary, TaskType, ) from .task_repository import CollectRerunError, TaskRepository +from .task_dispatcher import ( + TaskDispatcher, + admin_task_to_new_claimed_task, +) from .task_detail_view import TaskDetailWindow @@ -88,42 +91,8 @@ TASK_STATUS_TEXT = { } -def admin_task_to_new_claimed_task(task: AdminTask) -> NewClaimedTask: - """把 Admin 任务显式映射为本地任务，避免字段名自动展开出错。""" - - if task.task_type is not TaskType.COLLECT: - raise ValueError(f"任务 {task.task_id} 不是采集任务") - - payload = dict(task.payload) - goods_url = payload.get("goods_url") - goods_id = payload.get("goods_id") - if not isinstance(goods_url, str) or not goods_url.strip(): - raise ValueError("payload.goods_url 不能为空") - if goods_id is not None and not isinstance(goods_id, str): - raise ValueError("payload.goods_id 必须是文本") - - original_task = { - "id": task.task_id, - "type": task.task_type.value, - "version": task.version, - "priority": task.priority, - "payload": payload, - "created_at": task.created_at, - "updated_at": task.updated_at, - } - return NewClaimedTask( - remote_task_id=task.task_id, - task_type=task.task_type, - goods_url=goods_url, - goods_id=goods_id.strip() if isinstance(goods_id, str) else None, - priority=task.priority, - version=task.version, - admin_payload=original_task, - ) - - class ClaimTaskWorker(QObject): - """在后台补交或执行至多一条采集任务。""" + """在后台补交、领取或执行至多一条任务。""" noTask = pyqtSignal() taskSaved = pyqtSignal(str) @@ -141,6 +110,7 @@ class ClaimTaskWorker(QObject): client_service: CurrentClientService, android_device_service: SelectedAndroidDeviceService, collect_service_factory: Optional[CollectServiceFactory] = None, + purchase_adapter_factory: Optional[PurchaseAdapterFactory] = None, selected_task_id: str = "", ) -> None: super().__init__() @@ -150,6 +120,7 @@ class ClaimTaskWorker(QObject): self._android_device_service = android_device_service self._cancelled = False self._collect_service_factory = collect_service_factory + self._purchase_adapter_factory = purchase_adapter_factory self._selected_task_id = selected_task_id def cancel(self) -> None: @@ -168,22 +139,31 @@ class ClaimTaskWorker(QObject): return android_serial = self._android_device_service.load() - service = CollectTaskService( - self._gateway, - self._task_repository, - ClientInfo( - client_settings.client_id, - client_settings.client_name, - ), - android_serial or "", - cancelled=lambda: self._cancelled, - collect_service_factory=self._collect_service_factory, - ) - result = ( - service.execute_selected(self._selected_task_id) - if self._selected_task_id - else service.execute_one() + client = ClientInfo( + client_settings.client_id, + client_settings.client_name, ) + if self._selected_task_id: + service = CollectTaskService( + self._gateway, + self._task_repository, + client, + android_serial or "", + cancelled=lambda: self._cancelled, + collect_service_factory=self._collect_service_factory, + ) + result = service.execute_selected(self._selected_task_id) + else: + dispatcher = TaskDispatcher( + self._gateway, + self._task_repository, + client, + android_serial or "", + collect_service_factory=self._collect_service_factory, + purchase_adapter_factory=self._purchase_adapter_factory, + cancelled=lambda: self._cancelled, + ) + result = dispatcher.execute_one() if not self._cancelled or result.kind == "cancelled": self.outcome.emit(result.kind, result.message, result.task_id) except AdminGatewayError as exc: @@ -198,7 +178,7 @@ class ClaimTaskWorker(QObject): self.failed.emit(message) except Exception as exc: if not self._cancelled: - self.failed.emit(f"执行采集任务失败：{exc}") + self.failed.emit(f"执行任务失败：{exc}") finally: self.completed.emit() @@ -214,6 +194,7 @@ class PDDTaskPageEvent(QObject): claim_gateway: Optional[AdminGateway] = None, settings_repository: Optional[SettingsRepository] = None, collect_service_factory: Optional[CollectServiceFactory] = None, + purchase_adapter_factory: Optional[PurchaseAdapterFactory] = None, next_task_delay_ms: int = 500, no_task_delay_ms: int = 5_000, retry_delays_ms: tuple[int, ...] = (5_000, 10_000, 20_000, 30_000), @@ -233,6 +214,7 @@ class PDDTaskPageEvent(QObject): self._retry_count = 0 self._detail_windows: Dict[str, TaskDetailWindow] = {} self._collect_service_factory = collect_service_factory + self._purchase_adapter_factory = purchase_adapter_factory if next_task_delay_ms < 0 or no_task_delay_ms <= 0: raise ValueError("自动获取等待时间配置无效") if not retry_delays_ms or any(value <= 0 for value in retry_delays_ms): @@ -476,15 +458,16 @@ class PDDTaskPageEvent(QObject): self._claim_busy = True self._cycle_next_delay_ms = None self._page.set_auto_fetch_state("running") - self._page.set_engine_status("自动获取：运行中 · 正在处理一条采集任务…") + self._page.set_engine_status("自动获取：运行中 · 正在处理一条任务…") thread = QThread(self) worker = ClaimTaskWorker( - self._claim_gateway, - self._repository, - self._client_service, - self._selected_android_device_service, - self._collect_service_factory, + gateway=self._claim_gateway, + task_repository=self._repository, + client_service=self._client_service, + android_device_service=self._selected_android_device_service, + collect_service_factory=self._collect_service_factory, + purchase_adapter_factory=self._purchase_adapter_factory, ) worker.moveToThread(thread) thread.started.connect(worker.run) @@ -630,7 +613,7 @@ class PDDTaskPageEvent(QObject): self._show_retry_paused(content) self._stop_after_current(content) elif kind in {"manual_review", "failed"}: - self._show_claim_error("采集任务需要处理", message) + self._show_claim_error("任务需要处理", message) self._stop_after_current(message) elif kind == "cancelled": self._stop_after_current(message or "自动获取已停止") @@ -648,7 +631,11 @@ class PDDTaskPageEvent(QObject): task = self._repository.get_task(task_id) except Exception: return False - return task is not None and task.status is TaskStatus.RETRY_WAIT + return ( + task is not None + and task.task_type is TaskType.COLLECT + and task.status is TaskStatus.RETRY_WAIT + ) def _show_retry_paused(self, content: str) -> None: """用持久警告说明任务不会自行倒计时重试。""" diff --git a/client/src/task_dispatcher.py b/client/src/task_dispatcher.py new file mode 100644 index 0000000..238396f --- /dev/null +++ b/client/src/task_dispatcher.py @@ -0,0 +1,272 @@ +"""串行领取、落库并按任务类型分派一条任务。""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Callable, Mapping, Optional + +from .admin_gateway import ( + AdminGateway, + AdminGatewayError, + AdminTask, + AndroidDeviceInfo, + ClaimCapabilities, + ClientInfo, +) +from .collect_task_service import CollectServiceFactory, CollectTaskService +from .purchase_task_service import ( + PurchaseAdapterFactory, + PurchaseTaskService, +) +from .task_models import ( + NewClaimedTask, + OutboxEventRecord, + OutboxEventType, + TaskType, +) +from .task_repository import DuplicateTaskError, TaskRepository + + +@dataclass(frozen=True) +class TaskDispatchOutcome: + """一轮串行任务处理的结果。""" + + kind: str + message: str + task_id: str = "" + + +def admin_task_to_new_claimed_task(task: AdminTask) -> NewClaimedTask: + """严格校验 Admin 任务并映射为本地已领取任务。""" + + payload = dict(task.payload) + goods_url = payload.get("goods_url") + if not isinstance(goods_url, str) or not goods_url.strip(): + raise ValueError("payload.goods_url 不能为空") + goods_id = payload.get("goods_id") + if goods_id is not None and not isinstance(goods_id, str): + raise ValueError("payload.goods_id 必须是文本") + + target_color = None + target_size = None + price_cent = None + quantity = None + if task.task_type is TaskType.PURCHASE: + if not isinstance(goods_id, str) or not goods_id.strip(): + raise ValueError("采购任务 payload.goods_id 不能为空") + options = payload.get("options") + if not isinstance(options, Mapping) or not options: + raise ValueError("采购任务 payload.options 必须是非空对象") + normalized_options: dict[str, str] = {} + for key, value in options.items(): + if not isinstance(key, str) or not key.strip(): + raise ValueError("采购任务 options 的名称不能为空") + if not isinstance(value, str) or not value.strip(): + raise ValueError("采购任务 options 的值必须是非空文本") + normalized_options[key.strip()] = value.strip() + payload["options"] = normalized_options + quantity = payload.get("quantity") + price_cent = payload.get("max_price_cent") + if ( + isinstance(quantity, bool) + or not isinstance(quantity, int) + or quantity <= 0 + ): + raise ValueError("采购任务 payload.quantity 必须是大于 0 的整数") + if ( + isinstance(price_cent, bool) + or not isinstance(price_cent, int) + or price_cent <= 0 + ): + raise ValueError( + "采购任务 payload.max_price_cent 必须是大于 0 的整数分" + ) + target_color = normalized_options.get("color") + target_size = normalized_options.get("size") + + original_task = { + "id": task.task_id, + "type": task.task_type.value, + "version": task.version, + "priority": task.priority, + "payload": payload, + "created_at": task.created_at, + "updated_at": task.updated_at, + } + return NewClaimedTask( + remote_task_id=task.task_id, + task_type=task.task_type, + goods_url=goods_url.strip(), + goods_id=goods_id.strip() if isinstance(goods_id, str) else None, + target_color=target_color, + target_size=target_size, + price_cent=price_cent, + quantity=quantity, + priority=task.priority, + version=task.version, + admin_payload=original_task, + ) + + +class TaskDispatcher: + """一次只补交、执行或领取并执行一条任务。""" + + def __init__( + self, + gateway: AdminGateway, + repository: TaskRepository, + client: ClientInfo, + device_address: str, + *, + collect_service_factory: Optional[CollectServiceFactory] = None, + purchase_adapter_factory: Optional[PurchaseAdapterFactory] = None, + cancelled: Callable[[], bool] = lambda: False, + ) -> None: + self._gateway = gateway + self._repository = repository + self._client = client + self._device_address = str(device_address or "").strip() + self._collect_factory = collect_service_factory + self._purchase_factory = purchase_adapter_factory + self._cancelled = cancelled + + @property + def purchase_ready(self) -> bool: + """只有显式提供采购演练适配器时才声明采购能力。""" + + return self._purchase_factory is not None and bool( + self._device_address + ) + + def claim_capabilities(self) -> ClaimCapabilities: + """集中构造不会意外开放 live 的领取能力。""" + + supported = [TaskType.COLLECT] + if self.purchase_ready: + supported.append(TaskType.PURCHASE) + device = ( + AndroidDeviceInfo(self._device_address) + if self._device_address + else None + ) + return ClaimCapabilities( + device=device, + supported_types=tuple(supported), + purchase_mode="dry_run", + schema_versions=(1,), + ) + + def execute_one(self) -> TaskDispatchOutcome: + """严格按 Outbox、本地任务、Admin 新任务的顺序处理。""" + + pending = self._repository.next_pending_outbox() + if pending is not None: + return self._submit_pending(pending) + if not self._device_address: + raise ValueError("请先在设置页选择并保存 Android 设备") + + if not self.purchase_ready: + pending_purchase = self._repository.next_purchase_task() + if pending_purchase is not None: + raise RuntimeError( + f"本地采购任务 {pending_purchase.remote_task_id} 等待执行，" + "但采购演练执行器未就绪；已停止领取新任务" + ) + + task = self._repository.next_runnable_task( + include_purchase=self.purchase_ready + ) + if task is None: + remote = self._gateway.claim_next( + self._client, self.claim_capabilities() + ) + if remote is None: + return TaskDispatchOutcome("no_task", "暂无可领取的任务") + try: + self._repository.add_claimed_task( + admin_task_to_new_claimed_task(remote) + ) + 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 task.task_type is TaskType.COLLECT: + service = CollectTaskService( + self._gateway, + self._repository, + self._client, + self._device_address, + cancelled=self._cancelled, + collect_service_factory=self._collect_factory, + ) + outcome = service.execute_selected(task.remote_task_id) + else: + if self._purchase_factory is None: + raise RuntimeError( + "采购演练执行器未就绪，已停止领取以避免任务卡住" + ) + purchase_service = PurchaseTaskService( + self._gateway, + self._repository, + self._client, + self._device_address, + self._purchase_factory, + cancelled=self._cancelled, + ) + outcome = purchase_service.execute_selected(task.remote_task_id) + return TaskDispatchOutcome( + outcome.kind, outcome.message, outcome.task_id + ) + + def _submit_pending( + self, event: OutboxEventRecord + ) -> TaskDispatchOutcome: + """只补交已落库事件，不重新访问 PDD 页面。""" + + 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: + if exc.retryable: + self._repository.mark_outbox_retry(event.id, str(exc)) + return TaskDispatchOutcome( + "result_pending", + f"任务 {task_id} 结果已保存在本地，等待提交 Admin：{exc}", + task_id, + ) + self._repository.mark_outbox_failed(event.id, str(exc)) + return TaskDispatchOutcome( + "manual_review", + f"任务 {task_id} 结果被 Admin 拒绝：{exc}", + task_id, + ) + self._repository.mark_outbox_sent(event.id) + if event.event_type is OutboxEventType.TASK_FAILURE: + return TaskDispatchOutcome( + "failed", f"任务 {task_id} 失败信息已提交 Admin", task_id + ) + return TaskDispatchOutcome( + "succeeded", f"任务 {task_id} 结果已提交 Admin", task_id + ) diff --git a/client/src/task_repository.py b/client/src/task_repository.py index 6bbd0d2..01c3c49 100644 --- a/client/src/task_repository.py +++ b/client/src/task_repository.py @@ -224,6 +224,32 @@ class TaskRepository: connection.close() return self._to_detail(row) if row is not None else None + def next_runnable_task( + self, *, include_purchase: bool + ) -> Optional[TaskDetail]: + """按领取顺序返回当前执行器能够处理的最早本地任务。""" + + if include_purchase: + where = ( + "((task_type = 'collect'" + " AND status IN ('claimed', 'retry_wait'))" + " OR (task_type = 'purchase' AND status = 'claimed'))" + ) + else: + where = ( + "task_type = 'collect'" + " AND status IN ('claimed', 'retry_wait')" + ) + connection = open_database(self._db_path) + try: + row = connection.execute( + f"SELECT * FROM pdd_tasks WHERE {where}" + " 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 validate_collect_rerun(self, remote_task_id: str) -> TaskDetail: """校验任务能否重新采集，成功时返回任务详情。""" diff --git a/client/test/test_admin_gateway_contract.py b/client/test/test_admin_gateway_contract.py index a9a04f9..f21bd38 100644 --- a/client/test/test_admin_gateway_contract.py +++ b/client/test/test_admin_gateway_contract.py @@ -19,7 +19,8 @@ class MockAdminGatewayContractTest(unittest.TestCase): self.gateway = MockAdminGateway() self.client = ClientInfo("client-001") self.all_capabilities = ClaimCapabilities( - device=AndroidDeviceInfo("192.168.0.173:5555") + device=AndroidDeviceInfo("192.168.0.173:5555"), + supported_types=(TaskType.COLLECT, TaskType.PURCHASE), ) @staticmethod diff --git a/client/test/test_http_admin_gateway.py b/client/test/test_http_admin_gateway.py index 33e3847..96ec09f 100644 --- a/client/test/test_http_admin_gateway.py +++ b/client/test/test_http_admin_gateway.py @@ -14,6 +14,7 @@ from src.admin_gateway import ( ClientInfo, ) from src.http_admin_gateway import HttpAdminGateway +from src.task_models import TaskType class FakeResponse: @@ -88,7 +89,7 @@ class HttpAdminGatewayTest(unittest.TestCase): self.assertEqual(headers["authorization"], "Bearer secret-token") body = json.loads(opener.request.data.decode("utf-8")) self.assertEqual(body["client"]["name"], "办公室电脑") - self.assertEqual(body["supported_types"], ["collect", "purchase"]) + self.assertEqual(body["supported_types"], ["collect"]) self.assertEqual(body["device"]["platform"], "android") self.assertEqual(body["capabilities"]["purchase_mode"], "dry_run") self.assertEqual(opener.timeout, 2.5) @@ -218,7 +219,7 @@ class HttpAdminGatewayTest(unittest.TestCase): self.assertTrue(raised.exception.retryable) self.assertIn("可能已接收", str(raised.exception)) - def test_claim_maps_real_admin_payload_and_only_reports_collect(self): + def test_claim_maps_payload_and_serializes_confirmed_capabilities(self): opener = RecordingOpener( FakeResponse( 200, @@ -267,6 +268,56 @@ class HttpAdminGatewayTest(unittest.TestCase): self.assertEqual(body["supported_types"], ["collect"]) self.assertEqual(body["capabilities"]["purchase_mode"], "dry_run") + def test_claim_purchase_requires_all_safety_fields(self): + valid_task = { + "id": "PUR-001", + "type": "purchase", + "version": 1, + "priority": 0, + "payload": { + "goods_url": "https://example.test/goods/PUR-001", + "goods_id": "737116531267", + "options": {"color": "黑色", "size": "L"}, + "quantity": 2, + "max_price_cent": 4200, + }, + "created_at": "2026-08-09T08:00:00Z", + "updated_at": "2026-08-09T08:00:00Z", + } + gateway = HttpAdminGateway( + opener=RecordingOpener(FakeResponse(200, {"task": valid_task})) + ) + + task = gateway.claim_next( + ClientInfo("CLIENT-001"), + ClaimCapabilities( + supported_types=(TaskType.COLLECT, TaskType.PURCHASE), + purchase_mode="dry_run", + ), + ) + + self.assertEqual(task.task_type, TaskType.PURCHASE) + self.assertEqual(task.payload["max_price_cent"], 4200) + + for missing in ("goods_id", "options", "quantity", "max_price_cent"): + with self.subTest(missing=missing): + invalid_task = dict(valid_task) + invalid_payload = dict(valid_task["payload"]) + invalid_payload.pop(missing) + invalid_task["payload"] = invalid_payload + invalid_gateway = HttpAdminGateway( + opener=RecordingOpener( + FakeResponse(200, {"task": invalid_task}) + ) + ) + with self.assertRaises(AdminGatewayError) as raised: + invalid_gateway.claim_next( + ClientInfo("CLIENT-001"), self._capabilities() + ) + self.assertEqual( + raised.exception.code, "ADMIN_INVALID_RESPONSE" + ) + def test_claim_204_returns_none(self): gateway = HttpAdminGateway(opener=RecordingOpener(FakeResponse(204, {}))) diff --git a/client/test/test_pdd_ui_event.py b/client/test/test_pdd_ui_event.py index bc9faa5..208b5b3 100644 --- a/client/test/test_pdd_ui_event.py +++ b/client/test/test_pdd_ui_event.py @@ -20,6 +20,7 @@ from src.pdd_ui_event import ( summary_to_row, ) from src.mock_admin_gateway import MockAdminGateway +from src.pdd_purchase_adapter import PddPurchaseAdapter, PurchasePageState from src.settings_repository import SettingsRepository from src.task_models import NewClaimedTask, TaskStatus, TaskSummary, TaskType from src.task_repository import TaskRepository @@ -45,6 +46,12 @@ class BrokenSaveRepository(BrokenRepository): def next_collect_task(self): return None + def next_runnable_task(self, *, include_purchase): + return None + + def next_purchase_task(self): + return None + class RecordingClaimGateway: """记录领取参数并返回预设结果。""" @@ -148,6 +155,63 @@ def collect_admin_task(task_id="COL-001"): ) +def purchase_admin_task(task_id="PUR-001"): + return AdminTask( + task_id=task_id, + task_type=TaskType.PURCHASE, + version=1, + priority=10, + payload={ + "goods_id": "737116531267", + "goods_url": "https://example.test/737116531267", + "options": {"color": "黑色", "size": "L"}, + "quantity": 2, + "max_price_cent": 1200, + }, + created_at="2026-08-09T08:00:00Z", + updated_at="2026-08-09T08:00:00Z", + ) + + +class FakePurchaseAdapter(PddPurchaseAdapter): + def __init__(self): + self.options = {} + self.quantity = 0 + self.page_kind = "goods" + + def open_goods(self, _goods_url): + pass + + def read_state(self): + return PurchasePageState( + self.page_kind, + "737116531267", + dict(self.options), + self.quantity, + 990, + 1, + ) + + def select_options(self, options): + self.options = dict(options) + + def set_quantity(self, quantity): + self.quantity = quantity + + def enter_confirmation(self): + self.page_kind = "order_confirmation" + + def stop_before_submit(self): + pass + + def close(self): + pass + + +def fake_purchase_factory(_address, _cancelled): + return FakePurchaseAdapter() + + def wait_until(application, predicate, timeout=3.0): deadline = time.monotonic() + timeout while time.monotonic() < deadline: @@ -442,6 +506,35 @@ class PDDTaskPageEventTest(unittest.TestCase): events.shutdown() page.deleteLater() + def test_ready_runtime_claims_and_dispatches_purchase_in_worker(self): + gateway = RecordingClaimGateway(purchase_admin_task()) + page = PDDTaskPage() + events = PDDTaskPageEvent( + page, + self.repository, + claim_gateway=gateway, + settings_repository=self._saved_settings(), + collect_service_factory=fake_collect_factory, + purchase_adapter_factory=fake_purchase_factory, + ) + + page.autoFetchRequested.emit() + + self.assertTrue(wait_until(self.app, lambda: not events._claim_busy)) + _, capabilities = gateway.calls[0] + self.assertEqual( + capabilities.supported_types, + (TaskType.COLLECT, TaskType.PURCHASE), + ) + self.assertEqual(capabilities.purchase_mode, "dry_run") + detail = self.repository.get_task("PUR-001") + assert detail is not None + self.assertEqual(detail.task_type, TaskType.PURCHASE) + self.assertEqual(detail.status, TaskStatus.SUCCEEDED) + self.assertIn("演练完成", page.statusLabel.text()) + events.shutdown() + page.deleteLater() + def test_no_task_schedules_next_claim_and_stop_cancels_timer(self): gateway = RecordingClaimGateway(None) page = PDDTaskPage() diff --git a/client/test/test_task_dispatcher.py b/client/test/test_task_dispatcher.py new file mode 100644 index 0000000..cb20f43 --- /dev/null +++ b/client/test/test_task_dispatcher.py @@ -0,0 +1,215 @@ +"""统一任务分派器测试；不连接真实 Admin 或手机。""" + +import tempfile +import unittest +from pathlib import Path + +from src.admin_gateway import AdminTask, ClientInfo +from src.mock_admin_gateway import MockAdminGateway +from src.pdd_purchase_adapter import PddPurchaseAdapter, PurchasePageState +from src.task_dispatcher import TaskDispatcher, admin_task_to_new_claimed_task +from src.task_models import TaskStatus, TaskType +from src.task_repository import TaskRepository + + +class FakeCollectResult: + def to_pdd_data(self): + return { + "schema_version": 1, + "goods_id": "COL-GOODS", + "title": "采集测试商品", + "price_granularity": "color", + "dimensions": [], + "skus": [], + } + + +class RecordingCollector: + def __init__(self, calls): + self.calls = calls + + def collect(self, task): + self.calls.append(("collect", task.remote_task_id)) + return FakeCollectResult() + + +class ReadyPurchaseAdapter(PddPurchaseAdapter): + def __init__(self, calls): + self.calls = calls + self.options = {} + self.quantity = 0 + self.page_kind = "goods" + + def open_goods(self, goods_url): + self.calls.append(("purchase", "open", goods_url)) + + def read_state(self): + return PurchasePageState( + self.page_kind, + "PUR-GOODS", + dict(self.options), + self.quantity, + 990, + 1, + ) + + def select_options(self, options): + self.options = dict(options) + + def set_quantity(self, quantity): + self.quantity = quantity + + def enter_confirmation(self): + self.page_kind = "order_confirmation" + + def stop_before_submit(self): + self.calls.append(("purchase", "stopped")) + + def close(self): + pass + + +def purchase_task(task_id="PUR-001"): + return AdminTask( + task_id=task_id, + task_type=TaskType.PURCHASE, + version=1, + priority=10, + payload={ + "goods_url": "https://example.test/PUR-GOODS", + "goods_id": "PUR-GOODS", + "options": { + "color": "黑色", + "size": "L", + "bundle": "标准版", + }, + "quantity": 2, + "max_price_cent": 1200, + }, + ) + + +class TaskDispatcherTest(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.calls = [] + + def tearDown(self): + self.temporary.cleanup() + + def _dispatcher(self, *, purchase_ready): + purchase_factory = None + if purchase_ready: + purchase_factory = ( + lambda _address, _cancelled: ReadyPurchaseAdapter(self.calls) + ) + return TaskDispatcher( + self.gateway, + self.repository, + self.client, + "USB-001", + collect_service_factory=( + lambda *_args: RecordingCollector(self.calls) + ), + purchase_adapter_factory=purchase_factory, + ) + + def test_capability_only_includes_purchase_when_adapter_is_ready(self): + collect_only = self._dispatcher(purchase_ready=False) + ready = self._dispatcher(purchase_ready=True) + + self.assertEqual( + collect_only.claim_capabilities().supported_types, + (TaskType.COLLECT,), + ) + self.assertEqual( + ready.claim_capabilities().supported_types, + (TaskType.COLLECT, TaskType.PURCHASE), + ) + self.assertEqual( + ready.claim_capabilities().purchase_mode, "dry_run" + ) + + def test_claims_saves_then_dispatches_purchase_dry_run(self): + task = purchase_task() + self.gateway.enqueue_task(task, self.client.client_id) + + outcome = self._dispatcher(purchase_ready=True).execute_one() + + self.assertEqual(outcome.kind, "succeeded") + detail = self.repository.get_task(task.task_id) + assert detail is not None + self.assertEqual(detail.task_type, TaskType.PURCHASE) + self.assertEqual(detail.status, TaskStatus.SUCCEEDED) + self.assertEqual( + detail.admin_payload["payload"]["options"]["bundle"], + "标准版", + ) + self.assertIn(("purchase", "stopped"), self.calls) + + def test_purchase_is_not_claimed_when_runtime_is_not_ready(self): + self.gateway.enqueue_task(purchase_task(), self.client.client_id) + + outcome = self._dispatcher(purchase_ready=False).execute_one() + + self.assertEqual(outcome.kind, "no_task") + self.assertEqual(self.repository.count_tasks(), 0) + + def test_local_purchase_stops_dispatch_when_runtime_is_not_ready(self): + task = purchase_task() + self.repository.add_claimed_task( + admin_task_to_new_claimed_task(task) + ) + + with self.assertRaisesRegex(RuntimeError, "停止领取新任务"): + self._dispatcher(purchase_ready=False).execute_one() + + detail = self.repository.get_task(task.task_id) + assert detail is not None + self.assertEqual(detail.status, TaskStatus.CLAIMED) + + def test_purchase_mapper_rejects_missing_or_invalid_safety_fields(self): + invalid_payloads = ( + {"goods_url": "https://example.test"}, + { + "goods_url": "https://example.test", + "goods_id": "G", + "options": {}, + "quantity": 1, + "max_price_cent": 100, + }, + { + "goods_url": "https://example.test", + "goods_id": "G", + "options": {"size": "L"}, + "quantity": 0, + "max_price_cent": 100, + }, + { + "goods_url": "https://example.test", + "goods_id": "G", + "options": {"size": "L"}, + "quantity": 1, + "max_price_cent": 0, + }, + ) + for index, payload in enumerate(invalid_payloads): + with self.subTest(index=index): + task = AdminTask( + f"PUR-BAD-{index}", + TaskType.PURCHASE, + 1, + 0, + payload, + ) + with self.assertRaises(ValueError): + admin_task_to_new_claimed_task(task) + + +if __name__ == "__main__": + unittest.main() diff --git a/docs/client/02-architecture.md b/docs/client/02-architecture.md index 5bf86ed..da233b4 100644 --- a/docs/client/02-architecture.md +++ b/docs/client/02-architecture.md @@ -89,7 +89,7 @@ client/ | 目录分层 | domain / application / infrastructure / workers / ui | 平铺在 `src/` 下：`db.py`、`db_schema.py`、`task_repository.py`、`settings_repository.py`、`task_models.py` 等，另有 `src/util/`、`src/demo1/` | | 主按钮文案 | 「开始自动获取」⇄「停止自动获取」 | 已按持续串行模式实现 | | Admin 网关 | `AdminGateway` + Mock/HTTP 两实现 | 登记、领取、结果和失败提交已实现 | -| 任务应用服务 | `CollectTaskService` / `PurchaseTaskService` | 采集已接入自动获取；采购已实现本地 `dry_run` 演练，领取与分派由后续工单接入 | +| 任务应用服务 | `TaskDispatcher` / `CollectTaskService` / `PurchaseTaskService` | 自动获取先补交 Outbox，再按领取时间执行本地任务，最后按安全能力领取并分派新任务 | | Outbox 提交 | 从 `outbox_events` 取件重试 | 采集结果与失败已实现，重试不重复采集 | | PDD 自动化 | `infrastructure/pdd/` 适配层 | `pdd_device_service.py` 与 `pdd_collect_service.py` 已接入采集主链 | @@ -330,6 +330,11 @@ Client 与 Admin 的任务交互只有三种调用：领一个任务、提交结 6. Outbox 提交成功、Admin 返回 `accepted: true` 后标记 `succeeded`。 7. 回到第 1 步领下一个任务。**同一时间只做一个任务。** +`TaskDispatcher` 是能力声明的唯一入口。没有可用采购演练 Adapter、没有已保存 +Android 设备或本地持久化未准备好时，只声明 `collect`；条件满足时才声明 +`collect,purchase`，且 `purchase_mode` 永远是 `dry_run`。领取响应必须先完整校验并 +写入 SQLite，Repository 提交成功后才能分派，避免任务已在 Admin 领取却在本地丢失。 + 中途 Admin 是否取消了这个任务、是否重派给了别人，Client 不查也不管，做完照样提交—— Admin 侧必须无条件接受，见 [04](04-admin-api-contract.md) §6.1。 diff --git a/docs/client/04-admin-api-contract.md b/docs/client/04-admin-api-contract.md index 6becdff..be2aa01 100644 --- a/docs/client/04-admin-api-contract.md +++ b/docs/client/04-admin-api-contract.md @@ -199,10 +199,16 @@ POST /api/v1/client/tasks/claim 无可领取任务时返回 `204 No Content`。 -Client 的 `HttpAdminGateway.claim_next` 已实现本接口。当前只声明 -`supported_types: ["collect"]`，一次调用最多领取一个采集任务；采集结束后通过 +Client 的 `HttpAdminGateway.claim_next` 已实现本接口，并原样序列化应用层已经 +安全确认的 `ClaimCapabilities`。没有采购演练 Adapter 或设备未准备好时只声明 +`supported_types: ["collect"]`；条件满足时声明 `collect,purchase`，但 +`purchase_mode` 固定为 `dry_run`。一次调用最多领取一个任务，结果或失败通过 本页 §6 或 §7 提交，完整请求会先进入本地 Outbox。 +采购响应在写入本地前必须校验非空 `goods_url`、`goods_id`、动态 `options`、 +正整数 `quantity` 和正整数分 `max_price_cent`。字段不完整时停止新的领取并向 +操作员显示协议错误；不得让不完整任务进入手机执行。 + `[必须]` **`204` 不是错误。** Client 要把它当"暂时没活干"处理， 不要报错，也不要因此触发重试风暴。首次 claim 通常返回 204；如果该编号已经预先分配任务，也可以直接返回 200。 diff --git a/docs/client/05-ui-specification.md b/docs/client/05-ui-specification.md index f601060..764d2ca 100644 --- a/docs/client/05-ui-specification.md +++ b/docs/client/05-ui-specification.md @@ -64,7 +64,8 @@ 才向 Admin 领取一条。领取、手机采集和提交都在单独的工作线程完成，一轮结束且 线程完全退出后，由主线程的单次定时器安排下一轮。暂无任务时默认 5 秒后重试； 可恢复的 Admin 错误按 5、10、20、30 秒退避。需要人工处理、不可恢复错误、设备 -或配置错误会停止自动获取。当前仍不执行采购任务。 +或配置错误会停止自动获取。采购演练执行器和 Android 设备都准备好时，可以领取 +采购任务；界面必须明确显示“演练”，任何路径都不得显示成真实下单。 ### 4.2 搜索与筛选 @@ -86,6 +87,7 @@ - 确认后只执行选中的稳定任务编号，不领取新任务，不先处理其他任务或 Outbox。 - 重新采集在工作线程运行。执行期间禁用“获取任务”和“重新执行”，完成后刷新列表。 - 每次重新采集创建新的执行记录和幂等键；当前结果更新，旧结果保存在历史执行记录中。 +- 采购任务始终不能通过“重新执行”入口启动；需要处理时由自动获取的安全恢复流程决定。 - “等待重试”当前没有倒计时。自动获取因可恢复采集错误停止时，底部状态显示 “重试已暂停”，并提示选择任务点击“重新执行”或重新启动获取任务。