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

feat: 增加采集任务重新执行入口 (#63)

Browse Source
This commit is contained in:
chengma
2026-08-09 21:44:29 +08:00
parent 72821eaaef
commit eda3661604
12 changed files with 509 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/collect_task_service.py
+19 -4
View File
@@ -111,11 +111,26 @@ class CollectTaskService:
if task is None:
raise RuntimeError(f"任务 {remote.task_id} 未能保存到本地")
return self._execute_task(task.remote_task_id)
def execute_selected(self, remote_task_id: str) -> CollectTaskOutcome:
"""只执行指定的本地采集任务，不领取任务或补交其他结果。"""
if not self._device_address.strip():
raise ValueError("请先在设置页选择并保存 Android 设备")
task = self._repository.get_task(remote_task_id)
if task is None:
raise ValueError(f"任务 {remote_task_id} 不存在")
return self._execute_task(task.remote_task_id)
def _execute_task(self, remote_task_id: str) -> CollectTaskOutcome:
"""执行一条已经保存在本地且处于待执行状态的采集任务。"""
if self._cancelled():
return CollectTaskOutcome("cancelled", "本次采集已取消", task.remote_task_id)
return CollectTaskOutcome("cancelled", "本次采集已取消", remote_task_id)
started = self._repository.start_collect_run(
task.remote_task_id, self._device_address
remote_task_id, self._device_address
)
collector = self._factory(
self._device_address, self._client.client_id, self._cancelled
@@ -123,13 +138,13 @@ class CollectTaskService:
try:
result = collector.collect(started.task)
event = self._repository.save_collect_result(
task.remote_task_id, started.attempt_id, result.to_pdd_data()
remote_task_id, started.attempt_id, result.to_pdd_data()
)
except PddCollectError as exc:
status, retryable = self._classify_error(exc.code)
report_code = self._report_error_code(exc.code)
event = self._repository.save_collect_failure(
task.remote_task_id,
remote_task_id,
started.attempt_id,
status,
report_code,
client/src/db_schema.py
+9 -1
View File
@@ -4,7 +4,7 @@
``MIGRATIONS`` 末尾增加版本，不能修改已经发布的迁移。
"""
SCHEMA_VERSION = 1
SCHEMA_VERSION = 2
MIGRATION_1 = (
@@ -127,6 +127,14 @@ MIGRATION_1 = (
)
MIGRATION_2 = (
"""
ALTER TABLE task_runs ADD COLUMN result_data TEXT
""",
)
MIGRATIONS = {
1: MIGRATION_1,
2: MIGRATION_2,
}
client/src/pdd_ui.py
+31 -2
View File
@@ -21,8 +21,8 @@
- 图标用 `FluentIcon`，**不得用表情符号**。
- 页面 `objectName` 固定为 `pddTaskPage`，不要改。
注意：本页唯一会产生外部后果（真的去操作手机、可能下单）的命令是
“开始自动获取”。其余操作全部只读本地数据库。
注意：“开始自动获取”和“重新执行”会真的操作手机。重新执行当前只允许
采集任务，采购任务必须在事件层拦截，不能从本入口下单。
"""
from dataclasses import dataclass
@@ -276,6 +276,7 @@ class PDDTaskPage(QWidget):
autoFetchRequested = pyqtSignal()
searchRequested = pyqtSignal(dict)
refreshRequested = pyqtSignal()
rerunRequested = pyqtSignal(str)
detailRequested = pyqtSignal(str)
def __init__(self, parent=None):
@@ -323,6 +324,10 @@ class PDDTaskPage(QWidget):
self.refreshButton = PushButton(FIF.SYNC, "刷新", self)
self.refreshButton.setAccessibleName("刷新本地任务表格")
self.rerunButton = PushButton(FIF.UPDATE, "重新执行", self)
self.rerunButton.setAccessibleName("重新执行当前选中的采集任务")
self.rerunButton.setEnabled(False)
self.commandCard = CardWidget(self)
commandLayout = QGridLayout(self.commandCard)
commandLayout.setContentsMargins(20, 14, 20, 14)
@@ -338,6 +343,7 @@ class PDDTaskPage(QWidget):
commandLayout.setColumnStretch(6, 4)
commandLayout.setColumnStretch(8, 1)
commandLayout.addWidget(self.refreshButton, 0, 9)
commandLayout.addWidget(self.rerunButton, 0, 10)
def _build_content_area(self) -> None:
self.taskTable = TableView(self)
@@ -410,9 +416,13 @@ class PDDTaskPage(QWidget):
self.searchButton.clicked.connect(self._apply_filters)
self.keywordInput.returnPressed.connect(self._apply_filters)
self.refreshButton.clicked.connect(self.refreshRequested.emit)
self.rerunButton.clicked.connect(self._request_rerun)
self.clearFiltersButton.clicked.connect(self.clear_filters)
self.taskTable.clicked.connect(self._on_table_clicked)
self.taskTable.activated.connect(self._on_table_activated)
self.taskTable.selectionModel().selectionChanged.connect(
self._update_rerun_button
)
self.findShortcut = QShortcut(QKeySequence.Find, self)
self.findShortcut.activated.connect(self.keywordInput.setFocus)
@@ -497,6 +507,25 @@ class PDDTaskPage(QWidget):
task = self.taskModel.row_at(index.row()) if index.isValid() else None
return task.remote_task_id if task else ""
def set_rerun_running(self, running: bool) -> None:
"""重新采集期间锁定两个会操作手机的入口。"""
self.rerunButton.setText("正在重新采集" if running else "重新执行")
self.autoFetchButton.setEnabled(not running)
self._update_rerun_button()
def _request_rerun(self) -> None:
task_id = self.current_task_id()
if task_id:
self.rerunRequested.emit(task_id)
def _update_rerun_button(self) -> None:
running = self.rerunButton.text() == "正在重新采集"
has_selection = self.taskTable.selectionModel().hasSelection()
self.rerunButton.setEnabled(
has_selection and bool(self.current_task_id()) and not running
)
def _apply_filters(self) -> None:
filters = {
"task_type": self.taskTypeCombo.currentText(),
client/src/pdd_ui_event.py
+150 -4
View File
@@ -30,7 +30,7 @@ from PyQt5.QtCore import (
pyqtSignal,
pyqtSlot,
)
from qfluentwidgets import InfoBar, InfoBarPosition
from qfluentwidgets import InfoBar, InfoBarPosition, MessageBox
from .admin_gateway import (
AdminGatewayError,
@@ -51,7 +51,7 @@ from .task_models import (
TaskSummary,
TaskType,
)
from .task_repository import TaskRepository
from .task_repository import CollectRerunError, TaskRepository
from .task_detail_view import TaskDetailWindow
@@ -141,6 +141,7 @@ class ClaimTaskWorker(QObject):
client_service: CurrentClientService,
android_device_service: SelectedAndroidDeviceService,
collect_service_factory: Optional[CollectServiceFactory] = None,
selected_task_id: str = "",
) -> None:
super().__init__()
self._gateway = gateway
@@ -149,6 +150,7 @@ class ClaimTaskWorker(QObject):
self._android_device_service = android_device_service
self._cancelled = False
self._collect_service_factory = collect_service_factory
self._selected_task_id = selected_task_id
def cancel(self) -> None:
"""阻止尚未开始的领取；已领取的任务仍必须保存到本地。"""
@@ -166,7 +168,7 @@ class ClaimTaskWorker(QObject):
return
android_serial = self._android_device_service.load()
result = CollectTaskService(
service = CollectTaskService(
self._gateway,
self._task_repository,
ClientInfo(
@@ -176,7 +178,12 @@ class ClaimTaskWorker(QObject):
android_serial or "",
cancelled=lambda: self._cancelled,
collect_service_factory=self._collect_service_factory,
).execute_one()
)
result = (
service.execute_selected(self._selected_task_id)
if self._selected_task_id
else service.execute_one()
)
if not self._cancelled or result.kind == "cancelled":
self.outcome.emit(result.kind, result.message, result.task_id)
except AdminGatewayError as exc:
@@ -265,6 +272,7 @@ class PDDTaskPageEvent(QObject):
page.searchRequested.connect(self.search_tasks)
page.refreshRequested.connect(self.refresh_tasks)
page.rerunRequested.connect(self.request_rerun)
page.autoFetchRequested.connect(self._request_claim_task)
page.detailRequested.connect(self.show_task_detail)
page.taskModel.loadMoreRequested.connect(self._load_page)
@@ -294,6 +302,144 @@ class PDDTaskPageEvent(QObject):
self._reload()
@pyqtSlot(str)
def request_rerun(self, task_id: str) -> None:
"""确认后在后台重新采集当前选中的一条终态任务。"""
if self._closing or not task_id:
return
if self._auto_fetch_running or self._claim_busy:
self._show_rerun_warning(
"暂时不能重新执行",
"自动获取或其他采集正在运行，请停止并等待当前任务结束。",
)
return
if self._claim_gateway_error or self._claim_gateway is None:
self._show_rerun_warning(
"重新执行不可用",
self._claim_gateway_error or "Admin 提交服务未初始化",
)
return
try:
client_settings = self._client_service.load()
android_serial = self._selected_android_device_service.load()
except Exception:
self._show_rerun_warning(
"重新执行不可用", "无法读取本地设备设置，请到设置页检查。"
)
return
if not client_settings.client_id:
self._show_rerun_warning(
"重新执行不可用", "请先在设置页保存当前设备号和设备名。"
)
return
if not android_serial:
self._show_rerun_warning(
"重新执行不可用", "请先在设置页选择并保存 Android 设备。"
)
return
try:
detail = self._repository.validate_collect_rerun(task_id)
except (CollectRerunError, ValueError) as exc:
self._show_rerun_warning("不能重新执行", str(exc))
return
except Exception:
self._show_rerun_warning(
"不能重新执行", "无法读取任务状态，请检查数据库后重试。"
)
return
title = detail.title or "尚未获取标题"
dialog = MessageBox(
"确认重新采集",
f"任务：{detail.remote_task_id}\n商品：{title}\n\n"
"新结果会覆盖 Client 和 Admin 的当前采集数据，旧结果仍保留在执行记录中。",
self._page.window(),
)
dialog.yesButton.setText("重新采集")
dialog.cancelButton.setText("取消")
dialog.cancelButton.setFocus()
if not dialog.exec():
return
try:
self._repository.prepare_collect_rerun(task_id)
except (CollectRerunError, ValueError) as exc:
self._show_rerun_warning("不能重新执行", str(exc))
self._reload()
return
except Exception:
self._show_rerun_warning(
"重新执行失败", "无法更新本地任务状态，请检查数据库后重试。"
)
return
self._start_rerun_worker(task_id)
def _start_rerun_worker(self, task_id: str) -> None:
"""启动只处理指定任务的工作线程。"""
assert self._claim_gateway is not None
self._claim_busy = True
self._page.set_rerun_running(True)
self._page.set_engine_status(f"正在重新采集任务 {task_id}…")
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,
)
worker.moveToThread(thread)
thread.started.connect(worker.run)
worker.retryableFailed.connect(self._on_rerun_failed)
worker.failed.connect(self._on_rerun_failed)
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()
@pyqtSlot(str, str, str)
def _on_rerun_outcome(self, kind: str, message: str, _task_id: str) -> None:
if self._closing:
return
self._reload()
self._page.set_engine_status(message)
if kind != "succeeded":
self._show_claim_error("重新采集需要处理", message)
@pyqtSlot(str)
def _on_rerun_failed(self, message: str) -> None:
if self._closing:
return
self._page.set_engine_status(message)
self._show_claim_error("重新采集失败", message)
@pyqtSlot()
def _on_rerun_thread_finished(self) -> None:
self._claim_worker = None
self._claim_thread = None
self._claim_busy = False
if not self._closing:
self._page.set_rerun_running(False)
def _show_rerun_warning(self, title: str, content: str) -> None:
InfoBar.warning(
title=title,
content=content,
isClosable=True,
duration=5000,
position=InfoBarPosition.TOP_RIGHT,
parent=self._page,
)
@pyqtSlot()
def _request_claim_task(self) -> None:
"""切换持续自动获取；每一轮仍只启动一个后台 Worker。"""
client/src/task_models.py
+1
View File
@@ -161,6 +161,7 @@ class TaskRunRecord:
error_code: Optional[str]
error_message: Optional[str]
diagnostics_json: Optional[Dict[str, Any]]
result_data: Optional[Dict[str, Any]]
artifact_directory: Optional[str]
created_at: str
updated_at: str
client/src/task_repository.py
+84 -3
View File
@@ -34,6 +34,10 @@ class DuplicateTaskError(ValueError):
"""相同远程任务编号已经存在，不能覆盖。"""
class CollectRerunError(ValueError):
"""当前任务不满足重新采集条件。"""
def utc_now_iso() -> str:
"""返回精确到秒的 UTC ISO 8601 时间。"""
@@ -206,6 +210,81 @@ class TaskRepository:
connection.close()
return self._to_detail(row) if row is not None else None
def validate_collect_rerun(self, remote_task_id: str) -> TaskDetail:
"""校验任务能否重新采集，成功时返回任务详情。"""
connection = open_database(self._db_path)
try:
row = connection.execute(
"SELECT * FROM pdd_tasks WHERE remote_task_id = ?",
(remote_task_id,),
).fetchone()
if row is None:
raise CollectRerunError(f"任务 {remote_task_id} 不存在")
self._check_collect_rerun(connection, row)
finally:
connection.close()
return self._to_detail(row)
def prepare_collect_rerun(self, remote_task_id: str) -> TaskDetail:
"""事务内把已结束的采集任务恢复为待执行，保留旧结果。"""
now = utc_now_iso()
connection = open_database(self._db_path)
try:
with connection:
row = connection.execute(
"SELECT * FROM pdd_tasks WHERE remote_task_id = ?",
(remote_task_id,),
).fetchone()
if row is None:
raise CollectRerunError(f"任务 {remote_task_id} 不存在")
self._check_collect_rerun(connection, row)
connection.execute(
"UPDATE pdd_tasks SET status = 'claimed',"
" current_step = 'rerun_requested', finished_at = NULL,"
" last_error_code = NULL, last_error_message = NULL,"
" updated_at = ? WHERE id = ?",
(now, row["id"]),
)
finally:
connection.close()
task = self.get_task(remote_task_id)
assert task is not None
return task
@staticmethod
def _check_collect_rerun(
connection: sqlite3.Connection, row: sqlite3.Row
) -> None:
"""检查重新采集的类型、终态和 Outbox 约束。"""
if row["task_type"] != TaskType.COLLECT.value:
raise CollectRerunError("采购任务不能重新执行，以免重复下单")
allowed_statuses = {
TaskStatus.SUCCEEDED.value,
TaskStatus.FAILED.value,
TaskStatus.CANCELLED.value,
TaskStatus.MANUAL_REVIEW.value,
}
if row["status"] not in allowed_statuses:
status_name = {
TaskStatus.CLAIMED.value: "待执行",
TaskStatus.RUNNING.value: "执行中",
TaskStatus.RESULT_PENDING.value: "结果待提交",
TaskStatus.RETRY_WAIT.value: "等待重试",
}.get(row["status"], row["status"])
raise CollectRerunError(f"任务当前为“{status_name}”，不能重新采集")
unsent_count = int(
connection.execute(
"SELECT COUNT(*) FROM outbox_events"
" WHERE task_id = ? AND status != 'sent'",
(row["id"],),
).fetchone()[0]
)
if unsent_count:
raise CollectRerunError("任务仍有未发送的结果，请先完成提交")
def start_collect_run(
self, remote_task_id: str, device_address: str
) -> StartedTaskRun:
@@ -284,22 +363,24 @@ class TaskRepository:
"pdd_data": pdd_data,
}
idempotency_key = f"{remote_task_id}:{attempt_id}:result-v1"
result_json = json.dumps(pdd_data, ensure_ascii=False)
connection.execute(
"UPDATE pdd_tasks SET status = 'result_pending',"
" current_step = 'submit_result', pdd_data = ?, goods_id = ?,"
" title = ?, price_cent = ?, finished_at = ?, updated_at = ?"
" WHERE id = ?",
(
json.dumps(pdd_data, ensure_ascii=False),
result_json,
pdd_data.get("goods_id"), pdd_data.get("title"),
self._summary_price(pdd_data), now, now, task["id"],
),
)
connection.execute(
"UPDATE task_runs SET run_status = 'succeeded',"
" current_step = 'submit_result', finished_at = ?, updated_at = ?"
" current_step = 'submit_result', result_data = ?,"
" finished_at = ?, updated_at = ?"
" WHERE attempt_id = ?",
(now, now, attempt_id),
(result_json, now, now, attempt_id),
)
cursor = connection.execute(
"INSERT INTO outbox_events (task_id, event_type, idempotency_key,"