feat: 实现 PDD 任务详情窗口 (#39)

This commit is contained in:
chengma
2026-08-08 09:46:58 +08:00
parent 6c12e058c3
commit cb323d37ba
5 changed files with 654 additions and 0 deletions
+56
View File
@@ -45,6 +45,7 @@ from .task_models import (
TaskType,
)
from .task_repository import TaskRepository
from .task_detail_view import TaskDetailWindow
TASK_TYPE_BY_TEXT = {
@@ -203,6 +204,7 @@ class PDDTaskPageEvent(QObject):
self._claim_busy = False
self._claim_thread: Optional[QThread] = None
self._claim_worker: Optional[ClaimTaskWorker] = None
self._detail_windows: Dict[str, TaskDetailWindow] = {}
self._collect_service_factory = collect_service_factory
try:
@@ -234,6 +236,7 @@ class PDDTaskPageEvent(QObject):
page.searchRequested.connect(self.search_tasks)
page.refreshRequested.connect(self.refresh_tasks)
page.autoFetchRequested.connect(self._request_claim_task)
page.detailRequested.connect(self.show_task_detail)
page.taskModel.loadMoreRequested.connect(self._load_page)
page.autoFetchButton.setText("获取任务")
page.destroyed.connect(self.shutdown)
@@ -387,6 +390,55 @@ class PDDTaskPageEvent(QObject):
rows = [summary_to_row(summary) for summary in summaries]
self._page.append_task_page(rows, has_more=len(rows) == limit)
@pyqtSlot(str)
def show_task_detail(self, task_id: str) -> None:
"""读取本地完整任务,并打开一个非模态详情窗口。"""
if self._closing or not task_id:
return
current = self._detail_windows.get(task_id)
if current is not None:
current.showNormal()
current.raise_()
current.activateWindow()
return
try:
detail = self._repository.get_task(task_id)
except Exception:
InfoBar.error(
title="无法打开任务详情",
content="无法读取本地任务,请检查数据库后重试。",
isClosable=True,
duration=-1,
position=InfoBarPosition.TOP_RIGHT,
parent=self._page,
)
return
if detail is None:
InfoBar.warning(
title="任务不存在",
content=f"本地找不到任务 {task_id},请刷新任务列表。",
isClosable=True,
duration=5000,
position=InfoBarPosition.TOP_RIGHT,
parent=self._page,
)
return
window = TaskDetailWindow(detail, self._page.window())
window.closed.connect(self._on_detail_closed)
self._detail_windows[task_id] = window
window.show()
window.raise_()
window.activateWindow()
@pyqtSlot(str)
def _on_detail_closed(self, task_id: str) -> None:
self._detail_windows.pop(task_id, None)
if not self._closing:
self._page.taskTable.setFocus()
@pyqtSlot()
def shutdown(self) -> None:
"""窗口关闭时停止新的领取,并等待已领取任务完成本地保存。"""
@@ -395,6 +447,10 @@ class PDDTaskPageEvent(QObject):
return
self._closing = True
for window in list(self._detail_windows.values()):
window.close()
self._detail_windows.clear()
worker = self._claim_worker
thread = self._claim_thread
if worker is not None:
+405
View File
@@ -0,0 +1,405 @@
"""PDD 任务详情的数据整理和只读窗口。
本文件只负责把一条 ``TaskDetail`` 整理成容易阅读的内容并显示,
不访问数据库、Admin 或 Android 设备。
"""
from dataclasses import dataclass
from typing import Any, Dict, Iterable, List, Mapping
from PyQt5.QtCore import Qt, pyqtSignal
from PyQt5.QtGui import QCloseEvent, QKeySequence
from PyQt5.QtWidgets import (
QGridLayout,
QHBoxLayout,
QShortcut,
QSizePolicy,
QVBoxLayout,
QWidget,
)
from qfluentwidgets import (
BodyLabel,
CaptionLabel,
CardWidget,
FlowLayout,
PushButton,
ScrollArea,
StrongBodyLabel,
SubtitleLabel,
TitleLabel,
)
from .task_models import TaskDetail, TaskStatus, TaskType
TASK_TYPE_TEXT = {
TaskType.COLLECT: "采集",
TaskType.PURCHASE: "采购",
}
TASK_STATUS_TEXT = {
TaskStatus.CLAIMED: "待执行",
TaskStatus.RUNNING: "执行中",
TaskStatus.RESULT_PENDING: "结果待提交",
TaskStatus.RETRY_WAIT: "等待重试",
TaskStatus.MANUAL_REVIEW: "需要人工处理",
TaskStatus.SUCCEEDED: "已完成",
TaskStatus.FAILED: "失败",
TaskStatus.CANCELLED: "已取消",
}
@dataclass(frozen=True)
class ColorPriceView:
"""一个颜色及其采集价格。"""
text: str
price_text: str
available: bool = True
price_inconsistent: bool = False
@dataclass(frozen=True)
class TaskDetailViewData:
"""详情窗口直接使用的展示数据。"""
task_id: str
task_type: str
status: str
title: str
goods_id: str
goods_url: str
current_step: str
target_spec: str
quantity: str
updated_at: str
shop_name: str
sales: str
reviews: str
captured_at: str
last_error: str
colors: List[ColorPriceView]
sizes: List[str]
def _non_empty_text(value: Any) -> str:
return value.strip() if isinstance(value, str) else ""
def _deduplicate(values: Iterable[str]) -> List[str]:
result: List[str] = []
seen = set()
for value in values:
text = _non_empty_text(value)
if text and text not in seen:
seen.add(text)
result.append(text)
return result
def _dimension_values(
pdd_data: Mapping[str, Any], key: str
) -> List[Dict[str, Any]]:
for dimension in pdd_data.get("dimensions") or []:
if not isinstance(dimension, Mapping) or dimension.get("key") != key:
continue
result = []
for value in dimension.get("values") or []:
if isinstance(value, Mapping) and _non_empty_text(value.get("text")):
result.append(dict(value))
return result
return []
def _sku_option(sku: Mapping[str, Any], key: str) -> str:
options = sku.get("options")
if not isinstance(options, Mapping):
return ""
return _non_empty_text(options.get(key))
def _format_price(prices: Iterable[int]) -> tuple[str, bool]:
values = sorted(set(prices))
if not values:
return "价格未采集", False
if len(values) == 1:
return f"¥{values[0] / 100:.2f}", False
return f"¥{values[0] / 100:.2f}~¥{values[-1] / 100:.2f}", True
def _format_metric(value: Any) -> str:
if not isinstance(value, Mapping):
return "未采集"
raw = _non_empty_text(value.get("raw"))
if raw:
return raw
number = value.get("value")
if isinstance(number, (int, float)) and not isinstance(number, bool):
return str(number)
return "未采集"
def build_task_detail_view_data(detail: TaskDetail) -> TaskDetailViewData:
"""从完整任务记录生成详情展示数据。
颜色和尺码优先按 ``dimensions`` 的采集顺序显示;旧数据没有维度时,
再从 SKU 中补齐。同一颜色出现多个有效价格时显示价格范围。
"""
pdd_data: Mapping[str, Any] = detail.pdd_data or {}
skus = [
item for item in (pdd_data.get("skus") or [])
if isinstance(item, Mapping)
]
color_values = _dimension_values(pdd_data, "color")
color_names = _deduplicate(
[item.get("text", "") for item in color_values]
+ [_sku_option(sku, "color") for sku in skus]
)
availability = {
_non_empty_text(item.get("text")): item.get("available") is not False
for item in color_values
}
colors: List[ColorPriceView] = []
for color in color_names:
prices = []
for sku in skus:
if _sku_option(sku, "color") != color:
continue
price = sku.get("price_cent")
if isinstance(price, int) and not isinstance(price, bool) and price >= 0:
prices.append(price)
price_text, inconsistent = _format_price(prices)
colors.append(
ColorPriceView(
color,
price_text,
availability.get(color, True),
inconsistent,
)
)
size_values = _dimension_values(pdd_data, "size")
sizes = _deduplicate(
[item.get("text", "") for item in size_values]
+ [_sku_option(sku, "size") for sku in skus]
)
metrics = pdd_data.get("metrics")
if not isinstance(metrics, Mapping):
metrics = {}
target_spec = " / ".join(
value for value in (detail.target_color, detail.target_size) if value
) or "—"
last_error = ""
if detail.last_error_code or detail.last_error_message:
last_error = ":".join(
value
for value in (detail.last_error_code, detail.last_error_message)
if value
)
return TaskDetailViewData(
task_id=detail.remote_task_id,
task_type=TASK_TYPE_TEXT.get(detail.task_type, detail.task_type.value),
status=TASK_STATUS_TEXT.get(detail.status, detail.status.value),
title=_non_empty_text(pdd_data.get("title")) or detail.title or "未采集",
goods_id=_non_empty_text(pdd_data.get("goods_id")) or detail.goods_id or "—",
goods_url=detail.goods_url,
current_step=detail.current_step or "—",
target_spec=target_spec,
quantity=str(detail.quantity) if detail.quantity is not None else "—",
updated_at=detail.updated_at,
shop_name=_non_empty_text(pdd_data.get("shop_name")) or "未采集",
sales=_format_metric(metrics.get("sales")),
reviews=_format_metric(metrics.get("reviews")),
captured_at=_non_empty_text(pdd_data.get("captured_at")) or "未采集",
last_error=last_error,
colors=colors,
sizes=sizes,
)
class _ColorCard(CardWidget):
"""详情中的只读颜色卡片。"""
def __init__(self, color: ColorPriceView, parent=None):
super().__init__(parent)
self.setFocusPolicy(Qt.NoFocus)
self.setCursor(Qt.ArrowCursor)
self.setSizePolicy(QSizePolicy.Minimum, QSizePolicy.Fixed)
layout = QVBoxLayout(self)
layout.setContentsMargins(14, 10, 14, 10)
layout.setSpacing(3)
name = BodyLabel(color.text, self)
price = StrongBodyLabel(color.price_text, self)
layout.addWidget(name)
layout.addWidget(price)
notes = []
if not color.available:
notes.append("不可用")
if color.price_inconsistent:
notes.append("价格不一致")
if notes:
layout.addWidget(CaptionLabel(" · ".join(notes), self))
self.setAccessibleName(f"颜色 {color.text},{color.price_text}")
class _SizeCard(CardWidget):
"""详情中的只读尺码卡片。"""
def __init__(self, size: str, parent=None):
super().__init__(parent)
self.setFocusPolicy(Qt.NoFocus)
self.setCursor(Qt.ArrowCursor)
layout = QHBoxLayout(self)
layout.setContentsMargins(14, 8, 14, 8)
layout.addWidget(BodyLabel(size, self))
self.setAccessibleName(f"尺码 {size}")
class TaskDetailWindow(QWidget):
"""可调整大小、不会阻塞任务列表的详情窗口。"""
closed = pyqtSignal(str)
def __init__(self, detail: TaskDetail, parent=None):
super().__init__(parent, Qt.Window)
self.data = build_task_detail_view_data(detail)
self.colorItems: List[_ColorCard] = []
self.sizeItems: List[_SizeCard] = []
self.setAttribute(Qt.WA_DeleteOnClose)
self.setObjectName("taskDetailWindow")
self.setWindowTitle(f"任务详情 · {self.data.task_id}")
self.setMinimumSize(520, 420)
self.resize(760, 680)
self.setAccessibleName(f"任务 {self.data.task_id} 的详情")
QShortcut(QKeySequence.Cancel, self, activated=self.close)
self._build_ui()
def _build_ui(self) -> None:
root = QVBoxLayout(self)
root.setContentsMargins(0, 0, 0, 16)
root.setSpacing(0)
scroll = ScrollArea(self)
scroll.setWidgetResizable(True)
scroll.setHorizontalScrollBarPolicy(Qt.ScrollBarAlwaysOff)
scroll.enableTransparentBackground()
content = QWidget(scroll)
content_layout = QVBoxLayout(content)
content_layout.setContentsMargins(24, 22, 24, 18)
content_layout.setSpacing(14)
content_layout.addWidget(TitleLabel("任务详情", content))
content_layout.addWidget(
CaptionLabel(
f"{self.data.task_id} · {self.data.task_type} · {self.data.status}",
content,
)
)
content_layout.addWidget(self._build_basic_card(content))
content_layout.addWidget(
self._build_spec_card("颜色", self.data.colors, content)
)
content_layout.addWidget(
self._build_spec_card("尺码", self.data.sizes, content)
)
content_layout.addWidget(self._build_collect_card(content))
if self.data.last_error:
error_card = CardWidget(content)
error_layout = QVBoxLayout(error_card)
error_layout.setContentsMargins(16, 14, 16, 14)
error_layout.addWidget(SubtitleLabel("最近错误", error_card))
error_layout.addWidget(BodyLabel(self.data.last_error, error_card))
content_layout.addWidget(error_card)
content_layout.addStretch(1)
scroll.setWidget(content)
root.addWidget(scroll, 1)
buttons = QHBoxLayout()
buttons.setContentsMargins(24, 12, 24, 0)
buttons.addStretch(1)
close_button = PushButton("关闭", self)
close_button.setAccessibleName("关闭任务详情")
close_button.clicked.connect(self.close)
buttons.addWidget(close_button)
root.addLayout(buttons)
def _build_basic_card(self, parent: QWidget) -> CardWidget:
card = CardWidget(parent)
layout = QGridLayout(card)
layout.setContentsMargins(16, 14, 16, 14)
layout.setHorizontalSpacing(18)
layout.setVerticalSpacing(8)
rows = (
("商品标题", self.data.title),
("商品编号", self.data.goods_id),
("目标规格", self.data.target_spec),
("数量", self.data.quantity),
("当前步骤", self.data.current_step),
("更新时间", self.data.updated_at),
("商品链接", self.data.goods_url),
)
for row, (label, value) in enumerate(rows):
name = CaptionLabel(label, card)
text = BodyLabel(value, card)
text.setWordWrap(True)
layout.addWidget(name, row, 0, Qt.AlignTop)
layout.addWidget(text, row, 1)
layout.setColumnStretch(1, 1)
return card
def _build_spec_card(
self, title: str, values: list, parent: QWidget
) -> CardWidget:
card = CardWidget(parent)
layout = QVBoxLayout(card)
layout.setContentsMargins(16, 14, 16, 14)
layout.setSpacing(10)
layout.addWidget(SubtitleLabel(f"{title}({len(values)})", card))
if not values:
layout.addWidget(BodyLabel(f"暂无{title}数据", card))
return card
flow_host = QWidget(card)
flow = FlowLayout(flow_host, needAni=False)
flow.setContentsMargins(0, 0, 0, 0)
flow.setHorizontalSpacing(8)
flow.setVerticalSpacing(8)
if title == "颜色":
for value in values:
item = _ColorCard(value, flow_host)
self.colorItems.append(item)
flow.addWidget(item)
else:
for value in values:
item = _SizeCard(value, flow_host)
self.sizeItems.append(item)
flow.addWidget(item)
layout.addWidget(flow_host)
return card
def _build_collect_card(self, parent: QWidget) -> CardWidget:
card = CardWidget(parent)
layout = QGridLayout(card)
layout.setContentsMargins(16, 14, 16, 14)
layout.setHorizontalSpacing(18)
layout.setVerticalSpacing(8)
values = (
("店铺", self.data.shop_name),
("已拼数量", self.data.sales),
("评价数量", self.data.reviews),
("采集时间", self.data.captured_at),
)
layout.addWidget(SubtitleLabel("采集信息", card), 0, 0, 1, 2)
for row, (label, value) in enumerate(values, start=1):
layout.addWidget(CaptionLabel(label, card), row, 0)
layout.addWidget(BodyLabel(value, card), row, 1)
layout.setColumnStretch(1, 1)
return card
def closeEvent(self, event: QCloseEvent) -> None:
self.closed.emit(self.data.task_id)
super().closeEvent(event)
+59
View File
@@ -225,6 +225,65 @@ class PDDTaskPageEventTest(unittest.TestCase):
self.assertFalse(page.taskModel.canFetchMore())
page.deleteLater()
def test_detail_signal_opens_one_reusable_non_modal_window(self):
self._add_task(1)
started = self.repository.start_collect_run("PDD-001", "USB-001")
self.repository.save_collect_result(
"PDD-001",
started.attempt_id,
{
"goods_id": "1",
"title": "测试商品",
"dimensions": [
{
"key": "color",
"name": "颜色",
"values": [{"text": "黑色", "available": True}],
},
{
"key": "size",
"name": "尺码",
"values": [{"text": "M", "available": True}],
},
],
"skus": [
{
"options": {"color": "黑色", "size": "M"},
"price_cent": 3990,
}
],
},
)
page = PDDTaskPage()
events = PDDTaskPageEvent(page, self.repository)
page.detailRequested.emit("PDD-001")
self.app.processEvents()
first = events._detail_windows["PDD-001"]
self.assertTrue(first.isVisible())
self.assertFalse(first.isModal())
page.detailRequested.emit("PDD-001")
self.app.processEvents()
self.assertIs(events._detail_windows["PDD-001"], first)
first.close()
self.app.processEvents()
self.assertNotIn("PDD-001", events._detail_windows)
events.shutdown()
page.deleteLater()
def test_missing_detail_does_not_open_empty_window(self):
page = PDDTaskPage()
events = PDDTaskPageEvent(page, self.repository)
page.detailRequested.emit("PDD-NOT-FOUND")
self.app.processEvents()
self.assertEqual(events._detail_windows, {})
events.shutdown()
page.deleteLater()
def test_summary_to_row_does_not_copy_detail_json(self):
summary = TaskSummary(
id=1,
+125
View File
@@ -0,0 +1,125 @@
"""任务详情数据整理和窗口的离屏测试。"""
import os
import unittest
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
from PyQt5.QtWidgets import QApplication
from src.task_detail_view import TaskDetailWindow, build_task_detail_view_data
from src.task_models import TaskDetail, TaskStatus, TaskType
def make_detail(pdd_data=None) -> TaskDetail:
return TaskDetail(
id=1,
remote_task_id="COL-001",
task_type=TaskType.COLLECT,
goods_id="123",
goods_url="https://mobile.yangkeduo.com/goods.html?goods_id=123",
title="本地标题",
target_color=None,
target_size=None,
price_cent=None,
quantity=None,
status=TaskStatus.SUCCEEDED,
current_step="completed",
priority=0,
version=1,
admin_payload={},
pdd_data=pdd_data,
retry_count=0,
last_error_code=None,
last_error_message=None,
received_at="2026-08-08T08:00:00Z",
started_at="2026-08-08T08:01:00Z",
finished_at="2026-08-08T08:02:00Z",
created_at="2026-08-08T08:00:00Z",
updated_at="2026-08-08T08:02:00Z",
)
def collected_data():
return {
"goods_id": "123",
"title": "采集标题",
"shop_name": "测试店铺",
"captured_at": "2026-08-08T08:02:00Z",
"metrics": {
"sales": {"value": 1200, "raw": "已拼1.2千件"},
"reviews": {"value": 68, "raw": "68条评价"},
},
"dimensions": [
{
"key": "color",
"name": "颜色",
"values": [
{"text": "黑色", "available": True},
{"text": "白色", "available": False},
{"text": "黑色", "available": True},
],
},
{
"key": "size",
"name": "尺码",
"values": [
{"text": "M", "available": True},
{"text": "L", "available": True},
{"text": "M", "available": True},
],
},
],
"skus": [
{"options": {"color": "黑色", "size": "M"}, "price_cent": 1000},
{"options": {"color": "黑色", "size": "L"}, "price_cent": 1200},
{"options": {"color": "白色", "size": "M"}, "price_cent": None},
],
}
class TaskDetailViewTest(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.app = QApplication.instance() or QApplication([])
def test_colors_keep_order_deduplicate_and_show_price_state(self):
data = build_task_detail_view_data(make_detail(collected_data()))
self.assertEqual([item.text for item in data.colors], ["黑色", "白色"])
self.assertEqual(data.colors[0].price_text, "¥10.00~¥12.00")
self.assertTrue(data.colors[0].price_inconsistent)
self.assertEqual(data.colors[1].price_text, "价格未采集")
self.assertFalse(data.colors[1].available)
self.assertEqual(data.sizes, ["M", "L"])
def test_old_data_can_restore_dimensions_from_skus(self):
data = collected_data()
data["dimensions"] = []
view_data = build_task_detail_view_data(make_detail(data))
self.assertEqual([item.text for item in view_data.colors], ["黑色", "白色"])
self.assertEqual(view_data.sizes, ["M", "L"])
def test_empty_collection_data_has_clear_empty_values(self):
data = build_task_detail_view_data(make_detail())
self.assertEqual(data.colors, [])
self.assertEqual(data.sizes, [])
self.assertEqual(data.shop_name, "未采集")
self.assertEqual(data.sales, "未采集")
def test_window_builds_color_section_before_size_section(self):
window = TaskDetailWindow(make_detail(collected_data()))
self.assertEqual(len(window.colorItems), 2)
self.assertEqual(len(window.sizeItems), 2)
self.assertIn("COL-001", window.windowTitle())
self.assertTrue(window.isWindow())
window.close()
self.app.processEvents()
if __name__ == "__main__":
unittest.main()
+9
View File
@@ -184,6 +184,15 @@ class TaskTableModel(QAbstractTableModel):
详情是信息查看,不要求用户立即决策,优先使用非模态详情窗口或右侧详情面板,不使用简单消息框承载长 JSON。
当前 Client 使用可调整大小的非模态详情窗口。用户可以点击表格“详情”、双击任务行,或选中任务后按 Enter 打开。相同任务只保留一个详情窗口;再次打开时切回已有窗口。按 Esc 或“关闭”按钮返回任务列表。
采集规格按下面顺序展示:
1. **颜色:**放在上方,每个颜色旁边直接显示采集到的价格;未采集到价格时显示“价格未采集”。同一颜色意外出现多个价格时显示最低价到最高价,并显示“价格不一致”,方便检查采集结果。
2. **尺码:**放在颜色下方,只显示尺码文字。
颜色和尺码按采集顺序显示并去重。不可用颜色除文字外还要明确显示“不可用”,不能只改变颜色。没有规格数据时保留对应分区并显示空状态。
详情至少分为:
1. **基本信息:**任务编号、类型、商品、规格、数量、金额和状态。