diff --git a/client/src/admin_gateway.py b/client/src/admin_gateway.py index 9890a55..76d978e 100644 --- a/client/src/admin_gateway.py +++ b/client/src/admin_gateway.py @@ -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( diff --git a/client/src/http_admin_gateway.py b/client/src/http_admin_gateway.py index ccf23de..3ba01e4 100644 --- a/client/src/http_admin_gateway.py +++ b/client/src/http_admin_gateway.py @@ -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 = {} diff --git a/client/src/pdd_ui_event.py b/client/src/pdd_ui_event.py index 63346cb..1c31606 100644 --- a/client/src/pdd_ui_event.py +++ b/client/src/pdd_ui_event.py @@ -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: """把领域摘要转换成只供表格显示的轻量行。""" diff --git a/client/test/test_http_admin_gateway.py b/client/test/test_http_admin_gateway.py index 6052288..383d6e0 100644 --- a/client/test/test_http_admin_gateway.py +++ b/client/test/test_http_admin_gateway.py @@ -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( { diff --git a/client/test/test_pdd_ui_event.py b/client/test/test_pdd_ui_event.py index 63e7484..1598f5b 100644 --- a/client/test/test_pdd_ui_event.py +++ b/client/test/test_pdd_ui_event.py @@ -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() diff --git a/docs/client/04-admin-api-contract.md b/docs/client/04-admin-api-contract.md index e1155d1..7b0bfd8 100644 --- a/docs/client/04-admin-api-contract.md +++ b/docs/client/04-admin-api-contract.md @@ -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 +也按这个值模拟，等定了再按工单改。 | # | 待确认什么 | 临时默认值（先这么做） | |---|---|---| diff --git a/docs/client/05-ui-specification.md b/docs/client/05-ui-specification.md index ff13c87..d83bc89 100644 --- a/docs/client/05-ui-specification.md +++ b/docs/client/05-ui-specification.md @@ -60,6 +60,11 @@ - 停止表示不再领取新任务；当前任务进入安全停止流程。 - 如果设备、Admin 或必要设置无效，按钮可禁用，但附近必须说明缺少的条件。 +**当前过渡状态（#30）：**任务执行器尚未接入时，按钮文字固定为“获取任务”，每次 +点击只在后台领取一个采集任务、保存到 SQLite 并刷新表格。领取期间禁用重复点击， +底部状态区显示进度；不循环领取，也不显示“停止获取”。接入领取—执行闭环后再恢复 +本节上面的最终交互。 + ### 4.2 搜索与筛选 建议控件：