diff --git a/client/src/admin_gateway.py b/client/src/admin_gateway.py
index f856639..cda7a6f 100644
--- a/client/src/admin_gateway.py
+++ b/client/src/admin_gateway.py
@@ -56,8 +56,8 @@ class ClaimCapabilities:
raise ValueError("supported_types 不能为空")
if any(not isinstance(value, TaskType) for value in self.supported_types):
raise ValueError("supported_types 必须使用 TaskType")
- if self.purchase_mode != "dry_run":
- raise ValueError("当前版本只允许 purchase_mode=dry_run")
+ if self.purchase_mode not in {"dry_run", "live"}:
+ raise ValueError("purchase_mode 只允许 dry_run 或 live")
if not self.schema_versions or any(
version <= 0 for version in self.schema_versions
):
@@ -75,6 +75,7 @@ class AdminTask:
payload: Mapping[str, Any] = field(default_factory=dict)
created_at: str = ""
updated_at: str = ""
+ execution_mode: str = "dry_run"
def __post_init__(self) -> None:
if not self.task_id.strip():
@@ -83,6 +84,13 @@ class AdminTask:
raise ValueError("task_type 必须使用 TaskType")
if self.version <= 0:
raise ValueError("version 必须大于 0")
+ if self.execution_mode not in {"dry_run", "live"}:
+ raise ValueError("execution_mode 只允许 dry_run 或 live")
+ if (
+ self.task_type is not TaskType.PURCHASE
+ and self.execution_mode != "dry_run"
+ ):
+ raise ValueError("只有采购任务允许 execution_mode=live")
if not isinstance(self.payload, Mapping):
raise ValueError("payload 必须是对象")
diff --git a/client/src/db_schema.py b/client/src/db_schema.py
index 2efe68b..0220da4 100644
--- a/client/src/db_schema.py
+++ b/client/src/db_schema.py
@@ -4,7 +4,7 @@
``MIGRATIONS`` 末尾增加版本，不能修改已经发布的迁移。
"""
-SCHEMA_VERSION = 2
+SCHEMA_VERSION = 3
MIGRATION_1 = (
@@ -134,7 +134,16 @@ MIGRATION_2 = (
)
+MIGRATION_3 = (
+ """
+ ALTER TABLE pdd_tasks ADD COLUMN execution_mode TEXT NOT NULL DEFAULT 'dry_run'
+ CHECK (execution_mode IN ('dry_run', 'live'))
+ """,
+)
+
+
MIGRATIONS = {
1: MIGRATION_1,
2: MIGRATION_2,
+ 3: MIGRATION_3,
}
diff --git a/client/src/http_admin_gateway.py b/client/src/http_admin_gateway.py
index 0211d60..867b827 100644
--- a/client/src/http_admin_gateway.py
+++ b/client/src/http_admin_gateway.py
@@ -361,6 +361,7 @@ class HttpAdminGateway(AdminGateway):
raw_type = task.get("type")
version = task.get("version")
priority = task.get("priority")
+ execution_mode = task.get("execution_mode", "dry_run")
task_payload = task.get("payload")
created_at = task.get("created_at")
updated_at = task.get("updated_at")
@@ -373,6 +374,7 @@ class HttpAdminGateway(AdminGateway):
and version > 0
and isinstance(priority, int)
and not isinstance(priority, bool)
+ and isinstance(execution_mode, str)
and isinstance(task_payload, Mapping)
and isinstance(created_at, str)
and isinstance(updated_at, str)
@@ -395,6 +397,16 @@ class HttpAdminGateway(AdminGateway):
request_id,
) from exc
+ if execution_mode not in {"dry_run", "live"} or (
+ task_type is not TaskType.PURCHASE and execution_mode != "dry_run"
+ ):
+ raise AdminGatewayError(
+ "ADMIN_INVALID_RESPONSE",
+ "Admin 返回了不支持的任务执行模式",
+ False,
+ request_id,
+ )
+
cls._validate_claim_payload(task_type, task_payload, request_id)
return AdminTask(
@@ -402,6 +414,7 @@ class HttpAdminGateway(AdminGateway):
task_type=task_type,
version=version,
priority=priority,
+ execution_mode=execution_mode,
payload=dict(task_payload),
created_at=created_at,
updated_at=updated_at,
diff --git a/client/src/live_purchase_authorization.py b/client/src/live_purchase_authorization.py
new file mode 100644
index 0000000..a92daef
--- /dev/null
+++ b/client/src/live_purchase_authorization.py
@@ -0,0 +1,126 @@
+"""真实下单的本地授权状态。
+
+授权只来自设置页的明确确认，并绑定当前 Client ID 和 Android 设备。
+缺少、损坏或不匹配的设置一律按关闭处理。
+"""
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+
+from .settings_repository import SettingsRepository
+
+
+LIVE_ENABLED_KEY = "purchase.live_enabled"
+LIVE_CLIENT_ID_KEY = "purchase.live_client_id"
+LIVE_DEVICE_SERIAL_KEY = "purchase.live_device_serial"
+LIVE_CONFIRMED_AT_KEY = "purchase.live_confirmed_at"
+LIVE_CONFIRMATION_TEXT = "创建未付款订单"
+
+
+def _utc_now_iso() -> str:
+ return datetime.now(timezone.utc).isoformat(timespec="seconds").replace(
+ "+00:00", "Z"
+ )
+
+
+@dataclass(frozen=True)
+class LivePurchaseAuthorization:
+ """设置页展示和运行时核对使用的授权快照。"""
+
+ enabled: bool = False
+ client_id: str = ""
+ device_serial: str = ""
+ confirmed_at: str = ""
+
+ def matches(self, client_id: str, device_serial: str) -> bool:
+ """只有完整授权和两项绑定完全一致时才返回 True。"""
+
+ return (
+ self.enabled
+ and bool(self.confirmed_at)
+ and self.client_id == str(client_id or "").strip()
+ and self.device_serial == str(device_serial or "").strip()
+ )
+
+
+class LivePurchaseAuthorizationService:
+ """保存、关闭并校验本机真实下单授权。"""
+
+ def __init__(self, repository: SettingsRepository):
+ self._repository = repository
+
+ def load(self) -> LivePurchaseAuthorization:
+ """读取授权；任何字段异常都安全降级为关闭。"""
+
+ enabled = self._repository.get(LIVE_ENABLED_KEY, False)
+ client_id = self._repository.get(LIVE_CLIENT_ID_KEY, "")
+ serial = self._repository.get(LIVE_DEVICE_SERIAL_KEY, "")
+ confirmed_at = self._repository.get(LIVE_CONFIRMED_AT_KEY, "")
+ if enabled is not True:
+ return LivePurchaseAuthorization()
+ if not all(
+ isinstance(value, str)
+ for value in (client_id, serial, confirmed_at)
+ ):
+ return LivePurchaseAuthorization()
+ normalized = LivePurchaseAuthorization(
+ True,
+ client_id.strip(),
+ serial.strip(),
+ confirmed_at.strip(),
+ )
+ if (
+ not normalized.client_id
+ or not normalized.device_serial
+ or not normalized.confirmed_at
+ ):
+ return LivePurchaseAuthorization()
+ return normalized
+
+ def enable(
+ self,
+ client_id: str,
+ device_serial: str,
+ confirmation_text: str,
+ ) -> LivePurchaseAuthorization:
+ """精确核对确认文字后，原子保存绑定和确认时间。"""
+
+ checked_client = str(client_id or "").strip()
+ checked_serial = str(device_serial or "").strip()
+ if not checked_client:
+ raise ValueError("请先保存当前 Client 设备号")
+ if not checked_serial:
+ raise ValueError("请先选择并保存 Android 设备")
+ if confirmation_text.strip() != LIVE_CONFIRMATION_TEXT:
+ raise ValueError(f"请输入“{LIVE_CONFIRMATION_TEXT}”确认")
+ confirmed_at = _utc_now_iso()
+ self._repository.set_many(
+ {
+ LIVE_ENABLED_KEY: True,
+ LIVE_CLIENT_ID_KEY: checked_client,
+ LIVE_DEVICE_SERIAL_KEY: checked_serial,
+ LIVE_CONFIRMED_AT_KEY: confirmed_at,
+ }
+ )
+ return LivePurchaseAuthorization(
+ True, checked_client, checked_serial, confirmed_at
+ )
+
+ def disable(self) -> LivePurchaseAuthorization:
+ """立即关闭能力；历史绑定不用于重新启用。"""
+
+ self._repository.set(LIVE_ENABLED_KEY, False)
+ return LivePurchaseAuthorization()
+
+ def purchase_mode_for(
+ self,
+ client_id: str,
+ device_serial: str,
+ *,
+ live_adapter_ready: bool,
+ ) -> str:
+ """集中计算对 Admin 声明的能力，默认永远是 dry_run。"""
+
+ if live_adapter_ready and self.load().matches(client_id, device_serial):
+ return "live"
+ return "dry_run"
diff --git a/client/src/mock_admin_gateway.py b/client/src/mock_admin_gateway.py
index d16fbfa..b3cb622 100644
--- a/client/src/mock_admin_gateway.py
+++ b/client/src/mock_admin_gateway.py
@@ -143,6 +143,11 @@ class MockAdminGateway(AdminGateway):
continue
if item.task.task_type not in capabilities.supported_types:
continue
+ if (
+ item.task.execution_mode == "live"
+ and capabilities.purchase_mode != "live"
+ ):
+ continue
item.claimed = True
self._claimed_task_ids.add(item.task.task_id)
diff --git a/client/src/pdd_purchase_adapter.py b/client/src/pdd_purchase_adapter.py
index 2fd4e44..93822c2 100644
--- a/client/src/pdd_purchase_adapter.py
+++ b/client/src/pdd_purchase_adapter.py
@@ -21,6 +21,8 @@ class PurchasePageState:
quantity: int = 0
price_cent: int = 0
candidate_count: int = 1
+ in_stock: bool = True
+ submit_candidate_count: int = 0
class PddPurchaseError(RuntimeError):
@@ -73,3 +75,11 @@ class PddPurchaseAdapter(ABC):
@abstractmethod
def close(self) -> None:
"""释放设备会话；不得在此方法中产生页面点击。"""
+
+
+class PddLivePurchaseAdapter(PddPurchaseAdapter):
+ """受控真实采购接口；只增加一次性提交，不提供付款或取消。"""
+
+ @abstractmethod
+ def submit_order_once(self) -> None:
+ """用最新页面状态确认唯一按钮并单击一次；调用后禁止重试。"""
diff --git a/client/src/pdd_u2_purchase_adapter.py b/client/src/pdd_u2_purchase_adapter.py
index bb12bff..a8b8f11 100644
--- a/client/src/pdd_u2_purchase_adapter.py
+++ b/client/src/pdd_u2_purchase_adapter.py
@@ -1,7 +1,7 @@
-"""uiautomator2 采购演练 Adapter。
+"""uiautomator2 采购 Adapter。
-本模块只到 PDD 最终提交订单按钮前。代码中没有点击提交订单
-或付款的方法。
+演练 factory 返回没有提交方法的窄接口；live factory 单独返回只允许一次提交的
+接口。两条路径都不提供付款、取消订单或绕过安全校验的方法。
"""
from __future__ import annotations
@@ -19,6 +19,7 @@ from .pdd_device_service import (
PddDeviceService,
)
from .pdd_purchase_adapter import (
+ PddLivePurchaseAdapter,
PddPurchaseAdapter,
PddPurchaseError,
PurchasePageState,
@@ -36,6 +37,7 @@ _CAPTCHA_MARKERS = ("请完成验证", "拖动滑块", "安全验证", "点击
_RISK_MARKERS = ("操作频繁", "异常请求", "风险提示", "账号异常")
_PAYMENT_MARKERS = ("输入支付密码", "立即支付", "支付成功", "支付失败")
_FINAL_SUBMIT_MARKERS = ("提交订单", "现在买，仅", "确认购买")
+_OUT_OF_STOCK_MARKERS = ("已售罄", "暂时缺货", "库存不足", "该商品已售罄")
_SUPPORTED_OPTION_KEYS = frozenset({"color", "size"})
@@ -200,6 +202,33 @@ def _price_cent(root: ET.Element) -> int:
return max(parsed_nodes)[2]
+def _final_submit_targets(root: ET.Element) -> list[Bounds]:
+ """返回底部可见、启用且文字明确的唯一提交按钮坐标。"""
+
+ screen_bottom = 0
+ for node in root.iter("node"):
+ bounds = _parse_bounds(node.get("bounds", ""))
+ if bounds is not None:
+ screen_bottom = max(screen_bottom, bounds[3])
+ if screen_bottom <= 0:
+ return []
+
+ targets = set()
+ for node in root.iter("node"):
+ label = _label(node)
+ bounds = _parse_bounds(node.get("bounds", ""))
+ if not label or bounds is None:
+ continue
+ if not any(marker in label for marker in _FINAL_SUBMIT_MARKERS):
+ continue
+ if node.get("visible-to-user") != "true" or node.get("enabled") != "true":
+ continue
+ if (bounds[1] + bounds[3]) // 2 < screen_bottom * 0.6:
+ continue
+ targets.add(bounds)
+ return sorted(targets)
+
+
class U2PddPurchaseAdapter(PddPurchaseAdapter):
"""PDD 真机采购演练会话，不提供真实下单能力。"""
@@ -229,6 +258,7 @@ class U2PddPurchaseAdapter(PddPurchaseAdapter):
self._device = None
self._goods_id = ""
self._requested_options: dict[str, str] = {}
+ self._submit_attempted = False
def open_goods(self, goods_url: str) -> None:
self._goods_id = _goods_id_from_url(goods_url)
@@ -268,6 +298,8 @@ class U2PddPurchaseAdapter(PddPurchaseAdapter):
xml_data = self._dump_hierarchy()
root = _parse_xml(xml_data)
kind = _page_kind(root, str(current.get("package") or ""))
+ labels = _labels(root)
+ submit_targets = _final_submit_targets(root)
selected = {
key: value
for key, value in self._requested_options.items()
@@ -280,6 +312,11 @@ class U2PddPurchaseAdapter(PddPurchaseAdapter):
quantity=_quantity(root),
price_cent=_price_cent(root),
candidate_count=1 if kind != "unknown" else 0,
+ in_stock=not any(
+ marker in " ".join(labels)
+ for marker in _OUT_OF_STOCK_MARKERS
+ ),
+ submit_candidate_count=len(submit_targets),
)
except PddPurchaseError:
raise
@@ -512,3 +549,57 @@ def create_u2_purchase_adapter(
device_service=_PURCHASE_DEVICE_SERVICE,
cancelled=cancelled,
)
+
+
+class U2PddLivePurchaseAdapter(U2PddPurchaseAdapter, PddLivePurchaseAdapter):
+ """只允许一次最终提交点击的真机 Adapter，不包含付款路径。"""
+
+ def submit_order_once(self) -> None:
+ if self._submit_attempted:
+ raise PddPurchaseError(
+ "PURCHASE_SUBMIT_ALREADY_ATTEMPTED",
+ "本次采购已经尝试提交，禁止再次点击",
+ step="purchase_submit_once",
+ )
+ self._check_cancelled("purchase_submit_once")
+ device = self._require_device()
+ try:
+ current = device.app_current()
+ root = _parse_xml(self._dump_hierarchy())
+ kind = _page_kind(root, str(current.get("package") or ""))
+ if kind in {"captcha", "login_required", "risk_control", "payment"}:
+ self._raise_special_page(kind)
+ if kind != "order_confirmation":
+ raise PddPurchaseError(
+ "PURCHASE_CONFIRMATION_LOST",
+ "最终提交前页面已经变化，禁止提交订单",
+ step="purchase_submit_once",
+ )
+ targets = _final_submit_targets(root)
+ if len(targets) != 1:
+ raise PddPurchaseError(
+ "PURCHASE_SUBMIT_TARGET_AMBIGUOUS",
+ "最终提交按钮不是唯一可靠目标，禁止提交订单",
+ step="purchase_submit_once",
+ diagnostics={"candidate_count": len(targets)},
+ )
+ left, top, right, bottom = targets[0]
+ self._submit_attempted = True
+ device.click((left + right) // 2, (top + bottom) // 2)
+ except PddPurchaseError:
+ raise
+ except Exception as exc:
+ self._submit_attempted = True
+ self._raise_device_or_page_error(exc, "purchase_submit_once")
+
+
+def create_u2_live_purchase_adapter(
+ device_address: str, cancelled: Callable[[], bool]
+) -> PddLivePurchaseAdapter:
+ """为已通过本地绑定授权的任务创建一次 live 会话。"""
+
+ return U2PddLivePurchaseAdapter(
+ device_address,
+ device_service=_PURCHASE_DEVICE_SERVICE,
+ cancelled=cancelled,
+ )
diff --git a/client/src/pdd_ui_event.py b/client/src/pdd_ui_event.py
index 171d248..d259327 100644
--- a/client/src/pdd_ui_event.py
+++ b/client/src/pdd_ui_event.py
@@ -46,6 +46,8 @@ 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 .purchase_task_service import LivePurchaseAdapterFactory
+from .live_purchase_authorization import LivePurchaseAuthorizationService
from .purchase_reconcile_service import PurchaseReconcileFactory
from .selected_android_device_service import SelectedAndroidDeviceService
from .settings_repository import SettingsRepository
@@ -125,6 +127,12 @@ class ClaimTaskWorker(QObject):
android_device_service: SelectedAndroidDeviceService,
collect_service_factory: Optional[CollectServiceFactory] = None,
purchase_adapter_factory: Optional[PurchaseAdapterFactory] = None,
+ live_purchase_adapter_factory: Optional[
+ LivePurchaseAdapterFactory
+ ] = None,
+ live_authorization_service: Optional[
+ LivePurchaseAuthorizationService
+ ] = None,
purchase_reconcile_factory: Optional[PurchaseReconcileFactory] = None,
selected_task_id: str = "",
device_connection_checker: Optional[Callable[[str], None]] = None,
@@ -137,6 +145,8 @@ class ClaimTaskWorker(QObject):
self._cancelled = False
self._collect_service_factory = collect_service_factory
self._purchase_adapter_factory = purchase_adapter_factory
+ self._live_purchase_adapter_factory = live_purchase_adapter_factory
+ self._live_authorization_service = live_authorization_service
self._purchase_reconcile_factory = purchase_reconcile_factory
self._selected_task_id = selected_task_id
self._device_connection_checker = (
@@ -179,6 +189,17 @@ class ClaimTaskWorker(QObject):
)
result = service.execute_selected(self._selected_task_id)
else:
+ purchase_mode = "dry_run"
+ if self._live_authorization_service is not None:
+ purchase_mode = (
+ self._live_authorization_service.purchase_mode_for(
+ client.client_id,
+ android_serial or "",
+ live_adapter_ready=(
+ self._live_purchase_adapter_factory is not None
+ ),
+ )
+ )
dispatcher = TaskDispatcher(
self._gateway,
self._task_repository,
@@ -186,6 +207,10 @@ class ClaimTaskWorker(QObject):
android_serial or "",
collect_service_factory=self._collect_service_factory,
purchase_adapter_factory=self._purchase_adapter_factory,
+ live_purchase_adapter_factory=(
+ self._live_purchase_adapter_factory
+ ),
+ purchase_mode=purchase_mode,
purchase_reconcile_factory=self._purchase_reconcile_factory,
cancelled=lambda: self._cancelled,
device_connection_checker=self._device_connection_checker,
@@ -308,6 +333,9 @@ class PDDTaskPageEvent(QObject):
settings_repository: Optional[SettingsRepository] = None,
collect_service_factory: Optional[CollectServiceFactory] = None,
purchase_adapter_factory: Optional[PurchaseAdapterFactory] = None,
+ live_purchase_adapter_factory: Optional[
+ LivePurchaseAdapterFactory
+ ] = None,
purchase_reconcile_factory: Optional[PurchaseReconcileFactory] = None,
device_connection_checker: Optional[Callable[[str], None]] = None,
next_task_delay_ms: int = 500,
@@ -337,6 +365,7 @@ class PDDTaskPageEvent(QObject):
self._detail_windows: Dict[str, TaskDetailWindow] = {}
self._collect_service_factory = collect_service_factory
self._purchase_adapter_factory = purchase_adapter_factory
+ self._live_purchase_adapter_factory = live_purchase_adapter_factory
self._purchase_reconcile_factory = purchase_reconcile_factory
self._device_connection_checker = (
device_connection_checker
@@ -360,6 +389,9 @@ class PDDTaskPageEvent(QObject):
pass
settings = settings_repository or SettingsRepository()
+ self._live_authorization_service = LivePurchaseAuthorizationService(
+ settings
+ )
self._client_service = CurrentClientService(settings)
self._selected_android_device_service = SelectedAndroidDeviceService(
settings
@@ -788,6 +820,8 @@ class PDDTaskPageEvent(QObject):
android_device_service=self._selected_android_device_service,
collect_service_factory=self._collect_service_factory,
purchase_adapter_factory=self._purchase_adapter_factory,
+ live_purchase_adapter_factory=self._live_purchase_adapter_factory,
+ live_authorization_service=self._live_authorization_service,
purchase_reconcile_factory=self._purchase_reconcile_factory,
device_connection_checker=self._device_connection_checker,
)
diff --git a/client/src/purchase_task_service.py b/client/src/purchase_task_service.py
index fba6c5a..2cd72f9 100644
--- a/client/src/purchase_task_service.py
+++ b/client/src/purchase_task_service.py
@@ -1,7 +1,7 @@
-"""一条本地采购任务的安全演练流程。
+"""一条本地采购任务的安全执行流程。
-本模块固定为 ``dry_run``。它会核对商品、动态规格、数量和价格，并停在最终
-提交订单之前；代码中没有提交订单或付款入口。
+``dry_run`` 始终停在提交前；``live`` 只在独立 Adapter 就绪时允许一次提交，
+并且必须先持久化不可逆标记。任何路径都不包含付款。
"""
from __future__ import annotations
@@ -12,6 +12,7 @@ from typing import Callable, Mapping
from .admin_gateway import AdminGateway, AdminGatewayError, ClientInfo
from .pdd_purchase_adapter import (
+ PddLivePurchaseAdapter,
PddPurchaseAdapter,
PddPurchaseError,
PurchasePageState,
@@ -51,6 +52,9 @@ class PurchaseTaskOutcome:
PurchaseAdapterFactory = Callable[
[str, Callable[[], bool]], PddPurchaseAdapter
]
+LivePurchaseAdapterFactory = Callable[
+ [str, Callable[[], bool]], PddLivePurchaseAdapter
+]
class PurchaseTaskService:
@@ -87,7 +91,7 @@ class PurchaseTaskService:
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 设备")
@@ -104,6 +108,15 @@ class PurchaseTaskService:
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)
@@ -150,6 +163,15 @@ class PurchaseTaskService:
)
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()
@@ -182,6 +204,64 @@ class PurchaseTaskService:
adapter.close()
return self._submit(event)
+ 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
+ )
+ 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(
+ "manual_review",
+ f"任务 {remote_task_id} {message}",
+ remote_task_id,
+ )
+
def _enter_step(
self, remote_task_id: str, attempt_id: str, step: str
) -> None:
@@ -328,6 +408,35 @@ class PurchaseTaskService:
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":
diff --git a/client/src/settings_ui.py b/client/src/settings_ui.py
index 747fe2d..78085bf 100644
--- a/client/src/settings_ui.py
+++ b/client/src/settings_ui.py
@@ -212,6 +212,7 @@ class SettingsPage(QWidget):
android_device_service=None,
update_service=None,
update_credential_store=None,
+ live_purchase_adapter_ready: bool = False,
):
super().__init__(parent)
self.setObjectName("settingsPage")
@@ -227,6 +228,7 @@ class SettingsPage(QWidget):
android_device_service=android_device_service,
update_service=update_service,
update_credential_store=update_credential_store,
+ live_purchase_adapter_ready=live_purchase_adapter_ready,
)
def _build_ui(self) -> None:
@@ -296,6 +298,34 @@ class SettingsPage(QWidget):
self.pddAppStatusLabel.setWordWrap(True)
self.androidDeviceCard = self._build_android_device_card()
+ self.livePurchaseStatusLabel = CaptionLabel(
+ "真实下单默认关闭，仅允许创建未付款订单", self
+ )
+ self.livePurchaseStatusLabel.setAccessibleName("真实下单授权状态")
+ self.livePurchaseStatusLabel.setWordWrap(True)
+ self.livePurchaseClientLabel = CaptionLabel("—", self)
+ self.livePurchaseDeviceLabel = CaptionLabel("—", self)
+ self.livePurchaseConfirmedAtLabel = CaptionLabel("—", self)
+ self.livePurchaseConfirmationInput = LineEdit(self)
+ self.livePurchaseConfirmationInput.setPlaceholderText(
+ "请输入“创建未付款订单”"
+ )
+ self.livePurchaseConfirmationInput.setClearButtonEnabled(True)
+ self.livePurchaseConfirmationInput.setAccessibleName(
+ "真实下单确认文字"
+ )
+ self.livePurchaseEnableButton = PushButton(
+ FIF.ACCEPT, "启用真实下单", self
+ )
+ self.livePurchaseEnableButton.setAccessibleName(
+ "为当前 Client 和 Android 设备启用真实下单"
+ )
+ self.livePurchaseDisableButton = PushButton(
+ FIF.CANCEL, "关闭真实下单", self
+ )
+ self.livePurchaseDisableButton.setAccessibleName("关闭真实下单")
+ self.livePurchaseCard = self._build_live_purchase_card()
+
self.currentVersionLabel = CaptionLabel(__version__, self)
self.currentVersionLabel.setAccessibleName("当前软件版本")
self.updateManifestUrlInput = LineEdit(self)
@@ -331,6 +361,7 @@ class SettingsPage(QWidget):
contentLayout.addWidget(TitleLabel("设置", content))
contentLayout.addWidget(self.currentDeviceCard)
contentLayout.addWidget(self.androidDeviceCard)
+ contentLayout.addWidget(self.livePurchaseCard)
contentLayout.addWidget(self.softwareUpdateCard)
contentLayout.addStretch(1)
@@ -455,6 +486,52 @@ class SettingsPage(QWidget):
layout.addLayout(commandLayout)
return card
+ def _build_live_purchase_card(self) -> CardWidget:
+ card = CardWidget(self)
+ card.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Fixed)
+ layout = QVBoxLayout(card)
+ layout.setContentsMargins(24, 20, 24, 22)
+ layout.setSpacing(12)
+
+ titleLayout = QHBoxLayout()
+ titleLayout.setSpacing(12)
+ titleLayout.addWidget(SubtitleLabel("真实下单（不支付）", card))
+ titleLayout.addStretch(1)
+ titleLayout.addWidget(self.livePurchaseStatusLabel, 1)
+ layout.addLayout(titleLayout)
+
+ form = QFormLayout()
+ form.setHorizontalSpacing(16)
+ form.setVerticalSpacing(12)
+ form.addRow(
+ CaptionLabel("绑定 Client", card), self.livePurchaseClientLabel
+ )
+ form.addRow(
+ CaptionLabel("绑定设备", card), self.livePurchaseDeviceLabel
+ )
+ form.addRow(
+ CaptionLabel("确认时间", card), self.livePurchaseConfirmedAtLabel
+ )
+ confirmationLabel = CaptionLabel("确认文字", card)
+ confirmationLabel.setBuddy(self.livePurchaseConfirmationInput)
+ form.addRow(confirmationLabel, self.livePurchaseConfirmationInput)
+ layout.addLayout(form)
+
+ warning = CaptionLabel(
+ "启用后只允许提交一次订单并停在支付前；验证码、风控、登录失效、"
+ "规格或价格不一致时立即停止。",
+ card,
+ )
+ warning.setWordWrap(True)
+ layout.addWidget(warning)
+
+ commandLayout = QHBoxLayout()
+ commandLayout.addStretch(1)
+ commandLayout.addWidget(self.livePurchaseDisableButton)
+ commandLayout.addWidget(self.livePurchaseEnableButton)
+ layout.addLayout(commandLayout)
+ return card
+
def set_client_info(self, device_id: str, device_name: str) -> None:
"""显示后续设备身份服务提供的当前客户端信息。"""
@@ -482,6 +559,28 @@ class SettingsPage(QWidget):
self.updateStatusLabel.setText(message)
+ def set_live_purchase_authorization(
+ self,
+ *,
+ enabled: bool,
+ client_id: str = "",
+ device_serial: str = "",
+ confirmed_at: str = "",
+ message: str = "",
+ ) -> None:
+ """显示真实下单授权，不通过颜色单独表达状态。"""
+
+ self.livePurchaseClientLabel.setText(client_id or "—")
+ self.livePurchaseDeviceLabel.setText(device_serial or "—")
+ self.livePurchaseConfirmedAtLabel.setText(confirmed_at or "—")
+ if message:
+ status = message
+ elif enabled:
+ status = "已启用：仅限绑定 Client 和设备，且不会自动支付"
+ else:
+ status = "已关闭：所有采购任务只允许演练"
+ self.livePurchaseStatusLabel.setText(status)
+
def set_saved_android_device(self, serial: str) -> None:
"""显示已经保存并实际用于自动化的 Android 设备。"""
diff --git a/client/src/settings_ui_event.py b/client/src/settings_ui_event.py
index aefc101..f37ba23 100644
--- a/client/src/settings_ui_event.py
+++ b/client/src/settings_ui_event.py
@@ -35,8 +35,14 @@ from .current_client_service import (
)
from .http_admin_gateway import DEFAULT_ADMIN_BASE_URL, HttpAdminGateway
from .selected_android_device_service import SelectedAndroidDeviceService
+from .live_purchase_authorization import (
+ LIVE_CONFIRMATION_TEXT,
+ LivePurchaseAuthorization,
+ LivePurchaseAuthorizationService,
+)
from .settings_repository import SettingsRepository
from .settings_ui import AndroidDeviceRow
+from .task_models import TaskType
from .update_ui_event import UpdateUiEventBinder
DEVICE_ID_PLACEHOLDER = "待生成"
@@ -294,6 +300,85 @@ class AndroidDeviceSettingWorker(QObject):
self.completed.emit()
+class LivePurchaseAuthorizationWorker(QObject):
+ """后台保存 live 授权，并尽力把最新能力登记到 Admin。"""
+
+ saved = pyqtSignal(object)
+ failed = pyqtSignal(str)
+ registrationFailed = pyqtSignal(str)
+ completed = pyqtSignal()
+
+ def __init__(
+ self,
+ service: LivePurchaseAuthorizationService,
+ gateway: Optional[ClientRegistrationGateway],
+ client_id: str,
+ client_name: str,
+ device_serial: str,
+ action: str,
+ confirmation_text: str,
+ gateway_error: str = "",
+ ) -> None:
+ super().__init__()
+ self._service = service
+ self._gateway = gateway
+ self._client_id = str(client_id or "").strip()
+ self._client_name = str(client_name or "").strip()
+ self._device_serial = device_serial
+ self._action = action
+ self._confirmation_text = confirmation_text
+ self._gateway_error = gateway_error
+
+ @pyqtSlot()
+ def run(self) -> None:
+ try:
+ try:
+ if self._action == "enable":
+ authorization = self._service.enable(
+ self._client_id,
+ self._device_serial,
+ self._confirmation_text,
+ )
+ else:
+ authorization = self._service.disable()
+ except Exception as exc:
+ self.failed.emit(str(exc) or "真实下单授权保存失败")
+ return
+
+ self.saved.emit(authorization)
+ if not self._client_id:
+ return
+ if self._gateway is None:
+ self.registrationFailed.emit(
+ self._gateway_error or "Admin Gateway 尚未配置"
+ )
+ return
+ device = (
+ AndroidDeviceInfo(self._device_serial)
+ if self._device_serial
+ else None
+ )
+ supported_types = (
+ (TaskType.COLLECT, TaskType.PURCHASE)
+ if device is not None
+ else (TaskType.COLLECT,)
+ )
+ capabilities = ClaimCapabilities(
+ device=device,
+ supported_types=supported_types,
+ purchase_mode="live" if authorization.enabled else "dry_run",
+ )
+ try:
+ self._gateway.register_client(
+ ClientInfo(self._client_id, self._client_name),
+ capabilities,
+ )
+ except Exception as exc:
+ self.registrationFailed.emit(str(exc) or "Admin 登记失败")
+ finally:
+ self.completed.emit()
+
+
class SettingsPageEventBinder(QObject):
"""绑定设备管理控件，并向应用层发出稳定事件。"""
@@ -302,6 +387,7 @@ class SettingsPageEventBinder(QObject):
currentDeviceSaveRequested = pyqtSignal(str, str)
saveRequested = pyqtSignal(str, str)
deleteRequested = pyqtSignal(str)
+ liveAuthorizationRequested = pyqtSignal(str, str)
def __init__(
self,
@@ -311,6 +397,7 @@ class SettingsPageEventBinder(QObject):
android_device_service: Optional[AndroidDeviceService] = None,
update_service=None,
update_credential_store=None,
+ live_purchase_adapter_ready: bool = False,
):
super().__init__(page)
self._page = page
@@ -341,6 +428,14 @@ class SettingsPageEventBinder(QObject):
self._android_setting_thread: Optional[QThread] = None
self._android_setting_worker: Optional[AndroidDeviceSettingWorker] = None
self._saved_android_serial = ""
+ self._live_purchase_busy = False
+ self._live_purchase_thread: Optional[QThread] = None
+ self._live_purchase_worker: Optional[
+ LivePurchaseAuthorizationWorker
+ ] = None
+ self._live_purchase_adapter_ready = bool(
+ live_purchase_adapter_ready
+ )
self._android_device_service = (
android_device_service or AndroidDeviceService()
)
@@ -354,6 +449,10 @@ class SettingsPageEventBinder(QObject):
parent=self,
)
self._client_service = CurrentClientService(repository)
+ self._live_purchase_service = LivePurchaseAuthorizationService(
+ repository
+ )
+ self._live_authorization = self._live_purchase_service.load()
self._selected_android_device_service = SelectedAndroidDeviceService(
repository
)
@@ -384,6 +483,18 @@ class SettingsPageEventBinder(QObject):
page.saveButton.clicked.connect(self._request_save)
self.saveRequested.connect(self._start_save_android_device)
page.deleteButton.clicked.connect(self._request_delete)
+ page.livePurchaseEnableButton.clicked.connect(
+ self._request_enable_live_purchase
+ )
+ page.livePurchaseDisableButton.clicked.connect(
+ self._request_disable_live_purchase
+ )
+ page.livePurchaseConfirmationInput.textChanged.connect(
+ self._sync_button_state
+ )
+ self.liveAuthorizationRequested.connect(
+ self._start_live_authorization
+ )
self.deleteRequested.connect(self._start_delete_android_device)
page.deviceTableModel.checkedDeviceChanged.connect(
self._on_checked_device_changed
@@ -395,6 +506,7 @@ class SettingsPageEventBinder(QObject):
self._load_current_client()
self._load_selected_android_device()
+ self._show_live_authorization(self._live_authorization)
self._sync_button_state()
if self._saved_android_serial:
QTimer.singleShot(0, self._request_restore_saved_android_device)
@@ -430,7 +542,22 @@ class SettingsPageEventBinder(QObject):
try:
device = self._selected_android_device()
- capabilities = ClaimCapabilities(device=device)
+ displayed_client_id = self._page.deviceIdInput.text().strip()
+ purchase_mode = self._live_purchase_service.purchase_mode_for(
+ displayed_client_id,
+ device.address if device is not None else "",
+ live_adapter_ready=self._live_purchase_adapter_ready,
+ )
+ supported_types = (
+ (TaskType.COLLECT, TaskType.PURCHASE)
+ if device is not None and self._live_purchase_adapter_ready
+ else (TaskType.COLLECT,)
+ )
+ capabilities = ClaimCapabilities(
+ device=device,
+ supported_types=supported_types,
+ purchase_mode=purchase_mode,
+ )
except ValueError as exc:
self._page.set_current_device_status(f"保存失败：{exc}")
return
@@ -463,6 +590,128 @@ class SettingsPageEventBinder(QObject):
self._worker = worker
thread.start()
+ @pyqtSlot()
+ def _request_enable_live_purchase(self) -> None:
+ if self._live_purchase_busy:
+ return
+ if not self._live_purchase_adapter_ready:
+ self._page.set_live_purchase_authorization(
+ enabled=False,
+ message="真实下单执行器未就绪，只允许采购演练",
+ )
+ return
+ confirmation = self._page.livePurchaseConfirmationInput.text()
+ self.liveAuthorizationRequested.emit("enable", confirmation)
+
+ @pyqtSlot()
+ def _request_disable_live_purchase(self) -> None:
+ if not self._live_purchase_busy:
+ self.liveAuthorizationRequested.emit("disable", "")
+
+ @pyqtSlot(str, str)
+ def _start_live_authorization(
+ self, action: str, confirmation_text: str
+ ) -> None:
+ if self._closing or self._live_purchase_busy:
+ return
+ current = self._client_service.load()
+ serial = self._selected_android_device_service.load()
+ if action == "enable" and (
+ not current.client_id or not serial
+ ):
+ self._page.set_live_purchase_authorization(
+ enabled=False,
+ message="请先保存当前 Client 和 Android 设备",
+ )
+ return
+
+ self._live_purchase_busy = True
+ self._sync_button_state()
+ self._page.set_live_purchase_authorization(
+ enabled=self._live_authorization.enabled,
+ client_id=self._live_authorization.client_id,
+ device_serial=self._live_authorization.device_serial,
+ confirmed_at=self._live_authorization.confirmed_at,
+ message=(
+ "正在启用真实下单…"
+ if action == "enable"
+ else "正在关闭真实下单…"
+ ),
+ )
+ thread = QThread(self)
+ worker = LivePurchaseAuthorizationWorker(
+ self._live_purchase_service,
+ self._admin_gateway,
+ current.client_id,
+ current.client_name,
+ serial,
+ action,
+ confirmation_text,
+ self._gateway_error,
+ )
+ worker.moveToThread(thread)
+ thread.started.connect(worker.run)
+ worker.saved.connect(self._on_live_authorization_saved)
+ worker.failed.connect(self._on_live_authorization_failed)
+ worker.registrationFailed.connect(
+ self._on_live_registration_failed
+ )
+ worker.completed.connect(thread.quit)
+ worker.completed.connect(worker.deleteLater)
+ thread.finished.connect(self._on_live_authorization_finished)
+ thread.finished.connect(thread.deleteLater)
+ self._live_purchase_thread = thread
+ self._live_purchase_worker = worker
+ thread.start()
+
+ @pyqtSlot(object)
+ def _on_live_authorization_saved(
+ self, authorization: LivePurchaseAuthorization
+ ) -> None:
+ if self._closing:
+ return
+ self._live_authorization = authorization
+ self._page.livePurchaseConfirmationInput.clear()
+ self._show_live_authorization(authorization)
+
+ @pyqtSlot(str)
+ def _on_live_authorization_failed(self, message: str) -> None:
+ if not self._closing:
+ self._show_live_authorization(
+ self._live_authorization,
+ f"授权修改失败：{message}；原设置未改变",
+ )
+
+ @pyqtSlot(str)
+ def _on_live_registration_failed(self, message: str) -> None:
+ if not self._closing:
+ state = "已启用" if self._live_authorization.enabled else "已关闭"
+ self._show_live_authorization(
+ self._live_authorization,
+ f"本地{state}，Admin 登记失败：{message}；领取时会再次声明能力",
+ )
+
+ @pyqtSlot()
+ def _on_live_authorization_finished(self) -> None:
+ self._live_purchase_worker = None
+ self._live_purchase_thread = None
+ self._live_purchase_busy = False
+ if not self._closing:
+ self._sync_button_state()
+
+ def _show_live_authorization(
+ self,
+ authorization: LivePurchaseAuthorization,
+ message: str = "",
+ ) -> None:
+ self._page.set_live_purchase_authorization(
+ enabled=authorization.enabled,
+ client_id=authorization.client_id,
+ device_serial=authorization.device_serial,
+ confirmed_at=authorization.confirmed_at,
+ message=message,
+ )
+
@pyqtSlot()
def _request_search(self) -> None:
if (
@@ -871,6 +1120,25 @@ class SettingsPageEventBinder(QObject):
self._page.deleteButton.setEnabled(
device_commands_enabled and bool(self._saved_android_serial)
)
+ live_commands_enabled = (
+ not self._closing
+ and not self._busy
+ and not self._current_device_busy
+ and not self._live_purchase_busy
+ )
+ confirmation_matches = (
+ self._page.livePurchaseConfirmationInput.text().strip()
+ == LIVE_CONFIRMATION_TEXT
+ )
+ self._page.livePurchaseEnableButton.setEnabled(
+ live_commands_enabled
+ and self._live_purchase_adapter_ready
+ and not self._live_authorization.enabled
+ and confirmation_matches
+ )
+ self._page.livePurchaseDisableButton.setEnabled(
+ live_commands_enabled and self._live_authorization.enabled
+ )
def set_busy(self, busy: bool, message: str = "") -> None:
"""切换界面忙碌状态，防止用户重复提交设备命令。"""
@@ -1178,6 +1446,11 @@ class SettingsPageEventBinder(QObject):
thread.quit()
thread.wait(4000)
+ live_thread = self._live_purchase_thread
+ if live_thread is not None and live_thread.isRunning():
+ live_thread.quit()
+ live_thread.wait(4000)
+
search_worker = self._search_worker
search_thread = self._search_thread
if search_worker is not None:
diff --git a/client/src/task_detail_view.py b/client/src/task_detail_view.py
index 58f918d..c8a0cc4 100644
--- a/client/src/task_detail_view.py
+++ b/client/src/task_detail_view.py
@@ -68,6 +68,9 @@ CURRENT_STEP_TEXT = {
"purchase_verify_quantity_price": "正在核对数量和价格",
"purchase_enter_confirmation": "正在进入提交前确认页",
"purchase_verify_confirmation": "正在核对提交前确认页",
+ "purchase_live_final_check": "正在执行真实下单最终安全核对",
+ "purchase_irreversible_step_entered": "已进入不可逆阶段，禁止重新下单",
+ "purchase_submit_once": "已单击一次提交订单",
"purchase_dry_run_stopped": "采购演练已在提交前停止",
"purchase_recovery_ready": "上次演练中断，已安全等待恢复",
"reconcile_purchase": "只允许核对订单",
diff --git a/client/src/task_dispatcher.py b/client/src/task_dispatcher.py
index 5c37294..4b354d2 100644
--- a/client/src/task_dispatcher.py
+++ b/client/src/task_dispatcher.py
@@ -19,6 +19,7 @@ from .android_device_service import (
AndroidDeviceService,
)
from .purchase_task_service import (
+ LivePurchaseAdapterFactory,
PurchaseAdapterFactory,
PurchaseTaskService,
)
@@ -95,6 +96,7 @@ def admin_task_to_new_claimed_task(task: AdminTask) -> NewClaimedTask:
original_task = {
"id": task.task_id,
"type": task.task_type.value,
+ "execution_mode": task.execution_mode,
"version": task.version,
"priority": task.priority,
"payload": payload,
@@ -105,6 +107,7 @@ def admin_task_to_new_claimed_task(task: AdminTask) -> NewClaimedTask:
remote_task_id=task.task_id,
task_type=task.task_type,
goods_url=goods_url.strip(),
+ execution_mode=task.execution_mode,
goods_id=goods_id.strip() if isinstance(goods_id, str) else None,
target_color=target_color,
target_size=target_size,
@@ -128,6 +131,10 @@ class TaskDispatcher:
*,
collect_service_factory: Optional[CollectServiceFactory] = None,
purchase_adapter_factory: Optional[PurchaseAdapterFactory] = None,
+ live_purchase_adapter_factory: Optional[
+ LivePurchaseAdapterFactory
+ ] = None,
+ purchase_mode: str = "dry_run",
purchase_reconcile_factory: Optional[PurchaseReconcileFactory] = None,
cancelled: Callable[[], bool] = lambda: False,
device_connection_checker: Optional[Callable[[str], None]] = None,
@@ -138,6 +145,10 @@ class TaskDispatcher:
self._device_address = str(device_address or "").strip()
self._collect_factory = collect_service_factory
self._purchase_factory = purchase_adapter_factory
+ self._live_purchase_factory = live_purchase_adapter_factory
+ if purchase_mode not in {"dry_run", "live"}:
+ raise ValueError("purchase_mode 只允许 dry_run 或 live")
+ self._purchase_mode = purchase_mode
self._reconcile_factory = purchase_reconcile_factory
self._cancelled = cancelled
self._device_connection_checker = (
@@ -167,24 +178,38 @@ class TaskDispatcher:
return ClaimCapabilities(
device=device,
supported_types=tuple(supported),
- purchase_mode="dry_run",
+ purchase_mode=(
+ "live"
+ if self.purchase_ready
+ and self._purchase_mode == "live"
+ and self._live_purchase_factory is not None
+ else "dry_run"
+ ),
schema_versions=(1,),
)
def execute_one(self) -> TaskDispatchOutcome:
- """严格按 Outbox、本地任务、Admin 新任务的顺序处理。"""
+ """不可逆采购优先核单，其余按 Outbox、本地任务、Admin 顺序处理。"""
- pending = self._repository.next_pending_outbox()
- if pending is not None:
- return self._submit_pending(pending)
- if not self._device_address:
- raise AndroidDeviceSearchError(
- "请先在设置页选择并保存 Android 设备"
+ unresolved_reader = getattr(
+ self._repository, "unresolved_irreversible_purchase", None
+ )
+ unresolved = (
+ unresolved_reader() if unresolved_reader is not None else None
+ )
+ if unresolved is not None:
+ raise RuntimeError(
+ f"任务 {unresolved.remote_task_id} 已进入不可逆阶段但尚未转入核单；"
+ "已停止所有采购，绝不重新下单"
)
- self._device_connection_checker(self._device_address)
reconcile_task = self._repository.next_purchase_reconcile_task()
if reconcile_task is not None:
+ if not self._device_address:
+ raise AndroidDeviceSearchError(
+ "存在只允许核对的采购任务；请先连接并保存原 Android 设备"
+ )
+ self._device_connection_checker(self._device_address)
if self._reconcile_factory is None:
raise RuntimeError(
f"任务 {reconcile_task.remote_task_id} 只允许核对订单，"
@@ -200,6 +225,15 @@ class TaskDispatcher:
outcome.kind, outcome.message, outcome.task_id
)
+ pending = self._repository.next_pending_outbox()
+ if pending is not None:
+ return self._submit_pending(pending)
+ if not self._device_address:
+ raise AndroidDeviceSearchError(
+ "请先在设置页选择并保存 Android 设备"
+ )
+ self._device_connection_checker(self._device_address)
+
if not self.purchase_ready:
pending_purchase = self._repository.next_purchase_task()
if pending_purchase is not None:
@@ -244,16 +278,25 @@ class TaskDispatcher:
)
outcome = service.execute_selected(task.remote_task_id)
else:
- if self._purchase_factory is None:
+ factory = self._purchase_factory
+ if task.execution_mode == "live":
+ if self.claim_capabilities().purchase_mode != "live":
+ raise RuntimeError(
+ f"真实采购任务 {task.remote_task_id} 的本地授权已关闭或绑定不一致；"
+ "任务保持待执行，不会降级为演练"
+ )
+ factory = self._live_purchase_factory
+ if factory is None:
raise RuntimeError(
- "采购演练执行器未就绪，已停止领取以避免任务卡住"
+ f"采购任务 {task.remote_task_id} 的 {task.execution_mode} "
+ "执行器未就绪，已停止且不会改变任务模式"
)
purchase_service = PurchaseTaskService(
self._gateway,
self._repository,
self._client,
self._device_address,
- self._purchase_factory,
+ factory,
cancelled=self._cancelled,
)
outcome = purchase_service.execute_selected(task.remote_task_id)
diff --git a/client/src/task_models.py b/client/src/task_models.py
index f0ec5cb..42dec99 100644
--- a/client/src/task_models.py
+++ b/client/src/task_models.py
@@ -62,6 +62,7 @@ class NewClaimedTask:
remote_task_id: str
task_type: TaskType
goods_url: str
+ execution_mode: str = "dry_run"
goods_id: Optional[str] = None
title: Optional[str] = None
target_color: Optional[str] = None
@@ -79,6 +80,13 @@ class NewClaimedTask:
raise ValueError("remote_task_id 不能为空")
if not self.goods_url.strip():
raise ValueError("goods_url 不能为空")
+ if self.execution_mode not in {"dry_run", "live"}:
+ raise ValueError("execution_mode 只允许 dry_run 或 live")
+ if (
+ self.task_type is not TaskType.PURCHASE
+ and self.execution_mode != "dry_run"
+ ):
+ raise ValueError("只有采购任务允许 execution_mode=live")
if self.price_cent is not None and self.price_cent < 0:
raise ValueError("price_cent 不能小于 0")
if self.quantity is not None and self.quantity <= 0:
@@ -141,6 +149,7 @@ class TaskDetail:
finished_at: Optional[str]
created_at: str
updated_at: str
+ execution_mode: str = "dry_run"
@dataclass(frozen=True)
diff --git a/client/src/task_repository.py b/client/src/task_repository.py
index 3cc66f7..3c1c1bd 100644
--- a/client/src/task_repository.py
+++ b/client/src/task_repository.py
@@ -70,14 +70,16 @@ class TaskRepository:
with connection:
cursor = connection.execute(
"INSERT INTO pdd_tasks ("
- " remote_task_id, task_type, goods_id, goods_url, title,"
+ " remote_task_id, task_type, execution_mode, goods_id,"
+ " goods_url, title,"
" target_color, target_size, price_cent, quantity, status,"
" priority, version, admin_payload, received_at, created_at,"
" updated_at"
- ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+ ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(
task.remote_task_id.strip(),
task.task_type.value,
+ task.execution_mode,
task.goods_id,
task.goods_url.strip(),
task.title,
@@ -298,6 +300,22 @@ class TaskRepository:
connection.close()
return self._to_detail(row) if row is not None else None
+ def unresolved_irreversible_purchase(self) -> Optional[TaskDetail]:
+ """查找仍在运行且已有不可逆标记的采购，防止继续控制设备。"""
+
+ connection = open_database(self._db_path)
+ try:
+ row = connection.execute(
+ "SELECT t.* FROM pdd_tasks t JOIN task_runs r ON r.task_id = t.id"
+ " WHERE t.task_type = 'purchase' AND t.status = 'running'"
+ " AND r.run_status = 'running'"
+ " AND r.irreversible_action_at IS NOT NULL"
+ " ORDER BY r.attempt_no DESC LIMIT 1"
+ ).fetchone()
+ finally:
+ connection.close()
+ return self._to_detail(row) if row is not None else None
+
def latest_task_run(
self, remote_task_id: str
) -> Optional[TaskRunRecord]:
@@ -583,6 +601,102 @@ class TaskRepository:
finally:
connection.close()
+ def mark_purchase_irreversible(
+ self, remote_task_id: str, attempt_id: str
+ ) -> str:
+ """事务写入不可逆时间；成功返回后才允许点击提交订单。"""
+
+ now = utc_now_iso()
+ step = "purchase_irreversible_step_entered"
+ connection = open_database(self._db_path)
+ try:
+ with connection:
+ task = connection.execute(
+ "SELECT id, task_type, status, execution_mode FROM pdd_tasks"
+ " WHERE remote_task_id = ?",
+ (remote_task_id,),
+ ).fetchone()
+ if task is None:
+ raise ValueError(f"任务 {remote_task_id} 不存在")
+ if task["task_type"] != TaskType.PURCHASE.value:
+ raise ValueError("当前任务不是采购任务")
+ if task["execution_mode"] != "live":
+ raise ValueError("演练任务不能进入不可逆阶段")
+ if task["status"] != TaskStatus.RUNNING.value:
+ raise ValueError("采购任务当前不在执行中")
+ cursor = connection.execute(
+ "UPDATE task_runs SET irreversible_action_at = ?,"
+ " current_step = ?, updated_at = ?"
+ " WHERE task_id = ? AND attempt_id = ?"
+ " AND run_status = 'running'"
+ " AND irreversible_action_at IS NULL",
+ (now, step, now, task["id"], attempt_id),
+ )
+ if cursor.rowcount != 1:
+ raise ValueError("不可逆标记写入失败或已经存在")
+ connection.execute(
+ "UPDATE pdd_tasks SET current_step = ?, updated_at = ?"
+ " WHERE id = ?",
+ (step, now, task["id"]),
+ )
+ return now
+ finally:
+ connection.close()
+
+ def move_purchase_to_reconcile(
+ self,
+ remote_task_id: str,
+ attempt_id: str,
+ *,
+ order_submitted_at: Optional[str] = None,
+ message: str = "订单提交结果待核对，绝不重新下单",
+ ) -> None:
+ """不可逆点击后转入只读核单状态，不创建可重试下单路径。"""
+
+ now = utc_now_iso()
+ connection = open_database(self._db_path)
+ try:
+ with connection:
+ task = connection.execute(
+ "SELECT id, task_type, status FROM pdd_tasks"
+ " WHERE remote_task_id = ?",
+ (remote_task_id,),
+ ).fetchone()
+ if task is None:
+ raise ValueError(f"任务 {remote_task_id} 不存在")
+ if task["task_type"] != TaskType.PURCHASE.value:
+ raise ValueError("当前任务不是采购任务")
+ cursor = connection.execute(
+ "UPDATE task_runs SET run_status = 'manual_review',"
+ " current_step = 'reconcile_purchase',"
+ " order_submitted_at = COALESCE(order_submitted_at, ?),"
+ " error_code = 'PURCHASE_OUTCOME_UNKNOWN',"
+ " error_message = ?, finished_at = ?, updated_at = ?"
+ " WHERE task_id = ? AND attempt_id = ?"
+ " AND run_status = 'running'"
+ " AND irreversible_action_at IS NOT NULL",
+ (
+ order_submitted_at,
+ message,
+ now,
+ now,
+ task["id"],
+ attempt_id,
+ ),
+ )
+ if cursor.rowcount != 1:
+ raise ValueError("不可逆采购执行记录不存在或已经结束")
+ connection.execute(
+ "UPDATE pdd_tasks SET status = 'manual_review',"
+ " current_step = 'reconcile_purchase',"
+ " last_error_code = 'PURCHASE_OUTCOME_UNKNOWN',"
+ " last_error_message = ?, finished_at = ?, updated_at = ?"
+ " WHERE id = ?",
+ (message, now, now, task["id"]),
+ )
+ finally:
+ connection.close()
+
def save_purchase_reconciliation(
self,
remote_task_id: str,
@@ -1369,6 +1483,7 @@ class TaskRepository:
id=row["id"],
remote_task_id=row["remote_task_id"],
task_type=TaskType(row["task_type"]),
+ execution_mode=row["execution_mode"],
goods_id=row["goods_id"],
goods_url=row["goods_url"],
title=row["title"],
diff --git a/client/src/ui_main.py b/client/src/ui_main.py
index 9e33de9..5496758 100644
--- a/client/src/ui_main.py
+++ b/client/src/ui_main.py
@@ -29,7 +29,10 @@ from qfluentwidgets import (
from .pdd_ui import PDDTaskPage
from .pdd_ui_event import PDDTaskPageEvent
-from .pdd_u2_purchase_adapter import create_u2_purchase_adapter
+from .pdd_u2_purchase_adapter import (
+ create_u2_live_purchase_adapter,
+ create_u2_purchase_adapter,
+)
from .settings_ui import SettingsPage
from .task_repository import TaskRepository
from .update_service import mark_current_version_healthy
@@ -51,12 +54,14 @@ class MainWindow(FluentWindow):
self,
settings_repository=settings_repository,
admin_gateway=admin_gateway,
+ live_purchase_adapter_ready=True,
)
self.pddTaskPageEvent = PDDTaskPageEvent(
self.pddTaskPage,
task_repository or TaskRepository(),
self,
purchase_adapter_factory=create_u2_purchase_adapter,
+ live_purchase_adapter_factory=create_u2_live_purchase_adapter,
)
self.pddTaskPage.openSettingsRequested.connect(
diff --git a/client/test/test_db.py b/client/test/test_db.py
index ccfddb4..ed9a6ff 100644
--- a/client/test/test_db.py
+++ b/client/test/test_db.py
@@ -6,7 +6,7 @@ import unittest
from pathlib import Path
from src.db import DatabaseVersionError, initialize_database, open_database
-from src.db_schema import MIGRATION_1
+from src.db_schema import MIGRATION_1, MIGRATION_2
EXPECTED_TABLES = {
@@ -63,7 +63,7 @@ class DatabaseInitializationTests(unittest.TestCase):
self.assertTrue(EXPECTED_TABLES.issubset(tables))
self.assertTrue(EXPECTED_INDEXES.issubset(indexes))
- self.assertEqual(version, 2)
+ self.assertEqual(version, 3)
def test_v1_database_is_upgraded_without_losing_task_runs(self) -> None:
connection = open_database(self.db_path)
@@ -106,7 +106,7 @@ class DatabaseInitializationTests(unittest.TestCase):
connection.close()
self.assertIn("result_data", columns)
self.assertEqual(attempt_id, "ATTEMPT-OLD")
- self.assertEqual(version, 2)
+ self.assertEqual(version, 3)
def test_initialize_can_run_twice_without_losing_data(self) -> None:
initialize_database(self.db_path)
@@ -133,6 +133,44 @@ class DatabaseInitializationTests(unittest.TestCase):
self.assertEqual(value, "true")
+ def test_v2_database_adds_safe_execution_mode_default(self) -> None:
+ connection = open_database(self.db_path)
+ try:
+ with connection:
+ for statement in MIGRATION_1 + MIGRATION_2:
+ connection.execute(statement)
+ connection.execute("PRAGMA user_version = 2")
+ connection.execute(
+ "INSERT INTO pdd_tasks"
+ " (remote_task_id, task_type, goods_url, status, received_at,"
+ " created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
+ (
+ "PUR-OLD",
+ "purchase",
+ "https://example.test",
+ "claimed",
+ "2026-08-10T00:00:00Z",
+ "2026-08-10T00:00:00Z",
+ "2026-08-10T00:00:00Z",
+ ),
+ )
+ finally:
+ connection.close()
+
+ initialize_database(self.db_path)
+
+ connection = open_database(self.db_path)
+ try:
+ row = connection.execute(
+ "SELECT execution_mode FROM pdd_tasks"
+ " WHERE remote_task_id = 'PUR-OLD'"
+ ).fetchone()
+ version = connection.execute("PRAGMA user_version").fetchone()[0]
+ finally:
+ connection.close()
+ self.assertEqual(row[0], "dry_run")
+ self.assertEqual(version, 3)
+
def test_new_connection_uses_required_pragmas(self) -> None:
initialize_database(self.db_path)
connection = open_database(self.db_path)
@@ -181,6 +219,30 @@ class DatabaseInitializationTests(unittest.TestCase):
finally:
connection.close()
+ def test_execution_mode_constraint_is_enforced(self) -> None:
+ initialize_database(self.db_path)
+ connection = open_database(self.db_path)
+ try:
+ with self.assertRaises(sqlite3.IntegrityError):
+ connection.execute(
+ "INSERT INTO pdd_tasks"
+ " (remote_task_id, task_type, execution_mode, goods_url,"
+ " status, received_at, created_at, updated_at)"
+ " VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+ (
+ "PUR-INVALID-MODE",
+ "purchase",
+ "unknown",
+ "https://example.test/goods",
+ "claimed",
+ "2026-08-10T00:00:00Z",
+ "2026-08-10T00:00:00Z",
+ "2026-08-10T00:00:00Z",
+ ),
+ )
+ finally:
+ connection.close()
+
def test_task_run_and_outbox_foreign_keys_are_enforced(self) -> None:
initialize_database(self.db_path)
connection = open_database(self.db_path)
diff --git a/client/test/test_http_admin_gateway.py b/client/test/test_http_admin_gateway.py
index 96ec09f..9fb7944 100644
--- a/client/test/test_http_admin_gateway.py
+++ b/client/test/test_http_admin_gateway.py
@@ -274,6 +274,7 @@ class HttpAdminGatewayTest(unittest.TestCase):
"type": "purchase",
"version": 1,
"priority": 0,
+ "execution_mode": "live",
"payload": {
"goods_url": "https://example.test/goods/PUR-001",
"goods_id": "737116531267",
@@ -292,11 +293,12 @@ class HttpAdminGatewayTest(unittest.TestCase):
ClientInfo("CLIENT-001"),
ClaimCapabilities(
supported_types=(TaskType.COLLECT, TaskType.PURCHASE),
- purchase_mode="dry_run",
+ purchase_mode="live",
),
)
self.assertEqual(task.task_type, TaskType.PURCHASE)
+ self.assertEqual(task.execution_mode, "live")
self.assertEqual(task.payload["max_price_cent"], 4200)
for missing in ("goods_id", "options", "quantity", "max_price_cent"):
@@ -318,6 +320,23 @@ class HttpAdminGatewayTest(unittest.TestCase):
raised.exception.code, "ADMIN_INVALID_RESPONSE"
)
+ invalid_mode_task = dict(valid_task)
+ invalid_mode_task["execution_mode"] = "unsafe"
+ invalid_gateway = HttpAdminGateway(
+ opener=RecordingOpener(
+ FakeResponse(200, {"task": invalid_mode_task})
+ )
+ )
+ with self.assertRaises(AdminGatewayError) as raised:
+ invalid_gateway.claim_next(
+ ClientInfo("CLIENT-001"),
+ ClaimCapabilities(
+ supported_types=(TaskType.COLLECT, TaskType.PURCHASE),
+ purchase_mode="live",
+ ),
+ )
+ 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_live_purchase_authorization.py b/client/test/test_live_purchase_authorization.py
new file mode 100644
index 0000000..df0c4fb
--- /dev/null
+++ b/client/test/test_live_purchase_authorization.py
@@ -0,0 +1,87 @@
+"""真实下单授权服务测试；不连接 Admin 或手机。"""
+
+import os
+import tempfile
+import unittest
+from pathlib import Path
+
+from src.live_purchase_authorization import (
+ LIVE_CONFIRMATION_TEXT,
+ LIVE_ENABLED_KEY,
+ LivePurchaseAuthorizationService,
+)
+from src.settings_repository import SettingsRepository
+
+
+class LivePurchaseAuthorizationTest(unittest.TestCase):
+ def setUp(self) -> None:
+ self.temporary_directory = tempfile.TemporaryDirectory()
+ repository = SettingsRepository(
+ Path(self.temporary_directory.name) / "client.db"
+ )
+ self.repository = repository
+ self.service = LivePurchaseAuthorizationService(repository)
+
+ def tearDown(self) -> None:
+ self.temporary_directory.cleanup()
+
+ def test_missing_setting_and_environment_always_default_to_dry_run(self):
+ os.environ["CMAUTOBUY_PURCHASE_MODE"] = "live"
+ try:
+ self.assertFalse(self.service.load().enabled)
+ self.assertEqual(
+ self.service.purchase_mode_for(
+ "CLIENT-1", "USB-1", live_adapter_ready=True
+ ),
+ "dry_run",
+ )
+ finally:
+ os.environ.pop("CMAUTOBUY_PURCHASE_MODE", None)
+
+ def test_enable_requires_exact_confirmation_and_binds_identity(self):
+ with self.assertRaisesRegex(ValueError, LIVE_CONFIRMATION_TEXT):
+ self.service.enable("CLIENT-1", "USB-1", "确认")
+ self.assertIsNone(self.repository.get(LIVE_ENABLED_KEY))
+
+ authorization = self.service.enable(
+ "CLIENT-1", "USB-1", LIVE_CONFIRMATION_TEXT
+ )
+
+ self.assertTrue(authorization.enabled)
+ self.assertTrue(authorization.matches("CLIENT-1", "USB-1"))
+ self.assertEqual(
+ self.service.purchase_mode_for(
+ "CLIENT-1", "USB-1", live_adapter_ready=True
+ ),
+ "live",
+ )
+
+ def test_binding_mismatch_or_missing_adapter_stays_dry_run(self):
+ self.service.enable("CLIENT-1", "USB-1", LIVE_CONFIRMATION_TEXT)
+
+ self.assertEqual(
+ self.service.purchase_mode_for(
+ "CLIENT-2", "USB-1", live_adapter_ready=True
+ ),
+ "dry_run",
+ )
+ self.assertEqual(
+ self.service.purchase_mode_for(
+ "CLIENT-1", "USB-2", live_adapter_ready=True
+ ),
+ "dry_run",
+ )
+ self.assertEqual(
+ self.service.purchase_mode_for(
+ "CLIENT-1", "USB-1", live_adapter_ready=False
+ ),
+ "dry_run",
+ )
+
+ def test_disable_is_immediate_and_does_not_require_confirmation(self):
+ self.service.enable("CLIENT-1", "USB-1", LIVE_CONFIRMATION_TEXT)
+
+ authorization = self.service.disable()
+
+ self.assertFalse(authorization.enabled)
+ self.assertFalse(self.service.load().enabled)
diff --git a/client/test/test_pdd_u2_purchase_adapter.py b/client/test/test_pdd_u2_purchase_adapter.py
index 50e8652..63c7ff8 100644
--- a/client/test/test_pdd_u2_purchase_adapter.py
+++ b/client/test/test_pdd_u2_purchase_adapter.py
@@ -5,6 +5,7 @@ import unittest
from src.pdd_device_service import PddDeviceService
from src.pdd_purchase_adapter import PddPurchaseError
from src.pdd_u2_purchase_adapter import U2PddPurchaseAdapter
+from src.pdd_u2_purchase_adapter import U2PddLivePurchaseAdapter
GOODS_URL = "https://mobile.yangkeduo.com/goods.html?goods_id=753136429979"
@@ -35,7 +36,8 @@ def panel_xml(quantity: int = 1) -> str:
-
+
"""
@@ -103,6 +105,23 @@ class U2PddPurchaseAdapterTest(unittest.TestCase):
select_size_fn=select_size_fn,
)
+ def _live_adapter(self, device, calls):
+ def select_color_fn(_device, _xml, target, **_kwargs):
+ calls.append(("color", target))
+ return target == "黑色"
+
+ def select_size_fn(_device, _xml, target, **_kwargs):
+ calls.append(("size", target))
+ return target == "3XL【140-165斤】"
+
+ return U2PddLivePurchaseAdapter(
+ "USB-001",
+ device_service=PddDeviceService(connector=lambda _serial: device),
+ sleeper=lambda _seconds: None,
+ select_color_fn=select_color_fn,
+ select_size_fn=select_size_fn,
+ )
+
def test_dry_run_reaches_confirmation_without_clicking_submit(self):
device = FakeDevice()
calls = []
@@ -168,6 +187,47 @@ class U2PddPurchaseAdapterTest(unittest.TestCase):
self.assertEqual(raised.exception.code, "PURCHASE_GOODS_URL_INVALID")
self.assertEqual(device.opened_urls, [])
+ def test_live_adapter_clicks_unique_submit_target_only_once(self):
+ device = FakeDevice()
+ adapter = self._live_adapter(device, [])
+ adapter.open_goods(GOODS_URL)
+ adapter.select_options(
+ {"color": "黑色", "size": "3XL【140-165斤】"}
+ )
+ adapter.set_quantity(2)
+
+ state = adapter.read_state()
+ adapter.submit_order_once()
+ with self.assertRaises(PddPurchaseError) as raised:
+ adapter.submit_order_once()
+
+ self.assertEqual(state.submit_candidate_count, 1)
+ self.assertEqual(raised.exception.code, "PURCHASE_SUBMIT_ALREADY_ATTEMPTED")
+ # 第一次点击采购入口，第二次且仅一次点击最终提交。
+ self.assertEqual(len(device.clicks), 2)
+ adapter.close()
+
+ def test_live_adapter_rejects_multiple_submit_targets_without_click(self):
+ device = FakeDevice()
+ adapter = self._live_adapter(device, [])
+ adapter.open_goods(GOODS_URL)
+ adapter.select_options(
+ {"color": "黑色", "size": "3XL【140-165斤】"}
+ )
+ device.special_xml = panel_xml().replace(
+ "",
+ '',
+ )
+ device.mode = "special"
+
+ with self.assertRaises(PddPurchaseError) as raised:
+ adapter.submit_order_once()
+
+ self.assertEqual(raised.exception.code, "PURCHASE_SUBMIT_TARGET_AMBIGUOUS")
+ self.assertEqual(len(device.clicks), 1)
+ adapter.close()
+
if __name__ == "__main__":
unittest.main()
diff --git a/client/test/test_purchase_recovery.py b/client/test/test_purchase_recovery.py
index 245d66d..c85e0ee 100644
--- a/client/test/test_purchase_recovery.py
+++ b/client/test/test_purchase_recovery.py
@@ -8,6 +8,7 @@ from src.admin_gateway import AdminTask, ClaimCapabilities, ClientInfo
from src.db import open_database
from src.mock_admin_gateway import MockAdminGateway
from src.pdd_purchase_adapter import (
+ PddLivePurchaseAdapter,
PddPurchaseAdapter,
PddPurchaseError,
PurchasePageState,
@@ -279,11 +280,13 @@ class PurchaseRecoveryTest(unittest.TestCase):
self.assertEqual(detail.current_step, "reconcile_manual_review")
self.assertIn("核对设备已断开", run.diagnostics_json["error"])
- def test_live_mode_and_order_submission_methods_are_unavailable(self):
- with self.assertRaisesRegex(ValueError, "dry_run"):
- ClaimCapabilities(purchase_mode="live")
- for name in ("submit_order", "pay", "payment"):
+ def test_live_submit_is_isolated_and_payment_methods_are_unavailable(self):
+ self.assertEqual(ClaimCapabilities(purchase_mode="live").purchase_mode, "live")
+ self.assertFalse(hasattr(PddPurchaseAdapter, "submit_order_once"))
+ self.assertTrue(hasattr(PddLivePurchaseAdapter, "submit_order_once"))
+ for name in ("submit_order", "pay", "payment", "cancel_order"):
self.assertFalse(hasattr(PddPurchaseAdapter, name))
+ self.assertFalse(hasattr(PddLivePurchaseAdapter, name))
self.assertFalse(hasattr(PddPurchaseReconcileAdapter, name))
diff --git a/client/test/test_purchase_task_service.py b/client/test/test_purchase_task_service.py
index 4346206..3344b3a 100644
--- a/client/test/test_purchase_task_service.py
+++ b/client/test/test_purchase_task_service.py
@@ -11,7 +11,12 @@ from src.admin_gateway import (
ClientInfo,
)
from src.mock_admin_gateway import MockAdminGateway
-from src.pdd_purchase_adapter import PddPurchaseAdapter, PurchasePageState
+from src.pdd_purchase_adapter import (
+ PddLivePurchaseAdapter,
+ PddPurchaseAdapter,
+ PddPurchaseError,
+ PurchasePageState,
+)
from src.purchase_task_service import PurchaseTaskService
from src.task_models import NewClaimedTask, TaskStatus, TaskType
from src.task_repository import TaskRepository
@@ -31,12 +36,16 @@ class RecordingDryRunAdapter(PddPurchaseAdapter):
candidate_count: int = 1,
forced_page: str = "",
wrong_options: bool = False,
+ in_stock: bool = True,
+ submit_candidate_count: int = 1,
) -> None:
self.goods_id = goods_id
self.price_cent = price_cent
self.candidate_count = candidate_count
self.forced_page = forced_page
self.wrong_options = wrong_options
+ self.in_stock = in_stock
+ self.submit_candidate_count = submit_candidate_count
self.options = {}
self.quantity = 0
self.page_kind = "goods"
@@ -57,6 +66,8 @@ class RecordingDryRunAdapter(PddPurchaseAdapter):
quantity=self.quantity,
price_cent=self.price_cent,
candidate_count=self.candidate_count,
+ in_stock=self.in_stock,
+ submit_candidate_count=self.submit_candidate_count,
)
def select_options(self, options) -> None:
@@ -78,6 +89,25 @@ class RecordingDryRunAdapter(PddPurchaseAdapter):
self.calls.append(("close",))
+class RecordingLiveAdapter(RecordingDryRunAdapter, PddLivePurchaseAdapter):
+ """只记录一次提交调用的 live 测试 Adapter。"""
+
+ def __init__(self, *, submit_error: bool = False, **kwargs) -> None:
+ super().__init__(**kwargs)
+ self.submit_error = submit_error
+ self.submit_count = 0
+
+ def submit_order_once(self) -> None:
+ self.submit_count += 1
+ self.calls.append(("submit_order_once",))
+ if self.submit_error:
+ raise PddPurchaseError(
+ "DEVICE_DISCONNECTED",
+ "点击后设备断开",
+ step="purchase_submit_once",
+ )
+
+
class PurchaseTaskServiceTest(unittest.TestCase):
def setUp(self) -> None:
self.temp_dir = tempfile.TemporaryDirectory()
@@ -90,12 +120,18 @@ class PurchaseTaskServiceTest(unittest.TestCase):
def tearDown(self) -> None:
self.temp_dir.cleanup()
- def _prepare_task(self, *, task_id: str = "PUR-001") -> None:
+ def _prepare_task(
+ self,
+ *,
+ task_id: str = "PUR-001",
+ execution_mode: str = "dry_run",
+ ) -> None:
task = AdminTask(
task_id=task_id,
task_type=TaskType.PURCHASE,
version=1,
priority=10,
+ execution_mode=execution_mode,
payload={
"goods_url": (
"https://mobile.yangkeduo.com/goods.html?"
@@ -113,7 +149,9 @@ class PurchaseTaskServiceTest(unittest.TestCase):
ClaimCapabilities(
device=AndroidDeviceInfo("USB-001"),
supported_types=(TaskType.PURCHASE,),
- purchase_mode="dry_run",
+ purchase_mode=(
+ "live" if execution_mode == "live" else "dry_run"
+ ),
),
)
assert claimed is not None
@@ -122,6 +160,7 @@ class PurchaseTaskServiceTest(unittest.TestCase):
remote_task_id=claimed.task_id,
task_type=claimed.task_type,
goods_url=str(claimed.payload["goods_url"]),
+ execution_mode=claimed.execution_mode,
goods_id=str(claimed.payload["goods_id"]),
quantity=int(claimed.payload["quantity"]),
priority=claimed.priority,
@@ -236,6 +275,59 @@ class PurchaseTaskServiceTest(unittest.TestCase):
assert detail is not None
self.assertEqual(detail.last_error_code, expected_code)
+ def test_live_marks_irreversible_before_single_submit_and_enters_reconcile(self):
+ self._prepare_task(execution_mode="live")
+ adapter = RecordingLiveAdapter()
+
+ outcome = self._service(adapter).execute_one_local()
+
+ self.assertEqual(outcome.kind, "manual_review")
+ self.assertEqual(adapter.submit_count, 1)
+ detail = self.repository.get_task("PUR-001")
+ run = self.repository.latest_task_run("PUR-001")
+ self.assertEqual(detail.status, TaskStatus.MANUAL_REVIEW)
+ self.assertEqual(detail.current_step, "reconcile_purchase")
+ self.assertIsNotNone(run.irreversible_action_at)
+ self.assertIsNotNone(run.order_submitted_at)
+ self.assertEqual(run.current_step, "reconcile_purchase")
+ self.assertIsNone(self.repository.next_pending_outbox())
+
+ def test_live_ambiguous_submit_target_stops_before_irreversible(self):
+ self._prepare_task(execution_mode="live")
+ adapter = RecordingLiveAdapter(submit_candidate_count=2)
+
+ outcome = self._service(adapter).execute_one_local()
+
+ self.assertEqual(outcome.kind, "failed")
+ self.assertEqual(adapter.submit_count, 0)
+ run = self.repository.latest_task_run("PUR-001")
+ self.assertIsNone(run.irreversible_action_at)
+
+ def test_live_submit_error_never_retries_and_still_enters_reconcile(self):
+ self._prepare_task(execution_mode="live")
+ adapter = RecordingLiveAdapter(submit_error=True)
+
+ outcome = self._service(adapter).execute_one_local()
+
+ self.assertEqual(outcome.kind, "manual_review")
+ self.assertEqual(adapter.submit_count, 1)
+ run = self.repository.latest_task_run("PUR-001")
+ self.assertIsNotNone(run.irreversible_action_at)
+ self.assertIsNone(run.order_submitted_at)
+ self.assertEqual(run.current_step, "reconcile_purchase")
+ self.assertIsNone(self.repository.next_pending_outbox())
+
+ def test_live_task_rejects_dry_adapter_before_device_action(self):
+ self._prepare_task(execution_mode="live")
+ adapter = RecordingDryRunAdapter()
+
+ outcome = self._service(adapter).execute_one_local()
+
+ self.assertEqual(outcome.kind, "failed")
+ self.assertEqual(adapter.calls, [("close",)])
+ run = self.repository.latest_task_run("PUR-001")
+ self.assertIsNone(run.irreversible_action_at)
+
def test_purchase_task_with_missing_safety_fields_is_reported(self):
task_id = "PUR-INVALID"
task = AdminTask(
diff --git a/client/test/test_settings_ui_event.py b/client/test/test_settings_ui_event.py
index f5dd9f7..7a4dd83 100644
--- a/client/test/test_settings_ui_event.py
+++ b/client/test/test_settings_ui_event.py
@@ -19,6 +19,10 @@ from src.android_device_service import (
AndroidWifiConversionResult,
)
from src.mock_admin_gateway import MockAdminGateway
+from src.live_purchase_authorization import (
+ LIVE_CONFIRMATION_TEXT,
+ LIVE_ENABLED_KEY,
+)
from src.selected_android_device_service import SELECTED_ANDROID_SERIAL_KEY
from src.settings_repository import SettingsRepository
from src.settings_ui import AndroidDeviceRow, SettingsPage
@@ -215,6 +219,63 @@ class SettingsPageEventTest(unittest.TestCase):
page.eventBinder.shutdown()
page.deleteLater()
+ def test_live_purchase_defaults_off_and_requires_exact_confirmation(self):
+ page = SettingsPage(
+ settings_repository=self.repository,
+ admin_gateway=MockAdminGateway(),
+ live_purchase_adapter_ready=True,
+ )
+
+ self.assertIn("已关闭", page.livePurchaseStatusLabel.text())
+ self.assertFalse(page.livePurchaseEnableButton.isEnabled())
+ page.livePurchaseConfirmationInput.setText("确认")
+ self.assertFalse(page.livePurchaseEnableButton.isEnabled())
+ page.livePurchaseConfirmationInput.setText(LIVE_CONFIRMATION_TEXT)
+ self.assertTrue(page.livePurchaseEnableButton.isEnabled())
+ page.eventBinder.shutdown()
+ page.deleteLater()
+
+ def test_enable_and_disable_live_purchase_updates_admin_capability(self):
+ self.repository.set_many(
+ {
+ "admin.client_id": "CLIENT-LIVE",
+ "admin.client_name": "采购电脑",
+ SELECTED_ANDROID_SERIAL_KEY: "USB-LIVE",
+ }
+ )
+ gateway = MockAdminGateway()
+ service = SequenceAndroidDeviceService([[]])
+ page = SettingsPage(
+ settings_repository=self.repository,
+ admin_gateway=gateway,
+ android_device_service=service,
+ live_purchase_adapter_ready=True,
+ )
+ self._wait_until(lambda: page.eventBinder._search_thread is None)
+ page.livePurchaseConfirmationInput.setText(LIVE_CONFIRMATION_TEXT)
+
+ page.livePurchaseEnableButton.click()
+ self._wait_until(
+ lambda: page.eventBinder._live_purchase_thread is None
+ )
+
+ self.assertIs(self.repository.get(LIVE_ENABLED_KEY), True)
+ self.assertIn("已启用", page.livePurchaseStatusLabel.text())
+ registered = gateway.registered_client("CLIENT-LIVE")
+ self.assertEqual(registered[1].purchase_mode, "live")
+
+ page.livePurchaseDisableButton.click()
+ self._wait_until(
+ lambda: page.eventBinder._live_purchase_thread is None
+ )
+
+ self.assertIs(self.repository.get(LIVE_ENABLED_KEY), False)
+ self.assertIn("已关闭", page.livePurchaseStatusLabel.text())
+ registered = gateway.registered_client("CLIENT-LIVE")
+ self.assertEqual(registered[1].purchase_mode, "dry_run")
+ page.eventBinder.shutdown()
+ page.deleteLater()
+
def test_save_persists_locally_and_registers_with_mock(self):
gateway = MockAdminGateway()
page = SettingsPage(
diff --git a/client/test/test_task_dispatcher.py b/client/test/test_task_dispatcher.py
index 89353af..6371658 100644
--- a/client/test/test_task_dispatcher.py
+++ b/client/test/test_task_dispatcher.py
@@ -7,7 +7,11 @@ from pathlib import Path
from src.android_device_service import AndroidDeviceSearchError
from src.admin_gateway import AdminTask, ClientInfo, SubmissionReceipt
from src.mock_admin_gateway import MockAdminGateway
-from src.pdd_purchase_adapter import PddPurchaseAdapter, PurchasePageState
+from src.pdd_purchase_adapter import (
+ PddLivePurchaseAdapter,
+ 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
@@ -70,12 +74,31 @@ class ReadyPurchaseAdapter(PddPurchaseAdapter):
pass
-def purchase_task(task_id="PUR-001"):
+class ReadyLivePurchaseAdapter(ReadyPurchaseAdapter, PddLivePurchaseAdapter):
+ def read_state(self):
+ state = super().read_state()
+ return PurchasePageState(
+ page_kind=state.page_kind,
+ goods_id=state.goods_id,
+ selected_options=state.selected_options,
+ quantity=state.quantity,
+ price_cent=state.price_cent,
+ candidate_count=state.candidate_count,
+ in_stock=True,
+ submit_candidate_count=1,
+ )
+
+ def submit_order_once(self):
+ self.calls.append(("purchase", "submitted_once"))
+
+
+def purchase_task(task_id="PUR-001", execution_mode="dry_run"):
return AdminTask(
task_id=task_id,
task_type=TaskType.PURCHASE,
version=1,
priority=10,
+ execution_mode=execution_mode,
payload={
"goods_url": "https://example.test/PUR-GOODS",
"goods_id": "PUR-GOODS",
@@ -103,7 +126,14 @@ class TaskDispatcherTest(unittest.TestCase):
def tearDown(self):
self.temporary.cleanup()
- def _dispatcher(self, *, purchase_ready, device_checker=lambda _serial: None):
+ def _dispatcher(
+ self,
+ *,
+ purchase_ready,
+ live_ready=False,
+ purchase_mode="dry_run",
+ device_checker=lambda _serial: None,
+ ):
purchase_factory = None
if purchase_ready:
purchase_factory = (
@@ -118,6 +148,16 @@ class TaskDispatcherTest(unittest.TestCase):
lambda *_args: RecordingCollector(self.calls)
),
purchase_adapter_factory=purchase_factory,
+ live_purchase_adapter_factory=(
+ (
+ lambda _address, _cancelled: ReadyLivePurchaseAdapter(
+ self.calls
+ )
+ )
+ if live_ready
+ else None
+ ),
+ purchase_mode=purchase_mode,
device_connection_checker=device_checker,
)
@@ -137,6 +177,61 @@ class TaskDispatcherTest(unittest.TestCase):
ready.claim_capabilities().purchase_mode, "dry_run"
)
+ def test_live_capability_requires_authorization_and_live_factory(self):
+ no_factory = self._dispatcher(
+ purchase_ready=True, purchase_mode="live"
+ )
+ authorized = self._dispatcher(
+ purchase_ready=True,
+ live_ready=True,
+ purchase_mode="live",
+ )
+
+ self.assertEqual(
+ no_factory.claim_capabilities().purchase_mode, "dry_run"
+ )
+ self.assertEqual(
+ authorized.claim_capabilities().purchase_mode, "live"
+ )
+
+ def test_local_live_task_never_runs_after_authorization_is_disabled(self):
+ task = purchase_task(execution_mode="live")
+ self.repository.add_claimed_task(admin_task_to_new_claimed_task(task))
+
+ with self.assertRaisesRegex(RuntimeError, "授权已关闭"):
+ self._dispatcher(
+ purchase_ready=True,
+ live_ready=True,
+ purchase_mode="dry_run",
+ ).execute_one()
+
+ self.assertNotIn(("purchase", "submitted_once"), self.calls)
+ self.assertEqual(
+ self.repository.get_task(task.task_id).status,
+ TaskStatus.CLAIMED,
+ )
+
+ def test_live_task_is_claimed_only_by_authorized_live_dispatcher(self):
+ task = purchase_task(execution_mode="live")
+ self.gateway.enqueue_task(task, self.client.client_id)
+
+ dry_outcome = self._dispatcher(purchase_ready=True).execute_one()
+ live_outcome = self._dispatcher(
+ purchase_ready=True,
+ live_ready=True,
+ purchase_mode="live",
+ ).execute_one()
+
+ self.assertEqual(dry_outcome.kind, "no_task")
+ self.assertEqual(live_outcome.kind, "manual_review")
+ self.assertEqual(
+ self.repository.get_task(task.task_id).execution_mode,
+ "live",
+ )
+ self.assertEqual(
+ self.calls.count(("purchase", "submitted_once")), 1
+ )
+
def test_claims_saves_then_dispatches_purchase_dry_run(self):
task = purchase_task()
self.gateway.enqueue_task(task, self.client.client_id)
diff --git a/client/test/test_task_repository.py b/client/test/test_task_repository.py
index efe1a83..8b5581e 100644
--- a/client/test/test_task_repository.py
+++ b/client/test/test_task_repository.py
@@ -32,10 +32,12 @@ class TaskRepositoryTests(unittest.TestCase):
task_type: TaskType = TaskType.COLLECT,
title: str = "测试商品",
goods_id: str = "10001",
+ execution_mode: str = "dry_run",
) -> NewClaimedTask:
return NewClaimedTask(
remote_task_id=remote_task_id,
task_type=task_type,
+ execution_mode=execution_mode,
goods_id=goods_id,
goods_url=f"https://example.test/goods/{goods_id}",
title=title,
@@ -96,6 +98,43 @@ class TaskRepositoryTests(unittest.TestCase):
self.assertEqual(detail.admin_payload["task_id"], "TASK-001")
self.assertEqual(detail.pdd_data["schema_version"], 1)
+ def test_live_mode_and_irreversible_marker_are_persisted_once(self) -> None:
+ self.repository.add_claimed_task(
+ self._task(
+ "PURCHASE-LIVE",
+ TaskType.PURCHASE,
+ execution_mode="live",
+ )
+ )
+ detail = self.repository.get_task("PURCHASE-LIVE")
+ self.assertEqual(detail.execution_mode, "live")
+
+ started = self.repository.start_purchase_run(
+ "PURCHASE-LIVE", "192.168.0.173:5555"
+ )
+ marked_at = self.repository.mark_purchase_irreversible(
+ "PURCHASE-LIVE", started.attempt_id
+ )
+ run = self.repository.latest_task_run("PURCHASE-LIVE")
+ self.assertEqual(run.irreversible_action_at, marked_at)
+ with self.assertRaisesRegex(ValueError, "已经存在"):
+ self.repository.mark_purchase_irreversible(
+ "PURCHASE-LIVE", started.attempt_id
+ )
+
+ def test_dry_run_cannot_enter_irreversible_stage(self) -> None:
+ self.repository.add_claimed_task(
+ self._task("PURCHASE-DRY", TaskType.PURCHASE)
+ )
+ started = self.repository.start_purchase_run(
+ "PURCHASE-DRY", "USB-001"
+ )
+
+ with self.assertRaisesRegex(ValueError, "演练任务"):
+ self.repository.mark_purchase_irreversible(
+ "PURCHASE-DRY", started.attempt_id
+ )
+
def test_duplicate_remote_task_id_does_not_overwrite(self) -> None:
self.repository.add_claimed_task(self._task("TASK-001"))
diff --git a/docs/client/02-architecture.md b/docs/client/02-architecture.md
index bc7a126..8f6e735 100644
--- a/docs/client/02-architecture.md
+++ b/docs/client/02-architecture.md
@@ -344,19 +344,21 @@ Client 与 Admin 的任务交互只有三种调用：领一个任务、提交结
### 任务领取与执行
-1. 优先补交一条本地 Outbox。提交结果不依赖 Android 设备。
-2. 没有待提交结果时，在工作线程通过 ADB 检查已保存设备号是否仍为 `device` 状态。检查失败立即停止，不执行本地任务，也不领取新任务。
-3. 执行最早的本地待处理任务；没有本地任务时才调用 Admin `claim`。返回 204 表示暂时没活，按轮询周期退避后再试。
-4. Client 持久化新领取的任务，状态置 `claimed`。
-5. 根据 `task_type` 分派给采集或采购执行器。
-6. 工作线程执行，持续更新 `current_step`（只写本地，不上报 Admin）。
-7. 完成后在同一事务里写入结果与 Outbox，状态置 `result_pending`。
-8. Outbox 提交成功、Admin 返回 `accepted: true` 后标记 `succeeded`。
-9. 回到第 1 步处理下一轮。**同一时间只做一个任务。**
+1. 若存在 `irreversible_action_at`，只允许先核对该采购订单；不得补交其他结果、执行任务或领取任务。
+2. 没有待核对采购时，优先补交一条本地 Outbox；提交结果不依赖 Android 设备。
+3. 没有待提交结果时，在工作线程通过 ADB 检查已保存设备号是否仍为 `device` 状态。检查失败立即停止，不执行本地任务，也不领取新任务。
+4. 执行最早的本地待处理任务；没有本地任务时才调用 Admin `claim`。返回 204 表示暂时没活，按轮询周期退避后再试。
+5. Client 持久化新领取的任务，状态置 `claimed`。
+6. 根据 `task_type` 分派给采集或采购执行器。
+7. 工作线程执行，持续更新 `current_step`（只写本地，不上报 Admin）。
+8. 完成后在同一事务里写入结果与 Outbox，状态置 `result_pending`。
+9. Outbox 提交成功、Admin 返回 `accepted: true` 后标记 `succeeded`，然后继续下一轮。**同一时间只做一个任务。**
-`TaskDispatcher` 是能力声明的唯一入口。没有可用采购演练 Adapter、没有已保存
+`TaskDispatcher` 是能力声明的唯一入口。没有可用采购 Adapter、没有已保存
Android 设备或本地持久化未准备好时，只声明 `collect`；条件满足时才声明
-`collect,purchase`，且 `purchase_mode` 永远是 `dry_run`。领取响应必须先完整校验并
+`collect,purchase`。真实下单授权默认关闭；只有本地授权同时匹配当前 Client ID、
+已保存 Android 设备且 live Adapter 就绪时才声明 `purchase_mode=live`，否则声明
+`dry_run`。领取响应必须先完整校验并
写入 SQLite，Repository 提交成功后才能分派，避免任务已在 Admin 领取却在本地丢失。
中途 Admin 是否取消了这个任务、是否重派给了别人，Client 不查也不管，做完照样提交——
@@ -385,11 +387,15 @@ reconcile_purchase(task, run) -> PurchaseResult | ManualReview
采购演练通过 `PddPurchaseAdapter` 的窄接口逐步读取最新页面状态。该接口只提供
打开商品、读取状态、精确选择动态规格、设置数量、进入提交前确认页和停止，
**不提供提交订单或付款方法**。这样即使应用层调用错误，也没有可误触的真实下单入口。
+真实采购另用 `PddLivePurchaseAdapter`，只增加 `submit_order_once`。服务层在最新
+页面复核规格、数量、价格、库存和唯一提交目标后，先提交
+`irreversible_action_at` 事务，再允许 Adapter 点击一次；之后无论点击结果是否明确，
+都只进入订单核对。该接口不提供付款或取消订单方法。
采购规格使用完整 `options` 对象精确比较，不假定只有颜色和尺码两个维度。
只读核对另用 `PddPurchaseReconcileAdapter`，只暴露 `read_order_match`
和 `close`，不暴露选规格、设数量、下单或付款方法。
-正式 Client 由 `pdd_u2_purchase_adapter.py` 实现上述演练接口，并在
+正式 Client 由 `pdd_u2_purchase_adapter.py` 分别实现演练和 live 接口，并在
`ui_main.py` 注入工厂。Adapter 通过 `PddDeviceService` 独占连接，每次判断
都重新读取当前包名和控件树。当前 Admin 下发的 `color` 和 `size`
使用精确文字匹配；其他动态维度直接停止，不做相似匹配。原生 PDD
diff --git a/docs/client/03-data-model.md b/docs/client/03-data-model.md
index cd3b6fc..3245924 100644
--- a/docs/client/03-data-model.md
+++ b/docs/client/03-data-model.md
@@ -125,6 +125,8 @@ CREATE TABLE pdd_tasks (
remote_task_id TEXT NOT NULL UNIQUE,
task_type TEXT NOT NULL
CHECK (task_type IN ('collect', 'purchase')),
+ execution_mode TEXT NOT NULL DEFAULT 'dry_run'
+ CHECK (execution_mode IN ('dry_run', 'live')),
goods_id TEXT,
goods_url TEXT NOT NULL,
title TEXT,
@@ -167,6 +169,7 @@ CREATE INDEX idx_pdd_tasks_goods_id
### 字段语义
- `remote_task_id`：跨 Admin、Client、日志、提交和归档使用的稳定任务编号。Mock 任务也必须分配稳定编号。
+- `execution_mode`：Admin 领取时确定的不可变执行模式；历史记录默认为 `dry_run`，Client 不得把任务自行提升为 `live`。
- `price_cent`：列表摘要价格；采集任务通常为最低可用 SKU 价格，详细价格以 `pdd_data.skus` 为准。
- `status`：**本机执行状态**，只由本机的执行流程和 Outbox 提交响应驱动。Admin 那边把任务标成什么，本地不知道也不需要知道（见 §3.1）。
- `current_step`：当前安全步骤，例如 `open_goods`、`collect_skus`、`select_options`、`placing_order`、`reconcile_order`。
@@ -407,12 +410,17 @@ CREATE TABLE app_settings (
- `automation.poll_interval_seconds`
- `automation.max_retries`
- `automation.dry_run`
+- `purchase.live_enabled`
+- `purchase.live_client_id`
+- `purchase.live_device_serial`
+- `purchase.live_confirmed_at`
- `safety.max_quantity`
- `safety.price_tolerance_cent`
- `diagnostics.artifact_directory`
- `diagnostics.retention_days`
访问令牌、密码和 Cookie 不得存入本表。
+`purchase.live_*` 只保存非敏感的本地安全门禁状态；缺失、损坏或绑定不一致时一律按关闭处理。
## 7. 任务状态转换
diff --git a/docs/client/04-admin-api-contract.md b/docs/client/04-admin-api-contract.md
index 2b89449..2a18f8d 100644
--- a/docs/client/04-admin-api-contract.md
+++ b/docs/client/04-admin-api-contract.md
@@ -210,9 +210,10 @@ POST /api/v1/client/tasks/claim
无可领取任务时返回 `204 No Content`。
Client 的 `HttpAdminGateway.claim_next` 已实现本接口，并原样序列化应用层已经
-安全确认的 `ClaimCapabilities`。没有采购演练 Adapter 或设备未准备好时只声明
-`supported_types: ["collect"]`；条件满足时声明 `collect,purchase`，但
-`purchase_mode` 固定为 `dry_run`。一次调用最多领取一个任务，结果或失败通过
+安全确认的 `ClaimCapabilities`。没有采购 Adapter 或设备未准备好时只声明
+`supported_types: ["collect"]`；条件满足时声明 `collect,purchase`。只有设置页
+授权同时匹配当前 Client ID、已保存 Android 设备且 live Adapter 就绪时才声明
+`purchase_mode: "live"`，其他情况均为 `dry_run`。一次调用最多领取一个任务，结果或失败通过
本页 §6 或 §7 提交，完整请求会先进入本地 Outbox。
采购响应在写入本地前必须校验非空 `goods_url`、`goods_id`、动态 `options`、
diff --git a/docs/client/05-ui-specification.md b/docs/client/05-ui-specification.md
index 1288eb2..7b9b7ce 100644
--- a/docs/client/05-ui-specification.md
+++ b/docs/client/05-ui-specification.md
@@ -64,8 +64,8 @@
才向 Admin 领取一条。领取、手机采集和提交都在单独的工作线程完成，一轮结束且
线程完全退出后，由主线程的单次定时器安排下一轮。暂无任务时默认 5 秒后重试；
可恢复的 Admin 错误按 5、10、20、30 秒退避。需要人工处理、不可恢复错误、设备
-或配置错误会停止自动获取。采购演练执行器和 Android 设备都准备好时，可以领取
-采购任务；界面必须明确显示“演练”，任何路径都不得显示成真实下单。
+或配置错误会停止自动获取。采购 Adapter 和 Android 设备都准备好时，可以领取
+采购任务；任务详情必须按任务的不可变执行模式明确显示“演练”或“真实下单”。
- 补交本地 Outbox 不依赖手机，可以在 Android 设备断开时继续执行。
- 执行本地任务或向 Admin 领取新任务前，工作线程必须用 `adb devices -l` 检查已保存的精确设备号。USB 和 Wi-Fi 设备都只接受 `device` 状态。
@@ -288,6 +288,14 @@ class TaskTableModel(QAbstractTableModel):
- 最大重试次数；
- 演练模式开关。
+### 真实下单（不支付）
+
+- 默认关闭，并明确显示当前状态、绑定 Client、绑定 Android 设备和确认时间；
+- 只有输入完整确认文字“创建未付款订单”后才能启用，不能通过环境变量或调试参数开启；
+- 启用和关闭在后台线程保存，执行期间禁用重复操作，失败时保留原状态并说明恢复办法；
+- 启用仅代表允许创建一笔未付款订单，不允许自动付款；设备或 Client 绑定变化后自动降级为演练；
+- 关闭操作始终可见，状态不能只用颜色表达。
+
### 安全与诊断
- 最大购买数量；
diff --git a/docs/client/06-quality-security.md b/docs/client/06-quality-security.md
index 019be64..57b28c6 100644
--- a/docs/client/06-quality-security.md
+++ b/docs/client/06-quality-security.md
@@ -81,10 +81,11 @@ PDD 解析测试优先使用脱敏的 XML 固件，不要求每次连接真实
7. 演练模式在代表性商品和设备上通过验收。
8. 操作人员明确确认真实下单范围和安全设置。
-真实下单不得通过调试参数、默认配置或界面误操作意外开启。
-当前版本的 `ClaimCapabilities` 仅接受 `purchase_mode=dry_run`，采购
-Adapter 不提供提交订单或付款方法。开发者不能把自动化测试通过当成真实
-下单门禁放行；仍需项目负责人和实际操作人员共同确认。
+真实下单不得通过调试参数、默认配置或界面误操作意外开启。当前版本仅在设置页
+精确输入确认文字后保存授权，并绑定 Client ID 和已保存 Android 设备；任一绑定
+不一致都降级为 `purchase_mode=dry_run`。演练 Adapter 不提供提交方法，live Adapter
+只提供一次性提交方法，两者都不提供付款或取消订单方法。开发者不能把自动化测试
+通过当成真机门禁放行；仍需项目负责人和实际操作人员共同确认。
### 3.1 当前自动化安全检查
@@ -93,12 +94,14 @@ Adapter 不提供提交订单或付款方法。开发者不能把自动化测试
- 有不可逆标记的中断只调用只读核对，采购 Adapter 调用次数为 0。
- 已落库 Outbox 只补交，不重跑手机流程。
- 停止、设备断开、验证码、登录失效和结果不确定都保留稳定错误和诊断。
-- `purchase_mode=live` 会在数据对象创建时被拒绝。
+- `execution_mode=live` 只允许采购任务，且 Client 不得自行升级 Admin 下发的模式。
+- 最终点击前重新读取页面并核对规格、数量、价格、库存和唯一提交目标。
+- `irreversible_action_at` 必须在点击前以事务提交；点击最多一次，返回或抛错后均只进入核单。
- 正式 uiautomator2 采购 Adapter 只点击商品页采购入口和目标规格，
- `enter_confirmation` 与 `stop_before_submit` 均为只读检查。
-- 单元测试必须断言到达最终确认页后没有产生第二次底部按钮点击。
+ 演练的 `enter_confirmation` 与 `stop_before_submit` 均为只读检查。
+- 单元测试必须断言演练没有最终点击、live 只有一次最终点击，且模糊目标不会点击。
-上述只证明演练和恢复代码的默认安全性，**不代表真实下单已放行**。
+上述只证明本地安全边界；真实设备联调仍须按独立验收工单获得当次明确授权。
## 4. 自动化防护