441 lines
15 KiB
Python
441 lines
15 KiB
Python
"""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: "已取消",
|
||
}
|
||
|
||
CURRENT_STEP_TEXT = {
|
||
"open_goods": "正在打开商品页面",
|
||
"collecting": "正在采集商品信息",
|
||
"collect_skus": "正在采集颜色、价格和尺码",
|
||
"select_options": "正在选择商品规格",
|
||
"placing_order": "正在提交订单",
|
||
"reconcile_order": "正在核对订单",
|
||
"submit_result": "正在提交任务结果",
|
||
"completed": "已完成",
|
||
"failed": "执行失败",
|
||
"interrupted": "上次执行中断",
|
||
"purchase_prepare": "正在准备采购演练",
|
||
"purchase_open_goods": "正在打开采购商品页",
|
||
"purchase_verify_goods": "正在核对商品",
|
||
"purchase_select_options": "正在选择采购规格",
|
||
"purchase_verify_options": "正在核对采购规格",
|
||
"purchase_set_quantity": "正在设置采购数量",
|
||
"purchase_verify_quantity_price": "正在核对数量和价格",
|
||
"purchase_enter_confirmation": "正在进入提交前确认页",
|
||
"purchase_verify_confirmation": "正在核对提交前确认页",
|
||
"purchase_live_final_check": "正在执行真实下单最终安全核对",
|
||
"purchase_irreversible_step_entered": "已进入不可逆阶段，禁止重新下单",
|
||
"purchase_submit_once": "已单击一次提交订单",
|
||
"purchase_dry_run_stopped": "采购演练已在提交前停止",
|
||
"purchase_recovery_ready": "上次演练中断，已安全等待恢复",
|
||
"reconcile_purchase": "订单已提交，只允许核对未付款订单",
|
||
"purchase_order_matched_pending_report": "已提交待付款，等待向 Admin 上报",
|
||
"reconcile_completed": "已核对到订单，等待人工确认",
|
||
"reconcile_manual_review": "订单结果不确定或存在多个候选，需人工处理",
|
||
}
|
||
|
||
|
||
@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=(
|
||
CURRENT_STEP_TEXT.get(detail.current_step, "未知步骤")
|
||
if detail.current_step
|
||
else "—"
|
||
),
|
||
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)
|