feat: 领取 Admin 采集任务并保存本地 (#30)

2026-08-07 16:42:31 +08:00
parent 107366f1f0
commit 51062f4c7b
7 changed files with 841 additions and 15 deletions
@@ -136,14 +136,18 @@ class ClientRegistrationGateway(ABC):
        """幂等登记或更新当前 Client。"""


-class AdminGateway(ClientRegistrationGateway):
-    """完整 Admin 边界；不得增加状态查询、心跳或租约方法。"""
+class TaskClaimGateway(ABC):
+    """任务页只依赖领取能力，不依赖登记和结果提交。"""

    @abstractmethod
    def claim_next(
        self, client: ClientInfo, capabilities: ClaimCapabilities
    ) -> Optional[AdminTask]:
-        """领取至多一个分配给当前 Client 的任务。"""
+        """领取至多一个任务；暂无任务时返回 ``None``。"""
+
+
+class AdminGateway(ClientRegistrationGateway, TaskClaimGateway):
+    """完整 Admin 边界；不得增加状态查询、心跳或租约方法。"""

    @abstractmethod
    def submit_result(
@@ -1,28 +1,31 @@
-"""使用 Python 标准库调用 Admin 登记接口。"""
+"""使用 Python 标准库调用 Admin 登记和任务领取接口。"""

 import json
 import socket
 from http.client import RemoteDisconnected
-from typing import Callable, Optional
+from typing import Callable, Mapping, Optional
 from urllib.error import HTTPError, URLError
 from urllib.parse import urlparse
 from urllib.request import ProxyHandler, Request, build_opener
 from uuid import uuid4

 from .admin_gateway import (
+    AdminTask,
    AdminGatewayError,
    ClaimCapabilities,
    ClientInfo,
    ClientRegistrationGateway,
    RegistrationReceipt,
+    TaskClaimGateway,
 )
+from .task_models import TaskType


 DEFAULT_ADMIN_BASE_URL = "http://127.0.0.1:8080"


-class HttpAdminGateway(ClientRegistrationGateway):
-    """通过 HTTP 登记 Client；访问令牌只保存在内存。"""
+class HttpAdminGateway(ClientRegistrationGateway, TaskClaimGateway):
+    """通过 HTTP 登记 Client 和领取采集任务；令牌只保存在内存。"""

    def __init__(
        self,
@@ -135,6 +138,151 @@ class HttpAdminGateway(ClientRegistrationGateway):
            registered_at=data["registered_at"],
        )

+    def claim_next(
+        self,
+        client: ClientInfo,
+        capabilities: ClaimCapabilities,
+    ) -> Optional[AdminTask]:
+        """领取一个采集任务；Admin 返回 204 时返回 ``None``。"""
+
+        request_id = str(uuid4())
+        payload = {
+            "client": {"name": client.name.strip()},
+            # #30 只允许领取采集任务。采购能力必须由安全门禁工单开启。
+            "supported_types": [TaskType.COLLECT.value],
+            "capabilities": {
+                "purchase_mode": "dry_run",
+                "schema_versions": list(capabilities.schema_versions),
+            },
+        }
+        if capabilities.device is not None:
+            payload["device"] = {
+                "address": capabilities.device.address.strip(),
+                "platform": capabilities.device.platform,
+                "pdd_package": capabilities.device.pdd_package.strip(),
+            }
+
+        headers = {
+            "Content-Type": "application/json; charset=utf-8",
+            "Accept": "application/json",
+            "X-Client-Id": client.client_id.strip(),
+            "X-Request-Id": request_id,
+        }
+        if self._token:
+            headers["Authorization"] = f"Bearer {self._token}"
+
+        request = Request(
+            f"{self._base_url}/api/v1/client/tasks/claim",
+            data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
+            headers=headers,
+            method="POST",
+        )
+
+        try:
+            with self._opener(request, timeout=self._timeout_seconds) as response:
+                status = getattr(response, "status", None)
+                if status is None:
+                    status = response.getcode()
+                body = response.read()
+        except HTTPError as exc:
+            self._raise_http_error(exc, request_id, "领取任务")
+        except (
+            URLError,
+            RemoteDisconnected,
+            ConnectionError,
+            socket.timeout,
+            TimeoutError,
+        ) as exc:
+            reason = getattr(exc, "reason", exc)
+            if isinstance(reason, (socket.timeout, TimeoutError)):
+                raise AdminGatewayError(
+                    "ADMIN_TIMEOUT",
+                    "Admin 领取请求超时，请稍后重试",
+                    True,
+                    request_id,
+                ) from exc
+            raise AdminGatewayError(
+                "ADMIN_UNAVAILABLE",
+                "无法连接 Admin，请检查服务地址",
+                True,
+                request_id,
+            ) from exc
+
+        if status == 204:
+            return None
+        if status != 200:
+            raise AdminGatewayError(
+                "ADMIN_UNEXPECTED_RESPONSE",
+                f"Admin 返回了未预期的状态码 {status}",
+                status >= 500,
+                request_id,
+            )
+        return self._parse_claim_response(body, request_id)
+
+    @classmethod
+    def _parse_claim_response(
+        cls,
+        body: bytes,
+        request_id: str,
+    ) -> AdminTask:
+        data = cls._decode_json(body, request_id)
+        task = data.get("task")
+        if not isinstance(task, Mapping):
+            raise AdminGatewayError(
+                "ADMIN_INVALID_RESPONSE",
+                "Admin 领取响应缺少 task 对象",
+                False,
+                request_id,
+            )
+
+        task_id = task.get("id")
+        raw_type = task.get("type")
+        version = task.get("version")
+        priority = task.get("priority")
+        task_payload = task.get("payload")
+        created_at = task.get("created_at")
+        updated_at = task.get("updated_at")
+        valid = (
+            isinstance(task_id, str)
+            and bool(task_id.strip())
+            and isinstance(raw_type, str)
+            and isinstance(version, int)
+            and not isinstance(version, bool)
+            and version > 0
+            and isinstance(priority, int)
+            and not isinstance(priority, bool)
+            and isinstance(task_payload, Mapping)
+            and isinstance(created_at, str)
+            and isinstance(updated_at, str)
+        )
+        if not valid:
+            raise AdminGatewayError(
+                "ADMIN_INVALID_RESPONSE",
+                "Admin 领取响应的任务字段不完整",
+                False,
+                request_id,
+            )
+
+        try:
+            task_type = TaskType(raw_type)
+        except ValueError as exc:
+            raise AdminGatewayError(
+                "ADMIN_INVALID_RESPONSE",
+                f"Admin 返回了不支持的任务类型 {raw_type}",
+                False,
+                request_id,
+            ) from exc
+
+        return AdminTask(
+            task_id=task_id,
+            task_type=task_type,
+            version=version,
+            priority=priority,
+            payload=dict(task_payload),
+            created_at=created_at,
+            updated_at=updated_at,
+        )
+
    @staticmethod
    def _decode_json(body: bytes, request_id: str) -> dict:
        try:
@@ -156,7 +304,12 @@ class HttpAdminGateway(ClientRegistrationGateway):
        return data

    @classmethod
-    def _raise_http_error(cls, error: HTTPError, request_id: str) -> None:
+    def _raise_http_error(
+        cls,
+        error: HTTPError,
+        request_id: str,
+        operation: str = "登记",
+    ) -> None:
        try:
            body = error.read()
        except OSError:
@@ -168,7 +321,7 @@ class HttpAdminGateway(ClientRegistrationGateway):
            if not isinstance(details, dict):
                raise ValueError
            code = str(details.get("code") or "ADMIN_HTTP_ERROR")
-            message = str(details.get("message") or "Admin 拒绝了登记请求")
+            message = str(details.get("message") or f"Admin 拒绝了{operation}请求")
            retryable = bool(details.get("retryable", error.code >= 500))
            response_request_id = str(details.get("request_id") or request_id)
            extra = details.get("details")
@@ -176,7 +329,7 @@ class HttpAdminGateway(ClientRegistrationGateway):
                extra = {}
        except (AdminGatewayError, ValueError):
            code = "ADMIN_HTTP_ERROR"
-            message = f"Admin 登记失败，状态码 {error.code}"
+            message = f"Admin {operation}失败，状态码 {error.code}"
            retryable = error.code >= 500
            response_request_id = request_id
            extra = {}
@@ -22,11 +22,30 @@

 from typing import Dict, Optional

-from PyQt5.QtCore import QObject
+from PyQt5.QtCore import QCoreApplication, QObject, QThread, pyqtSignal, pyqtSlot
+from qfluentwidgets import InfoBar, InfoBarPosition

+from .admin_gateway import (
+    AdminGatewayError,
+    AdminTask,
+    AndroidDeviceInfo,
+    ClaimCapabilities,
+    ClientInfo,
+    TaskClaimGateway,
+)
+from .current_client_service import CurrentClientService
+from .http_admin_gateway import DEFAULT_ADMIN_BASE_URL, HttpAdminGateway
 from .pdd_ui import PDDTaskPage, TaskRow
-from .task_models import TaskFilters, TaskStatus, TaskSummary, TaskType
-from .task_repository import TaskRepository
+from .selected_android_device_service import SelectedAndroidDeviceService
+from .settings_repository import SettingsRepository
+from .task_models import (
+    NewClaimedTask,
+    TaskFilters,
+    TaskStatus,
+    TaskSummary,
+    TaskType,
+)
+from .task_repository import DuplicateTaskError, TaskRepository


 TASK_TYPE_BY_TEXT = {
@@ -62,6 +81,132 @@ TASK_STATUS_TEXT = {
 }


+def admin_task_to_new_claimed_task(task: AdminTask) -> NewClaimedTask:
+    """把 Admin 任务显式映射为本地任务，避免字段名自动展开出错。"""
+
+    if task.task_type is not TaskType.COLLECT:
+        raise ValueError(f"任务 {task.task_id} 不是采集任务")
+
+    payload = dict(task.payload)
+    goods_url = payload.get("goods_url")
+    goods_id = payload.get("goods_id")
+    if not isinstance(goods_url, str) or not goods_url.strip():
+        raise ValueError("payload.goods_url 不能为空")
+    if goods_id is not None and not isinstance(goods_id, str):
+        raise ValueError("payload.goods_id 必须是文本")
+
+    original_task = {
+        "id": task.task_id,
+        "type": task.task_type.value,
+        "version": task.version,
+        "priority": task.priority,
+        "payload": payload,
+        "created_at": task.created_at,
+        "updated_at": task.updated_at,
+    }
+    return NewClaimedTask(
+        remote_task_id=task.task_id,
+        task_type=task.task_type,
+        goods_url=goods_url,
+        goods_id=goods_id.strip() if isinstance(goods_id, str) else None,
+        priority=task.priority,
+        version=task.version,
+        admin_payload=original_task,
+    )
+
+
+class ClaimTaskWorker(QObject):
+    """在后台领取至多一个采集任务，并先写入本地数据库。"""
+
+    noTask = pyqtSignal()
+    taskSaved = pyqtSignal(str)
+    duplicateTask = pyqtSignal(str)
+    localSaveFailed = pyqtSignal(str, str)
+    failed = pyqtSignal(str)
+    completed = pyqtSignal()
+
+    def __init__(
+        self,
+        gateway: TaskClaimGateway,
+        task_repository: TaskRepository,
+        client_service: CurrentClientService,
+        android_device_service: SelectedAndroidDeviceService,
+    ) -> None:
+        super().__init__()
+        self._gateway = gateway
+        self._task_repository = task_repository
+        self._client_service = client_service
+        self._android_device_service = android_device_service
+        self._cancelled = False
+
+    def cancel(self) -> None:
+        """阻止尚未开始的领取；已领取的任务仍必须保存到本地。"""
+
+        self._cancelled = True
+
+    @pyqtSlot()
+    def run(self) -> None:
+        claimed_task: Optional[AdminTask] = None
+        try:
+            if self._cancelled:
+                return
+            client_settings = self._client_service.load()
+            if not client_settings.client_id:
+                self.failed.emit("请先在设置页保存当前设备号和设备名")
+                return
+            android_serial = self._android_device_service.load()
+            if not android_serial:
+                self.failed.emit("请先在设置页选择并保存 Android 设备")
+                return
+
+            capabilities = ClaimCapabilities(
+                device=AndroidDeviceInfo(android_serial),
+                supported_types=(TaskType.COLLECT,),
+                purchase_mode="dry_run",
+                schema_versions=(1,),
+            )
+            claimed_task = self._gateway.claim_next(
+                ClientInfo(
+                    client_settings.client_id,
+                    client_settings.client_name,
+                ),
+                capabilities,
+            )
+            if claimed_task is None:
+                if not self._cancelled:
+                    self.noTask.emit()
+                return
+
+            local_task = admin_task_to_new_claimed_task(claimed_task)
+            try:
+                self._task_repository.add_claimed_task(local_task)
+            except DuplicateTaskError:
+                if not self._cancelled:
+                    self.duplicateTask.emit(claimed_task.task_id)
+                return
+            except Exception as exc:
+                if not self._cancelled:
+                    self.localSaveFailed.emit(claimed_task.task_id, str(exc))
+                return
+
+            if not self._cancelled:
+                self.taskSaved.emit(claimed_task.task_id)
+        except AdminGatewayError as exc:
+            if not self._cancelled:
+                request_hint = (
+                    f"，请求编号：{exc.request_id}" if exc.request_id else ""
+                )
+                self.failed.emit(f"{exc}{request_hint}")
+        except Exception as exc:
+            if not self._cancelled:
+                if claimed_task is not None:
+                    self.localSaveFailed.emit(claimed_task.task_id, str(exc))
+                else:
+                    self.failed.emit(f"领取任务失败：{exc}")
+        finally:
+            self.completed.emit()
+
+
 class PDDTaskPageEvent(QObject):
    """把 PDD 页面只读操作连接到本地任务 Repository。"""

@@ -70,15 +215,46 @@ class PDDTaskPageEvent(QObject):
        page: PDDTaskPage,
        repository: Optional[TaskRepository] = None,
        parent=None,
+        claim_gateway: Optional[TaskClaimGateway] = None,
+        settings_repository: Optional[SettingsRepository] = None,
    ):
        super().__init__(parent or page)
        self._page = page
        self._repository = repository or TaskRepository()
        self._filters = TaskFilters()
+        self._closing = False
+        self._claim_busy = False
+        self._claim_thread: Optional[QThread] = None
+        self._claim_worker: Optional[ClaimTaskWorker] = None
+
+        settings = settings_repository or SettingsRepository()
+        self._client_service = CurrentClientService(settings)
+        self._selected_android_device_service = SelectedAndroidDeviceService(
+            settings
+        )
+        self._claim_gateway = claim_gateway
+        self._claim_gateway_error = ""
+        if self._claim_gateway is None:
+            base_url = settings.get("admin.base_url", DEFAULT_ADMIN_BASE_URL)
+            timeout_value = settings.get("admin.request_timeout_seconds", 3.0)
+            try:
+                timeout_seconds = float(timeout_value)
+                self._claim_gateway = HttpAdminGateway(
+                    base_url if isinstance(base_url, str) else "",
+                    timeout_seconds=timeout_seconds,
+                )
+            except (TypeError, ValueError) as exc:
+                self._claim_gateway_error = str(exc)

        page.searchRequested.connect(self.search_tasks)
        page.refreshRequested.connect(self.refresh_tasks)
+        page.autoFetchRequested.connect(self._request_claim_task)
        page.taskModel.loadMoreRequested.connect(self._load_page)
+        page.autoFetchButton.setText("获取任务")
+        page.destroyed.connect(self.shutdown)
+        application = QCoreApplication.instance()
+        if application is not None:
+            application.aboutToQuit.connect(self.shutdown)

    def load_initial_tasks(self) -> None:
        """应用启动后读取第一页本地任务。"""
@@ -100,6 +276,103 @@ class PDDTaskPageEvent(QObject):

        self._reload()

+    @pyqtSlot()
+    def _request_claim_task(self) -> None:
+        """启动一次后台领取；重复点击和关闭期间直接忽略。"""
+
+        if self._closing or self._claim_busy:
+            return
+        if self._claim_gateway_error or self._claim_gateway is None:
+            message = self._claim_gateway_error or "Admin 领取服务未初始化"
+            self._page.set_engine_status(f"领取失败：{message}")
+            self._show_claim_error("领取任务失败", message)
+            return
+
+        self._claim_busy = True
+        self._page.autoFetchButton.setEnabled(False)
+        self._page.set_engine_status("正在领取一个采集任务，请稍候…")
+
+        thread = QThread(self)
+        worker = ClaimTaskWorker(
+            self._claim_gateway,
+            self._repository,
+            self._client_service,
+            self._selected_android_device_service,
+        )
+        worker.moveToThread(thread)
+        thread.started.connect(worker.run)
+        worker.noTask.connect(self._on_no_claimed_task)
+        worker.taskSaved.connect(self._on_claimed_task_saved)
+        worker.duplicateTask.connect(self._on_duplicate_claimed_task)
+        worker.localSaveFailed.connect(self._on_claimed_task_save_failed)
+        worker.failed.connect(self._on_claim_failed)
+        worker.completed.connect(thread.quit)
+        worker.completed.connect(worker.deleteLater)
+        thread.finished.connect(thread.deleteLater)
+        thread.finished.connect(self._on_claim_thread_finished)
+        self._claim_thread = thread
+        self._claim_worker = worker
+        thread.start()
+
+    @pyqtSlot()
+    def _on_no_claimed_task(self) -> None:
+        if not self._closing:
+            self._page.set_engine_status("暂无可领取的采集任务")
+
+    @pyqtSlot(str)
+    def _on_claimed_task_saved(self, task_id: str) -> None:
+        if self._closing:
+            return
+        self._page.set_engine_status(
+            f"已领取任务 {task_id}，已保存到本地任务列表"
+        )
+        self._reload()
+
+    @pyqtSlot(str)
+    def _on_duplicate_claimed_task(self, task_id: str) -> None:
+        if not self._closing:
+            self._page.set_engine_status(
+                f"任务 {task_id} 本地已有，未重复保存"
+            )
+
+    @pyqtSlot(str, str)
+    def _on_claimed_task_save_failed(self, task_id: str, message: str) -> None:
+        if not self._closing:
+            content = (
+                f"任务 {task_id} 已在服务端领取，但本地保存失败：{message}。"
+                "请记下这个任务号联系维护者。"
+            )
+            self._page.set_engine_status(content)
+            self._show_claim_error("本地任务保存失败", content)
+
+    @pyqtSlot(str)
+    def _on_claim_failed(self, message: str) -> None:
+        if not self._closing:
+            content = message or "领取任务失败，请稍后重试"
+            self._page.set_engine_status(content)
+            self._show_claim_error("领取任务失败", content)
+
+    def _show_claim_error(self, title: str, content: str) -> None:
+        """显示不会自动消失的可恢复错误，同时保留底部状态文字。"""
+
+        InfoBar.error(
+            title=title,
+            content=content,
+            isClosable=True,
+            duration=-1,
+            position=InfoBarPosition.TOP_RIGHT,
+            parent=self._page,
+        )
+
+    @pyqtSlot()
+    def _on_claim_thread_finished(self) -> None:
+        self._claim_worker = None
+        self._claim_thread = None
+        self._claim_busy = False
+        if not self._closing:
+            self._page.autoFetchButton.setText("获取任务")
+            self._page.autoFetchButton.setEnabled(True)
+
    def _reload(self) -> None:
        self._page.begin_task_reload()
        self._page.taskModel.fetchMore()
@@ -118,6 +391,41 @@ class PDDTaskPageEvent(QObject):
        rows = [summary_to_row(summary) for summary in summaries]
        self._page.append_task_page(rows, has_more=len(rows) == limit)

+    @pyqtSlot()
+    def shutdown(self) -> None:
+        """窗口关闭时停止新的领取，并等待已领取任务完成本地保存。"""
+
+        if self._closing:
+            return
+        self._closing = True
+
+        worker = self._claim_worker
+        thread = self._claim_thread
+        if worker is not None:
+            try:
+                worker.cancel()
+                signal_slots = (
+                    (worker.noTask, self._on_no_claimed_task),
+                    (worker.taskSaved, self._on_claimed_task_saved),
+                    (worker.duplicateTask, self._on_duplicate_claimed_task),
+                    (
+                        worker.localSaveFailed,
+                        self._on_claimed_task_save_failed,
+                    ),
+                    (worker.failed, self._on_claim_failed),
+                )
+            except RuntimeError:
+                signal_slots = ()
+            for signal, slot in signal_slots:
+                try:
+                    signal.disconnect(slot)
+                except (TypeError, RuntimeError):
+                    pass
+
+        if thread is not None and thread.isRunning():
+            thread.quit()
+            thread.wait(10_000)
+

 def summary_to_row(summary: TaskSummary) -> TaskRow:
    """把领域摘要转换成只供表格显示的轻量行。"""
@@ -116,6 +116,109 @@ class HttpAdminGatewayTest(unittest.TestCase):
            key.lower(): value for key, value in opener.request.header_items()
        })

+    def test_claim_maps_real_admin_payload_and_only_reports_collect(self):
+        opener = RecordingOpener(
+            FakeResponse(
+                200,
+                {
+                    "task": {
+                        "id": "COL-8020a8729f111c15",
+                        "type": "collect",
+                        "version": 1,
+                        "priority": 0,
+                        "payload": {
+                            "goods_id": "737116531267",
+                            "goods_url": (
+                                "https://mobile.yangkeduo.com/goods.html"
+                                "?goods_id=737116531267"
+                            ),
+                        },
+                        "created_at": "2026-08-07T03:19:49Z",
+                        "updated_at": "2026-08-07T03:19:49Z",
+                    }
+                },
+            )
+        )
+        gateway = HttpAdminGateway(opener=opener)
+
+        task = gateway.claim_next(
+            ClientInfo("CLIENT-001", "办公室电脑"),
+            self._capabilities(),
+        )
+
+        self.assertIsNotNone(task)
+        self.assertEqual(task.task_id, "COL-8020a8729f111c15")
+        self.assertEqual(task.task_type.value, "collect")
+        self.assertEqual(task.payload["goods_id"], "737116531267")
+        self.assertEqual(opener.request.get_method(), "POST")
+        self.assertEqual(
+            opener.request.full_url,
+            "http://127.0.0.1:8080/api/v1/client/tasks/claim",
+        )
+        headers = {
+            key.lower(): value for key, value in opener.request.header_items()
+        }
+        self.assertEqual(headers["x-client-id"], "CLIENT-001")
+        self.assertTrue(headers["x-request-id"])
+        self.assertIn("application/json", headers["content-type"])
+        body = json.loads(opener.request.data.decode("utf-8"))
+        self.assertEqual(body["supported_types"], ["collect"])
+        self.assertEqual(body["capabilities"]["purchase_mode"], "dry_run")
+
+    def test_claim_204_returns_none(self):
+        gateway = HttpAdminGateway(opener=RecordingOpener(FakeResponse(204, {})))
+
+        task = gateway.claim_next(
+            ClientInfo("CLIENT-001"),
+            self._capabilities(),
+        )
+
+        self.assertIsNone(task)
+
+    def test_claim_invalid_task_fields_include_request_id(self):
+        gateway = HttpAdminGateway(
+            opener=RecordingOpener(FakeResponse(200, {"task": {"id": None}}))
+        )
+
+        with self.assertRaises(AdminGatewayError) as context:
+            gateway.claim_next(
+                ClientInfo("CLIENT-001"),
+                self._capabilities(),
+            )
+
+        self.assertEqual(context.exception.code, "ADMIN_INVALID_RESPONSE")
+        self.assertTrue(context.exception.request_id)
+
+    def test_claim_http_error_keeps_server_request_id(self):
+        error_body = json.dumps(
+            {
+                "error": {
+                    "code": "TASK_CLAIM_FAILED",
+                    "message": "领取任务失败",
+                    "retryable": True,
+                    "request_id": "claim-request-id",
+                }
+            }
+        ).encode("utf-8")
+        error = HTTPError(
+            "http://admin/api/v1/client/tasks/claim",
+            500,
+            "Internal Server Error",
+            {},
+            io.BytesIO(error_body),
+        )
+        gateway = HttpAdminGateway(opener=RecordingOpener(error))
+
+        with self.assertRaises(AdminGatewayError) as context:
+            gateway.claim_next(
+                ClientInfo("CLIENT-001"),
+                self._capabilities(),
+            )
+
+        self.assertEqual(context.exception.code, "TASK_CLAIM_FAILED")
+        self.assertEqual(context.exception.request_id, "claim-request-id")
+        self.assertTrue(context.exception.retryable)
+
    def test_admin_error_preserves_code_retry_and_request_id(self):
        error_body = json.dumps(
            {
@@ -2,6 +2,8 @@

 import os
 import tempfile
+import threading
+import time
 import unittest
 from pathlib import Path

@@ -10,7 +12,12 @@ os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
 from PyQt5.QtWidgets import QApplication

 from src.pdd_ui import PDDTaskPage
-from src.pdd_ui_event import PDDTaskPageEvent, summary_to_row
+from src.admin_gateway import AdminTask
+from src.pdd_ui_event import (
+    PDDTaskPageEvent,
+    admin_task_to_new_claimed_task,
+    summary_to_row,
+)
 from src.mock_admin_gateway import MockAdminGateway
 from src.settings_repository import SettingsRepository
 from src.task_models import NewClaimedTask, TaskStatus, TaskSummary, TaskType
@@ -25,6 +32,70 @@ class BrokenRepository:
        raise RuntimeError("database is unavailable")


+class BrokenSaveRepository(BrokenRepository):
+    """模拟任务已经在 Admin 领取，但本地写入失败。"""
+
+    def add_claimed_task(self, _task):
+        raise RuntimeError("disk is full")
+
+
+class RecordingClaimGateway:
+    """记录领取参数并返回预设结果。"""
+
+    def __init__(self, response=None):
+        self.response = response
+        self.calls = []
+        self.thread_ids = []
+
+    def claim_next(self, client, capabilities):
+        self.calls.append((client, capabilities))
+        self.thread_ids.append(threading.get_ident())
+        return self.response
+
+
+class SlowClaimGateway(RecordingClaimGateway):
+    """让关闭测试能稳定发生在 HTTP 返回之前。"""
+
+    def __init__(self, response):
+        super().__init__(response)
+        self.started = threading.Event()
+
+    def claim_next(self, client, capabilities):
+        self.started.set()
+        time.sleep(0.1)
+        return super().claim_next(client, capabilities)
+
+
+def collect_admin_task(task_id="COL-001"):
+    return AdminTask(
+        task_id=task_id,
+        task_type=TaskType.COLLECT,
+        version=1,
+        priority=0,
+        payload={
+            "goods_id": "737116531267",
+            "goods_url": (
+                "https://mobile.yangkeduo.com/goods.html"
+                "?goods_id=737116531267"
+            ),
+        },
+        created_at="2026-08-07T03:19:49Z",
+        updated_at="2026-08-07T03:19:49Z",
+    )
+
+
+def wait_until(application, predicate, timeout=3.0):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        application.processEvents()
+        if predicate():
+            application.processEvents()
+            return True
+        time.sleep(0.01)
+    application.processEvents()
+    return predicate()
+
+
 class PDDTaskPageEventTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
@@ -59,6 +130,17 @@ class PDDTaskPageEventTest(unittest.TestCase):
            received_at="2026-08-06T08:00:00Z",
        )

+    def _saved_settings(self):
+        settings = SettingsRepository(self.db_path)
+        settings.set_many(
+            {
+                "admin.client_id": "CLIENT-001",
+                "admin.client_name": "办公室电脑",
+                "android.selected_serial": "USB-001",
+            }
+        )
+        return settings
+
    def test_initial_load_and_fetch_next_page(self):
        for number in range(51):
            self._add_task(number)
@@ -143,6 +225,164 @@ class PDDTaskPageEventTest(unittest.TestCase):
        window.close()
        window.deleteLater()

+    def test_admin_task_mapping_uses_explicit_real_field_names(self):
+        local_task = admin_task_to_new_claimed_task(
+            collect_admin_task("COL-8020a8729f111c15")
+        )
+
+        self.assertEqual(local_task.remote_task_id, "COL-8020a8729f111c15")
+        self.assertEqual(local_task.task_type, TaskType.COLLECT)
+        self.assertEqual(local_task.goods_id, "737116531267")
+        self.assertEqual(
+            local_task.admin_payload["payload"]["goods_id"],
+            "737116531267",
+        )
+
+    def test_click_claims_one_collect_task_saves_and_refreshes_table(self):
+        gateway = RecordingClaimGateway(collect_admin_task())
+        page = PDDTaskPage()
+        events = PDDTaskPageEvent(
+            page,
+            self.repository,
+            claim_gateway=gateway,
+            settings_repository=self._saved_settings(),
+        )
+
+        page.autoFetchRequested.emit()
+
+        self.assertTrue(
+            wait_until(self.app, lambda: not events._claim_busy)
+        )
+        self.assertEqual(len(gateway.calls), 1)
+        self.assertNotEqual(gateway.thread_ids[0], threading.get_ident())
+        _, capabilities = gateway.calls[0]
+        self.assertEqual(
+            [task_type.value for task_type in capabilities.supported_types],
+            ["collect"],
+        )
+        self.assertEqual(self.repository.count_tasks(), 1)
+        self.assertEqual(page.taskModel.data_row_count(), 1)
+        self.assertEqual(page.taskModel.row_at(0).remote_task_id, "COL-001")
+        self.assertEqual(page.autoFetchButton.text(), "获取任务")
+        self.assertIn("COL-001", page.statusLabel.text())
+        events.shutdown()
+        page.deleteLater()
+
+    def test_204_is_neutral_and_each_click_only_calls_once(self):
+        gateway = RecordingClaimGateway(None)
+        page = PDDTaskPage()
+        events = PDDTaskPageEvent(
+            page,
+            self.repository,
+            claim_gateway=gateway,
+            settings_repository=self._saved_settings(),
+        )
+
+        page.autoFetchRequested.emit()
+
+        self.assertTrue(wait_until(self.app, lambda: not events._claim_busy))
+        self.assertEqual(len(gateway.calls), 1)
+        self.assertEqual(page.statusLabel.text(), "暂无可领取的采集任务")
+        self.assertEqual(self.repository.count_tasks(), 0)
+        events.shutdown()
+        page.deleteLater()
+
+    def test_repeated_click_while_claiming_does_not_start_second_request(self):
+        gateway = SlowClaimGateway(None)
+        page = PDDTaskPage()
+        events = PDDTaskPageEvent(
+            page,
+            self.repository,
+            claim_gateway=gateway,
+            settings_repository=self._saved_settings(),
+        )
+
+        page.autoFetchRequested.emit()
+        page.autoFetchRequested.emit()
+
+        self.assertFalse(page.autoFetchButton.isEnabled())
+        self.assertTrue(wait_until(self.app, lambda: not events._claim_busy))
+        self.assertEqual(len(gateway.calls), 1)
+        events.shutdown()
+        page.deleteLater()
+
+    def test_duplicate_task_is_normal_status(self):
+        task = collect_admin_task()
+        self.repository.add_claimed_task(admin_task_to_new_claimed_task(task))
+        gateway = RecordingClaimGateway(task)
+        page = PDDTaskPage()
+        events = PDDTaskPageEvent(
+            page,
+            self.repository,
+            claim_gateway=gateway,
+            settings_repository=self._saved_settings(),
+        )
+
+        page.autoFetchRequested.emit()
+
+        self.assertTrue(wait_until(self.app, lambda: not events._claim_busy))
+        self.assertIn("本地已有", page.statusLabel.text())
+        self.assertEqual(self.repository.count_tasks(), 1)
+        events.shutdown()
+        page.deleteLater()
+
+    def test_local_save_failure_status_contains_claimed_task_id(self):
+        gateway = RecordingClaimGateway(collect_admin_task("COL-LOST"))
+        page = PDDTaskPage()
+        events = PDDTaskPageEvent(
+            page,
+            BrokenSaveRepository(),
+            claim_gateway=gateway,
+            settings_repository=self._saved_settings(),
+        )
+
+        page.autoFetchRequested.emit()
+
+        self.assertTrue(wait_until(self.app, lambda: not events._claim_busy))
+        self.assertIn("COL-LOST", page.statusLabel.text())
+        self.assertIn("本地保存失败", page.statusLabel.text())
+        self.assertIn("disk is full", page.statusLabel.text())
+        events.shutdown()
+        page.deleteLater()
+
+    def test_missing_client_settings_do_not_call_admin(self):
+        gateway = RecordingClaimGateway(collect_admin_task())
+        page = PDDTaskPage()
+        events = PDDTaskPageEvent(
+            page,
+            self.repository,
+            claim_gateway=gateway,
+            settings_repository=SettingsRepository(self.db_path),
+        )
+
+        page.autoFetchRequested.emit()
+
+        self.assertTrue(wait_until(self.app, lambda: not events._claim_busy))
+        self.assertEqual(gateway.calls, [])
+        self.assertIn("设置页", page.statusLabel.text())
+        events.shutdown()
+        page.deleteLater()
+
+    def test_shutdown_after_claim_started_still_saves_task_without_ui_callback(self):
+        gateway = SlowClaimGateway(collect_admin_task("COL-CLOSE"))
+        page = PDDTaskPage()
+        events = PDDTaskPageEvent(
+            page,
+            self.repository,
+            claim_gateway=gateway,
+            settings_repository=self._saved_settings(),
+        )
+        page.autoFetchRequested.emit()
+        self.assertTrue(wait_until(self.app, gateway.started.is_set))
+        status_before_close = page.statusLabel.text()
+
+        events.shutdown()
+        self.app.processEvents()
+
+        self.assertIsNotNone(self.repository.get_task("COL-CLOSE"))
+        self.assertEqual(page.statusLabel.text(), status_before_close)
+        page.deleteLater()
+

 if __name__ == "__main__":
    unittest.main()
@@ -199,6 +199,10 @@ POST /api/v1/client/tasks/claim

 无可领取任务时返回 `204 No Content`。

+Client 的 `HttpAdminGateway.claim_next` 已实现本接口。当前只声明
+`supported_types: ["collect"]`，一次调用最多领取一个采集任务；结果提交和失败提交
+仍由后续工单实现。
+
 `[必须]` **`204` 不是错误。** Client 要把它当"暂时没活干"处理，
 不要报错，也不要因此触发重试风暴。首次 claim 通常返回 204；如果该编号已经预先分配任务，也可以直接返回 200。

@@ -243,6 +247,14 @@ Client 侧对这两种没有任何区别：调 `claim`，拿到任务就做，

 Admin 可以在超时后把任务重派给别的 Client，Client 侧对此**无感知也不需要感知**。

+### 5.2 已知缺口：领取成功但本地保存失败
+
+Admin 返回任务时已经把它改成 `claimed`。Client 随后写 SQLite，如果磁盘或数据库
+此时失败，服务端任务会处于已领取、本机却没有执行记录的状态。
+
+当前处理方式：界面持续显示任务编号和本地保存错误，要求操作人员记录编号并联系
+维护者；不得静默继续领取下一条。自动补偿或 Admin 侧回收由后续独立工单处理。
+
 ## 6. 提交成功结果

 ```http
@@ -361,7 +373,8 @@ Mock 测试通过不能替代与真实 Admin 的契约测试。

 ## 10. 待联合确认

-Admin 还没做完，下面这些要和 Admin 一起定。**不许因此停工**——先按"临时默认值"实现，Mock Gateway 也按这个值模拟，等定了再按工单改。
+下面这些仍需联合确认。**不许因此停工**——先按“临时默认值”实现，Mock Gateway
+也按这个值模拟，等定了再按工单改。

 | # | 待确认什么 | 临时默认值（先这么做） |
 |---|---|---|
@@ -60,6 +60,11 @@
 - 停止表示不再领取新任务；当前任务进入安全停止流程。
 - 如果设备、Admin 或必要设置无效，按钮可禁用，但附近必须说明缺少的条件。

+**当前过渡状态（#30）：**任务执行器尚未接入时，按钮文字固定为“获取任务”，每次
+点击只在后台领取一个采集任务、保存到 SQLite 并刷新表格。领取期间禁用重复点击，
+底部状态区显示进度；不循环领取，也不显示“停止获取”。接入领取—执行闭环后再恢复
+本节上面的最终交互。
+
 ### 4.2 搜索与筛选

 建议控件：