feat: 增加任务批量勾选和重新上报 (#89)

This commit is contained in:
chengma
2026-08-10 10:56:48 +08:00
parent d017062d82
commit 34ff39a200
8 changed files with 734 additions and 60 deletions
+1 -1
View File
@@ -72,7 +72,7 @@
- 优先使用 `FluentWindow`、Fluent 导航、主题和图标;不得使用表情符号充当结构图标。
- 使用布局、尺寸策略和伸缩项,不用固定坐标排列常规界面。
- 任务表格使用模型/视图和稳定任务编号,不把完整 `pdd_data` 放入隐藏列,也不为每个单元格创建常驻 QWidget。
- “获取任务”(启动后变为“停止获取”)是界面上唯一会产生外部后果的命令;其余操作只读本地数据库。
- 会产生外部后果的命令只有“获取任务”“重新执行”和“重新上报”:重新执行只允许单条采集任务;重新上报只重发既有结果 Outbox,不操作手机、不创建新结果。搜索、筛选、刷新和详情仍只读本地数据库。
- 本地只保存已领取的任务,不缓存 Admin 任务池;已完成任务永久保留,不得清理。
- 普通成功更新页面状态即可;可恢复错误使用 `InfoBar`,只有必须阻断决策时才使用模态对话框。
- 主要流程必须支持键盘;表单具有可见标签;状态和错误不能只依赖颜色。
+223 -46
View File
@@ -56,7 +56,8 @@ from qfluentwidgets import (
PAGE_SIZE = 50
DETAIL_COLUMN = 8
CHECK_COLUMN = 0
DETAIL_COLUMN = 9
REMOTE_TASK_ID_ROLE = Qt.UserRole + 1
@@ -81,8 +82,10 @@ class TaskTableModel(QAbstractTableModel):
"""PDD 任务表格模型,只保存已经从数据库读取的轻量行。"""
loadMoreRequested = pyqtSignal(int, int)
checkedTaskIdsChanged = pyqtSignal(object)
HEADERS = (
"选择",
"任务类型",
"商品标题",
"颜色",
@@ -100,6 +103,7 @@ class TaskTableModel(QAbstractTableModel):
self._has_more = False
self._loading = False
self._placeholder_row_count = 1
self._checked_task_ids: set[str] = set()
def rowCount(self, parent=QModelIndex()) -> int:
if parent.isValid():
@@ -121,18 +125,25 @@ class TaskTableModel(QAbstractTableModel):
task = self._rows[index.row()]
column = index.column()
if role == Qt.CheckStateRole and column == CHECK_COLUMN:
return (
Qt.Checked
if task.remote_task_id in self._checked_task_ids
else Qt.Unchecked
)
if role == Qt.DisplayRole:
return self._display_value(task, column)
if role == Qt.TextAlignmentRole:
if column in (0, 6, 8):
if column in (0, 1, 7, 9):
return Qt.AlignCenter
if column in (4, 5):
if column in (5, 6):
return Qt.AlignRight | Qt.AlignVCenter
return Qt.AlignLeft | Qt.AlignVCenter
if role == Qt.ToolTipRole:
if column == 1:
if column == 2:
return task.title or "尚未获取标题"
if column == DETAIL_COLUMN:
return f"查看任务 {task.remote_task_id} 的详情"
@@ -145,7 +156,30 @@ class TaskTableModel(QAbstractTableModel):
def flags(self, index):
if not index.isValid() or index.row() >= len(self._rows):
return Qt.NoItemFlags
return Qt.ItemIsEnabled | Qt.ItemIsSelectable
flags = Qt.ItemIsEnabled | Qt.ItemIsSelectable
if index.column() == CHECK_COLUMN:
flags |= Qt.ItemIsUserCheckable
return flags
def setData(self, index, value, role=Qt.EditRole):
"""切换任务勾选状态;业务身份始终使用远端任务编号。"""
if (
not index.isValid()
or index.column() != CHECK_COLUMN
or index.row() >= len(self._rows)
or role != Qt.CheckStateRole
):
return False
task_id = self._rows[index.row()].remote_task_id
if value == Qt.Checked:
self._checked_task_ids.add(task_id)
else:
self._checked_task_ids.discard(task_id)
self.dataChanged.emit(index, index, [Qt.CheckStateRole])
self.checkedTaskIdsChanged.emit(self.checked_task_ids())
return True
def canFetchMore(self, parent=QModelIndex()) -> bool:
"""告诉 Qt 数据库中是否还有下一页。"""
@@ -172,6 +206,7 @@ class TaskTableModel(QAbstractTableModel):
self._has_more = False
self._loading = False
self.endResetModel()
self._prune_checked_task_ids()
def reset_for_reload(self) -> None:
"""清空旧结果,准备从数据库重新读取第一页。"""
@@ -192,6 +227,8 @@ class TaskTableModel(QAbstractTableModel):
self._has_more = has_more
self._loading = False
self.endResetModel()
if not has_more:
self._prune_checked_task_ids()
return
if page:
@@ -201,6 +238,8 @@ class TaskTableModel(QAbstractTableModel):
self.endInsertRows()
self._has_more = has_more
self._loading = False
if not has_more:
self._prune_checked_task_ids()
def finish_loading(self) -> None:
"""查询失败时停止继续请求,等待用户刷新或重新搜索。"""
@@ -231,6 +270,54 @@ class TaskTableModel(QAbstractTableModel):
return len(self._rows)
def checked_task_ids(self) -> tuple[str, ...]:
"""按当前列表顺序返回已勾选的稳定任务编号。"""
loaded_order = [
task.remote_task_id
for task in self._rows
if task.remote_task_id in self._checked_task_ids
]
loaded = set(loaded_order)
return tuple(loaded_order) + tuple(sorted(self._checked_task_ids - loaded))
def clear_checked_tasks(self) -> None:
"""清空全部勾选,并只刷新复选框列。"""
if not self._checked_task_ids:
return
self._checked_task_ids.clear()
if self._rows:
self.dataChanged.emit(
self.index(0, CHECK_COLUMN),
self.index(len(self._rows) - 1, CHECK_COLUMN),
[Qt.CheckStateRole],
)
self.checkedTaskIdsChanged.emit(())
def uncheck_tasks(self, remote_task_ids: Iterable[str]) -> None:
"""取消指定任务的勾选,供批量操作成功后更新界面。"""
removed = self._checked_task_ids.intersection(remote_task_ids)
if not removed:
return
self._checked_task_ids.difference_update(removed)
for row, task in enumerate(self._rows):
if task.remote_task_id in removed:
index = self.index(row, CHECK_COLUMN)
self.dataChanged.emit(index, index, [Qt.CheckStateRole])
self.checkedTaskIdsChanged.emit(self.checked_task_ids())
def _prune_checked_task_ids(self) -> None:
"""完整加载结束后移除本地已不存在的勾选。"""
loaded = {task.remote_task_id for task in self._rows}
removed = self._checked_task_ids - loaded
if not removed:
return
self._checked_task_ids.intersection_update(loaded)
self.checkedTaskIdsChanged.emit(self.checked_task_ids())
def set_placeholder_row_count(self, count: int) -> None:
"""设置无数据时显示的禁用空行数量。"""
@@ -248,24 +335,26 @@ class TaskTableModel(QAbstractTableModel):
@staticmethod
def _display_value(task: TaskRow, column: int) -> str:
if column == 0:
return task.task_type
if column == CHECK_COLUMN:
return ""
if column == 1:
return task.title or "尚未获取标题"
return task.task_type
if column == 2:
return task.color or "—"
return task.title or "尚未获取标题"
if column == 3:
return task.size or "—"
return task.color or "—"
if column == 4:
return task.size or "—"
if column == 5:
if task.price_cents is None:
return "—"
suffix = " 起" if task.price_is_starting else ""
return f"¥{task.price_cents / 100:.2f}{suffix}"
if column == 5:
return "—" if task.quantity is None else str(task.quantity)
if column == 6:
return task.status
return "—" if task.quantity is None else str(task.quantity)
if column == 7:
return task.status
if column == 8:
return task.updated_at or "—"
return "详情"
@@ -278,6 +367,7 @@ class PDDTaskPage(QWidget):
refreshRequested = pyqtSignal()
rerunRequested = pyqtSignal(str)
rerunCancelRequested = pyqtSignal()
resubmitRequested = pyqtSignal(object)
openSettingsRequested = pyqtSignal()
detailRequested = pyqtSignal(str)
@@ -286,6 +376,7 @@ class PDDTaskPage(QWidget):
self.setObjectName("pddTaskPage")
self._rerun_state = "idle"
self._resubmit_running = False
self.taskModel = TaskTableModel(self)
self._build_command_area()
self._build_content_area()
@@ -301,11 +392,13 @@ class PDDTaskPage(QWidget):
self.autoFetchButton.setMinimumWidth(132)
self.autoFetchButton.setAccessibleName("开始自动获取 PDD 任务")
self.taskTypeLabel = CaptionLabel("任务类型", self)
self.taskTypeCombo = ComboBox(self)
self.taskTypeCombo.addItems(("全部", "采集", "采购"))
self.taskTypeCombo.setMinimumWidth(100)
self.taskTypeCombo.setAccessibleName("任务类型")
self.statusLabelCaption = CaptionLabel("任务状态", self)
self.statusCombo = ComboBox(self)
self.statusCombo.addItems(
("全部", "待执行", "执行中", "已完成", "失败", "需要人工处理")
@@ -313,6 +406,7 @@ class PDDTaskPage(QWidget):
self.statusCombo.setMinimumWidth(130)
self.statusCombo.setAccessibleName("任务状态")
self.keywordLabel = CaptionLabel("关键词", self)
self.keywordInput = SearchLineEdit(self)
self.keywordInput.setPlaceholderText("任务编号、商品编号或商品标题")
self.keywordInput.setClearButtonEnabled(True)
@@ -331,22 +425,68 @@ class PDDTaskPage(QWidget):
self.rerunButton.setAccessibleName("重新执行当前选中的采集任务")
self.rerunButton.setEnabled(False)
self.resubmitButton = PushButton(FIF.SEND, "重新上报", self)
self.resubmitButton.setAccessibleName("重新上报勾选任务的已有结果")
self.resubmitButton.setEnabled(False)
self.commandCard = CardWidget(self)
commandLayout = QGridLayout(self.commandCard)
commandLayout.setContentsMargins(20, 14, 20, 14)
commandLayout.setHorizontalSpacing(12)
commandLayout.addWidget(self.autoFetchButton, 0, 0)
commandLayout.addWidget(CaptionLabel("任务类型", self.commandCard), 0, 1)
commandLayout.addWidget(self.taskTypeCombo, 0, 2)
commandLayout.addWidget(CaptionLabel("任务状态", self.commandCard), 0, 3)
commandLayout.addWidget(self.statusCombo, 0, 4)
commandLayout.addWidget(CaptionLabel("关键词", self.commandCard), 0, 5)
commandLayout.addWidget(self.keywordInput, 0, 6)
commandLayout.addWidget(self.searchButton, 0, 7)
commandLayout.setColumnStretch(6, 4)
commandLayout.setColumnStretch(8, 1)
commandLayout.addWidget(self.refreshButton, 0, 9)
commandLayout.addWidget(self.rerunButton, 0, 10)
self.checkedCountLabel = CaptionLabel("已选 0 条", self.commandCard)
self.checkedCountLabel.setAccessibleName("已勾选任务数量")
self.commandLayout = QGridLayout(self.commandCard)
self.commandLayout.setContentsMargins(20, 14, 20, 14)
self.commandLayout.setHorizontalSpacing(12)
self.commandLayout.setVerticalSpacing(10)
self._compact_commands = False
# 先使用安全的双行布局,窗口真正变宽后再切换为单行。
self._update_command_layout(compact=True)
def _update_command_layout(self, compact: bool) -> None:
"""在窄窗口把搜索区换到第二行,不重建任何控件。"""
if compact == self._compact_commands and self.commandLayout.count():
return
self._compact_commands = compact
for column in range(13):
self.commandLayout.setColumnStretch(column, 0)
if compact:
positions = (
(self.autoFetchButton, 0, 0, 1, 1),
(self.taskTypeLabel, 0, 1, 1, 1),
(self.taskTypeCombo, 0, 2, 1, 1),
(self.statusLabelCaption, 0, 3, 1, 1),
(self.statusCombo, 0, 4, 1, 1),
(self.checkedCountLabel, 0, 6, 1, 1),
(self.rerunButton, 0, 7, 1, 1),
(self.resubmitButton, 0, 8, 1, 1),
(self.keywordLabel, 1, 0, 1, 1),
(self.keywordInput, 1, 1, 1, 5),
(self.searchButton, 1, 6, 1, 1),
(self.refreshButton, 1, 7, 1, 1),
)
self.commandLayout.setColumnStretch(5, 1)
else:
positions = (
(self.autoFetchButton, 0, 0, 1, 1),
(self.taskTypeLabel, 0, 1, 1, 1),
(self.taskTypeCombo, 0, 2, 1, 1),
(self.statusLabelCaption, 0, 3, 1, 1),
(self.statusCombo, 0, 4, 1, 1),
(self.keywordLabel, 0, 5, 1, 1),
(self.keywordInput, 0, 6, 1, 1),
(self.searchButton, 0, 7, 1, 1),
(self.refreshButton, 0, 8, 1, 1),
(self.checkedCountLabel, 0, 10, 1, 1),
(self.rerunButton, 0, 11, 1, 1),
(self.resubmitButton, 0, 12, 1, 1),
)
self.commandLayout.setColumnStretch(6, 4)
self.commandLayout.setColumnStretch(9, 1)
for widget, row, column, row_span, column_span in positions:
self.commandLayout.addWidget(
widget, row, column, row_span, column_span
)
def _build_content_area(self) -> None:
self.taskTable = TableView(self)
@@ -362,16 +502,17 @@ class PDDTaskPage(QWidget):
header = self.taskTable.horizontalHeader()
header.setSectionResizeMode(QHeaderView.Interactive)
header.setSectionResizeMode(1, QHeaderView.Stretch)
header.setSectionResizeMode(2, QHeaderView.Stretch)
header.setMinimumSectionSize(64)
self.taskTable.setColumnWidth(0, 82)
self.taskTable.setColumnWidth(2, 108)
self.taskTable.setColumnWidth(3, 82)
self.taskTable.setColumnWidth(4, 96)
self.taskTable.setColumnWidth(5, 70)
self.taskTable.setColumnWidth(6, 116)
self.taskTable.setColumnWidth(7, 156)
self.taskTable.setColumnWidth(8, 72)
self.taskTable.setColumnWidth(0, 64)
self.taskTable.setColumnWidth(1, 82)
self.taskTable.setColumnWidth(3, 108)
self.taskTable.setColumnWidth(4, 82)
self.taskTable.setColumnWidth(5, 96)
self.taskTable.setColumnWidth(6, 70)
self.taskTable.setColumnWidth(7, 116)
self.taskTable.setColumnWidth(8, 156)
self.taskTable.setColumnWidth(9, 72)
self.taskTable.viewport().installEventFilter(self)
self.emptyStateCard = CardWidget(self)
@@ -420,12 +561,11 @@ class PDDTaskPage(QWidget):
self.keywordInput.returnPressed.connect(self._apply_filters)
self.refreshButton.clicked.connect(self.refreshRequested.emit)
self.rerunButton.clicked.connect(self._request_rerun)
self.resubmitButton.clicked.connect(self._request_resubmit)
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.taskModel.checkedTaskIdsChanged.connect(self._update_checked_actions)
self.findShortcut = QShortcut(QKeySequence.Find, self)
self.findShortcut.activated.connect(self.keywordInput.setFocus)
@@ -435,6 +575,12 @@ class PDDTaskPage(QWidget):
self._update_placeholder_rows()
return super().eventFilter(watched, event)
def resizeEvent(self, event) -> None:
"""按内容区宽度切换顶部命令区的单行或双行布局。"""
self._update_command_layout(compact=event.size().width() < 1180)
super().resizeEvent(event)
def set_tasks(self, tasks: Iterable[TaskRow]) -> None:
"""显示任务数据,并尽量恢复原来选中的稳定任务编号。"""
@@ -516,6 +662,12 @@ class PDDTaskPage(QWidget):
self._rerun_state = "running" if running else "idle"
self._update_rerun_button()
def set_resubmit_running(self, running: bool) -> None:
"""设置重新上报忙碌状态并防止重复提交。"""
self._resubmit_running = running
self._update_checked_actions(self.taskModel.checked_task_ids())
def set_rerun_cancelling(self) -> None:
"""显示取消请求已发出,防止用户重复点击。"""
@@ -528,33 +680,58 @@ class PDDTaskPage(QWidget):
return
if self._rerun_state != "idle":
return
task_id = self.current_task_id()
if task_id:
self.rerunRequested.emit(task_id)
task_ids = self.taskModel.checked_task_ids()
if len(task_ids) == 1:
self.rerunRequested.emit(task_ids[0])
def _request_resubmit(self) -> None:
if self._resubmit_running:
return
task_ids = self.taskModel.checked_task_ids()
if task_ids:
self.resubmitRequested.emit(task_ids)
def _update_rerun_button(self) -> None:
if self._rerun_state == "running":
self.rerunButton.setText("停止重新采集")
self.rerunButton.setAccessibleName("停止当前重新采集")
self.rerunButton.setEnabled(True)
self.resubmitButton.setEnabled(False)
self.autoFetchButton.setEnabled(False)
return
if self._rerun_state == "cancelling":
self.rerunButton.setText("正在停止…")
self.rerunButton.setAccessibleName("正在停止重新采集")
self.rerunButton.setEnabled(False)
self.resubmitButton.setEnabled(False)
self.autoFetchButton.setEnabled(False)
return
self.rerunButton.setText("重新执行")
self.rerunButton.setAccessibleName("重新执行当前选中的采集任务")
self.autoFetchButton.setEnabled(True)
has_selection = self.taskTable.selectionModel().hasSelection()
self.autoFetchButton.setEnabled(not self._resubmit_running)
self.rerunButton.setEnabled(
has_selection and bool(self.current_task_id())
not self._resubmit_running
and len(self.taskModel.checked_task_ids()) == 1
)
self.resubmitButton.setEnabled(
not self._resubmit_running
and bool(self.taskModel.checked_task_ids())
)
def _update_checked_actions(self, task_ids=None) -> None:
"""同步勾选数量和批量命令启用状态。"""
checked = tuple(task_ids or ())
self.checkedCountLabel.setText(f"已选 {len(checked)} 条")
self.resubmitButton.setText(
"正在上报…" if self._resubmit_running else "重新上报"
)
self.resubmitButton.setEnabled(bool(checked) and not self._resubmit_running)
self._update_rerun_button()
def _apply_filters(self) -> None:
self.taskModel.clear_checked_tasks()
filters = {
"task_type": self.taskTypeCombo.currentText(),
"status": self.statusCombo.currentText(),
+195 -1
View File
@@ -50,6 +50,7 @@ from .purchase_reconcile_service import PurchaseReconcileFactory
from .selected_android_device_service import SelectedAndroidDeviceService
from .settings_repository import SettingsRepository
from .task_models import (
OutboxStatus,
TaskFilters,
TaskStatus,
TaskSummary,
@@ -211,6 +212,82 @@ class ClaimTaskWorker(QObject):
self.completed.emit()
class ResultResubmitWorker(QObject):
"""在后台逐条重发既有结果 Outbox,不执行任何手机操作。"""
progress = pyqtSignal(int, int, str)
finished = pyqtSignal(object, object, object)
completed = pyqtSignal()
def __init__(
self,
gateway: AdminGateway,
repository: TaskRepository,
task_ids: tuple[str, ...],
) -> None:
super().__init__()
self._gateway = gateway
self._repository = repository
self._task_ids = task_ids
self._cancelled = False
def cancel(self) -> None:
"""当前网络请求结束后停止处理后续任务。"""
self._cancelled = True
@pyqtSlot()
def run(self) -> None:
succeeded: list[str] = []
failed: list[str] = []
skipped: list[str] = []
total = len(self._task_ids)
try:
for current, task_id in enumerate(self._task_ids, 1):
if self._cancelled:
break
self.progress.emit(current, total, task_id)
try:
event = self._repository.latest_result_outbox(task_id)
except Exception:
skipped.append(task_id)
continue
if event is None or event.status is OutboxStatus.SENDING:
skipped.append(task_id)
continue
self._repository.mark_outbox_sending(event.id)
try:
receipt = self._gateway.submit_result(
task_id,
event.idempotency_key,
event.payload_json,
)
if not receipt.accepted:
raise AdminGatewayError(
"ADMIN_RESULT_NOT_ACCEPTED",
"Admin 未确认接收任务结果",
False,
)
except AdminGatewayError as exc:
if exc.retryable:
self._repository.mark_outbox_retry(event.id, str(exc))
else:
self._repository.mark_outbox_failed(event.id, str(exc))
failed.append(task_id)
continue
except Exception as exc:
self._repository.mark_outbox_retry(event.id, str(exc))
failed.append(task_id)
continue
self._repository.mark_outbox_sent(event.id)
succeeded.append(task_id)
self.finished.emit(succeeded, failed, skipped)
finally:
self.completed.emit()
class PDDTaskPageEvent(QObject):
"""把 PDD 页面只读操作连接到本地任务 Repository。"""
@@ -242,6 +319,9 @@ class PDDTaskPageEvent(QObject):
self._claim_feedback: Optional[InfoBar] = None
self._device_feedback: Optional[InfoBar] = None
self._auto_fetch_running = False
self._resubmit_busy = False
self._resubmit_thread: Optional[QThread] = None
self._resubmit_worker: Optional[ResultResubmitWorker] = None
self._stop_requested = False
self._cycle_next_delay_ms: Optional[int] = None
self._stop_status = "自动获取:已停止 · 当前没有执行中的任务"
@@ -295,6 +375,7 @@ class PDDTaskPageEvent(QObject):
page.refreshRequested.connect(self.refresh_tasks)
page.rerunRequested.connect(self.request_rerun)
page.rerunCancelRequested.connect(self.request_cancel_rerun)
page.resubmitRequested.connect(self.request_resubmit)
page.autoFetchRequested.connect(self._request_claim_task)
page.detailRequested.connect(self.show_task_detail)
page.taskModel.loadMoreRequested.connect(self._load_page)
@@ -330,7 +411,7 @@ class PDDTaskPageEvent(QObject):
if self._closing or not task_id:
return
if self._auto_fetch_running or self._claim_busy:
if self._auto_fetch_running or self._claim_busy or self._resubmit_busy:
self._show_rerun_warning(
"暂时不能重新执行",
"自动获取或其他采集正在运行,请停止并等待当前任务结束。",
@@ -386,6 +467,100 @@ class PDDTaskPageEvent(QObject):
self._start_rerun_worker(task_id)
@pyqtSlot(object)
def request_resubmit(self, task_ids) -> None:
"""确认后在后台重发勾选任务的原始结果 Outbox。"""
stable_ids = tuple(
dict.fromkeys(str(value) for value in task_ids if value)
)
if self._closing or not stable_ids:
return
if self._auto_fetch_running or self._claim_busy or self._resubmit_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
dialog = MessageBox(
f"重新上报 {len(stable_ids)} 条任务结果?",
"只会重新提交本地已经保存的原始结果,"
"不会重新采集、采购或操作 Android 手机,也不会修改本地结果。",
self._page.window(),
)
dialog.yesButton.setText(f"上报 {len(stable_ids)} 条")
dialog.cancelButton.setText("暂不上报")
dialog.cancelButton.setFocus()
if not dialog.exec():
return
self._start_resubmit_worker(stable_ids)
def _start_resubmit_worker(self, task_ids: tuple[str, ...]) -> None:
"""启动只处理指定结果 Outbox 的工作线程。"""
assert self._claim_gateway is not None
self._resubmit_busy = True
self._page.set_resubmit_running(True)
self._page.set_engine_status(
f"正在准备重新上报 {len(task_ids)} 条任务结果…"
)
thread = QThread(self)
worker = ResultResubmitWorker(
self._claim_gateway,
self._repository,
task_ids,
)
worker.moveToThread(thread)
thread.started.connect(worker.run)
worker.progress.connect(self._on_resubmit_progress)
worker.finished.connect(self._on_resubmit_finished)
worker.completed.connect(thread.quit)
worker.completed.connect(worker.deleteLater)
thread.finished.connect(thread.deleteLater)
thread.finished.connect(self._on_resubmit_thread_finished)
self._resubmit_thread = thread
self._resubmit_worker = worker
thread.start()
@pyqtSlot(int, int, str)
def _on_resubmit_progress(self, current: int, total: int, task_id: str) -> None:
if not self._closing:
self._page.set_engine_status(
f"正在重新上报 {current}/{total}:{task_id}"
)
@pyqtSlot(object, object, object)
def _on_resubmit_finished(self, succeeded, failed, skipped) -> None:
if self._closing:
return
succeeded_ids = tuple(succeeded)
self._page.taskModel.uncheck_tasks(succeeded_ids)
message = (
f"重新上报完成:成功 {len(succeeded_ids)} 条,"
f"失败 {len(failed)} 条,跳过 {len(skipped)} 条"
)
self._page.set_engine_status(message)
if failed or skipped:
self._show_claim_error("重新上报未全部完成", message)
self._reload()
@pyqtSlot()
def _on_resubmit_thread_finished(self) -> None:
self._resubmit_worker = None
self._resubmit_thread = None
self._resubmit_busy = False
if not self._closing:
self._page.set_resubmit_running(False)
def _start_rerun_worker(self, task_id: str) -> None:
"""启动只处理指定任务的工作线程。"""
@@ -557,6 +732,12 @@ class PDDTaskPageEvent(QObject):
if self._closing:
return
if self._resubmit_busy:
self._show_rerun_warning(
"暂时不能获取任务",
"任务结果正在重新上报,请等待当前操作结束。",
)
return
if self._auto_fetch_running:
self._request_stop_auto_fetch()
return
@@ -1045,6 +1226,19 @@ class PDDTaskPageEvent(QObject):
# 避免窗口销毁时出现 "QThread destroyed while running"。
thread.wait(60_000)
resubmit_worker = self._resubmit_worker
resubmit_thread = self._resubmit_thread
if resubmit_worker is not None:
try:
resubmit_worker.cancel()
resubmit_worker.progress.disconnect(self._on_resubmit_progress)
resubmit_worker.finished.disconnect(self._on_resubmit_finished)
except (TypeError, RuntimeError):
pass
if resubmit_thread is not None and resubmit_thread.isRunning():
resubmit_thread.quit()
resubmit_thread.wait(60_000)
def summary_to_row(summary: TaskSummary) -> TaskRow:
"""把领域摘要转换成只供表格显示的轻量行。"""
+19
View File
@@ -998,6 +998,25 @@ class TaskRepository:
connection.close()
return self._to_outbox(row) if row is not None else None
def latest_result_outbox(
self, remote_task_id: str
) -> Optional[OutboxEventRecord]:
"""读取任务最新的成功结果事件,不返回失败上报事件。"""
connection = open_database(self._db_path)
try:
row = connection.execute(
"SELECT o.* FROM outbox_events o"
" JOIN pdd_tasks t ON t.id = o.task_id"
" WHERE t.remote_task_id = ?"
" AND o.event_type IN ('collect_result', 'purchase_result')"
" ORDER BY o.id DESC LIMIT 1",
(remote_task_id,),
).fetchone()
finally:
connection.close()
return self._to_outbox(row) if row is not None else None
def get_outbox_event(self, event_id: int) -> Optional[OutboxEventRecord]:
connection = open_database(self._db_path)
try:
+85
View File
@@ -0,0 +1,85 @@
"""PDD 任务表勾选模型的离屏测试。"""
import os
import unittest
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
from PyQt5.QtCore import Qt
from PyQt5.QtWidgets import QApplication
from src.pdd_ui import CHECK_COLUMN, PDDTaskPage, TaskRow, TaskTableModel
class TaskTableModelCheckTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.app = QApplication.instance() or QApplication([])
def test_check_uses_stable_task_id_and_survives_reorder(self):
model = TaskTableModel()
model.set_tasks(
[TaskRow("TASK-1", "采集"), TaskRow("TASK-2", "采集")]
)
self.assertTrue(
model.setData(
model.index(1, CHECK_COLUMN), Qt.Checked, Qt.CheckStateRole
)
)
model.set_tasks(
[TaskRow("TASK-2", "采集"), TaskRow("TASK-1", "采集")]
)
self.assertEqual(model.checked_task_ids(), ("TASK-2",))
self.assertEqual(
model.data(model.index(0, CHECK_COLUMN), Qt.CheckStateRole),
Qt.Checked,
)
def test_placeholder_is_not_checkable_and_removed_task_is_pruned(self):
model = TaskTableModel()
self.assertEqual(model.flags(model.index(0, CHECK_COLUMN)), Qt.NoItemFlags)
model.set_tasks([TaskRow("TASK-1", "采集")])
model.setData(
model.index(0, CHECK_COLUMN), Qt.Checked, Qt.CheckStateRole
)
model.set_tasks([])
self.assertEqual(model.checked_task_ids(), ())
def test_search_clears_checks_and_refresh_is_next_to_search(self):
page = PDDTaskPage()
page.set_tasks([TaskRow("TASK-1", "采集")])
page.taskModel.setData(
page.taskModel.index(0, CHECK_COLUMN), Qt.Checked, Qt.CheckStateRole
)
page.resize(1300, 650)
self.app.processEvents()
layout = page.commandCard.layout()
search_position = layout.getItemPosition(layout.indexOf(page.searchButton))
refresh_position = layout.getItemPosition(layout.indexOf(page.refreshButton))
rerun_position = layout.getItemPosition(layout.indexOf(page.rerunButton))
resubmit_position = layout.getItemPosition(layout.indexOf(page.resubmitButton))
self.assertEqual(refresh_position[1], search_position[1] + 1)
self.assertEqual(resubmit_position[1], rerun_position[1] + 1)
self.assertEqual(page.checkedCountLabel.text(), "已选 1 条")
page.resize(1000, 650)
self.app.processEvents()
compact_search = layout.getItemPosition(layout.indexOf(page.searchButton))
compact_refresh = layout.getItemPosition(layout.indexOf(page.refreshButton))
self.assertEqual(compact_search[0], 1)
self.assertEqual(compact_refresh[1], compact_search[1] + 1)
page.searchButton.click()
self.assertEqual(page.taskModel.checked_task_ids(), ())
self.assertEqual(page.checkedCountLabel.text(), "已选 0 条")
page.deleteLater()
if __name__ == "__main__":
unittest.main()
+169 -6
View File
@@ -10,6 +10,7 @@ from unittest.mock import patch
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
from PyQt5.QtCore import Qt
from PyQt5.QtWidgets import QApplication
from qfluentwidgets import InfoBarPosition
@@ -127,6 +128,25 @@ class RetryableClaimGateway(RecordingClaimGateway):
raise AdminGatewayError("ADMIN_UNAVAILABLE", "Admin 暂时不可用", True)
class RecordingResubmitGateway(RecordingClaimGateway):
"""记录结果重新上报,并可按任务编号模拟失败。"""
def __init__(self, failures=None, delay=0.0):
super().__init__()
self.failures = dict(failures or {})
self.delay = delay
self.submit_calls = []
def submit_result(self, task_id, idempotency_key, result):
self.submit_calls.append((task_id, idempotency_key, result))
if self.delay:
time.sleep(self.delay)
error = self.failures.get(task_id)
if error is not None:
raise error
return SubmissionReceipt(True, f"RESULT-{task_id}", "2026-08-10T08:00:00Z")
class FakeCollectResult:
def to_pdd_data(self):
return {
@@ -249,7 +269,7 @@ class PDDTaskPageEventTest(unittest.TestCase):
"src.pdd_ui_event.AndroidDeviceService.require_connected",
autospec=True,
)
self.device_checker_patch.start()
self.device_checker = self.device_checker_patch.start()
def tearDown(self):
self.device_checker_patch.stop()
@@ -325,24 +345,167 @@ class PDDTaskPageEventTest(unittest.TestCase):
self.assertEqual(row.price_cents, 3990)
page.deleteLater()
def test_rerun_button_follows_current_row_selection(self):
def test_rerun_button_requires_exactly_one_checked_task(self):
self._add_task(1)
self._add_task(2)
page = PDDTaskPage()
events = PDDTaskPageEvent(page, self.repository)
events.load_initial_tasks()
page.taskTable.clearSelection()
self.app.processEvents()
self.assertFalse(page.rerunButton.isEnabled())
page.taskTable.selectRow(0)
page.taskModel.setData(
page.taskModel.index(0, 0), Qt.Checked, Qt.CheckStateRole
)
self.app.processEvents()
self.assertTrue(page.rerunButton.isEnabled())
page.taskTable.clearSelection()
page.taskModel.setData(
page.taskModel.index(1, 0), Qt.Checked, Qt.CheckStateRole
)
self.app.processEvents()
self.assertFalse(page.rerunButton.isEnabled())
events.shutdown()
page.deleteLater()
def test_batch_resubmit_reuses_outbox_and_keeps_skipped_checked(self):
for number in range(1, 4):
self._add_task(number)
expected = {}
for task_id in ("PDD-001", "PDD-002"):
started = self.repository.start_collect_run(task_id, "USB-001")
event = self.repository.save_collect_result(
task_id, started.attempt_id, FakeCollectResult().to_pdd_data()
)
self.repository.mark_outbox_sent(event.id)
expected[task_id] = (event.idempotency_key, event.payload_json)
page = PDDTaskPage()
gateway = RecordingResubmitGateway()
events = PDDTaskPageEvent(
page,
self.repository,
claim_gateway=gateway,
settings_repository=self._saved_settings(),
)
events.load_initial_tasks()
for row in range(3):
page.taskModel.setData(
page.taskModel.index(row, 0), Qt.Checked, Qt.CheckStateRole
)
with patch("src.pdd_ui_event.MessageBox") as message_box:
message_box.return_value.exec.return_value = True
page.resubmitButton.click()
self.assertTrue(
wait_until(self.app, lambda: not events._resubmit_busy),
"重新上报线程没有按时结束",
)
calls = {
task_id: (key, payload)
for task_id, key, payload in gateway.submit_calls
}
self.assertEqual(calls, expected)
self.assertEqual(page.taskModel.checked_task_ids(), ("PDD-003",))
self.assertIn("成功 2 条,失败 0 条,跳过 1 条", page.statusLabel.text())
self.device_checker.assert_not_called()
events.shutdown()
page.deleteLater()
def test_batch_resubmit_records_retryable_and_permanent_failures(self):
for number in range(1, 3):
self._add_task(number)
event_ids = {}
for task_id in ("PDD-001", "PDD-002"):
started = self.repository.start_collect_run(task_id, "USB-001")
event = self.repository.save_collect_result(
task_id, started.attempt_id, FakeCollectResult().to_pdd_data()
)
self.repository.mark_outbox_sent(event.id)
event_ids[task_id] = event.id
gateway = RecordingResubmitGateway(
{
"PDD-001": AdminGatewayError("TEMP", "暂时不可用", True),
"PDD-002": AdminGatewayError("REJECTED", "结果被拒绝", False),
}
)
page = PDDTaskPage()
events = PDDTaskPageEvent(page, self.repository, claim_gateway=gateway)
events.load_initial_tasks()
for row in range(2):
page.taskModel.setData(
page.taskModel.index(row, 0), Qt.Checked, Qt.CheckStateRole
)
with patch("src.pdd_ui_event.MessageBox") as message_box:
message_box.return_value.exec.return_value = True
page.resubmitButton.click()
self.assertTrue(wait_until(self.app, lambda: not events._resubmit_busy))
self.assertEqual(
self.repository.get_outbox_event(event_ids["PDD-001"]).status.value,
"pending",
)
self.assertEqual(
self.repository.get_outbox_event(event_ids["PDD-002"]).status.value,
"failed",
)
self.assertEqual(
set(page.taskModel.checked_task_ids()), {"PDD-001", "PDD-002"}
)
self.assertIn("失败 2 条", page.statusLabel.text())
events.shutdown()
page.deleteLater()
def test_resubmit_confirmation_cancel_does_not_submit(self):
self._add_task(1)
page = PDDTaskPage()
gateway = RecordingResubmitGateway()
events = PDDTaskPageEvent(page, self.repository, claim_gateway=gateway)
events.load_initial_tasks()
page.taskModel.setData(
page.taskModel.index(0, 0), Qt.Checked, Qt.CheckStateRole
)
with patch("src.pdd_ui_event.MessageBox") as message_box:
message_box.return_value.exec.return_value = False
page.resubmitButton.click()
self.assertEqual(gateway.submit_calls, [])
self.assertFalse(events._resubmit_busy)
events.shutdown()
page.deleteLater()
def test_resubmit_rejects_duplicate_start_and_shutdown_waits_worker(self):
self._add_task(1)
started = self.repository.start_collect_run("PDD-001", "USB-001")
event = self.repository.save_collect_result(
"PDD-001", started.attempt_id, FakeCollectResult().to_pdd_data()
)
self.repository.mark_outbox_sent(event.id)
page = PDDTaskPage()
gateway = RecordingResubmitGateway(delay=0.1)
events = PDDTaskPageEvent(page, self.repository, claim_gateway=gateway)
events.load_initial_tasks()
page.taskModel.setData(
page.taskModel.index(0, 0), Qt.Checked, Qt.CheckStateRole
)
with patch("src.pdd_ui_event.MessageBox") as message_box:
message_box.return_value.exec.return_value = True
page.resubmitButton.click()
events.request_resubmit(("PDD-001",))
self.assertTrue(
wait_until(self.app, lambda: bool(gateway.submit_calls))
)
events.shutdown()
self.assertEqual(len(gateway.submit_calls), 1)
self.assertFalse(
events._resubmit_thread and events._resubmit_thread.isRunning()
)
page.deleteLater()
def test_confirmed_rerun_executes_selected_terminal_collect_task(self):
self._add_task(1)
first = self.repository.start_collect_run("PDD-001", "USB-001")
+27
View File
@@ -209,6 +209,33 @@ class TaskRepositoryTests(unittest.TestCase):
TaskStatus.SUCCEEDED,
)
def test_latest_result_outbox_excludes_newer_failure_event(self):
self.repository.add_claimed_task(self._task("TASK-REPORT"))
first = self.repository.start_collect_run("TASK-REPORT", "USB-001")
result = self.repository.save_collect_result(
"TASK-REPORT",
first.attempt_id,
{"goods_id": "10001", "title": "测试商品", "skus": []},
)
self.repository.mark_outbox_sent(result.id)
self.repository.prepare_collect_rerun("TASK-REPORT")
second = self.repository.start_collect_run("TASK-REPORT", "USB-001")
failure = self.repository.save_collect_failure(
"TASK-REPORT",
second.attempt_id,
TaskStatus.FAILED,
"TEST_FAILURE",
"测试失败",
False,
)
latest = self.repository.latest_result_outbox("TASK-REPORT")
self.assertEqual(latest.id, result.id)
self.assertEqual(latest.event_type, OutboxEventType.COLLECT_RESULT)
self.assertNotEqual(latest.id, failure.id)
self.assertIsNone(self.repository.latest_result_outbox("MISSING"))
def test_prepare_rerun_preserves_old_result_and_creates_new_attempt(self):
self.repository.add_claimed_task(self._task("TASK-RERUN"))
first = self.repository.start_collect_run("TASK-RERUN", "USB-001")
+15 -6
View File
@@ -9,7 +9,7 @@
## 1. 设计目标
- 让操作人员在一个页面完成任务监控、搜索和异常定位。
- “获取任务”和“重新执行”会产生外部后果;后者当前只允许重新采集,不允许采购。
- “获取任务”“重新执行”和“重新上报”会产生外部后果;重新执行只允许单条采集任务,重新上报只重发既有结果 Outbox。
- 长任务状态始终可找到,不使用连续模态弹窗打断工作。
- 任务表格在数据增长后仍保持响应速度、稳定选择和可访问性。
- 界面只展示任务状态,不在 Qt 主线程执行 Admin 或手机自动化。
@@ -35,7 +35,7 @@
```text
┌─────────────────────────────────────────────────────────────┐
│ PDD 任务 │
│ [开始自动获取] [类型▼] [状态▼] [关键词...] [搜索] [刷新] [重新执行] │
│ [开始自动获取] [类型▼] [状态▼] [关键词...] [搜索] [刷新] [已选0条] [重新执行] [重新上报] │
├─────────────────────────────────────────────────────────────┤
│ 类型 │ 商品标题 │ 颜色 │ 尺码 │ 价格 │ 数量 │ 状态 │ 更新时间 │详情│
│ │
@@ -83,10 +83,11 @@
筛选条件之间采用 AND。点击搜索或在关键词输入框按 Enter 执行本地数据库查询,不请求领取任务。活动筛选必须可见,筛选无结果时保留条件并提供清除入口。
### 4.3 刷新与重新执行
### 4.3 刷新、重新执行与重新上报
- “刷新”只重新读取本地任务列表,不请求 Admin,也不操作手机。
- “重新执行”位于刷新右侧;没有选中任务时禁用。
- “刷新”紧邻“搜索”右侧,只重新读取本地任务列表,不请求 Admin,也不操作手机。
- 页面右侧显示“已选 N 条”,“重新执行”后面是“重新上报”。
- “重新执行”只在恰好勾选一条任务时启用;多选不允许批量操作手机。
- 点击后先校验任务,再显示明确的“重新采集”确认弹窗。弹窗显示任务编号、商品标题,并说明新结果会覆盖 Client 和 Admin 的当前采集数据。“暂不重新采集”是默认聚焦的安全操作,点击该按钮、按 `Escape` 或关闭弹窗都不得启动任务。
- 用户确认后,工作线程必须先检查已保存 Android 设备的实际连接状态,再重置任务和创建执行记录。检查失败时保留原任务状态、采集结果和执行历史。
- 允许重新采集已经结束或处于“等待重试”的采集任务。采购、执行中、结果待提交、仍有未发送 Outbox 或自动获取忙碌时必须阻止,并用中文说明原因。
@@ -98,6 +99,11 @@
- 采购任务始终不能通过“重新执行”入口启动;需要处理时由自动获取的安全恢复流程决定。
- “等待重试”当前没有倒计时。自动获取因可恢复采集错误停止时,底部状态显示
“重试已暂停”,并提示选择任务点击“重新执行”或重新启动获取任务。
- “重新上报”作用于当前已经加载并勾选的任务。执行前显示任务数量,并明确说明不会重新采集、采购或操作手机。
- 重新上报只查找每条任务最新的 `collect_result` 或 `purchase_result` Outbox,使用原来的 `idempotency_key` 和 `payload_json`;不重新组装数据、不创建新 Outbox,也不发送 `task_failure`。
- 重新上报不依赖 Android 设备,在独立工作线程中逐条提交。自动获取、重新执行或另一批上报运行时不得启动。
- 批量完成后显示成功、失败和跳过数量。成功任务取消勾选;失败及没有结果 Outbox 的跳过任务保留勾选,方便继续处理。
- 相同幂等键和相同内容只用于让 Admin 再次确认已接收,不代表创建新结果或覆盖 Admin 数据。
## 5. 任务表格
@@ -109,6 +115,7 @@
| 列 | 对齐 | 显示规则 |
|---|---|---|
| 选择 | 居中 | 行首复选框;占位空行不可勾选,身份使用 `remote_task_id` |
| 任务类型 | 居中 | “采集”或“采购”,颜色只作辅助 |
| 商品标题 | 左对齐、可伸展 | 空值显示“尚未获取标题”,截断时提供完整工具提示 |
| 颜色 | 左对齐 | 无值显示 `—` |
@@ -123,6 +130,8 @@
- 默认排序为 `updated_at DESC, id DESC`。
- 排序、筛选和数据变化后以稳定任务编号恢复当前行。
- 当前行和勾选集相互独立;点击复选框不打开详情,也不启动任何业务操作。
- 勾选集使用 `remote_task_id` 保存。普通刷新保留仍存在的勾选,搜索条件变化清空勾选,避免操作被筛选隐藏的旧任务。
- 初始加载有限批次,滚动时通过 `canFetchMore/fetchMore` 增量加载后续任务。
- 单击非交互区域只设置当前行和选择。
- 双击行、按 Enter 或点击“详情”执行相同的非破坏性详情命令。
@@ -362,7 +371,7 @@ self.show_recoverable_error(
## 10. 键盘与无障碍
- Tab 顺序:自动获取 → 类型 → 状态 → 关键词 → 搜索 → 刷新 → 重新执行 → 表格 → 状态区可操作项。
- Tab 顺序:自动获取 → 类型 → 状态 → 关键词 → 搜索 → 刷新 → 重新执行 → 重新上报 → 表格 → 状态区可操作项。
- `Ctrl+F` 聚焦关键词,Enter 打开当前行详情。不绑定 `F5`——界面上没有需要刷新的远端数据。
- 仅图标按钮必须设置准确的无障碍名称和工具提示。
- 表单具有可见标签,占位符不能替代标签。