fix: 修复重新上报任务状态不一致 (#91)

2026-08-10 11:18:06 +08:00
parent d81aa25b13
commit 077cba6831
7 changed files with 352 additions and 27 deletions
@@ -50,6 +50,7 @@ from .purchase_reconcile_service import PurchaseReconcileFactory
 from .selected_android_device_service import SelectedAndroidDeviceService
 from .settings_repository import SettingsRepository
 from .task_models import (
+    OutboxEventType,
    OutboxStatus,
    TaskFilters,
    TaskStatus,
@@ -213,7 +214,7 @@ class ClaimTaskWorker(QObject):


 class ResultResubmitWorker(QObject):
-    """在后台逐条重发既有结果 Outbox，不执行任何手机操作。"""
+    """在后台逐条重发既有 Outbox，不执行任何手机操作。"""

    progress = pyqtSignal(int, int, str)
    finished = pyqtSignal(object, object, object)
@@ -248,7 +249,7 @@ class ResultResubmitWorker(QObject):
                    break
                self.progress.emit(current, total, task_id)
                try:
-                    event = self._repository.latest_result_outbox(task_id)
+                    event = self._repository.outbox_for_resubmit(task_id)
                except Exception:
                    skipped.append(task_id)
                    continue
@@ -258,11 +259,18 @@ class ResultResubmitWorker(QObject):

                self._repository.mark_outbox_sending(event.id)
                try:
-                    receipt = self._gateway.submit_result(
-                        task_id,
-                        event.idempotency_key,
-                        event.payload_json,
-                    )
+                    if event.event_type is OutboxEventType.TASK_FAILURE:
+                        receipt = self._gateway.submit_failure(
+                            task_id,
+                            event.idempotency_key,
+                            event.payload_json,
+                        )
+                    else:
+                        receipt = self._gateway.submit_result(
+                            task_id,
+                            event.idempotency_key,
+                            event.payload_json,
+                        )
                    if not receipt.accepted:
                        raise AdminGatewayError(
                            "ADMIN_RESULT_NOT_ACCEPTED",
@@ -469,7 +477,7 @@ class PDDTaskPageEvent(QObject):

    @pyqtSlot(object)
    def request_resubmit(self, task_ids) -> None:
-        """确认后在后台重发勾选任务的原始结果 Outbox。"""
+        """确认后在后台重发勾选任务尚未发送或最新的 Outbox。"""

        stable_ids = tuple(
            dict.fromkeys(str(value) for value in task_ids if value)
@@ -490,8 +498,9 @@ class PDDTaskPageEvent(QObject):
            return

        dialog = MessageBox(
-            f"重新上报 {len(stable_ids)} 条任务结果？",
-            "只会重新提交本地已经保存的原始结果，"
+            f"重新上报 {len(stable_ids)} 条任务数据？",
+            "会优先提交尚未发送的结果或失败信息；没有待发送数据时，"
+            "才重新提交本地已经保存的最新结果。"
            "不会重新采集、采购或操作 Android 手机，也不会修改本地结果。",
            self._page.window(),
        )
@@ -504,13 +513,13 @@ class PDDTaskPageEvent(QObject):
        self._start_resubmit_worker(stable_ids)

    def _start_resubmit_worker(self, task_ids: tuple[str, ...]) -> None:
-        """启动只处理指定结果 Outbox 的工作线程。"""
+        """启动只处理指定任务 Outbox 的工作线程。"""

        assert self._claim_gateway is not None
        self._resubmit_busy = True
        self._page.set_resubmit_running(True)
        self._page.set_engine_status(
-            f"正在准备重新上报 {len(task_ids)} 条任务结果…"
+            f"正在准备重新上报 {len(task_ids)} 条任务数据…"
        )

        thread = QThread(self)
@@ -408,18 +408,30 @@ class TaskRepository:
            }.get(row["status"], row["status"])
            raise CollectRerunError(f"任务当前为“{status_name}”，不能重新采集")
        unsent = connection.execute(
-            "SELECT status, last_error FROM outbox_events"
+            "SELECT event_type, status, last_error FROM outbox_events"
            " WHERE task_id = ? AND status != 'sent'"
            " ORDER BY id DESC LIMIT 1",
            (row["id"],),
        ).fetchone()
        if unsent is not None and unsent["status"] == OutboxStatus.FAILED.value:
            reason = unsent["last_error"] or "Admin 未接收上次结果"
+            event_name = (
+                "失败信息"
+                if unsent["event_type"] == OutboxEventType.TASK_FAILURE.value
+                else "结果"
+            )
            raise CollectRerunError(
-                f"任务上次结果提交失败：{reason}；请先处理后再重新采集"
+                f"任务上次{event_name}提交失败：{reason}；"
+                "请先勾选任务点击“重新上报”"
            )
        if unsent is not None:
-            raise CollectRerunError("任务仍有未发送的结果，请先完成提交")
+            if unsent["event_type"] == OutboxEventType.TASK_FAILURE.value:
+                raise CollectRerunError(
+                    "任务仍有未上报的失败信息，请先勾选任务点击“重新上报”"
+                )
+            raise CollectRerunError(
+                "任务仍有未发送的结果，请先勾选任务点击“重新上报”"
+            )

    def start_collect_run(
        self, remote_task_id: str, device_address: str
@@ -1017,6 +1029,37 @@ class TaskRepository:
            connection.close()
        return self._to_outbox(row) if row is not None else None

+    def outbox_for_resubmit(
+        self, remote_task_id: str
+    ) -> Optional[OutboxEventRecord]:
+        """返回手工重新上报应发送的事件。
+
+        先返回最早一条尚未发送的事件，保证失败信息不会被旧结果跳过；
+        全部事件都已发送时，才返回最新的成功结果用于再次确认。
+        """
+
+        connection = open_database(self._db_path)
+        try:
+            row = connection.execute(
+                "SELECT o.* FROM outbox_events o"
+                " JOIN pdd_tasks t ON t.id = o.task_id"
+                " WHERE t.remote_task_id = ? AND o.status != 'sent'"
+                " ORDER BY o.id ASC LIMIT 1",
+                (remote_task_id,),
+            ).fetchone()
+            if row is None:
+                row = connection.execute(
+                    "SELECT o.* FROM outbox_events o"
+                    " JOIN pdd_tasks t ON t.id = o.task_id"
+                    " WHERE t.remote_task_id = ?"
+                    " AND o.event_type IN ('collect_result', 'purchase_result')"
+                    " ORDER BY o.id DESC LIMIT 1",
+                    (remote_task_id,),
+                ).fetchone()
+        finally:
+            connection.close()
+        return self._to_outbox(row) if row is not None else None
+
    def get_outbox_event(self, event_id: int) -> Optional[OutboxEventRecord]:
        connection = open_database(self._db_path)
        try:
@@ -1051,34 +1094,133 @@ class TaskRepository:
        self._update_outbox(event_id, "failed", message, increment=True)

    def mark_outbox_sent(self, event_id: int) -> None:
+        """标记事件已发送，并只按当前最新执行更新任务状态。
+
+        已经发送过的历史结果再次上报时，不得覆盖任务的最新状态。
+        最新失败信息发送成功时会恢复对应失败状态，用于修复旧版本
+        产生的“表格显示已完成、实际最新执行失败”记录。
+        """
+
        now = utc_now_iso()
        connection = open_database(self._db_path)
        try:
            with connection:
                row = connection.execute(
-                    "SELECT task_id, event_type FROM outbox_events WHERE id = ?",
+                    "SELECT task_id, event_type, payload_json, sent_at"
+                    " FROM outbox_events WHERE id = ?",
                    (event_id,),
                ).fetchone()
                if row is None:
                    raise ValueError(f"Outbox {event_id} 不存在")
+                was_already_sent = row["sent_at"] is not None
                connection.execute(
                    "UPDATE outbox_events SET status = 'sent', sent_at = ?,"
                    " updated_at = ?, attempt_count = attempt_count + 1"
                    " WHERE id = ?",
                    (now, now, event_id),
                )
-                if row["event_type"] in {
+                payload = self._load_outbox_payload(row["payload_json"])
+                if row["event_type"] == OutboxEventType.TASK_FAILURE.value:
+                    self._restore_current_failure_state(
+                        connection, row["task_id"], payload, now
+                    )
+                elif not was_already_sent and row["event_type"] in {
                    OutboxEventType.COLLECT_RESULT.value,
                    OutboxEventType.PURCHASE_RESULT.value,
                }:
-                    connection.execute(
-                        "UPDATE pdd_tasks SET status = 'succeeded',"
-                        " current_step = 'completed', updated_at = ? WHERE id = ?",
-                        (now, row["task_id"]),
+                    self._complete_current_result(
+                        connection, row["task_id"], payload, now
                    )
        finally:
            connection.close()

+    @staticmethod
+    def _load_outbox_payload(payload_json: str) -> Dict[str, object]:
+        """读取本地 Outbox JSON；损坏时返回空对象，不猜测任务状态。"""
+
+        try:
+            payload = json.loads(payload_json)
+        except (TypeError, ValueError):
+            return {}
+        return payload if isinstance(payload, dict) else {}
+
+    @staticmethod
+    def _latest_run_for_task(
+        connection: sqlite3.Connection, task_id: int
+    ) -> Optional[sqlite3.Row]:
+        return connection.execute(
+            "SELECT attempt_id, finished_at FROM task_runs"
+            " WHERE task_id = ? ORDER BY attempt_no DESC LIMIT 1",
+            (task_id,),
+        ).fetchone()
+
+    @classmethod
+    def _complete_current_result(
+        cls,
+        connection: sqlite3.Connection,
+        task_id: int,
+        payload: Dict[str, object],
+        now: str,
+    ) -> None:
+        """只有最新执行的待提交结果可以把任务改成已完成。"""
+
+        latest_run = cls._latest_run_for_task(connection, task_id)
+        if (
+            latest_run is None
+            or payload.get("attempt_id") != latest_run["attempt_id"]
+        ):
+            return
+        connection.execute(
+            "UPDATE pdd_tasks SET status = 'succeeded',"
+            " current_step = 'completed', updated_at = ?"
+            " WHERE id = ? AND status = 'result_pending'",
+            (now, task_id),
+        )
+
+    @classmethod
+    def _restore_current_failure_state(
+        cls,
+        connection: sqlite3.Connection,
+        task_id: int,
+        payload: Dict[str, object],
+        now: str,
+    ) -> None:
+        """最新失败事件上报后恢复任务失败状态，旧事件不覆盖新状态。"""
+
+        latest_run = cls._latest_run_for_task(connection, task_id)
+        if (
+            latest_run is None
+            or payload.get("attempt_id") != latest_run["attempt_id"]
+        ):
+            return
+        status = payload.get("status")
+        allowed_statuses = {
+            TaskStatus.RETRY_WAIT.value,
+            TaskStatus.MANUAL_REVIEW.value,
+            TaskStatus.FAILED.value,
+            TaskStatus.CANCELLED.value,
+        }
+        if status not in allowed_statuses:
+            return
+        error = payload.get("error")
+        error = error if isinstance(error, dict) else {}
+        finished_at = (
+            latest_run["finished_at"] or payload.get("reported_at") or now
+        )
+        connection.execute(
+            "UPDATE pdd_tasks SET status = ?, current_step = 'failed',"
+            " last_error_code = ?, last_error_message = ?,"
+            " finished_at = ?, updated_at = ? WHERE id = ?",
+            (
+                status,
+                error.get("code"),
+                error.get("message"),
+                finished_at,
+                now,
+                task_id,
+            ),
+        )
+
    def _update_outbox(
        self, event_id: int, status: str, message: Optional[str], increment: bool = False
    ) -> None: