"""PDD 任务页的界面结构。 这一页显示**本机已领取的全部任务**（正在做的和早已做完的）， 提供搜索、筛选、详情，以及启动/停止自动获取。 完整规范见 `docs/client/05-ui-specification.md`。 改动本文件前必读 `client/AGENTS.md`。以下几条最容易踩： - 本文件只负责显示和把用户操作转发出去，**不写业务逻辑**， 不发 Admin 请求、不查数据库、不碰手机。这些放 `pdd_ui_event.py`。 - 任务表格必须用**模型/视图**（`QAbstractTableModel` + `TableView`）， 并用 `remote_task_id` 作为任务身份。**不要用表格行号**—— 行号会随排序和筛选变化，拿它当身份必出错。 - 不要为每个单元格创建常驻 `QWidget`（几万行会把内存吃光）。 “详情”那一列用委托绘制。 - 不要把完整的 `pdd_data` 放进表格模型，那是详情数据， 查看详情时再从数据库读。 - 表格数据只增不减（已完成任务永久保留），所以增量加载是必须的， 照抄 `docs/client/05-ui-specification.md` §5.3 的模板。 - 空状态要分情况：从没领过任务、筛选无结果、加载失败，文案不能一样。 - 图标用 `FluentIcon`，**不得用表情符号**。 - 页面 `objectName` 固定为 `pddTaskPage`，不要改。 注意：本页唯一会产生外部后果（真的去操作手机、可能下单）的命令是 “开始自动获取”。其余操作全部只读本地数据库。 """ from dataclasses import dataclass from typing import Iterable, Optional from PyQt5.QtCore import QAbstractTableModel, QEvent, QModelIndex, Qt, pyqtSignal from PyQt5.QtGui import QKeySequence from PyQt5.QtWidgets import ( QAbstractItemView, QGridLayout, QHeaderView, QHBoxLayout, QShortcut, QSizePolicy, QVBoxLayout, QWidget, ) from qfluentwidgets import ( BodyLabel, CaptionLabel, CardWidget, ComboBox, FluentIcon as FIF, PrimaryPushButton, PushButton, SearchLineEdit, SubtitleLabel, TableView, TitleLabel, ) PAGE_SIZE = 50 DETAIL_COLUMN = 8 REMOTE_TASK_ID_ROLE = Qt.UserRole + 1 @dataclass(frozen=True) class TaskRow: """任务表格的一行，只保存列表展示和搜索需要的字段。""" remote_task_id: str task_type: str goods_id: str = "" title: str = "" color: str = "" size: str = "" price_cents: Optional[int] = None price_is_starting: bool = False quantity: Optional[int] = None status: str = "待执行" updated_at: str = "" class TaskTableModel(QAbstractTableModel): """PDD 任务表格模型，只保存已经从数据库读取的轻量行。""" loadMoreRequested = pyqtSignal(int, int) HEADERS = ( "任务类型", "商品标题", "颜色", "尺码", "价格", "数量", "状态", "更新时间", "操作", ) def __init__(self, parent=None): super().__init__(parent) self._rows: list[TaskRow] = [] self._has_more = False self._loading = False self._placeholder_row_count = 1 def rowCount(self, parent=QModelIndex()) -> int: if parent.isValid(): return 0 return len(self._rows) or self._placeholder_row_count def columnCount(self, parent=QModelIndex()) -> int: return 0 if parent.isValid() else len(self.HEADERS) def headerData(self, section, orientation, role=Qt.DisplayRole): if orientation == Qt.Horizontal and role == Qt.DisplayRole: return self.HEADERS[section] return super().headerData(section, orientation, role) def data(self, index, role=Qt.DisplayRole): if not index.isValid() or not 0 <= index.row() < len(self._rows): return None task = self._rows[index.row()] column = index.column() if role == Qt.DisplayRole: return self._display_value(task, column) if role == Qt.TextAlignmentRole: if column in (0, 6, 8): return Qt.AlignCenter if column in (4, 5): return Qt.AlignRight | Qt.AlignVCenter return Qt.AlignLeft | Qt.AlignVCenter if role == Qt.ToolTipRole: if column == 1: return task.title or "尚未获取标题" if column == DETAIL_COLUMN: return f"查看任务 {task.remote_task_id} 的详情" if role == REMOTE_TASK_ID_ROLE: return task.remote_task_id return None def flags(self, index): if not index.isValid() or index.row() >= len(self._rows): return Qt.NoItemFlags return Qt.ItemIsEnabled | Qt.ItemIsSelectable def canFetchMore(self, parent=QModelIndex()) -> bool: """告诉 Qt 数据库中是否还有下一页。""" return not parent.isValid() and self._has_more and not self._loading def fetchMore(self, parent=QModelIndex()) -> None: """请求事件层从数据库读取下一页。""" if parent.isValid() or not self.canFetchMore(parent): return self._loading = True self.loadMoreRequested.emit(len(self._rows), PAGE_SIZE) def sort(self, column, order=Qt.AscendingOrder) -> None: """列表顺序由 Repository 固定为更新时间倒序。""" def set_tasks(self, tasks: Iterable[TaskRow]) -> None: """替换全部任务；保留给简单测试和静态预览使用。""" self.beginResetModel() self._rows = list(tasks) self._has_more = False self._loading = False self.endResetModel() def reset_for_reload(self) -> None: """清空旧结果，准备从数据库重新读取第一页。""" self.beginResetModel() self._rows = [] self._has_more = True self._loading = False self.endResetModel() def append_tasks(self, tasks: Iterable[TaskRow], has_more: bool) -> None: """追加一页任务，并结束本次加载状态。""" page = list(tasks) if not self._rows: self.beginResetModel() self._rows = page self._has_more = has_more self._loading = False self.endResetModel() return if page: start = len(self._rows) self.beginInsertRows(QModelIndex(), start, start + len(page) - 1) self._rows.extend(page) self.endInsertRows() self._has_more = has_more self._loading = False def finish_loading(self) -> None: """查询失败时停止继续请求，等待用户刷新或重新搜索。""" self._has_more = False self._loading = False def row_at(self, row: int) -> Optional[TaskRow]: """根据当前可见行获取任务，越界时返回 None。""" if 0 <= row < len(self._rows): return self._rows[row] return None def find_row(self, remote_task_id: str) -> int: """根据稳定任务编号查找当前已加载行。""" for row, task in enumerate(self._rows): if task.remote_task_id == remote_task_id: return row return -1 def filtered_count(self) -> int: return len(self._rows) def data_row_count(self) -> int: """返回真实任务行数，不包含填满空表格的占位行。""" return len(self._rows) def set_placeholder_row_count(self, count: int) -> None: """设置无数据时显示的禁用空行数量。""" count = max(1, count) if count == self._placeholder_row_count: return if self._rows: self._placeholder_row_count = count return self.beginResetModel() self._placeholder_row_count = count self.endResetModel() @staticmethod def _display_value(task: TaskRow, column: int) -> str: if column == 0: return task.task_type if column == 1: return task.title or "尚未获取标题" if column == 2: return task.color or "—" if column == 3: return task.size or "—" if column == 4: 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 if column == 7: return task.updated_at or "—" return "详情" class PDDTaskPage(QWidget): """展示、筛选本机任务，并把用户操作转发给事件层。""" autoFetchRequested = pyqtSignal() searchRequested = pyqtSignal(dict) refreshRequested = pyqtSignal() detailRequested = pyqtSignal(str) def __init__(self, parent=None): super().__init__(parent) self.setObjectName("pddTaskPage") self.taskModel = TaskTableModel(self) self._build_command_area() self._build_content_area() self._build_status_area() self._build_page_layout() self._connect_ui_events() self._update_empty_state() def _build_command_area(self) -> None: self.autoFetchButton = PrimaryPushButton( FIF.PLAY, "开始自动获取", self ) self.autoFetchButton.setMinimumWidth(132) self.autoFetchButton.setAccessibleName("开始自动获取 PDD 任务") self.taskTypeCombo = ComboBox(self) self.taskTypeCombo.addItems(("全部", "采集", "采购")) self.taskTypeCombo.setMinimumWidth(100) self.taskTypeCombo.setAccessibleName("任务类型") self.statusCombo = ComboBox(self) self.statusCombo.addItems( ("全部", "待执行", "执行中", "已完成", "失败", "需要人工处理") ) self.statusCombo.setMinimumWidth(130) self.statusCombo.setAccessibleName("任务状态") self.keywordInput = SearchLineEdit(self) self.keywordInput.setPlaceholderText("任务编号、商品编号或商品标题") self.keywordInput.setClearButtonEnabled(True) self.keywordInput.setMinimumWidth(190) self.keywordInput.setMaximumWidth(480) self.keywordInput.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Fixed) self.keywordInput.setAccessibleName("任务关键词") self.searchButton = PushButton(FIF.SEARCH, "搜索", self) self.searchButton.setAccessibleName("搜索本地任务") self.refreshButton = PushButton(FIF.SYNC, "刷新", self) self.refreshButton.setAccessibleName("刷新本地任务表格") 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) def _build_content_area(self) -> None: self.taskTable = TableView(self) self.taskTable.setModel(self.taskModel) self.taskTable.setAccessibleName("本机 PDD 任务列表") self.taskTable.setSelectionBehavior(QAbstractItemView.SelectRows) self.taskTable.setSelectionMode(QAbstractItemView.SingleSelection) self.taskTable.setEditTriggers(QAbstractItemView.NoEditTriggers) self.taskTable.setAlternatingRowColors(True) self.taskTable.setWordWrap(False) self.taskTable.setSortingEnabled(False) self.taskTable.verticalHeader().setVisible(False) header = self.taskTable.horizontalHeader() header.setSectionResizeMode(QHeaderView.Interactive) header.setSectionResizeMode(1, 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.viewport().installEventFilter(self) self.emptyStateCard = CardWidget(self) emptyLayout = QHBoxLayout(self.emptyStateCard) emptyLayout.setContentsMargins(16, 10, 16, 10) emptyTextLayout = QVBoxLayout() emptyTextLayout.setSpacing(2) self.emptyTitleLabel = SubtitleLabel("", self.emptyStateCard) self.emptyMessageLabel = BodyLabel("", self.emptyStateCard) self.emptyMessageLabel.setWordWrap(True) emptyTextLayout.addWidget(self.emptyTitleLabel) emptyTextLayout.addWidget(self.emptyMessageLabel) self.clearFiltersButton = PushButton( "清除筛选条件", self.emptyStateCard ) self.clearFiltersButton.setAccessibleName("清除全部任务筛选条件") emptyLayout.addLayout(emptyTextLayout, 1) emptyLayout.addWidget(self.clearFiltersButton) def _build_status_area(self) -> None: self.statusCard = CardWidget(self) statusLayout = QHBoxLayout(self.statusCard) statusLayout.setContentsMargins(16, 10, 16, 10) self.statusLabel = CaptionLabel( "自动获取：已停止 · 当前没有执行中的任务", self.statusCard ) self.statusLabel.setTextInteractionFlags(Qt.TextSelectableByMouse) self.statusLabel.setAccessibleName("PDD 任务运行状态") statusLayout.addWidget(self.statusLabel, 1) def _build_page_layout(self) -> None: layout = QVBoxLayout(self) layout.setContentsMargins(36, 30, 36, 36) layout.setSpacing(16) layout.addWidget(TitleLabel("PDD 任务", self)) layout.addWidget(self.commandCard) layout.addWidget(self.emptyStateCard) layout.addWidget(self.taskTable, 1) layout.addWidget(self.statusCard) def _connect_ui_events(self) -> None: self.autoFetchButton.clicked.connect(self.autoFetchRequested.emit) self.searchButton.clicked.connect(self._apply_filters) self.keywordInput.returnPressed.connect(self._apply_filters) self.refreshButton.clicked.connect(self.refreshRequested.emit) self.clearFiltersButton.clicked.connect(self.clear_filters) self.taskTable.clicked.connect(self._on_table_clicked) self.taskTable.activated.connect(self._on_table_activated) self.findShortcut = QShortcut(QKeySequence.Find, self) self.findShortcut.activated.connect(self.keywordInput.setFocus) def eventFilter(self, watched, event): if watched is self.taskTable.viewport() and event.type() == QEvent.Resize: self._update_placeholder_rows() return super().eventFilter(watched, event) def set_tasks(self, tasks: Iterable[TaskRow]) -> None: """显示任务数据，并尽量恢复原来选中的稳定任务编号。""" current_task_id = self.current_task_id() self.taskModel.set_tasks(tasks) self._update_empty_state() self._restore_current_task(current_task_id) def begin_task_reload(self) -> None: """清空旧结果并显示本地数据库加载状态。""" self._task_id_before_reload = self.current_task_id() self.taskModel.reset_for_reload() self.set_loading() def append_task_page( self, tasks: Iterable[TaskRow], has_more: bool ) -> None: """把事件层读取的一页任务追加到表格。""" self.taskModel.append_tasks(tasks, has_more) self._update_empty_state() task_id = getattr(self, "_task_id_before_reload", "") self._restore_current_task(task_id) self._task_id_before_reload = "" def set_loading(self) -> None: """显示正在读取本地任务的状态。""" self._show_empty_state("正在加载任务", "正在读取本地任务，请稍候。", False) def set_load_error(self, message: str) -> None: """显示本地任务加载失败状态，不清除当前筛选条件。""" self.taskModel.finish_loading() self._show_empty_state( "任务加载失败", message or "无法读取本地任务，请检查日志后重试。", False, ) def set_auto_fetch_state(self, state: str) -> None: """更新自动获取按钮；state 为 stopped/starting/running/stopping。""" states = { "stopped": ("开始自动获取", FIF.PLAY, True), "starting": ("正在启动", FIF.SYNC, False), "running": ("停止自动获取", FIF.PAUSE, True), "stopping": ("正在停止", FIF.PAUSE, False), } text, icon, enabled = states.get(state, states["stopped"]) self.autoFetchButton.setText(text) self.autoFetchButton.setIcon(icon) self.autoFetchButton.setEnabled(enabled) def set_engine_status(self, text: str) -> None: """更新页面底部的任务引擎状态说明。""" self.statusLabel.setText(text or "自动获取：已停止 · 当前没有执行中的任务") def clear_filters(self) -> None: """清除全部搜索条件并重新查询本地任务。""" self.taskTypeCombo.setCurrentIndex(0) self.statusCombo.setCurrentIndex(0) self.keywordInput.clear() self._apply_filters() def current_task_id(self) -> str: """返回当前行的稳定任务编号，没有当前行时返回空字符串。""" index = self.taskTable.currentIndex() task = self.taskModel.row_at(index.row()) if index.isValid() else None return task.remote_task_id if task else "" def _apply_filters(self) -> None: filters = { "task_type": self.taskTypeCombo.currentText(), "status": self.statusCombo.currentText(), "keyword": self.keywordInput.text().strip(), } self.searchRequested.emit(filters) def _on_table_clicked(self, index: QModelIndex) -> None: if index.column() == DETAIL_COLUMN: self._emit_detail(index) def _on_table_activated(self, index: QModelIndex) -> None: self._emit_detail(index) def _emit_detail(self, index: QModelIndex) -> None: task = self.taskModel.row_at(index.row()) if task is not None: self.detailRequested.emit(task.remote_task_id) def _restore_current_task(self, remote_task_id: str) -> None: if not remote_task_id: return row = self.taskModel.find_row(remote_task_id) if row < 0: return index = self.taskModel.index(row, 0) self.taskTable.setCurrentIndex(index) self.taskTable.selectRow(row) self.taskTable.scrollTo(index) def _has_active_filters(self) -> bool: return ( self.taskTypeCombo.currentIndex() != 0 or self.statusCombo.currentIndex() != 0 or bool(self.keywordInput.text().strip()) ) def _update_empty_state(self) -> None: if self.taskModel.data_row_count() > 0: self.emptyStateCard.hide() return if self._has_active_filters(): self._show_empty_state( "没有符合条件的任务", "请调整搜索条件，或者清除筛选后查看全部本地任务。", True, ) return self.emptyStateCard.hide() self._update_placeholder_rows() def _show_empty_state( self, title: str, message: str, show_clear_button: bool ) -> None: self.emptyTitleLabel.setText(title) self.emptyMessageLabel.setText(message) self.clearFiltersButton.setVisible(show_clear_button) self.emptyStateCard.show() self._update_placeholder_rows() def _update_placeholder_rows(self) -> None: """用不可交互的空行填满无数据时的表格可视区域。""" row_height = max(1, self.taskTable.verticalHeader().defaultSectionSize()) viewport_height = max(row_height, self.taskTable.viewport().height()) row_count = max(1, viewport_height // row_height) self.taskModel.set_placeholder_row_count(row_count)