chengma/cmautobuy
2
0
Fork 0
Code Issues 89 Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: 复用持久设备工作线程 (#110)

Browse Source
This commit is contained in:
chengma
2026-08-10 18:00:38 +08:00
parent 553659208a
commit 70cedc34d4
11 changed files with 563 additions and 149 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/collect_task_service.py
+5 -2
View File
@@ -13,7 +13,10 @@ from .admin_gateway import (
ClientInfo,
)
from .pdd_collect_service import PddCollectError, PddCollectService
from .pdd_device_service import PddDeviceService
from .pdd_device_service import (
PddDeviceService,
current_thread_device_service,
)
from .db import data_dir
from .task_models import OutboxEventRecord, OutboxEventType, TaskStatus, TaskType
from .task_models import NewClaimedTask
@@ -258,7 +261,7 @@ class CollectTaskService:
cancelled: Callable[[], bool],
) -> PddCollectService:
return PddCollectService(
PddDeviceService(),
current_thread_device_service() or PddDeviceService(),
device_address,
client_id,
cancelled=cancelled,
client/src/pdd_device_service.py
+185 -4
View File
@@ -1,14 +1,16 @@
"""uiautomator2 设备连接边界。
每次自动化任务通过 ``connect`` 获取一个会话。会话只能在创建它的线程中
使用，退出 ``with`` 后立即释放，避免多个任务同时控制同一台手机。
普通服务每个任务建立一次连接；持久服务只复用 Device 连接，不复用页面数据。
两种会话都只能在创建它们的线程中使用。
"""
from __future__ import annotations
import threading
from contextlib import AbstractContextManager
from typing import Any, Callable, Optional
import time
from contextlib import AbstractContextManager, contextmanager
from contextvars import ContextVar
from typing import Any, Callable, Iterator, Optional
from .performance_timing import current_performance_trace
@@ -151,3 +153,182 @@ class PddDeviceService:
with self._state_lock:
if self._active_serial == serial:
self._active_serial = None
class PersistentPddDeviceService(PddDeviceService):
"""在固定工作线程内复用同一台设备的 Device 连接。
``connect`` 每次仍调用 ``app_current`` 做健康检查并建立全新的任务会话。
缓存连接失效时最多重连一次；页面 XML、选择器和业务数据均不缓存。
"""
def __init__(
self,
connector: Optional[Callable[[str], Any]] = None,
*,
ttl_seconds: float = 90.0,
monotonic: Callable[[], float] = time.monotonic,
) -> None:
super().__init__(connector)
if ttl_seconds <= 0:
raise ValueError("设备会话 TTL 必须大于 0")
self._ttl_seconds = float(ttl_seconds)
self._monotonic = monotonic
self._owner_thread_id: Optional[int] = None
self._cached_serial: Optional[str] = None
self._cached_device: Any = None
self._last_used_at = 0.0
@property
def ttl_seconds(self) -> float:
return self._ttl_seconds
@property
def has_cached_device(self) -> bool:
"""只供生命周期管理和测试判断，不返回 Device 对象。"""
return self._cached_device is not None
def connect(self, serial: str) -> PddDeviceSession:
"""在所有者线程取得独占任务会话，并按需复用底层连接。"""
self._assert_owner_thread()
checked_serial = self._validate_serial(serial)
with self._state_lock:
if self._active_serial is not None:
raise PddDeviceError(
"DEVICE_IN_USE",
f"设备 {self._active_serial} 正在执行其他自动化任务",
)
self._active_serial = checked_serial
try:
device, current = self._healthy_device(checked_serial)
except PddDeviceError:
self._release(checked_serial)
raise
except Exception as exc:
self._release(checked_serial)
details = str(exc).lower()
code = (
"DEVICE_OFFLINE"
if any(
word in details
for word in ("offline", "not found", "disconnected")
)
else "DEVICE_CONNECT_FAILED"
)
raise PddDeviceError(
code,
f"无法连接 Android 设备 {checked_serial}：{exc}",
) from exc
return PddDeviceSession(
device,
checked_serial,
current,
lambda: self._release_persistent(checked_serial),
)
def release_cached(self) -> None:
"""在所有者线程丢弃缓存连接；可重复调用。"""
self._assert_owner_thread()
with self._state_lock:
if self._active_serial is not None:
raise PddDeviceError(
"DEVICE_IN_USE",
"设备任务尚未到达安全结束点，暂不能释放连接",
)
self._discard_cached()
def _healthy_device(self, serial: str) -> tuple[Any, dict[str, Any]]:
now = self._monotonic()
expired = (
self._cached_device is not None
and now - self._last_used_at >= self._ttl_seconds
)
if self._cached_serial != serial or expired:
self._discard_cached()
if self._cached_device is not None:
try:
current = self._read_app_state(self._cached_device)
return self._cached_device, current
except Exception:
# 旧连接失效时只丢弃并重连一次，不能在这里无限重试。
self._discard_cached()
trace = current_performance_trace()
if trace is None:
device = self._connector(serial)
else:
with trace.stage("uiautomator2_connect"):
device = self._connector(serial)
current = self._read_app_state(device)
self._cached_serial = serial
self._cached_device = device
self._last_used_at = now
return device, current
@staticmethod
def _read_app_state_without_trace(device: Any) -> dict[str, Any]:
current = device.app_current()
if not isinstance(current, dict):
raise RuntimeError("uiautomator2 未返回有效的设备状态")
return current
def _read_app_state(self, device: Any) -> dict[str, Any]:
trace = current_performance_trace()
if trace is None:
return self._read_app_state_without_trace(device)
with trace.stage("first_app_current"):
return self._read_app_state_without_trace(device)
def _release_persistent(self, serial: str) -> None:
self._assert_owner_thread()
with self._state_lock:
if self._active_serial == serial:
self._active_serial = None
self._last_used_at = self._monotonic()
def _discard_cached(self) -> None:
# uiautomator2 Device 没有稳定的 close API；清除唯一引用即可让底层
# HTTP 客户端按库自身生命周期回收，不能猜测调用私有方法。
self._cached_serial = None
self._cached_device = None
self._last_used_at = 0.0
def _assert_owner_thread(self) -> None:
current = threading.get_ident()
if self._owner_thread_id is None:
self._owner_thread_id = current
elif self._owner_thread_id != current:
raise PddDeviceError(
"DEVICE_THREAD_VIOLATION",
"持久 uiautomator2 Device 只能在固定工作线程中使用和释放",
)
_CURRENT_DEVICE_SERVICE: ContextVar[Optional[PersistentPddDeviceService]] = (
ContextVar("pdd_device_service", default=None)
)
@contextmanager
def bind_thread_device_service(
service: PersistentPddDeviceService,
) -> Iterator[None]:
"""让当前工作线程创建的采集、采购和核单适配器使用同一连接服务。"""
token = _CURRENT_DEVICE_SERVICE.set(service)
try:
yield
finally:
_CURRENT_DEVICE_SERVICE.reset(token)
def current_thread_device_service() -> Optional[PersistentPddDeviceService]:
"""返回当前工作线程绑定的服务；未绑定时返回 ``None``。"""
return _CURRENT_DEVICE_SERVICE.get()
client/src/pdd_u2_purchase_adapter.py
+3 -5
View File
@@ -18,6 +18,7 @@ from .pdd_device_service import (
PDD_PACKAGE_NAME,
PddDeviceError,
PddDeviceService,
current_thread_device_service,
)
from .performance_timing import current_performance_trace
from .pdd_purchase_adapter import (
@@ -561,9 +562,6 @@ class U2PddPurchaseAdapter(PddPurchaseAdapter):
) from error
_PURCHASE_DEVICE_SERVICE = PddDeviceService()
def create_u2_purchase_adapter(
device_address: str, cancelled: Callable[[], bool]
) -> PddPurchaseAdapter:
@@ -571,7 +569,7 @@ def create_u2_purchase_adapter(
return U2PddPurchaseAdapter(
device_address,
device_service=_PURCHASE_DEVICE_SERVICE,
device_service=current_thread_device_service() or PddDeviceService(),
cancelled=cancelled,
)
@@ -625,6 +623,6 @@ def create_u2_live_purchase_adapter(
return U2PddLivePurchaseAdapter(
device_address,
device_service=_PURCHASE_DEVICE_SERVICE,
device_service=current_thread_device_service() or PddDeviceService(),
cancelled=cancelled,
)
client/src/pdd_u2_purchase_reconcile_adapter.py
+11 -6
View File
@@ -13,7 +13,11 @@ from datetime import datetime, timedelta, timezone
from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
from typing import Any, Callable, Optional
from .pdd_device_service import PDD_PACKAGE_NAME, PddDeviceService
from .pdd_device_service import (
PDD_PACKAGE_NAME,
PddDeviceService,
current_thread_device_service,
)
from .pdd_purchase_reconcile_adapter import (
PddPurchaseReconcileAdapter,
PurchaseOrderCandidate,
@@ -215,7 +219,11 @@ class U2PddPurchaseReconcileAdapter(PddPurchaseReconcileAdapter):
max_pages: int = 5,
) -> None:
self._device_address = str(device_address or "").strip()
self._device_service = device_service or _RECONCILE_DEVICE_SERVICE
self._device_service = (
device_service
or current_thread_device_service()
or PddDeviceService()
)
self._cancelled = cancelled
self._settle_seconds = max(0.0, float(settle_seconds))
self._max_pages = max(1, int(max_pages))
@@ -373,9 +381,6 @@ class U2PddPurchaseReconcileAdapter(PddPurchaseReconcileAdapter):
time.sleep(self._settle_seconds)
_RECONCILE_DEVICE_SERVICE = PddDeviceService()
def create_u2_purchase_reconcile_adapter(
device_address: str, cancelled: Callable[[], bool]
) -> PddPurchaseReconcileAdapter:
@@ -383,6 +388,6 @@ def create_u2_purchase_reconcile_adapter(
return U2PddPurchaseReconcileAdapter(
device_address,
device_service=_RECONCILE_DEVICE_SERVICE,
device_service=current_thread_device_service() or PddDeviceService(),
cancelled=cancelled,
)
client/src/pdd_ui_event.py
+212 -125
View File
@@ -25,6 +25,7 @@ from typing import Callable, Dict, Optional
from PyQt5.QtCore import (
QCoreApplication,
QObject,
Qt,
QThread,
QTimer,
pyqtSignal,
@@ -45,6 +46,10 @@ from .collect_task_service import CollectServiceFactory, CollectTaskService
from .current_client_service import CurrentClientService
from .http_admin_gateway import DEFAULT_ADMIN_BASE_URL, HttpAdminGateway
from .pdd_ui import PDDTaskPage, TaskRow
from .pdd_device_service import (
PersistentPddDeviceService,
bind_thread_device_service,
)
from .purchase_task_service import PurchaseAdapterFactory
from .purchase_task_service import LivePurchaseAdapterFactory
from .live_purchase_authorization import LivePurchaseAuthorizationService
@@ -111,7 +116,7 @@ ERROR_FEEDBACK_BUTTON_MIN_HEIGHT = 40
class ClaimTaskWorker(QObject):
"""在后台补交、领取或执行至多一条任务。"""
"""在固定后台线程中串行补交、领取或执行至多一条任务。"""
noTask = pyqtSignal()
taskSaved = pyqtSignal(str)
@@ -122,6 +127,7 @@ class ClaimTaskWorker(QObject):
deviceUnavailable = pyqtSignal(str)
outcome = pyqtSignal(str, str, str)
completed = pyqtSignal()
shutdownCompleted = pyqtSignal()
def __init__(
self,
@@ -157,6 +163,14 @@ class ClaimTaskWorker(QObject):
device_connection_checker
or AndroidDeviceService().require_connected
)
self._device_service = PersistentPddDeviceService()
self._idle_timer: Optional[QTimer] = None
self._accepting = True
def prepare_run(self) -> None:
"""主线程提交新命令前清除上一次的协作停止标志。"""
self._cancelled = False
def cancel(self) -> None:
"""阻止尚未开始的领取；已领取的任务仍必须保存到本地。"""
@@ -165,62 +179,82 @@ class ClaimTaskWorker(QObject):
@pyqtSlot()
def run(self) -> None:
try:
if self._cancelled and not self._selected_task_id:
return
client_settings = self._client_service.load()
if not client_settings.client_id:
self.failed.emit("请先在设置页保存当前设备号和设备名")
return
android_serial = self._android_device_service.load()
"""兼容旧调用；新代码通过 ``run_selected`` 提交命令。"""
client = ClientInfo(
client_settings.client_id,
client_settings.client_name,
)
if self._selected_task_id:
self._device_connection_checker(android_serial or "")
self._task_repository.prepare_collect_rerun(
self._selected_task_id
self.run_selected(self._selected_task_id)
@pyqtSlot(str)
def run_selected(self, selected_task_id: str = "") -> None:
"""执行一条命令；该槽始终由固定设备 QThread 串行调用。"""
if not self._accepting:
self.completed.emit()
return
self._ensure_idle_timer()
assert self._idle_timer is not None
self._idle_timer.stop()
try:
if self._cancelled and not selected_task_id:
return
with bind_thread_device_service(self._device_service):
client_settings = self._client_service.load()
if not client_settings.client_id:
self.failed.emit("请先在设置页保存当前设备号和设备名")
return
android_serial = self._android_device_service.load()
client = ClientInfo(
client_settings.client_id,
client_settings.client_name,
)
service = CollectTaskService(
self._gateway,
self._task_repository,
client,
android_serial or "",
cancelled=lambda: self._cancelled,
collect_service_factory=self._collect_service_factory,
)
result = service.execute_selected(self._selected_task_id)
else:
purchase_mode = "dry_run"
if self._live_authorization_service is not None:
purchase_mode = (
self._live_authorization_service.purchase_mode_for(
client.client_id,
android_serial or "",
live_adapter_ready=(
self._live_purchase_adapter_factory is not None
),
)
if selected_task_id:
self._device_connection_checker(android_serial or "")
self._task_repository.prepare_collect_rerun(
selected_task_id
)
dispatcher = TaskDispatcher(
self._gateway,
self._task_repository,
client,
android_serial or "",
collect_service_factory=self._collect_service_factory,
purchase_adapter_factory=self._purchase_adapter_factory,
live_purchase_adapter_factory=(
self._live_purchase_adapter_factory
),
purchase_mode=purchase_mode,
purchase_reconcile_factory=self._purchase_reconcile_factory,
cancelled=lambda: self._cancelled,
device_connection_checker=self._device_connection_checker,
task_saved=self.taskSaved.emit,
)
result = dispatcher.execute_one()
service = CollectTaskService(
self._gateway,
self._task_repository,
client,
android_serial or "",
cancelled=lambda: self._cancelled,
collect_service_factory=self._collect_service_factory,
)
result = service.execute_selected(selected_task_id)
else:
purchase_mode = "dry_run"
if self._live_authorization_service is not None:
purchase_mode = (
self._live_authorization_service.purchase_mode_for(
client.client_id,
android_serial or "",
live_adapter_ready=(
self._live_purchase_adapter_factory
is not None
),
)
)
dispatcher = TaskDispatcher(
self._gateway,
self._task_repository,
client,
android_serial or "",
collect_service_factory=self._collect_service_factory,
purchase_adapter_factory=self._purchase_adapter_factory,
live_purchase_adapter_factory=(
self._live_purchase_adapter_factory
),
purchase_mode=purchase_mode,
purchase_reconcile_factory=(
self._purchase_reconcile_factory
),
cancelled=lambda: self._cancelled,
device_connection_checker=(
self._device_connection_checker
),
task_saved=self.taskSaved.emit,
)
result = dispatcher.execute_one()
if not self._cancelled or result.kind == "cancelled":
self.outcome.emit(result.kind, result.message, result.task_id)
except AndroidDeviceSearchError as exc:
@@ -240,8 +274,38 @@ class ClaimTaskWorker(QObject):
if not self._cancelled:
self.failed.emit(f"执行任务失败：{exc}")
finally:
if self._accepting and self._idle_timer is not None:
self._idle_timer.start(
max(1, int(self._device_service.ttl_seconds * 1000))
)
self.completed.emit()
@pyqtSlot()
def release_device(self) -> None:
"""停止、换设备或配置变化后，在设备线程安全释放缓存。"""
self._ensure_idle_timer()
assert self._idle_timer is not None
self._idle_timer.stop()
self._device_service.release_cached()
@pyqtSlot()
def shutdown_worker(self) -> None:
"""停止接收命令并释放设备；调用方随后再 quit/wait。"""
self._accepting = False
self._cancelled = True
try:
self.release_device()
finally:
self.shutdownCompleted.emit()
def _ensure_idle_timer(self) -> None:
if self._idle_timer is None:
self._idle_timer = QTimer(self)
self._idle_timer.setSingleShot(True)
self._idle_timer.timeout.connect(self.release_device)
class ResultResubmitWorker(QObject):
"""在后台逐条重发既有 Outbox，不执行任何手机操作。"""
@@ -359,6 +423,10 @@ class TaskRemoveWorker(QObject):
class PDDTaskPageEvent(QObject):
"""把 PDD 页面只读操作连接到本地任务 Repository。"""
_claimRunRequested = pyqtSignal(str)
_claimReleaseRequested = pyqtSignal()
_claimShutdownRequested = pyqtSignal()
def __init__(
self,
page: PDDTaskPage,
@@ -385,6 +453,7 @@ class PDDTaskPageEvent(QObject):
self._claim_busy = False
self._claim_thread: Optional[QThread] = None
self._claim_worker: Optional[ClaimTaskWorker] = None
self._claim_operation = ""
self._rerun_cancel_requested = False
self._rerun_feedback: Optional[InfoBar] = None
self._claim_feedback: Optional[InfoBar] = None
@@ -744,7 +813,7 @@ class PDDTaskPageEvent(QObject):
self._page.set_resubmit_running(False)
def _start_rerun_worker(self, task_id: str) -> None:
"""启动只处理指定任务的工作线程。"""
"""把指定任务提交到持久设备工作线程。"""
assert self._claim_gateway is not None
self._close_rerun_feedback()
@@ -758,29 +827,10 @@ class PDDTaskPageEvent(QObject):
)
self._reload()
thread = QThread(self)
worker = ClaimTaskWorker(
self._claim_gateway,
self._repository,
self._client_service,
self._selected_android_device_service,
self._collect_service_factory,
selected_task_id=task_id,
device_connection_checker=self._device_connection_checker,
)
worker.moveToThread(thread)
thread.started.connect(worker.run)
worker.retryableFailed.connect(self._on_rerun_failed)
worker.failed.connect(self._on_rerun_failed)
worker.deviceUnavailable.connect(self._on_rerun_device_unavailable)
worker.outcome.connect(self._on_rerun_outcome)
worker.completed.connect(thread.quit)
worker.completed.connect(worker.deleteLater)
thread.finished.connect(thread.deleteLater)
thread.finished.connect(self._on_rerun_thread_finished)
self._claim_thread = thread
self._claim_worker = worker
thread.start()
worker = self._ensure_claim_executor()
self._claim_operation = "rerun"
worker.prepare_run()
self._claimRunRequested.emit(task_id)
@pyqtSlot()
def request_cancel_rerun(self) -> None:
@@ -839,8 +889,6 @@ class PDDTaskPageEvent(QObject):
@pyqtSlot()
def _on_rerun_thread_finished(self) -> None:
cancel_requested = self._rerun_cancel_requested
self._claim_worker = None
self._claim_thread = None
self._claim_busy = False
self._rerun_cancel_requested = False
if not self._closing:
@@ -939,7 +987,7 @@ class PDDTaskPageEvent(QObject):
@pyqtSlot()
def _start_claim_cycle(self) -> None:
"""开始一轮任务处理；只允许上一轮线程完全结束后进入。"""
"""开始一轮任务处理；设备命令由固定工作线程串行执行。"""
if (
self._closing
@@ -953,6 +1001,20 @@ class PDDTaskPageEvent(QObject):
self._page.set_auto_fetch_state("running")
self._page.set_engine_status("自动获取：运行中 · 正在处理一条任务…")
worker = self._ensure_claim_executor()
self._claim_operation = "auto"
worker.prepare_run()
self._claimRunRequested.emit("")
def _ensure_claim_executor(self) -> ClaimTaskWorker:
"""懒创建唯一的设备 Worker 和固定 QThread。"""
current = self._claim_worker
thread = self._claim_thread
if current is not None and thread is not None and thread.isRunning():
return current
assert self._claim_gateway is not None
thread = QThread(self)
worker = ClaimTaskWorker(
gateway=self._claim_gateway,
@@ -967,22 +1029,63 @@ class PDDTaskPageEvent(QObject):
device_connection_checker=self._device_connection_checker,
)
worker.moveToThread(thread)
thread.started.connect(worker.run)
worker.noTask.connect(self._on_no_claimed_task)
self._claimRunRequested.connect(worker.run_selected)
self._claimReleaseRequested.connect(worker.release_device)
self._claimShutdownRequested.connect(worker.shutdown_worker)
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.retryableFailed.connect(self._on_claim_retryable_failed)
worker.failed.connect(self._on_claim_failed)
worker.deviceUnavailable.connect(self._on_claim_device_unavailable)
worker.outcome.connect(self._on_collect_outcome)
worker.completed.connect(thread.quit)
worker.completed.connect(worker.deleteLater)
worker.retryableFailed.connect(self._route_claim_retryable_failed)
worker.failed.connect(self._route_claim_failed)
worker.deviceUnavailable.connect(self._route_device_unavailable)
worker.outcome.connect(self._route_claim_outcome)
worker.completed.connect(self._on_claim_cycle_completed)
# QThread 对象属于主线程；显式直连可在 shutdown_worker 完成释放后
# 立即停止工作线程事件循环，避免主线程 wait() 时互相等待。
worker.shutdownCompleted.connect(thread.quit, Qt.DirectConnection)
thread.finished.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()
return worker
@pyqtSlot(str)
def _route_claim_retryable_failed(self, message: str) -> None:
if self._claim_operation == "rerun":
self._on_rerun_failed(message)
else:
self._on_claim_retryable_failed(message)
@pyqtSlot(str)
def _route_claim_failed(self, message: str) -> None:
if self._claim_operation == "rerun":
self._on_rerun_failed(message)
else:
self._on_claim_failed(message)
@pyqtSlot(str)
def _route_device_unavailable(self, message: str) -> None:
if self._claim_operation == "rerun":
self._on_rerun_device_unavailable(message)
else:
self._on_claim_device_unavailable(message)
@pyqtSlot(str, str, str)
def _route_claim_outcome(
self, kind: str, message: str, task_id: str
) -> None:
if self._claim_operation == "rerun":
self._on_rerun_outcome(kind, message, task_id)
else:
self._on_collect_outcome(kind, message, task_id)
@pyqtSlot()
def _on_claim_cycle_completed(self) -> None:
operation = self._claim_operation
self._claim_operation = ""
if operation == "rerun":
self._on_rerun_thread_finished()
else:
self._on_claim_thread_finished()
def _request_stop_auto_fetch(self) -> None:
"""取消下一轮，并请求当前 Worker 在安全点停止。"""
@@ -992,7 +1095,7 @@ class PDDTaskPageEvent(QObject):
self._next_cycle_timer.stop()
self._stop_status = "自动获取：已停止 · 当前没有执行中的任务"
worker = self._claim_worker
if worker is not None:
if worker is not None and self._claim_busy:
self._page.set_auto_fetch_state("stopping")
self._page.set_engine_status(
"自动获取：正在停止 · 等待当前任务安全结束"
@@ -1010,6 +1113,8 @@ class PDDTaskPageEvent(QObject):
self._page.set_auto_fetch_state("stopped")
self._page.set_engine_status(self._stop_status)
self._stop_requested = False
if self._claim_worker is not None:
self._claimReleaseRequested.emit()
def _continue_after(self, delay_ms: int, status: str) -> None:
"""记录本轮结束后的下一次调度，不在 Worker 结束前启动。"""
@@ -1266,8 +1371,6 @@ class PDDTaskPageEvent(QObject):
@pyqtSlot()
def _on_claim_thread_finished(self) -> None:
self._claim_worker = None
self._claim_thread = None
self._claim_busy = False
if self._closing:
return
@@ -1349,6 +1452,13 @@ class PDDTaskPageEvent(QObject):
if not self._closing:
self._page.taskTable.setFocus()
@pyqtSlot(str)
def android_device_configuration_changed(self, _serial: str = "") -> None:
"""设置页保存、删除或切换设备后释放旧的持久连接。"""
if not self._closing and self._claim_worker is not None:
self._claimReleaseRequested.emit()
@pyqtSlot()
def shutdown(self) -> None:
"""窗口关闭时停止新的领取，并等待已领取任务完成本地保存。"""
@@ -1372,43 +1482,20 @@ class PDDTaskPageEvent(QObject):
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),
(
worker.deviceUnavailable,
self._on_claim_device_unavailable,
),
(
worker.deviceUnavailable,
self._on_rerun_device_unavailable,
),
(
worker.retryableFailed,
self._on_claim_retryable_failed,
),
(worker.outcome, self._on_collect_outcome),
)
except RuntimeError:
signal_slots = ()
for signal, slot in signal_slots:
try:
signal.disconnect(slot)
except (TypeError, RuntimeError):
pass
pass
try:
self._claimShutdownRequested.emit()
except RuntimeError:
pass
if thread is not None and thread.isRunning():
thread.quit()
# uiautomator2/ADB 的单次调用可能需要数秒才返回。先通过
# cancelled 标志让采集在下一个安全点退出，再等待工作线程收尾，
# 避免窗口销毁时出现 "QThread destroyed while running"。
thread.wait(60_000)
self._claim_worker = None
self._claim_thread = None
resubmit_worker = self._resubmit_worker
resubmit_thread = self._resubmit_thread
client/src/settings_ui_event.py
+3
View File
@@ -388,6 +388,7 @@ class SettingsPageEventBinder(QObject):
saveRequested = pyqtSignal(str, str)
deleteRequested = pyqtSignal(str)
liveAuthorizationRequested = pyqtSignal(str, str)
androidDeviceConfigurationChanged = pyqtSignal(str)
def __init__(
self,
@@ -1355,12 +1356,14 @@ class SettingsPageEventBinder(QObject):
self._page.deviceStatusLabel.setText(
f"当前使用设备：{serial}（已保存）"
)
self.androidDeviceConfigurationChanged.emit(serial)
else:
self._saved_android_serial = ""
self._page.deviceTableModel.set_checked_serial("")
self._page.deviceStatusLabel.setText(
"当前使用设备：未选择（本地配置已删除）"
)
self.androidDeviceConfigurationChanged.emit("")
@pyqtSlot(str, str)
def _on_android_device_setting_failed(
client/src/ui_main.py
+3
View File
@@ -71,6 +71,9 @@ class MainWindow(FluentWindow):
self.pddTaskPage.openSettingsRequested.connect(
lambda: self.switchTo(self.settingsPage)
)
self.settingsPage.eventBinder.androidDeviceConfigurationChanged.connect(
self.pddTaskPageEvent.android_device_configuration_changed
)
self.addSubInterface(self.pddTaskPage, FIF.HOME, "pdd")
self.addSubInterface(
self.settingsPage,
client/test/test_pdd_device_service.py
+88 -1
View File
@@ -3,7 +3,13 @@
import threading
import unittest
from src.pdd_device_service import PddDeviceError, PddDeviceService
from src.pdd_device_service import (
PddDeviceError,
PddDeviceService,
PersistentPddDeviceService,
bind_thread_device_service,
current_thread_device_service,
)
from src.performance_timing import TaskPerformanceTrace
@@ -89,5 +95,86 @@ class PddDeviceServiceTest(unittest.TestCase):
self.assertEqual(errors, ["DEVICE_THREAD_VIOLATION"])
class PersistentPddDeviceServiceTest(unittest.TestCase):
def test_same_serial_within_ttl_connects_once_but_checks_each_task(self):
connected = []
device = FakeDevice()
service = PersistentPddDeviceService(
lambda serial: connected.append(serial) or device,
ttl_seconds=90,
)
with service.connect("USB-001"):
pass
with service.connect("USB-001"):
pass
self.assertEqual(connected, ["USB-001"])
self.assertEqual(device.current_calls, 2)
def test_expired_or_changed_serial_discards_cached_connection(self):
now = [10.0]
connected = []
service = PersistentPddDeviceService(
lambda serial: connected.append(serial) or FakeDevice(),
ttl_seconds=5,
monotonic=lambda: now[0],
)
with service.connect("USB-001"):
pass
now[0] = 16.0
with service.connect("USB-001"):
pass
with service.connect("USB-002"):
pass
self.assertEqual(connected, ["USB-001", "USB-001", "USB-002"])
def test_failed_cached_health_check_reconnects_only_once(self):
class UnhealthyDevice(FakeDevice):
def app_current(self):
self.current_calls += 1
if self.current_calls >= 2:
raise OSError("offline")
return {"package": "com.xunmeng.pinduoduo"}
devices = [UnhealthyDevice(), FakeDevice()]
connected = []
service = PersistentPddDeviceService(
lambda serial: connected.append(serial) or devices.pop(0)
)
with service.connect("USB-001"):
pass
with service.connect("USB-001") as current:
self.assertIsInstance(current, FakeDevice)
self.assertEqual(connected, ["USB-001", "USB-001"])
def test_release_and_context_binding_stay_on_owner_thread(self):
service = PersistentPddDeviceService(lambda _serial: FakeDevice())
with service.connect("USB-001"):
pass
with bind_thread_device_service(service):
self.assertIs(current_thread_device_service(), service)
self.assertIsNone(current_thread_device_service())
errors = []
def release_from_other_thread():
try:
service.release_cached()
except PddDeviceError as exc:
errors.append(exc.code)
worker = threading.Thread(target=release_from_other_thread)
worker.start()
worker.join()
self.assertEqual(errors, ["DEVICE_THREAD_VIOLATION"])
service.release_cached()
self.assertFalse(service.has_cached_device)
if __name__ == "__main__":
unittest.main()
client/test/test_pdd_ui_event.py
+30
View File
@@ -1153,6 +1153,36 @@ class PDDTaskPageEventTest(unittest.TestCase):
events.shutdown()
page.deleteLater()
def test_auto_fetch_reuses_one_fixed_worker_thread_across_cycles(self):
gateway = RecordingClaimGateway()
page = PDDTaskPage()
events = PDDTaskPageEvent(
page,
self.repository,
claim_gateway=gateway,
settings_repository=self._saved_settings(),
collect_service_factory=fake_collect_factory,
no_task_delay_ms=10,
)
page.autoFetchRequested.emit()
self.assertTrue(
wait_until(self.app, lambda: len(gateway.thread_ids) >= 2),
"自动获取没有完成两轮串行调度",
)
executor_thread = events._claim_thread
page.autoFetchRequested.emit()
self.assertTrue(wait_until(self.app, lambda: not events._claim_busy))
self.assertIs(events._claim_thread, executor_thread)
self.assertTrue(
executor_thread is not None and executor_thread.isRunning()
)
self.assertEqual(len(set(gateway.thread_ids)), 1)
events.shutdown()
self.assertFalse(executor_thread.isRunning())
page.deleteLater()
def test_new_claim_refreshes_table_before_collection_finishes(self):
gateway = RecordingClaimGateway(collect_admin_task("COL-EARLY"))
collect_started = threading.Event()
client/test/test_settings_ui_event.py
+5
View File
@@ -513,6 +513,10 @@ class SettingsPageEventTest(unittest.TestCase):
)
page.deviceTableModel.set_checked_serial("USB-001")
self._wait_until(lambda: page.eventBinder._pdd_check_thread is None)
changed_serials = []
page.eventBinder.androidDeviceConfigurationChanged.connect(
changed_serials.append
)
page.saveButton.click()
self.assertFalse(page.saveButton.isEnabled())
@@ -530,6 +534,7 @@ class SettingsPageEventTest(unittest.TestCase):
"当前使用设备：USB-001（已保存）",
)
self.assertTrue(page.deleteButton.isEnabled())
self.assertEqual(changed_serials, ["USB-001"])
page.eventBinder.shutdown()
page.deleteLater()
docs/client/02-architecture.md
+18 -6
View File
@@ -156,11 +156,10 @@ Qt 主线程
├── 窗口、页面、表格模型和用户事件
└── 接收后台信号并更新界面
任务工作线程
├── Admin 请求
├── uiautomator2 调用
持久设备工作线程（单设备、固定 QThread）
├── uiautomator2 Device 的创建、健康检查、调用和释放
├── 页面等待与 XML 解析
└── 单个任务的串行执行
└── 采集、采购、订单核对和手动重新执行的串行命令
结果提交工作线程或同一任务线程的独立队列
└── Outbox 重试，不重复执行 PDD 操作
@@ -172,6 +171,12 @@ Qt 主线程
- `[必须]` QWidget 只能在 Qt 主线程创建和访问。
- `[必须]` 一个 Android 设备由一个工作线程独占，不跨线程共享 uiautomator2 Device 对象。
- `[必须]` 同一设备在 90 秒空闲期内只复用 Device 连接；每个任务必须重新读取
当前应用、页面、控件树、商品、价格、规格和采购安全状态。
- `[必须]` 停止自动获取、切换或删除已保存设备、设备断连、配置变化、窗口关闭
和空闲超时都会在设备线程释放连接。关闭只使用协作停止和 `quit()/wait()`。
- `[必须]` `task_runs.irreversible_action_at` 有值后只允许把核单命令放入队列，
不得重新执行采购下单步骤。
- `[必须]` 后台信号只传递不可变数据、稳定编号或轻量视图模型。
- `[必须]` 点“停止获取”后不再领取新任务，当前任务在定义的安全点退出。
- `[建议]` 关闭窗口时应选择停止、等待或后台继续；MVP 默认安全停止并持久化状态。
@@ -259,6 +264,12 @@ def start_task(self, remote_task_id: str) -> None:
| 用 `thread.terminate()` 停任务 | 数据库写一半、订单状态不明 | 用 `cancel()` 置标志位，在安全点退出 |
| 在 `run()` 里改界面 | 随机崩溃，且很难复现 | 只 emit 信号 |
PDD 设备执行器是这个模板的长生命周期版本：QThread 在第一次设备任务时创建，
后续通过队列信号提交一条命令，命令完成后线程继续等待；不能在 Worker 内写无限
轮询。设备 Worker 仍不访问界面。停止接单后等待当前命令到安全点结束，再在线程内
释放 Device，最后执行 `quit()/wait()`。Outbox 批量重新上报不操作手机，继续使用
独立的网络 Worker，不进入设备命令队列。
### 5.2 后台结果回主线程 / 迟到结果
窗口关掉了，后台任务还在跑，跑完再发信号——这时槽函数去访问已经销毁的控件，程序就崩了。
@@ -403,8 +414,9 @@ reconcile_purchase(task, run) -> PurchaseResult | ManualReview
或电话。最终匹配由领域服务完成，不能让页面解析层单独决定采购成功。
正式 Client 由 `pdd_u2_purchase_adapter.py` 分别实现演练和 live 接口，并在
`ui_main.py` 注入工厂。Adapter 通过 `PddDeviceService` 独占连接，每次判断
都重新读取当前包名和控件树。当前 Admin 下发的 `color` 和 `size`
`ui_main.py` 注入工厂。Adapter 通过设备工作线程绑定的
`PersistentPddDeviceService` 独占一次任务会话；同一设备可在 90 秒内复用底层
Device 连接，但每次判断都重新读取当前包名和控件树。当前 Admin 下发的 `color` 和 `size`
使用精确文字匹配；其他动态维度直接停止，不做相似匹配。原生 PDD
控件树不暴露商品编号，因此商品编号来自 Adapter 本次已校验并打开的
PDD URL，包名和页面类型仍以最新控件树确认。