605 lines
22 KiB
Python
605 lines
22 KiB
Python
"""一条本地采购任务的安全执行流程。
|
|
|
|
``dry_run`` 始终停在提交前；``live`` 只在独立 Adapter 就绪时允许一次提交，
|
|
并且必须先持久化不可逆标记。任何路径都不包含付款。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from typing import Callable, Mapping
|
|
|
|
from .admin_gateway import AdminGateway, AdminGatewayError, ClientInfo
|
|
from .pdd_purchase_adapter import (
|
|
PddLivePurchaseAdapter,
|
|
PddPurchaseAdapter,
|
|
PddPurchaseError,
|
|
PurchasePageState,
|
|
)
|
|
from .task_models import OutboxEventRecord, OutboxEventType, TaskStatus
|
|
from .task_repository import TaskRepository
|
|
|
|
|
|
def utc_now_iso() -> str:
|
|
"""返回精确到秒的 UTC ISO 8601 时间。"""
|
|
|
|
return datetime.now(timezone.utc).isoformat(timespec="seconds").replace(
|
|
"+00:00", "Z"
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PurchaseTarget:
|
|
"""从已领取任务中校验出的采购目标。"""
|
|
|
|
goods_id: str
|
|
goods_url: str
|
|
options: Mapping[str, str]
|
|
quantity: int
|
|
max_price_cent: int
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PurchaseTaskOutcome:
|
|
"""采购工作线程返回给协调层的简短结果。"""
|
|
|
|
kind: str
|
|
message: str
|
|
task_id: str = ""
|
|
|
|
|
|
PurchaseAdapterFactory = Callable[
|
|
[str, Callable[[], bool]], PddPurchaseAdapter
|
|
]
|
|
LivePurchaseAdapterFactory = Callable[
|
|
[str, Callable[[], bool]], PddLivePurchaseAdapter
|
|
]
|
|
|
|
|
|
class PurchaseTaskService:
|
|
"""只执行本地已保存的采购任务，不领取新任务。"""
|
|
|
|
def __init__(
|
|
self,
|
|
gateway: AdminGateway,
|
|
repository: TaskRepository,
|
|
client: ClientInfo,
|
|
device_address: str,
|
|
adapter_factory: PurchaseAdapterFactory,
|
|
*,
|
|
cancelled: Callable[[], bool] = lambda: False,
|
|
) -> None:
|
|
self._gateway = gateway
|
|
self._repository = repository
|
|
self._client = client
|
|
self._device_address = str(device_address or "").strip()
|
|
self._factory = adapter_factory
|
|
self._cancelled = cancelled
|
|
|
|
def execute_one_local(self) -> PurchaseTaskOutcome:
|
|
"""先补交结果，再执行最早的本地待执行采购任务。"""
|
|
|
|
pending = self._repository.next_pending_outbox()
|
|
if pending is not None:
|
|
return self._submit(pending)
|
|
if not self._device_address:
|
|
raise ValueError("请先在设置页选择并保存 Android 设备")
|
|
task = self._repository.next_purchase_task()
|
|
if task is None:
|
|
return PurchaseTaskOutcome("no_task", "暂无本地待执行的采购任务")
|
|
return self.execute_selected(task.remote_task_id)
|
|
|
|
def execute_selected(self, remote_task_id: str) -> PurchaseTaskOutcome:
|
|
"""按任务不可变执行模式安全执行一次采购。"""
|
|
|
|
if not self._device_address:
|
|
raise ValueError("请先在设置页选择并保存 Android 设备")
|
|
if self._cancelled():
|
|
return PurchaseTaskOutcome(
|
|
"cancelled", "采购演练已在开始前取消", remote_task_id
|
|
)
|
|
|
|
started = self._repository.start_purchase_run(
|
|
remote_task_id, self._device_address
|
|
)
|
|
adapter: PddPurchaseAdapter | None = None
|
|
step = "purchase_prepare"
|
|
business_failure_message = ""
|
|
try:
|
|
target = self._target_from_task(started.task)
|
|
adapter = self._factory(self._device_address, self._cancelled)
|
|
execution_mode = started.task.execution_mode
|
|
if execution_mode == "live" and not isinstance(
|
|
adapter, PddLivePurchaseAdapter
|
|
):
|
|
raise PddPurchaseError(
|
|
"PURCHASE_LIVE_ADAPTER_UNAVAILABLE",
|
|
"真实采购执行器未就绪，未进入商品页面",
|
|
step="purchase_prepare",
|
|
)
|
|
|
|
step = "purchase_open_goods"
|
|
self._enter_step(remote_task_id, started.attempt_id, step)
|
|
adapter.open_goods(target.goods_url)
|
|
|
|
step = "purchase_verify_goods"
|
|
self._enter_step(remote_task_id, started.attempt_id, step)
|
|
state = adapter.read_state()
|
|
self._validate_identity(state, target, "goods")
|
|
|
|
step = "purchase_select_options"
|
|
self._enter_step(remote_task_id, started.attempt_id, step)
|
|
self._validate_identity(adapter.read_state(), target, "goods")
|
|
adapter.select_options(target.options)
|
|
|
|
step = "purchase_verify_options"
|
|
self._enter_step(remote_task_id, started.attempt_id, step)
|
|
state = adapter.read_state()
|
|
self._validate_selected_options(state, target)
|
|
|
|
step = "purchase_set_quantity"
|
|
self._enter_step(remote_task_id, started.attempt_id, step)
|
|
self._validate_selected_options(adapter.read_state(), target)
|
|
adapter.set_quantity(target.quantity)
|
|
|
|
step = "purchase_verify_quantity_price"
|
|
self._enter_step(remote_task_id, started.attempt_id, step)
|
|
state = adapter.read_state()
|
|
self._validate_checkout_values(state, target)
|
|
|
|
step = "purchase_enter_confirmation"
|
|
self._enter_step(remote_task_id, started.attempt_id, step)
|
|
self._validate_checkout_values(adapter.read_state(), target)
|
|
adapter.enter_confirmation()
|
|
|
|
step = "purchase_verify_confirmation"
|
|
self._enter_step(remote_task_id, started.attempt_id, step)
|
|
state = adapter.read_state()
|
|
if state.page_kind != "order_confirmation":
|
|
raise PddPurchaseError(
|
|
"PURCHASE_CONFIRMATION_NOT_REACHED",
|
|
"没有到达最终提交前确认页，已停止演练",
|
|
step=step,
|
|
)
|
|
self._validate_checkout_values(state, target)
|
|
|
|
if execution_mode == "live":
|
|
assert isinstance(adapter, PddLivePurchaseAdapter)
|
|
return self._submit_live_once(
|
|
remote_task_id,
|
|
started.attempt_id,
|
|
target,
|
|
adapter,
|
|
)
|
|
|
|
step = "purchase_dry_run_stopped"
|
|
self._enter_step(remote_task_id, started.attempt_id, step)
|
|
adapter.stop_before_submit()
|
|
result = self._result_data(target, state)
|
|
event = self._repository.save_purchase_result(
|
|
remote_task_id, started.attempt_id, result
|
|
)
|
|
except PddPurchaseError as exc:
|
|
event = self._save_failure(
|
|
remote_task_id,
|
|
started.attempt_id,
|
|
exc.code,
|
|
exc.message,
|
|
exc.retryable,
|
|
step,
|
|
exc.diagnostics,
|
|
)
|
|
if exc.code == "PDD_GOODS_UNAVAILABLE":
|
|
business_failure_message = exc.message
|
|
except Exception as exc:
|
|
event = self._save_failure(
|
|
remote_task_id,
|
|
started.attempt_id,
|
|
"PURCHASE_UNEXPECTED",
|
|
f"采购演练在“{step}”发生未知错误：{exc}",
|
|
False,
|
|
step,
|
|
{},
|
|
)
|
|
finally:
|
|
if adapter is not None:
|
|
adapter.close()
|
|
outcome = self._submit(event)
|
|
if business_failure_message and outcome.kind == "failed":
|
|
return PurchaseTaskOutcome(
|
|
"business_failed", business_failure_message, remote_task_id
|
|
)
|
|
return outcome
|
|
|
|
def _submit_live_once(
|
|
self,
|
|
remote_task_id: str,
|
|
attempt_id: str,
|
|
target: PurchaseTarget,
|
|
adapter: PddLivePurchaseAdapter,
|
|
) -> PurchaseTaskOutcome:
|
|
"""最终复核、先落不可逆标记、单击一次，然后只转核单。"""
|
|
|
|
step = "purchase_live_final_check"
|
|
self._enter_step(remote_task_id, attempt_id, step)
|
|
state = adapter.read_state()
|
|
self._validate_live_confirmation(state, target)
|
|
if self._cancelled():
|
|
raise PddPurchaseError(
|
|
"PURCHASE_CANCELLED",
|
|
"用户已在不可逆标记写入前停止真实采购",
|
|
step=step,
|
|
)
|
|
|
|
irreversible_at = self._repository.mark_purchase_irreversible(
|
|
remote_task_id,
|
|
attempt_id,
|
|
{
|
|
"options": dict(state.selected_options),
|
|
"quantity": state.quantity,
|
|
"unit_price_cent": state.price_cent,
|
|
"total_price_cent": state.price_cent * state.quantity,
|
|
},
|
|
)
|
|
submitted_at = None
|
|
message = "订单提交点击已执行，结果待只读核对；绝不重新下单"
|
|
try:
|
|
adapter.submit_order_once()
|
|
submitted_at = utc_now_iso()
|
|
except Exception as exc:
|
|
message = (
|
|
"订单提交点击结果不确定，已转只读核对；"
|
|
f"绝不重新下单：{str(exc) or type(exc).__name__}"
|
|
)
|
|
|
|
try:
|
|
self._repository.move_purchase_to_reconcile(
|
|
remote_task_id,
|
|
attempt_id,
|
|
order_submitted_at=submitted_at,
|
|
message=message,
|
|
)
|
|
except Exception as exc:
|
|
# 不可逆标记已经提交成功。这里绝不能保存成普通失败或再次执行；
|
|
# 重启恢复会根据 irreversible_action_at 强制进入只读核单。
|
|
return PurchaseTaskOutcome(
|
|
"manual_review",
|
|
(
|
|
f"任务 {remote_task_id} 已在 {irreversible_at} 进入不可逆阶段，"
|
|
f"但核单状态保存失败：{exc}；请关闭自动获取并重启，绝不重下"
|
|
),
|
|
remote_task_id,
|
|
)
|
|
return PurchaseTaskOutcome(
|
|
"reconcile_pending",
|
|
f"任务 {remote_task_id} {message}",
|
|
remote_task_id,
|
|
)
|
|
|
|
def _enter_step(
|
|
self, remote_task_id: str, attempt_id: str, step: str
|
|
) -> None:
|
|
"""先持久化步骤，再检查停止请求。"""
|
|
|
|
self._repository.update_purchase_step(
|
|
remote_task_id, attempt_id, step
|
|
)
|
|
if self._cancelled():
|
|
raise PddPurchaseError(
|
|
"PURCHASE_CANCELLED",
|
|
"用户已请求停止采购演练",
|
|
step=step,
|
|
)
|
|
|
|
@staticmethod
|
|
def _target_from_task(task: object) -> PurchaseTarget:
|
|
payload_root = getattr(task, "admin_payload", None)
|
|
if not isinstance(payload_root, Mapping):
|
|
raise PddPurchaseError(
|
|
"PURCHASE_TASK_INVALID",
|
|
"采购任务原始数据不是对象",
|
|
step="purchase_prepare",
|
|
)
|
|
payload = payload_root.get("payload")
|
|
if not isinstance(payload, Mapping):
|
|
raise PddPurchaseError(
|
|
"PURCHASE_TASK_INVALID",
|
|
"采购任务缺少 payload 对象",
|
|
step="purchase_prepare",
|
|
)
|
|
goods_url = str(payload.get("goods_url") or "").strip()
|
|
goods_id = str(payload.get("goods_id") or "").strip()
|
|
options = payload.get("options")
|
|
quantity = payload.get("quantity")
|
|
max_price_cent = payload.get("max_price_cent")
|
|
if not goods_url or not goods_id:
|
|
raise PddPurchaseError(
|
|
"PURCHASE_TASK_INVALID",
|
|
"采购任务缺少商品链接或商品编号",
|
|
step="purchase_prepare",
|
|
)
|
|
if not isinstance(options, Mapping) or not options:
|
|
raise PddPurchaseError(
|
|
"PURCHASE_TASK_INVALID",
|
|
"采购任务缺少动态规格 options",
|
|
step="purchase_prepare",
|
|
)
|
|
normalized_options: dict[str, str] = {}
|
|
for key, value in options.items():
|
|
checked_key = str(key or "").strip()
|
|
checked_value = str(value or "").strip()
|
|
if not checked_key or not checked_value:
|
|
raise PddPurchaseError(
|
|
"PURCHASE_TASK_INVALID",
|
|
"采购任务 options 的名称和值都不能为空",
|
|
step="purchase_prepare",
|
|
)
|
|
normalized_options[checked_key] = checked_value
|
|
if isinstance(quantity, bool) or not isinstance(quantity, int) or quantity <= 0:
|
|
raise PddPurchaseError(
|
|
"PURCHASE_TASK_INVALID",
|
|
"采购数量必须是大于 0 的整数",
|
|
step="purchase_prepare",
|
|
)
|
|
if (
|
|
isinstance(max_price_cent, bool)
|
|
or not isinstance(max_price_cent, int)
|
|
or max_price_cent <= 0
|
|
):
|
|
raise PddPurchaseError(
|
|
"PURCHASE_TASK_INVALID",
|
|
"人民币价格上限必须是大于 0 的整数分",
|
|
step="purchase_prepare",
|
|
)
|
|
return PurchaseTarget(
|
|
goods_id,
|
|
goods_url,
|
|
normalized_options,
|
|
quantity,
|
|
max_price_cent,
|
|
)
|
|
|
|
@staticmethod
|
|
def _validate_identity(
|
|
state: PurchasePageState, target: PurchaseTarget, expected_page: str
|
|
) -> None:
|
|
PurchaseTaskService._validate_common_state(state)
|
|
if state.page_kind != expected_page:
|
|
raise PddPurchaseError(
|
|
"PDD_PAGE_UNKNOWN",
|
|
f"当前页面不是预期的 {expected_page} 页面，已停止演练",
|
|
step="purchase_verify_goods",
|
|
)
|
|
if state.goods_id != target.goods_id:
|
|
raise PddPurchaseError(
|
|
"PURCHASE_GOODS_MISMATCH",
|
|
"当前 PDD 商品与采购任务不一致，已停止演练",
|
|
step="purchase_verify_goods",
|
|
)
|
|
|
|
@staticmethod
|
|
def _validate_selected_options(
|
|
state: PurchasePageState, target: PurchaseTarget
|
|
) -> None:
|
|
PurchaseTaskService._validate_common_state(state)
|
|
if state.goods_id != target.goods_id:
|
|
raise PddPurchaseError(
|
|
"PURCHASE_GOODS_MISMATCH",
|
|
"选择规格后商品编号发生变化，已停止演练",
|
|
step="purchase_verify_options",
|
|
)
|
|
if dict(state.selected_options) != dict(target.options):
|
|
raise PddPurchaseError(
|
|
"PURCHASE_OPTIONS_MISMATCH",
|
|
"当前选中规格与采购任务不完全一致，已停止演练",
|
|
step="purchase_verify_options",
|
|
)
|
|
|
|
@staticmethod
|
|
def _validate_checkout_values(
|
|
state: PurchasePageState, target: PurchaseTarget
|
|
) -> None:
|
|
PurchaseTaskService._validate_selected_options(state, target)
|
|
if state.quantity != target.quantity:
|
|
raise PddPurchaseError(
|
|
"PURCHASE_QUANTITY_MISMATCH",
|
|
"当前数量与采购任务不一致，已停止演练",
|
|
step="purchase_verify_quantity_price",
|
|
)
|
|
if state.price_cent <= 0:
|
|
raise PddPurchaseError(
|
|
"PURCHASE_PRICE_MISSING",
|
|
"无法读取稳定的当前人民币价格，已停止演练",
|
|
step="purchase_verify_quantity_price",
|
|
)
|
|
if state.price_cent > target.max_price_cent:
|
|
raise PddPurchaseError(
|
|
"PURCHASE_PRICE_EXCEEDED",
|
|
(
|
|
f"当前价格 {state.price_cent} 分超过上限 "
|
|
f"{target.max_price_cent} 分，已停止演练"
|
|
),
|
|
step="purchase_verify_quantity_price",
|
|
)
|
|
|
|
@staticmethod
|
|
def _validate_live_confirmation(
|
|
state: PurchasePageState, target: PurchaseTarget
|
|
) -> None:
|
|
"""在不可逆标记前核对最新确认页和唯一提交目标。"""
|
|
|
|
if state.page_kind != "order_confirmation":
|
|
raise PddPurchaseError(
|
|
"PURCHASE_CONFIRMATION_LOST",
|
|
"最终提交前确认页已变化，禁止真实下单",
|
|
step="purchase_live_final_check",
|
|
)
|
|
PurchaseTaskService._validate_checkout_values(state, target)
|
|
if not state.in_stock:
|
|
raise PddPurchaseError(
|
|
"PURCHASE_OUT_OF_STOCK",
|
|
"当前规格库存不足或已售罄，禁止真实下单",
|
|
step="purchase_live_final_check",
|
|
)
|
|
if state.submit_candidate_count != 1:
|
|
raise PddPurchaseError(
|
|
"PURCHASE_SUBMIT_TARGET_AMBIGUOUS",
|
|
"最终提交按钮不是唯一可靠目标，禁止真实下单",
|
|
step="purchase_live_final_check",
|
|
diagnostics={
|
|
"candidate_count": state.submit_candidate_count
|
|
},
|
|
)
|
|
|
|
@staticmethod
|
|
def _validate_common_state(state: PurchasePageState) -> None:
|
|
if state.page_kind == "captcha":
|
|
raise PddPurchaseError(
|
|
"PDD_PAGE_CAPTCHA",
|
|
"PDD 出现安全验证，请人工处理",
|
|
step="purchase_page_check",
|
|
)
|
|
if state.page_kind == "login_required":
|
|
raise PddPurchaseError(
|
|
"PDD_PAGE_LOGIN_REQUIRED",
|
|
"PDD 登录状态失效，请人工登录后再处理",
|
|
step="purchase_page_check",
|
|
)
|
|
if state.page_kind in {"unknown", "risk_control", "payment"}:
|
|
raise PddPurchaseError(
|
|
"PDD_PAGE_UNKNOWN",
|
|
"PDD 出现未知、风控或支付页面，已停止演练",
|
|
step="purchase_page_check",
|
|
)
|
|
if state.candidate_count != 1:
|
|
raise PddPurchaseError(
|
|
"PURCHASE_AMBIGUOUS_TARGET",
|
|
"页面存在多个候选目标，无法安全确认，已停止演练",
|
|
step="purchase_page_check",
|
|
)
|
|
|
|
def _save_failure(
|
|
self,
|
|
remote_task_id: str,
|
|
attempt_id: str,
|
|
code: str,
|
|
message: str,
|
|
retryable: bool,
|
|
step: str,
|
|
diagnostics: Mapping[str, object],
|
|
) -> OutboxEventRecord:
|
|
if code == "PURCHASE_CANCELLED":
|
|
status = TaskStatus.CANCELLED
|
|
elif code == "PDD_GOODS_UNAVAILABLE":
|
|
status = TaskStatus.FAILED
|
|
else:
|
|
# 采购没有手动“重新执行”入口。即使错误属于
|
|
# 技术上可重试，也先留给人工判断，避免隐式重复采购。
|
|
status = TaskStatus.MANUAL_REVIEW
|
|
return self._repository.save_purchase_failure(
|
|
remote_task_id,
|
|
attempt_id,
|
|
status,
|
|
code,
|
|
message,
|
|
retryable,
|
|
step,
|
|
dict(diagnostics),
|
|
)
|
|
|
|
def _submit(self, event: OutboxEventRecord) -> PurchaseTaskOutcome:
|
|
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 PurchaseTaskOutcome(
|
|
"result_pending",
|
|
f"任务 {task_id} 演练结果已保存在本地，等待提交 Admin：{message}",
|
|
task_id,
|
|
)
|
|
self._repository.mark_outbox_failed(event.id, message)
|
|
return PurchaseTaskOutcome(
|
|
"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 PurchaseTaskOutcome(
|
|
"failed",
|
|
f"任务 {task_id} 采购演练未完成，失败信息已提交 Admin",
|
|
task_id,
|
|
)
|
|
purchase = event.payload_json.get("purchase")
|
|
if isinstance(purchase, Mapping) and purchase.get("mode") == "live":
|
|
return PurchaseTaskOutcome(
|
|
"succeeded",
|
|
f"任务 {task_id} 已核对订单并提交 Admin",
|
|
task_id,
|
|
)
|
|
return PurchaseTaskOutcome(
|
|
"succeeded",
|
|
f"任务 {task_id} 演练完成并已提交 Admin；没有提交订单",
|
|
task_id,
|
|
)
|
|
|
|
def submit_saved_event(self, event: OutboxEventRecord) -> PurchaseTaskOutcome:
|
|
"""提交核单后已经落库的 Outbox，供批量采购闭环复用。"""
|
|
|
|
return self._submit(event)
|
|
|
|
def _result_data(
|
|
self, target: PurchaseTarget, state: PurchasePageState
|
|
) -> dict[str, object]:
|
|
return {
|
|
"schema_version": 1,
|
|
"goods_id": target.goods_id,
|
|
"goods_url": target.goods_url,
|
|
"purchase": {
|
|
"mode": "dry_run",
|
|
"requested": {
|
|
"options": dict(target.options),
|
|
"quantity": target.quantity,
|
|
"max_price_cent": target.max_price_cent,
|
|
},
|
|
"confirmed": {
|
|
"options": dict(state.selected_options),
|
|
"quantity": state.quantity,
|
|
"unit_price_cent": state.price_cent,
|
|
"total_price_cent": state.price_cent * state.quantity,
|
|
},
|
|
"confirmation_reached": True,
|
|
"order_submitted": False,
|
|
"payment_attempted": False,
|
|
"order_no": None,
|
|
"ordered_at": None,
|
|
"ordered_at_raw": None,
|
|
"match_status": "not_submitted",
|
|
},
|
|
"captured_at": utc_now_iso(),
|
|
"source": {
|
|
"client_id": self._client.client_id,
|
|
"device_address": self._device_address,
|
|
"mode": "dry_run",
|
|
},
|
|
}
|