"""串行领取、落库并按任务类型分派一条任务。""" 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 )