"""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)