feat: 完善采购任务安全恢复 (#73)

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 not in {"dry_run", "live"}:
-            raise ValueError("purchase_mode 只能是 dry_run 或 live")
+        if self.purchase_mode != "dry_run":
+            raise ValueError("当前版本只允许 purchase_mode=dry_run")
         if not self.schema_versions or any(
             version <= 0 for version in self.schema_versions
         ):

client/src/pdd_purchase_reconcile_adapter.py

@@ -0,0 +1,48 @@
+"""采购结果只读核对适配层。
+
+该接口只允许读取订单候选，故意不提供商品选择、提交订单或付款方法。
+"""
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from typing import Mapping, Optional
+
+
+@dataclass(frozen=True)
+class PurchaseReconcileQuery:
+    """从本地任务与执行记录构造的核对条件。"""
+
+    goods_id: str
+    options: Mapping[str, str]
+    quantity: int
+    irreversible_action_at: str
+
+
+@dataclass(frozen=True)
+class PurchaseReconcileObservation:
+    """只读核对看到的结果。"""
+
+    match_status: str
+    order_no: Optional[str] = None
+    ordered_at: Optional[str] = None
+    diagnostics: Mapping[str, object] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        if self.match_status not in {
+            "matched", "not_found", "ambiguous", "unknown"
+        }:
+            raise ValueError("采购核对结果无效")
+
+
+class PddPurchaseReconcileAdapter(ABC):
+    """已进入不可逆阶段后使用的只读订单核对会话。"""
+
+    @abstractmethod
+    def read_order_match(
+        self, query: PurchaseReconcileQuery
+    ) -> PurchaseReconcileObservation:
+        """读取并核对订单候选；不得点击下单或付款。"""
+
+    @abstractmethod
+    def close(self) -> None:
+        """释放读取会话；不得在此方法中产生点击。"""

client/src/pdd_ui_event.py

@@ -42,6 +42,7 @@ 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_reconcile_service import PurchaseReconcileFactory
 from .selected_android_device_service import SelectedAndroidDeviceService
 from .settings_repository import SettingsRepository
 from .task_models import (
@@ -111,6 +112,7 @@ class ClaimTaskWorker(QObject):
         android_device_service: SelectedAndroidDeviceService,
         collect_service_factory: Optional[CollectServiceFactory] = None,
         purchase_adapter_factory: Optional[PurchaseAdapterFactory] = None,
+        purchase_reconcile_factory: Optional[PurchaseReconcileFactory] = None,
         selected_task_id: str = "",
     ) -> None:
         super().__init__()
@@ -121,6 +123,7 @@ class ClaimTaskWorker(QObject):
         self._cancelled = False
         self._collect_service_factory = collect_service_factory
         self._purchase_adapter_factory = purchase_adapter_factory
+        self._purchase_reconcile_factory = purchase_reconcile_factory
         self._selected_task_id = selected_task_id
 
     def cancel(self) -> None:
@@ -161,6 +164,7 @@ class ClaimTaskWorker(QObject):
                     android_serial or "",
                     collect_service_factory=self._collect_service_factory,
                     purchase_adapter_factory=self._purchase_adapter_factory,
+                    purchase_reconcile_factory=self._purchase_reconcile_factory,
                     cancelled=lambda: self._cancelled,
                 )
                 result = dispatcher.execute_one()
@@ -195,6 +199,7 @@ class PDDTaskPageEvent(QObject):
         settings_repository: Optional[SettingsRepository] = None,
         collect_service_factory: Optional[CollectServiceFactory] = None,
         purchase_adapter_factory: Optional[PurchaseAdapterFactory] = None,
+        purchase_reconcile_factory: Optional[PurchaseReconcileFactory] = None,
         next_task_delay_ms: int = 500,
         no_task_delay_ms: int = 5_000,
         retry_delays_ms: tuple[int, ...] = (5_000, 10_000, 20_000, 30_000),
@@ -215,6 +220,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._purchase_reconcile_factory = purchase_reconcile_factory
         if next_task_delay_ms < 0 or no_task_delay_ms <= 0:
             raise ValueError("自动获取等待时间配置无效")
         if not retry_delays_ms or any(value <= 0 for value in retry_delays_ms):
@@ -468,6 +474,7 @@ 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,
+            purchase_reconcile_factory=self._purchase_reconcile_factory,
         )
         worker.moveToThread(thread)
         thread.started.connect(worker.run)

client/src/purchase_reconcile_service.py

@@ -0,0 +1,138 @@
+"""不可逆阶段中断后的只读采购结果核对。"""
+
+from dataclasses import dataclass
+from typing import Callable, Mapping
+
+from .pdd_purchase_reconcile_adapter import (
+    PddPurchaseReconcileAdapter,
+    PurchaseReconcileObservation,
+    PurchaseReconcileQuery,
+)
+from .task_models import TaskDetail
+from .task_repository import TaskRepository
+
+
+PurchaseReconcileFactory = Callable[
+    [str, Callable[[], bool]], PddPurchaseReconcileAdapter
+]
+
+
+@dataclass(frozen=True)
+class PurchaseReconcileOutcome:
+    """只读核对的简短结果。"""
+
+    kind: str
+    message: str
+    task_id: str = ""
+
+
+class PurchaseReconcileService:
+    """只读核对一条已进入不可逆阶段的采购运行。"""
+
+    def __init__(
+        self,
+        repository: TaskRepository,
+        device_address: str,
+        adapter_factory: PurchaseReconcileFactory,
+        *,
+        cancelled: Callable[[], bool] = lambda: False,
+    ) -> None:
+        self._repository = repository
+        self._device_address = str(device_address or "").strip()
+        self._factory = adapter_factory
+        self._cancelled = cancelled
+
+    def execute_selected(
+        self, remote_task_id: str
+    ) -> PurchaseReconcileOutcome:
+        """执行一次只读核对，任何结果都交给人工最终确认。"""
+
+        if not self._device_address:
+            raise ValueError("请先在设置页选择并保存 Android 设备")
+        if self._cancelled():
+            return PurchaseReconcileOutcome(
+                "cancelled", "采购结果核对已取消", remote_task_id
+            )
+        task = self._repository.get_task(remote_task_id)
+        run = self._repository.latest_task_run(remote_task_id)
+        if task is None or run is None:
+            raise ValueError(f"任务 {remote_task_id} 或执行记录不存在")
+        if run.irreversible_action_at is None:
+            raise ValueError("未进入不可逆阶段，不应启动订单核对")
+
+        query = self._query(task, run.irreversible_action_at)
+        adapter = None
+        close_error = ""
+        try:
+            adapter = self._factory(self._device_address, self._cancelled)
+            observation = adapter.read_order_match(query)
+            if not isinstance(observation, PurchaseReconcileObservation):
+                raise TypeError("采购核对 Adapter 返回值无效")
+        except Exception as exc:
+            observation = PurchaseReconcileObservation(
+                "unknown", diagnostics={"error": str(exc)}
+            )
+        finally:
+            if adapter is not None:
+                try:
+                    adapter.close()
+                except Exception as exc:
+                    close_error = str(exc)
+
+        diagnostics = dict(observation.diagnostics)
+        diagnostics.update(
+            {
+                "order_no": observation.order_no,
+                "ordered_at": observation.ordered_at,
+                "mode": "reconcile_only",
+            }
+        )
+        if close_error:
+            diagnostics["close_error"] = close_error
+        self._repository.save_purchase_reconciliation(
+            remote_task_id,
+            run.attempt_id,
+            observation.match_status,
+            diagnostics,
+        )
+        if observation.match_status == "matched":
+            message = (
+                f"任务 {remote_task_id} 仅核对到唯一候选订单；"
+                "请人工确认，程序不会重新下单"
+            )
+        else:
+            message = (
+                f"任务 {remote_task_id} 核对结果不确定；"
+                "需人工处理，程序不会重新下单"
+            )
+        return PurchaseReconcileOutcome(
+            "manual_review", message, remote_task_id
+        )
+
+    @staticmethod
+    def _query(
+        task: TaskDetail, irreversible_action_at: str
+    ) -> PurchaseReconcileQuery:
+        payload_root = task.admin_payload
+        payload = payload_root.get("payload")
+        if not isinstance(payload, Mapping):
+            raise ValueError("采购任务缺少 payload")
+        options = payload.get("options")
+        if not isinstance(options, Mapping) or not options:
+            raise ValueError("采购任务缺少 options")
+        goods_id = str(payload.get("goods_id") or "").strip()
+        quantity = payload.get("quantity")
+        if not goods_id:
+            raise ValueError("采购任务缺少 goods_id")
+        if (
+            isinstance(quantity, bool)
+            or not isinstance(quantity, int)
+            or quantity <= 0
+        ):
+            raise ValueError("采购任务缺少有效 quantity")
+        return PurchaseReconcileQuery(
+            goods_id=goods_id,
+            options={str(k): str(v) for k, v in options.items()},
+            quantity=quantity,
+            irreversible_action_at=irreversible_action_at,
+        )

client/src/purchase_task_service.py

@@ -164,7 +164,7 @@ class PurchaseTaskService:
                 exc.code,
                 exc.message,
                 exc.retryable,
-                exc.step or step,
+                step,
                 exc.diagnostics,
             )
         except Exception as exc:
@@ -367,9 +367,9 @@ class PurchaseTaskService:
     ) -> OutboxEventRecord:
         if code == "PURCHASE_CANCELLED":
             status = TaskStatus.CANCELLED
-        elif retryable:
-            status = TaskStatus.RETRY_WAIT
         else:
+            # 采购没有手动“重新执行”入口。即使错误属于
+            # 技术上可重试，也先留给人工判断，避免隐式重复采购。
             status = TaskStatus.MANUAL_REVIEW
         return self._repository.save_purchase_failure(
             remote_task_id,

client/src/task_detail_view.py

@@ -59,6 +59,20 @@ CURRENT_STEP_TEXT = {
     "completed": "已完成",
     "failed": "执行失败",
     "interrupted": "上次执行中断",
+    "purchase_prepare": "正在准备采购演练",
+    "purchase_open_goods": "正在打开采购商品页",
+    "purchase_verify_goods": "正在核对商品",
+    "purchase_select_options": "正在选择采购规格",
+    "purchase_verify_options": "正在核对采购规格",
+    "purchase_set_quantity": "正在设置采购数量",
+    "purchase_verify_quantity_price": "正在核对数量和价格",
+    "purchase_enter_confirmation": "正在进入提交前确认页",
+    "purchase_verify_confirmation": "正在核对提交前确认页",
+    "purchase_dry_run_stopped": "采购演练已在提交前停止",
+    "purchase_recovery_ready": "上次演练中断，已安全等待恢复",
+    "reconcile_purchase": "只允许核对订单",
+    "reconcile_completed": "只读核对完成，等待人工确认",
+    "reconcile_manual_review": "核对结果不确定，需人工处理",
 }
 
 

client/src/task_dispatcher.py

@@ -18,6 +18,10 @@ from .purchase_task_service import (
     PurchaseAdapterFactory,
     PurchaseTaskService,
 )
+from .purchase_reconcile_service import (
+    PurchaseReconcileFactory,
+    PurchaseReconcileService,
+)
 from .task_models import (
     NewClaimedTask,
     OutboxEventRecord,
@@ -120,6 +124,7 @@ class TaskDispatcher:
         *,
         collect_service_factory: Optional[CollectServiceFactory] = None,
         purchase_adapter_factory: Optional[PurchaseAdapterFactory] = None,
+        purchase_reconcile_factory: Optional[PurchaseReconcileFactory] = None,
         cancelled: Callable[[], bool] = lambda: False,
     ) -> None:
         self._gateway = gateway
@@ -128,6 +133,7 @@ class TaskDispatcher:
         self._device_address = str(device_address or "").strip()
         self._collect_factory = collect_service_factory
         self._purchase_factory = purchase_adapter_factory
+        self._reconcile_factory = purchase_reconcile_factory
         self._cancelled = cancelled
 
     @property
@@ -165,6 +171,23 @@ class TaskDispatcher:
         if not self._device_address:
             raise ValueError("请先在设置页选择并保存 Android 设备")
 
+        reconcile_task = self._repository.next_purchase_reconcile_task()
+        if reconcile_task is not None:
+            if self._reconcile_factory is None:
+                raise RuntimeError(
+                    f"任务 {reconcile_task.remote_task_id} 只允许核对订单，"
+                    "但只读核对执行器未就绪；不会重新下单"
+                )
+            outcome = PurchaseReconcileService(
+                self._repository,
+                self._device_address,
+                self._reconcile_factory,
+                cancelled=self._cancelled,
+            ).execute_selected(reconcile_task.remote_task_id)
+            return TaskDispatchOutcome(
+                outcome.kind, outcome.message, outcome.task_id
+            )
+
         if not self.purchase_ready:
             pending_purchase = self._repository.next_purchase_task()
             if pending_purchase is not None:

client/src/task_repository.py

@@ -23,6 +23,7 @@ from .task_models import (
     TaskStatus,
     TaskSummary,
     TaskType,
+    TaskRunRecord,
 )
 
 
@@ -156,7 +157,7 @@ class TaskRepository:
         return self._to_detail(row) if row is not None else None
 
     def recover_interrupted_work(self) -> None:
-        """恢复上次异常退出留下的可重试状态。"""
+        """恢复上次异常退出留下的任务，不猜测内存状态。"""
 
         now = utc_now_iso()
         connection = open_database(self._db_path)
@@ -193,6 +194,64 @@ class TaskRepository:
                     " AND task_id IN (SELECT id FROM pdd_tasks WHERE task_type = 'collect')",
                     (now, now),
                 )
+                purchase_runs = connection.execute(
+                    "SELECT t.id AS task_id, r.irreversible_action_at"
+                    " 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'"
+                    " ORDER BY r.attempt_no DESC"
+                ).fetchall()
+                recovered_task_ids = set()
+                for run in purchase_runs:
+                    task_id = int(run["task_id"])
+                    if task_id in recovered_task_ids:
+                        continue
+                    recovered_task_ids.add(task_id)
+                    irreversible = connection.execute(
+                        "SELECT 1 FROM task_runs WHERE task_id = ?"
+                        " AND run_status = 'running'"
+                        " AND irreversible_action_at IS NOT NULL LIMIT 1",
+                        (task_id,),
+                    ).fetchone()
+                    if irreversible is not None:
+                        message = (
+                            "上次采购在不可逆阶段中断，"
+                            "只允许核对订单"
+                        )
+                        connection.execute(
+                            "UPDATE task_runs SET run_status = 'manual_review',"
+                            " current_step = 'reconcile_purchase',"
+                            " error_code = 'PURCHASE_OUTCOME_UNKNOWN',"
+                            " error_message = ?, finished_at = ?, updated_at = ?"
+                            " WHERE task_id = ? AND run_status = 'running'",
+                            (message, now, now, task_id),
+                        )
+                        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),
+                        )
+                    else:
+                        message = "采购演练上次执行中断，已关闭旧执行记录"
+                        connection.execute(
+                            "UPDATE task_runs SET run_status = 'failed',"
+                            " error_code = 'CLIENT_INTERRUPTED',"
+                            " error_message = ?, finished_at = ?, updated_at = ?"
+                            " WHERE task_id = ? AND run_status = 'running'",
+                            (message, now, now, task_id),
+                        )
+                        connection.execute(
+                            "UPDATE pdd_tasks SET status = 'claimed',"
+                            " current_step = 'purchase_recovery_ready',"
+                            " retry_count = retry_count + 1,"
+                            " last_error_code = 'CLIENT_INTERRUPTED',"
+                            " last_error_message = ?, finished_at = NULL,"
+                            " updated_at = ? WHERE id = ?",
+                            (message, now, task_id),
+                        )
         finally:
             connection.close()
 
@@ -224,6 +283,39 @@ class TaskRepository:
             connection.close()
         return self._to_detail(row) if row is not None else None
 
+    def next_purchase_reconcile_task(self) -> Optional[TaskDetail]:
+        """返回最早一条只允许读取核对的采购任务。"""
+
+        connection = open_database(self._db_path)
+        try:
+            row = connection.execute(
+                "SELECT * FROM pdd_tasks WHERE task_type = 'purchase'"
+                " AND status = 'manual_review'"
+                " AND current_step = 'reconcile_purchase'"
+                " ORDER BY received_at ASC, id ASC LIMIT 1"
+            ).fetchone()
+        finally:
+            connection.close()
+        return self._to_detail(row) if row is not None else None
+
+    def latest_task_run(
+        self, remote_task_id: str
+    ) -> Optional[TaskRunRecord]:
+        """返回任务最新的执行记录。"""
+
+        connection = open_database(self._db_path)
+        try:
+            row = connection.execute(
+                "SELECT r.* FROM task_runs r"
+                " JOIN pdd_tasks t ON t.id = r.task_id"
+                " WHERE t.remote_task_id = ?"
+                " ORDER BY r.attempt_no DESC LIMIT 1",
+                (remote_task_id,),
+            ).fetchone()
+        finally:
+            connection.close()
+        return self._to_task_run(row) if row is not None else None
+
     def next_runnable_task(
         self, *, include_purchase: bool
     ) -> Optional[TaskDetail]:
@@ -479,6 +571,92 @@ class TaskRepository:
         finally:
             connection.close()
 
+    def save_purchase_reconciliation(
+        self,
+        remote_task_id: str,
+        attempt_id: str,
+        match_status: str,
+        diagnostics: Dict[str, object],
+    ) -> None:
+        """保存一次只读订单核对结果，任务仍留给人工确认。"""
+
+        allowed = {"matched", "not_found", "ambiguous", "unknown"}
+        if match_status not in allowed:
+            raise ValueError("采购核对结果无效")
+        now = utc_now_iso()
+        step = (
+            "reconcile_completed"
+            if match_status == "matched"
+            else "reconcile_manual_review"
+        )
+        messages = {
+            "matched": "只读核对发现唯一候选订单，请人工确认",
+            "not_found": "只读核对未找到订单，不得重新下单",
+            "ambiguous": "只读核对发现多个候选订单，请人工确认",
+            "unknown": "无法确定采购结果，不得重新下单",
+        }
+        message = messages[match_status]
+        saved_diagnostics = dict(diagnostics)
+        saved_diagnostics["match_status"] = match_status
+        connection = open_database(self._db_path)
+        try:
+            with connection:
+                task = connection.execute(
+                    "SELECT id, task_type, status, current_step 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["status"] != TaskStatus.MANUAL_REVIEW.value
+                    or task["current_step"] != "reconcile_purchase"
+                ):
+                    raise ValueError("采购任务当前不在待核对状态")
+                run = connection.execute(
+                    "SELECT irreversible_action_at, run_status, current_step"
+                    " FROM task_runs"
+                    " WHERE task_id = ? AND attempt_id = ?",
+                    (task["id"], attempt_id),
+                ).fetchone()
+                if run is None or run["irreversible_action_at"] is None:
+                    raise ValueError("只有已进入不可逆阶段的运行才能核对")
+                if (
+                    run["run_status"] != RunStatus.MANUAL_REVIEW.value
+                    or run["current_step"] != "reconcile_purchase"
+                ):
+                    raise ValueError("采购执行记录已经核对或状态已变更")
+                run_cursor = connection.execute(
+                    "UPDATE task_runs SET run_status = 'manual_review',"
+                    " current_step = ?, error_code = 'PURCHASE_RECONCILED',"
+                    " error_message = ?, diagnostics_json = ?, updated_at = ?"
+                    " WHERE task_id = ? AND attempt_id = ?",
+                    (
+                        step,
+                        message,
+                        json.dumps(saved_diagnostics, ensure_ascii=False),
+                        now,
+                        task["id"],
+                        attempt_id,
+                    ),
+                )
+                if run_cursor.rowcount != 1:
+                    raise ValueError("采购执行记录核对保存失败")
+                task_cursor = connection.execute(
+                    "UPDATE pdd_tasks SET status = 'manual_review',"
+                    " current_step = ?, last_error_code = 'PURCHASE_RECONCILED',"
+                    " last_error_message = ?, updated_at = ? WHERE id = ?"
+                    " AND status = 'manual_review'"
+                    " AND current_step = 'reconcile_purchase'",
+                    (step, message, now, task["id"]),
+                )
+                if task_cursor.rowcount != 1:
+                    raise ValueError("采购任务状态已变更，核对结果未保存")
+        finally:
+            connection.close()
+
     def save_purchase_result(
         self,
         remote_task_id: str,
@@ -628,7 +806,8 @@ class TaskRepository:
                 }.get(status, RunStatus.FAILED)
                 cursor = connection.execute(
                     "UPDATE task_runs SET run_status = ?, current_step = ?,"
-                    " error_code = ?, error_message = ?, finished_at = ?,"
+                    " error_code = ?, error_message = ?, diagnostics_json = ?,"
+                    " finished_at = ?,"
                     " updated_at = ? WHERE task_id = ? AND attempt_id = ?"
                     " AND run_status = 'running'",
                     (
@@ -636,6 +815,7 @@ class TaskRepository:
                         checked_step,
                         error_code,
                         error_message,
+                        json.dumps(diagnostics or {}, ensure_ascii=False),
                         now,
                         now,
                         task["id"],
@@ -935,6 +1115,39 @@ class TaskRepository:
             sent_at=row["sent_at"],
         )
 
+    @staticmethod
+    def _to_task_run(row: sqlite3.Row) -> TaskRunRecord:
+        diagnostics = (
+            TaskRepository._load_json_object(row["diagnostics_json"])
+            if row["diagnostics_json"] is not None
+            else None
+        )
+        result_data = (
+            TaskRepository._load_json_object(row["result_data"])
+            if row["result_data"] is not None
+            else None
+        )
+        return TaskRunRecord(
+            id=row["id"],
+            task_id=row["task_id"],
+            attempt_id=row["attempt_id"],
+            attempt_no=row["attempt_no"],
+            device_address=row["device_address"],
+            run_status=RunStatus(row["run_status"]),
+            current_step=row["current_step"],
+            started_at=row["started_at"],
+            finished_at=row["finished_at"],
+            irreversible_action_at=row["irreversible_action_at"],
+            order_submitted_at=row["order_submitted_at"],
+            error_code=row["error_code"],
+            error_message=row["error_message"],
+            diagnostics_json=diagnostics,
+            result_data=result_data,
+            artifact_directory=row["artifact_directory"],
+            created_at=row["created_at"],
+            updated_at=row["updated_at"],
+        )
+
     @staticmethod
     def _validate_page(limit: int, offset: int) -> None:
         if not 1 <= limit <= MAX_PAGE_SIZE: