Files
cmautobuy/client/src/task_detail_view.py
T

648 lines
23 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""PDD 任务详情的数据整理和只读窗口。
本文件只负责把一条 ``TaskDetail`` 整理成容易阅读的内容并显示,
不访问数据库、Admin 或 Android 设备。
"""
from dataclasses import dataclass
from typing import Any, Dict, Iterable, List, Mapping, Optional
from PyQt5.QtCore import Qt, QTimer, pyqtSignal
from PyQt5.QtGui import QCloseEvent, QKeySequence
from PyQt5.QtWidgets import (
QApplication,
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": "订单结果不确定或存在多个候选,需人工处理",
}
class _CopyableLabelMixin:
"""让只读 Label 支持双击或 Ctrl+C 复制完整字段值。"""
copied = pyqtSignal(str, str)
def _init_copyable(self, field_name: str) -> None:
self._copy_field_name = field_name
self.setTextInteractionFlags(
Qt.TextSelectableByMouse | Qt.TextSelectableByKeyboard
)
self.setFocusPolicy(Qt.StrongFocus)
self.setCursor(Qt.IBeamCursor)
self.setToolTip(f"双击或按 Ctrl+C 复制{field_name}")
self.setAccessibleDescription(
f"可选择文本;双击或按 Ctrl+C 复制完整{field_name}"
)
def copy_full_text(self) -> None:
"""复制完整可见内容,并通知详情窗口显示反馈。"""
value = self.text()
QApplication.clipboard().setText(value)
self.copied.emit(self._copy_field_name, value)
def mouseDoubleClickEvent(self, event) -> None:
if event.button() == Qt.LeftButton:
self.copy_full_text()
event.accept()
return
super().mouseDoubleClickEvent(event)
def keyPressEvent(self, event) -> None:
if event.matches(QKeySequence.Copy):
self.copy_full_text()
event.accept()
return
super().keyPressEvent(event)
class CopyableBodyLabel(_CopyableLabelMixin, BodyLabel):
"""保持 BodyLabel 外观的可复制字段值。"""
def __init__(self, text: str, field_name: str, parent=None):
super().__init__(parent)
self.setText(text)
self._init_copyable(field_name)
class CopyableStrongBodyLabel(_CopyableLabelMixin, StrongBodyLabel):
"""保持强调文字外观的可复制字段值。"""
def __init__(self, text: str, field_name: str, parent=None):
super().__init__(parent)
self.setText(text)
self._init_copyable(field_name)
@dataclass(frozen=True)
class ColorPriceView:
"""一个颜色及其采集价格。"""
text: str
price_text: str
available: bool = True
price_inconsistent: bool = False
@dataclass(frozen=True)
class PurchaseResultView:
"""采购任务的下单与核单结果。"""
mode: str
confirmed_spec: str
quantity: str
unit_price: str
total_price: str
order_no: str
ordered_at: str
match_status: str
payment_status: str
@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]
purchase: Optional[PurchaseResultView] = None
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 _format_cent(value: Any) -> str:
if isinstance(value, int) and not isinstance(value, bool) and value >= 0:
return f"¥{value / 100:.2f}"
return "—"
def _purchase_result_view(
task_type: TaskType, pdd_data: Mapping[str, Any]
) -> Optional[PurchaseResultView]:
"""整理采购结果;非采购任务不显示采购信息卡片。"""
if task_type is not TaskType.PURCHASE:
return None
purchase = pdd_data.get("purchase")
if not isinstance(purchase, Mapping):
purchase = {}
confirmed = purchase.get("confirmed")
if not isinstance(confirmed, Mapping):
confirmed = {}
options = confirmed.get("options")
if not isinstance(options, Mapping):
options = {}
confirmed_spec = " / ".join(
_non_empty_text(value) for value in options.values()
if _non_empty_text(value)
) or "—"
mode_text = {"live": "真实采购", "dry_run": "采购演练"}
match_text = {
"matched": "已匹配",
"not_submitted": "未提交",
"not_found": "未找到",
"ambiguous": "多个候选",
"unknown": "结果不确定",
}
payment_text = {
"unpaid": "待付款",
"paid": "已支付",
"other": "其他状态",
"unknown": "未知",
}
quantity = confirmed.get("quantity")
return PurchaseResultView(
mode=mode_text.get(_non_empty_text(purchase.get("mode")), "—"),
confirmed_spec=confirmed_spec,
quantity=(
str(quantity)
if isinstance(quantity, int) and not isinstance(quantity, bool)
else "—"
),
unit_price=_format_cent(confirmed.get("unit_price_cent")),
total_price=_format_cent(confirmed.get("total_price_cent")),
order_no=_non_empty_text(purchase.get("order_no")) or "—",
ordered_at=(
_non_empty_text(purchase.get("ordered_at_raw"))
or _non_empty_text(purchase.get("ordered_at"))
or "—"
),
match_status=match_text.get(
_non_empty_text(purchase.get("match_status")), "—"
),
payment_status=payment_text.get(
_non_empty_text(purchase.get("payment_status")), "—"
),
)
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
)
status_text = TASK_STATUS_TEXT.get(detail.status, detail.status.value)
if detail.status in {TaskStatus.RETRY_WAIT, TaskStatus.FAILED}:
status_text = "采集失败" if detail.task_type is TaskType.COLLECT else "采购失败"
return TaskDetailViewData(
task_id=detail.remote_task_id,
task_type=TASK_TYPE_TEXT.get(detail.task_type, detail.task_type.value),
status=status_text,
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,
purchase=_purchase_result_view(detail.task_type, pdd_data),
)
class _ColorCard(CardWidget):
"""详情中的只读颜色卡片。"""
def __init__(self, color: ColorPriceView, on_copied, 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 = CopyableBodyLabel(color.text, "颜色", self)
price = CopyableStrongBodyLabel(color.price_text, "颜色价格", self)
name.copied.connect(on_copied)
price.copied.connect(on_copied)
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, on_copied, parent=None):
super().__init__(parent)
self.setFocusPolicy(Qt.NoFocus)
self.setCursor(Qt.ArrowCursor)
layout = QHBoxLayout(self)
layout.setContentsMargins(14, 8, 14, 8)
value = CopyableBodyLabel(size, "尺码", self)
value.copied.connect(on_copied)
layout.addWidget(value)
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._copy_feedback_timer = QTimer(self)
self._copy_feedback_timer.setSingleShot(True)
self._copy_feedback_timer.setInterval(2_000)
self._copy_feedback_timer.timeout.connect(self._reset_copy_feedback)
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))
if self.data.purchase is not None:
content_layout.addWidget(self._build_purchase_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 = self._copyable_label(
self.data.last_error, "最近错误", error_card
)
error.setWordWrap(True)
error_layout.addWidget(error)
content_layout.addWidget(error_card)
content_layout.addStretch(1)
scroll.setWidget(content)
root.addWidget(scroll, 1)
buttons = QHBoxLayout()
buttons.setContentsMargins(24, 12, 24, 0)
self.copyStatusLabel = CaptionLabel("双击字段值可复制", self)
self.copyStatusLabel.setAccessibleName("任务详情复制状态")
buttons.addWidget(self.copyStatusLabel)
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 = self._copyable_label(value, label, 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, self._on_field_copied, flow_host)
self.colorItems.append(item)
flow.addWidget(item)
else:
for value in values:
item = _SizeCard(value, self._on_field_copied, 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(self._copyable_label(value, label, card), row, 1)
layout.setColumnStretch(1, 1)
return card
def _build_purchase_card(self, parent: QWidget) -> CardWidget:
"""显示采购任务已经保存在本地的下单和核单结果。"""
purchase = self.data.purchase
assert purchase is not None
card = CardWidget(parent)
card.setObjectName("purchaseResultCard")
layout = QGridLayout(card)
layout.setContentsMargins(16, 14, 16, 14)
layout.setHorizontalSpacing(18)
layout.setVerticalSpacing(8)
values = (
("采购模式", purchase.mode),
("确认规格", purchase.confirmed_spec),
("确认数量", purchase.quantity),
("商品单价", purchase.unit_price),
("订单总价", purchase.total_price),
("订单号", purchase.order_no),
("下单时间", purchase.ordered_at),
("订单匹配", purchase.match_status),
("付款状态", purchase.payment_status),
)
layout.addWidget(SubtitleLabel("采购结果", card), 0, 0, 1, 2)
for row, (label, value) in enumerate(values, start=1):
layout.addWidget(CaptionLabel(label, card), row, 0)
text = self._copyable_label(value, label, card)
text.setWordWrap(True)
layout.addWidget(text, row, 1)
layout.setColumnStretch(1, 1)
return card
def _copyable_label(
self, value: str, field_name: str, parent: QWidget
) -> CopyableBodyLabel:
label = CopyableBodyLabel(value, field_name, parent)
label.copied.connect(self._on_field_copied)
return label
def _on_field_copied(self, field_name: str, _value: str) -> None:
self.copyStatusLabel.setText(f"已复制:{field_name}")
self._copy_feedback_timer.start()
def _reset_copy_feedback(self) -> None:
self.copyStatusLabel.setText("双击字段值可复制")
def closeEvent(self, event: QCloseEvent) -> None:
self.closed.emit(self.data.task_id)
super().closeEvent(event)