"""Tab 2: AI generation UI.""" from __future__ import annotations from PySide6.QtCore import QEvent, QModelIndex, QRectF, QSize, Signal from PySide6.QtGui import QColor, QPainter from ... import product_status from .. import file_manager from ..models import GenerateTaskTableModel from ..widgets import * from ..workers import GenerateWorker as _RealGenerateWorker def GenerateWorker(*args, **kwargs): return _call_package_attr("GenerateWorker", _RealGenerateWorker, *args, **kwargs) class _GenerateTaskTableView(QTableView): """Task table that exposes only left-button double-clicks for previews.""" leftDoubleClicked = Signal(QModelIndex) def mouseDoubleClickEvent(self, event): super().mouseDoubleClickEvent(event) if event.button() != Qt.LeftButton: return index = self.indexAt(event.position().toPoint()) if index.isValid(): self.leftDoubleClicked.emit(index) class _CoverThumbnailLabel(QLabel): def __init__(self, image_path, open_callback, parent=None): super().__init__(parent) self.image_path = image_path self.open_callback = open_callback self.setCursor(Qt.PointingHandCursor) def mouseDoubleClickEvent(self, event): if self.open_callback is not None: self.open_callback(self.image_path) event.accept() return super().mouseDoubleClickEvent(event) class _CoverCandidateCard(QWidget): EXTRA_WIDTH = 16 def __init__(self, thumbnail, radio, meta, parent=None): super().__init__(parent) self.thumbnail = thumbnail self.radio = radio self.meta = meta self.card_layout = QVBoxLayout(self) self.card_layout.setContentsMargins(0, 0, 0, 0) self.card_layout.setSpacing(6) self.card_layout.addWidget(self.thumbnail) self.card_layout.addWidget(self.radio) self.card_layout.addWidget(self.meta) @staticmethod def _widget_height(widget, width): height = widget.heightForWidth(width) if widget.hasHeightForWidth() else -1 if height < 0: height = widget.sizeHint().height() return max(height, widget.minimumHeight(), widget.minimumSizeHint().height()) def set_thumbnail_size(self, size): size = max(1, int(size)) self.setFixedWidth(size + self.EXTRA_WIDTH) self.thumbnail.setFixedSize(size, size) self.sync_geometry() def sync_geometry(self): margins = self.card_layout.contentsMargins() content_width = max(1, self.width() - margins.left() - margins.right()) widget_heights = [ self.thumbnail.height(), self._widget_height(self.radio, content_width), self._widget_height(self.meta, content_width), ] required_height = ( margins.top() + margins.bottom() + sum(widget_heights) + max(0, len(widget_heights) - 1) * max(0, self.card_layout.spacing()) ) self.card_layout.invalidate() self.setFixedHeight(required_height) self.card_layout.activate() self.updateGeometry() return required_height def control_height(self): return max(0, self.height() - self.thumbnail.height()) class OriginalImageDialog(QDialog): def __init__(self, image_path, parent=None): super().__init__(parent) self.image_path = image_path self.image = QImage(str(image_path)) self.fit_to_window = True self._display_size = QSize() self._resize_timer = QTimer(self) self._resize_timer.setSingleShot(True) self._resize_timer.setInterval(50) self._resize_timer.timeout.connect(self._update_image_display) self.setWindowTitle(os.path.basename(str(image_path or "")) or "原图") layout = QVBoxLayout(self) self.scroll = QScrollArea() self.scroll.setObjectName("originalImageScrollArea") self.scroll.setWidgetResizable(False) self.image_label = QLabel() self.image_label.setObjectName("originalImageLabel") self.image_label.setAlignment(Qt.AlignCenter) if self.image.isNull(): self.image_label.setText("图片读取失败") else: self.setWindowTitle( f"{os.path.basename(str(image_path))} · {self.image.width()}x{self.image.height()}" ) self.scroll.setWidget(self.image_label) layout.addWidget(self.scroll, 1) button_layout = QHBoxLayout() button_layout.addStretch(1) self.display_mode_button = QPushButton("原始尺寸") self.display_mode_button.setObjectName("originalImageDisplayModeButton") self.display_mode_button.setToolTip("按图片原始像素查看") self.display_mode_button.setEnabled(not self.image.isNull()) self.display_mode_button.clicked.connect(self.toggle_display_mode) button_layout.addWidget(self.display_mode_button) close_button = QPushButton("关闭") close_button.clicked.connect(self.reject) button_layout.addWidget(close_button) layout.addLayout(button_layout) self._fit_to_screen(self.image) self._update_image_display() QTimer.singleShot(0, self._update_image_display) def resizeEvent(self, event): super().resizeEvent(event) self._resize_timer.start() def toggle_display_mode(self, checked=False): self.fit_to_window = not self.fit_to_window if self.fit_to_window: self.display_mode_button.setText("原始尺寸") self.display_mode_button.setToolTip("按图片原始像素查看") else: self.display_mode_button.setText("适应窗口") self.display_mode_button.setToolTip("缩放图片以完整显示在窗口内") self._display_size = QSize() self._update_image_display() QTimer.singleShot(0, self._update_image_display) def _update_image_display(self): viewport_size = self.scroll.viewport().size() viewport_width = max(1, viewport_size.width()) viewport_height = max(1, viewport_size.height()) if self.image.isNull(): self.image_label.clear() self.image_label.setText("图片读取失败") self.image_label.resize(viewport_width, viewport_height) return if self.fit_to_window: target_size = QSize(viewport_width, viewport_height) pixmap = QPixmap.fromImage( self.image.scaled( target_size, Qt.KeepAspectRatio, Qt.SmoothTransformation, ) ) label_size = target_size else: pixmap = QPixmap.fromImage(self.image) label_size = pixmap.size() if self._display_size == label_size and self.image_label.pixmap() is not None: return self.image_label.setPixmap(pixmap) self.image_label.resize(label_size) self._display_size = QSize(label_size) def _fit_to_screen(self, image): available = _available_geometry() if available is None: self.resize(720, 520) return max_width = max(360, int(available.width() * 0.9)) max_height = max(300, int(available.height() * 0.9)) image_width = image.width() if not image.isNull() else 640 image_height = image.height() if not image.isNull() else 480 self.resize(min(max_width, image_width + 48), min(max_height, image_height + 92)) _center_dialog(self, available) class _SegmentedProgressBar(QProgressBar): def __init__(self, parent=None): super().__init__(parent) self._segment_success = 0 self._segment_failed = 0 self._segment_total = 0 self.setTextVisible(False) self.setRange(0, 1) self.setValue(0) self.setMinimumHeight(12) def set_segments(self, success, failed, total): total = max(0, int(total or 0)) success = min(max(0, int(success or 0)), total) failed = min(max(0, int(failed or 0)), max(0, total - success)) self._segment_success = success self._segment_failed = failed self._segment_total = total maximum = max(1, total) self.setRange(0, maximum) self.setValue(min(maximum, success + failed)) self.update() def segments(self): return self._segment_success, self._segment_failed, self._segment_total def paintEvent(self, event): painter = QPainter(self) painter.setRenderHint(QPainter.Antialiasing) rect = QRectF(self.rect()).adjusted(0.5, 0.5, -0.5, -0.5) painter.setPen(Qt.NoPen) painter.setBrush(QColor("#f6f8fa")) painter.drawRoundedRect(rect, 3, 3) total = max(0, self._segment_total) if total: inner = rect.adjusted(1, 1, -1, -1) unit_width = inner.width() / total if total else 0 success_width = unit_width * self._segment_success failed_width = unit_width * self._segment_failed if success_width > 0: painter.setBrush(QColor(COLOR_SUCCESS)) painter.drawRect(QRectF(inner.left(), inner.top(), success_width, inner.height())) if failed_width > 0: painter.setBrush(QColor(COLOR_DANGER)) painter.drawRect( QRectF( inner.left() + success_width, inner.top(), failed_width, inner.height(), ) ) painter.setBrush(Qt.NoBrush) painter.setPen(QColor("#d0d7de")) painter.drawRoundedRect(rect, 3, 3) class CoverGalleryDialog(QDialog): THUMBNAIL_SIZE = 180 MAX_THUMBNAIL_SIZE = 360 CANDIDATE_CONTROL_FALLBACK_HEIGHT = 72 OLD_THUMBNAIL_SIZE = 300 def __init__( self, task, image_root, db_path, account=None, parent=None, config=None, bulk_running_checker=None, gallery_updated_callback=None, visible_tasks=None, task_index=None, account_by_alias=None, ): super().__init__(parent) self.image_root = image_root self.db_path = db_path self.visible_tasks = list(visible_tasks or [task]) self.task_index = self._initial_task_index(task, task_index) self.task = self.visible_tasks[self.task_index] if self.visible_tasks else task self.account_by_alias = dict(account_by_alias or {}) self.fallback_account = account self.account = self._account_for_task(self.task) or account self.config = config or {} self.bulk_running_checker = bulk_running_checker self.gallery_updated_callback = gallery_updated_callback self.current_path = _normalize_file_path(getattr(task, "new_cover_path", None)) self.selected_path = None self.changed = False self.candidate_thumbnail_size = self.THUMBNAIL_SIZE self.candidate_images = {} self.candidate_thumbnails = {} self.candidate_items = {} self._candidate_resize_timer = QTimer(self) self._candidate_resize_timer.setSingleShot(True) self._candidate_resize_timer.setInterval(60) self._candidate_resize_timer.timeout.connect(self._update_candidate_thumbnail_layout) self.candidate_buttons = {} self.button_group = QButtonGroup(self) self.button_group.setExclusive(True) self.candidates = image_paths.list_task_cover_candidates(image_root, self.task, account=self.account) self.setWindowTitle(f"封面画廊：{getattr(task, 'item_id', '')}") layout = QVBoxLayout(self) body_layout = QHBoxLayout() body_layout.addWidget(self._old_cover_panel(), 0) body_layout.addWidget(self._candidate_panel(), 1) layout.addLayout(body_layout, 1) button_layout = QHBoxLayout() self.previous_button = QPushButton("上一条") self.next_button = QPushButton("下一条") self.previous_button.clicked.connect(lambda checked=False: self.switch_task(-1)) self.next_button.clicked.connect(lambda checked=False: self.switch_task(1)) button_layout.addWidget(self.previous_button) button_layout.addWidget(self.next_button) self.reset_cover_button = QPushButton("重置图片") self.reset_cover_button.clicked.connect(self.reset_cover_image) button_layout.addWidget(self.reset_cover_button) button_layout.addStretch(1) self.save_hint_label = QLabel("") self.save_hint_label.setObjectName("coverGallerySaveHintLabel") button_layout.addWidget(self.save_hint_label) self.save_button = QPushButton("保存") self.cancel_button = QPushButton("关闭") self.save_button.clicked.connect(self.save_selection) self.cancel_button.clicked.connect(self.reject) button_layout.addWidget(self.save_button) button_layout.addWidget(self.cancel_button) layout.addLayout(button_layout) self._sync_initial_selection() self._update_status() self._update_navigation_buttons() self._update_reset_cover_button_state() self._fit_to_screen() self._install_navigation_event_filters() QTimer.singleShot(0, self._update_candidate_thumbnail_layout) def _old_cover_panel(self): panel = QWidget() layout = QVBoxLayout(panel) layout.addWidget(QLabel("旧封面")) self.old_cover_label = _build_image_label( getattr(self.task, "old_cover_path", None), self.OLD_THUMBNAIL_SIZE, empty_text="暂无旧封面", ) layout.addWidget(self.old_cover_label) layout.addStretch(1) return panel def _initial_task_index(self, task, task_index): if self.visible_tasks: if task_index is not None and 0 <= int(task_index) < len(self.visible_tasks): return int(task_index) task_id = getattr(task, "id", None) for index, visible_task in enumerate(self.visible_tasks): if getattr(visible_task, "id", None) == task_id: return index return 0 def _account_for_task(self, task): alias = str(getattr(task, "alias", "") or "").strip() return self.account_by_alias.get(alias) or self.fallback_account def _candidate_panel(self): panel = QWidget() layout = QVBoxLayout(panel) layout.addWidget(QLabel("生成封面候选")) self.status_label = QLabel() self.status_label.setObjectName("coverGalleryStatusLabel") self.status_label.setWordWrap(True) layout.addWidget(self.status_label) self.candidate_scroll = QScrollArea() self.candidate_scroll.setObjectName("coverGalleryScrollArea") self.candidate_scroll.setWidgetResizable(False) self.candidate_scroll.setHorizontalScrollBarPolicy(Qt.ScrollBarAsNeeded) self.candidate_scroll.setVerticalScrollBarPolicy(Qt.ScrollBarAlwaysOff) self.candidate_content = QWidget() self.candidate_content.setObjectName("coverGalleryCandidateContent") self.candidate_layout = QHBoxLayout(self.candidate_content) self.candidate_layout.setContentsMargins(0, 0, 0, 0) self.candidate_layout.setSpacing(12) if self.candidates: for index, candidate_path in enumerate(self.candidates): self.candidate_layout.addWidget( self._candidate_item(candidate_path, index), 0, Qt.AlignTop, ) self.candidate_layout.addStretch(1) else: empty_label = QLabel("暂无生成封面图片") empty_label.setObjectName("coverGalleryEmptyLabel") empty_label.setMinimumSize(320, self.THUMBNAIL_SIZE) empty_label.setAlignment(Qt.AlignCenter) self.candidate_layout.addWidget(empty_label) self._sync_candidate_content_size() self.candidate_scroll.setWidget(self.candidate_content) layout.addWidget(self.candidate_scroll, 1) return panel def _candidate_item(self, candidate_path, index): thumbnail = _CoverThumbnailLabel(candidate_path, self.open_original_image) thumbnail.setObjectName("coverCandidateThumbnail") thumbnail.setAlignment(Qt.AlignCenter) image = _load_image(candidate_path) self.candidate_images[candidate_path] = image self.candidate_thumbnails[candidate_path] = thumbnail if image.isNull(): thumbnail.setText("图片读取失败") else: thumbnail.setPixmap( QPixmap.fromImage( image.scaled( self.candidate_thumbnail_size, self.candidate_thumbnail_size, Qt.KeepAspectRatio, Qt.SmoothTransformation, ) ) ) radio = QRadioButton(_candidate_label(candidate_path, self.current_path, index)) radio.setObjectName(f"coverCandidateRadio{index}") radio.setFocusPolicy(Qt.ClickFocus) radio.toggled.connect( lambda checked, path=candidate_path: self._select_candidate(path) if checked else None ) self.button_group.addButton(radio, index) self.candidate_buttons[candidate_path] = radio meta = QLabel(_candidate_meta(candidate_path, self.current_path)) meta.setWordWrap(True) meta.setObjectName("coverCandidateMetaLabel") basename = os.path.basename(str(candidate_path or "")) file_tooltip = f"完整文件名：{basename}" radio.setToolTip(file_tooltip) meta.setToolTip(file_tooltip) thumbnail.setToolTip(f"双击查看原图\n{file_tooltip}") item = _CoverCandidateCard(thumbnail, radio, meta) item.setObjectName("coverCandidateCard") item.setToolTip(file_tooltip) item.set_thumbnail_size(self.candidate_thumbnail_size) self.candidate_items[candidate_path] = item return item def _sync_candidate_content_size(self): layout = getattr(self, "candidate_layout", None) content = getattr(self, "candidate_content", None) if layout is None or content is None: return candidate_items = [ self.candidate_items[candidate_path] for candidate_path in self.candidates if candidate_path in self.candidate_items ] for item in candidate_items: if isinstance(item, _CoverCandidateCard): item.sync_geometry() layout.invalidate() layout.activate() margins = layout.contentsMargins() spacing = max(0, layout.spacing()) hint = layout.sizeHint() if candidate_items: min_width = ( margins.left() + margins.right() + sum(item.width() for item in candidate_items) + max(0, len(candidate_items) - 1) * spacing ) min_height = ( margins.top() + margins.bottom() + max(item.height() for item in candidate_items) ) else: min_width = margins.left() + margins.right() + 320 min_height = max(self.THUMBNAIL_SIZE, hint.height()) size = QSize(max(min_width, hint.width()), max(1, min_height)) content.setMinimumSize(size) content.resize(size) layout.invalidate() layout.setGeometry(content.rect()) layout.activate() content.updateGeometry() scroll = getattr(self, "candidate_scroll", None) if scroll is not None: scroll.updateGeometry() def resizeEvent(self, event): super().resizeEvent(event) self._candidate_resize_timer.start() def _candidate_thumbnail_target_size(self): scroll = getattr(self, "candidate_scroll", None) if scroll is None or not self.candidates: return self.THUMBNAIL_SIZE viewport = scroll.viewport() viewport_width = viewport.width() viewport_height = viewport.height() if viewport_width <= 0 or viewport_height <= 0: return self.candidate_thumbnail_size visible_count = min(3, max(1, len(self.candidates))) spacing = max(0, self.candidate_layout.spacing()) width_budget = ( viewport_width - max(0, visible_count - 1) * spacing - visible_count * _CoverCandidateCard.EXTRA_WIDTH ) // visible_count control_heights = [ item.control_height() for item in self.candidate_items.values() if isinstance(item, _CoverCandidateCard) ] control_height = max( control_heights, default=self.CANDIDATE_CONTROL_FALLBACK_HEIGHT, ) height_budget = viewport_height - control_height target = min(width_budget, height_budget, self.MAX_THUMBNAIL_SIZE) return max(self.THUMBNAIL_SIZE, int(target)) def _update_candidate_thumbnail_layout(self, force=False): if not self.candidates: self._sync_candidate_content_size() return target_size = self._candidate_thumbnail_target_size() size_changed = target_size != self.candidate_thumbnail_size self.candidate_thumbnail_size = target_size for candidate_path in self.candidates: item = self.candidate_items.get(candidate_path) thumbnail = self.candidate_thumbnails.get(candidate_path) image = self.candidate_images.get(candidate_path) if isinstance(item, _CoverCandidateCard): item.set_thumbnail_size(target_size) if thumbnail is None: continue if not size_changed and not force: continue thumbnail.clear() if image is None or image.isNull(): thumbnail.setText("图片读取失败") continue thumbnail.setPixmap( QPixmap.fromImage( image.scaled( target_size, target_size, Qt.KeepAspectRatio, Qt.SmoothTransformation, ) ) ) self._sync_candidate_content_size() def _sync_initial_selection(self): for candidate_path, radio in self.candidate_buttons.items(): if _same_file(candidate_path, self.current_path): radio.setChecked(True) self.selected_path = candidate_path break self.save_button.setEnabled(self.selected_path is not None) def _select_candidate(self, candidate_path): self.selected_path = candidate_path self.save_button.setEnabled(True) if not _same_file(candidate_path, self.current_path): self._set_save_hint("") self._update_status() def _update_status(self): if not self.candidates: self.status_label.setText("暂无生成封面图片") self.save_button.setEnabled(False) return if self.current_path and not os.path.exists(self.current_path): self.status_label.setText("当前生效封面文件不存在，请选择一张后保存") return if self.current_path and self.selected_path is None: self.status_label.setText("当前生效封面不在候选列表，请选择一张后保存") return if self.selected_path is None: self.status_label.setText("请选择一张生成封面后保存") return if _same_file(self.selected_path, self.current_path): self.status_label.setText("当前生效封面已选中") return self.status_label.setText("已选择新的生成封面，保存后用于后续更新蝦皮") def save_selection(self, checked=False): return self._save_selected_cover(close_on_success=False, show_success_hint=True) def _save_selected_cover(self, close_on_success=False, show_success_hint=False): if not self.selected_path: QMessageBox.warning(self, "封面画廊", "请先选择一张生成封面") return False if _same_file(self.selected_path, self.current_path): self.changed = False if show_success_hint: self._set_save_hint("已保存为本次更新封面") if close_on_success: self.accept() return True if self._needs_committed_confirmation() and not self._confirm_committed_save(): return False try: db.update_generated_cover(self.task.id, self.selected_path, path=self.db_path) except Exception as exc: self._set_save_hint("") QMessageBox.warning(self, "封面画廊", f"保存当前封面失败：{exc}") return False fresh_task = db.get_task(self.task.id, path=self.db_path) if fresh_task is not None: self.task = fresh_task self.visible_tasks[self.task_index] = fresh_task self.current_path = _normalize_file_path(self.selected_path) self.changed = True self._rebuild_candidate_items() self._sync_initial_selection() self._update_status() if show_success_hint: self._set_save_hint("已保存为本次更新封面") if self.gallery_updated_callback is not None: self.gallery_updated_callback(self.task) if close_on_success: self.accept() return True def _set_save_hint(self, message): self.save_hint_label.setText(str(message or "")) def reset_cover_image(self, checked=False): if self._is_bulk_generate_running(): self.status_label.setText("AI 生成正在进行，不能重置图片") self._update_reset_cover_button_state() return False if not str(getattr(self.task, "new_cover_path", "") or "").strip(): self.status_label.setText("当前任务没有可重置的新封面图片") return False if self._needs_committed_confirmation() and not self._confirm_committed_reset(): return False try: db.reset_generated( self.task.id, reset_title=False, reset_cover=True, path=self.db_path, ) fresh_task = db.get_task(self.task.id, path=self.db_path) except Exception as exc: QMessageBox.warning(self, "重置图片", str(exc)) self.status_label.setText(f"重置图片失败：{exc}") return False if fresh_task is None: QMessageBox.warning(self, "重置图片", "任务不存在或已删除") self.status_label.setText("重置图片失败：任务不存在或已删除") return False self._reload_task_and_candidates(fresh_task) self.status_label.setText( "已重置图片，该行图片状态已改为待生成；" "请在②选择「只生成封面」或「生成标题和封面」后点击开始生成" ) if self.gallery_updated_callback is not None: self.gallery_updated_callback(self.task) return True def _needs_committed_confirmation(self): return int(getattr(self.task, "committed", 0) or 0) == 1 or getattr(self.task, "stage", None) == "applied" def _confirm_committed_save(self): box = QMessageBox(self) box.setWindowTitle("确认保存封面") box.setText( "该商品已经提交过线上。本地换封面不会回滚蝦皮，重复更新会再次提交线上。" ) save_button = box.addButton("确认保存", QMessageBox.AcceptRole) cancel_button = box.addButton("取消", QMessageBox.RejectRole) box.setDefaultButton(cancel_button) box.exec() return box.clickedButton() is save_button def _confirm_committed_reset(self): box = QMessageBox(self) box.setWindowTitle("确认重置图片") box.setText( "该商品已经提交过线上。本地重置不会回滚蝦皮，重新生成并更新会再次提交线上。" ) reset_button = box.addButton("确认重置", QMessageBox.AcceptRole) cancel_button = box.addButton("取消", QMessageBox.RejectRole) box.setDefaultButton(cancel_button) box.exec() return box.clickedButton() is reset_button def switch_task(self, offset): target_index = self.task_index + int(offset) if target_index < 0 or target_index >= len(self.visible_tasks): return False if not self._confirm_unsaved_selection(): return False self._load_task_at_index(target_index) return True def _load_task_at_index(self, target_index): self.task_index = int(target_index) target_task = self.visible_tasks[self.task_index] fresh_task = db.get_task(getattr(target_task, "id", None), path=self.db_path) or target_task self.visible_tasks[self.task_index] = fresh_task self.task = fresh_task self.account = self._account_for_task(fresh_task) self.setWindowTitle(f"封面画廊：{getattr(fresh_task, 'item_id', '')}") _set_image_label( self.old_cover_label, getattr(fresh_task, "old_cover_path", None), self.OLD_THUMBNAIL_SIZE, "暂无旧封面", ) self._reload_task_and_candidates(fresh_task) self._update_navigation_buttons() self._update_reset_cover_button_state() def _update_navigation_buttons(self): self.previous_button.setEnabled(self.task_index > 0) self.next_button.setEnabled(self.task_index < len(self.visible_tasks) - 1) def _has_unsaved_selection(self): if not self.selected_path: return False return not _same_file(self.selected_path, self.current_path) def _confirm_unsaved_selection(self): if not self._has_unsaved_selection(): return True box = QMessageBox(self) box.setWindowTitle("未保存封面选择") box.setText("当前已选择另一张生成封面，但尚未保存。") save_button = box.addButton("保存", QMessageBox.AcceptRole) discard_button = box.addButton("不保存", QMessageBox.DestructiveRole) box.addButton("取消", QMessageBox.RejectRole) box.setDefaultButton(save_button) box.exec() clicked = box.clickedButton() if clicked is save_button: return self._save_selected_cover(close_on_success=False) if clicked is discard_button: return True return False def reject(self): if self._confirm_unsaved_selection(): super().reject() def closeEvent(self, event): if self._confirm_unsaved_selection(): event.accept() return event.ignore() def keyPressEvent(self, event): if self._handle_navigation_key(event, QApplication.focusWidget()): return super().keyPressEvent(event) def eventFilter(self, watched, event): focus = QApplication.focusWidget() if watched is self else watched if event.type() == QEvent.KeyPress and self._handle_navigation_key(event, focus): return True return super().eventFilter(watched, event) def _handle_navigation_key(self, event, focus): if event.key() == Qt.Key_Up and self._should_consume_navigation_key(focus): self.switch_task(-1) event.accept() return True if event.key() == Qt.Key_Down and self._should_consume_navigation_key(focus): self.switch_task(1) event.accept() return True return False def _should_consume_navigation_key(self, focus=None): focus = focus or QApplication.focusWidget() if focus is None or focus is self: return True return not isinstance( focus, ( QLineEdit, QPlainTextEdit, QComboBox, QSpinBox, ), ) def _install_navigation_event_filters(self): for widget in [self] + self.findChildren(QWidget): widget.installEventFilter(self) def open_original_image(self, image_path): dialog = OriginalImageDialog(image_path, self) dialog.exec() def _is_bulk_generate_running(self): return bool(self.bulk_running_checker and self.bulk_running_checker()) def _update_reset_cover_button_state(self): running = self._is_bulk_generate_running() self.reset_cover_button.setEnabled(not running) self.reset_cover_button.setToolTip( "AI 生成正在进行，完成后才能重置图片" if running else "" ) def _reload_task_and_candidates(self, task=None): if task is not None: self.task = task self.account = self._account_for_task(task) self.current_path = _normalize_file_path(getattr(self.task, "new_cover_path", None)) self.selected_path = None self._set_save_hint("") self.candidates = image_paths.list_task_cover_candidates(self.image_root, self.task, account=self.account) self._rebuild_candidate_items() self._sync_initial_selection() self._update_status() def _rebuild_candidate_items(self): _clear_layout(self.candidate_layout) self.candidate_buttons = {} self.candidate_images = {} self.candidate_thumbnails = {} self.candidate_items = {} self.button_group = QButtonGroup(self) self.button_group.setExclusive(True) if self.candidates: for index, candidate_path in enumerate(self.candidates): self.candidate_layout.addWidget( self._candidate_item(candidate_path, index), 0, Qt.AlignTop, ) self.candidate_layout.addStretch(1) else: empty_label = QLabel("暂无生成封面图片") empty_label.setObjectName("coverGalleryEmptyLabel") empty_label.setMinimumSize(320, self.THUMBNAIL_SIZE) empty_label.setAlignment(Qt.AlignCenter) self.candidate_layout.addWidget(empty_label) self._candidate_resize_timer.stop() self._update_candidate_thumbnail_layout(force=True) self.candidate_scroll.horizontalScrollBar().setValue(0) self._candidate_resize_timer.start() self._install_navigation_event_filters() def _fit_to_screen(self): available = _available_geometry() if available is None: self.resize(960, 580) return max_width = max(520, int(available.width() * 0.9)) max_height = max(420, int(available.height() * 0.9)) self.resize(min(1180, max_width), min(720, max_height)) _center_dialog(self, available) def _load_image(image_path): return QImage(str(image_path or "")) def _clear_layout(layout): while layout.count(): item = layout.takeAt(0) widget = item.widget() if widget is not None: widget.setParent(None) widget.deleteLater() def _build_image_label(image_path, size, empty_text): label = QLabel() label.setAlignment(Qt.AlignCenter) label.setMinimumSize(size, size) _set_image_label(label, image_path, size, empty_text) return label def _set_image_label(label, image_path, size, empty_text): label.clear() image = _load_image(image_path) if image_path else QImage() if image.isNull(): label.setText(empty_text) return label.setPixmap( QPixmap.fromImage( image.scaled( size, size, Qt.KeepAspectRatio, Qt.SmoothTransformation, ) ) ) def _candidate_label(candidate_path, current_path, index): return "当前生效" if _same_file(candidate_path, current_path) else f"候选 {int(index) + 1}" def _candidate_meta(candidate_path, current_path): if _same_file(candidate_path, current_path): return "当前生效封面" stem, _ext = os.path.splitext(os.path.basename(candidate_path)) parts = stem.rsplit("_", 2) if len(parts) >= 2 and len(parts[-1]) == 14 and parts[-1].isdigit(): return _format_archive_timestamp(parts[-1]) if len(parts) >= 3 and len(parts[-2]) == 14 and parts[-2].isdigit() and parts[-1].isdigit(): return f"{_format_archive_timestamp(parts[-2])} · 同秒第 {parts[-1]} 张" if stem.endswith("_new"): return "当前输出槽" return "历史候选" def _format_archive_timestamp(value): return f"{value[0:4]}-{value[4:6]}-{value[6:8]} {value[8:10]}:{value[10:12]}:{value[12:14]}" def _normalize_file_path(path): value = str(path or "").strip() return os.path.abspath(value) if value else None def _same_file(left, right): if not left or not right: return False return os.path.normcase(os.path.abspath(str(left))) == os.path.normcase(os.path.abspath(str(right))) def _available_geometry(): app = QApplication.instance() screen = app.primaryScreen() if app is not None else None return screen.availableGeometry() if screen is not None else None def _center_dialog(dialog, available): frame = dialog.frameGeometry() frame.moveCenter(available.center()) dialog.move(frame.topLeft()) class GenerateTab(QWidget): """Tab 2: prompt area plus generation task filters/list.""" STATUS_FILTERS = [ ("全部状态", "all"), ("待生成", "to_generate"), ("已生成", "generated"), ("失败", "failed"), ("略过", "skipped"), ("已更新", "applied"), ] RESET_MODE_LABELS = { "title": "标题", "cover": "封面", "all": "标题和封面", } GENERATE_MODE_ITEMS = [ ("只生成标题", "title"), ("只生成封面", "cover"), ("生成标题和封面", "title_cover"), ] TABLE_COLUMN_WEIGHTS = (1.0, 0.5, 1.8375, 1.8375, 0.4125, 0.4125) def __init__( self, parent=None, db_path=None, config=None, config_path=None, status_callback=None, title_prompt_path=None, cover_prompts_dir=None, title_templates_dir=None, open_accounts_callback=None, refresh_workflow_callback=None, subscription_preflight_callback=None, ): super().__init__(parent) self.config = appconfig.load_config(config_path or appconfig.CONFIG_PATH) if config is None else config self.config_path = ( config_path or self.config.get("config_path") or appconfig.CONFIG_PATH ) self.db_path = _database_path(db_path, self.config) self.status_callback = status_callback self.open_accounts_callback = open_accounts_callback self.refresh_workflow_callback = refresh_workflow_callback self.subscription_preflight_callback = subscription_preflight_callback self.title_prompt_path = title_prompt_path or appconfig.title_prompt_path(self.config) self.cover_prompts_dir = cover_prompts_dir or appconfig.cover_prompts_dir(self.config) self.title_templates_dir = title_templates_dir or appconfig.title_templates_dir(self.config) prompts.ensure_default_prompts( self.title_prompt_path, self.cover_prompts_dir, self.title_templates_dir, ) self.current_title_template = None self.current_cover_template = None self.generate_worker = None self.generate_thread = None self.generate_running = False self._title_elapsed_start = None self._cover_elapsed_start = None self._title_elapsed_seconds = 0 self._cover_elapsed_seconds = 0 self.title_prompt_edit = QPlainTextEdit() self.title_prompt_edit.setObjectName("titlePromptEdit") self.title_prompt_edit.setPlaceholderText("标题提示词") self.title_prompt_edit.setPlainText( prompts.load_title_prompt(self.title_prompt_path) ) self.title_template_combo = QComboBox() self.title_template_combo.setObjectName("titleTemplateCombo") self.title_template_combo.setToolTip("选择标题提示词模板并载入到输入框") self.new_title_template_button = QPushButton("新建") self.new_title_template_button.setObjectName("newTitleTemplateButton") self.new_title_template_button.setToolTip("新建一个标题提示词模板") self.save_title_template_button = QPushButton("保存模板") self.save_title_template_button.setObjectName("saveTitleTemplateButton") self.save_title_template_button.setToolTip("保存当前选中的标题提示词模板；未选中模板时会另存为新模板") self.title_template_actions_button = QPushButton("模板操作") self.title_template_actions_button.setObjectName("titleTemplateActionsButton") self.title_template_actions_menu = QMenu(self) self.save_title_template_as_action = self.title_template_actions_menu.addAction("另存为") self.save_title_template_as_action.setObjectName("saveTitleTemplateAsAction") self.rename_title_template_action = self.title_template_actions_menu.addAction("重命名") self.rename_title_template_action.setObjectName("renameTitleTemplateAction") self.delete_title_template_action = self.title_template_actions_menu.addAction("删除") self.delete_title_template_action.setObjectName("deleteTitleTemplateAction") self.title_template_actions_button.setMenu(self.title_template_actions_menu) self.save_title_button = QPushButton("保存标题提示词") self.save_title_button.setToolTip("保存当前标题提示词工作文本，生成时会使用这份内容") self.insert_old_title_button = QPushButton("插入旧标题") self.insert_old_title_button.setObjectName("insertOldTitleButton") self.cover_prompt_edit = QPlainTextEdit() self.cover_prompt_edit.setObjectName("coverPromptEdit") self.cover_prompt_edit.setPlaceholderText("封面提示词") self.cover_template_combo = QComboBox() self.cover_template_combo.setObjectName("coverTemplateCombo") self.new_cover_template_button = QPushButton("新建") self.save_cover_template_button = QPushButton("保存模板") self.save_cover_template_button.setToolTip("保存当前选中的封面提示词模板；未选中模板时会另存为新模板") self.cover_template_actions_button = QPushButton("模板操作") self.cover_template_actions_button.setObjectName("coverTemplateActionsButton") self.cover_template_actions_menu = QMenu(self) self.save_cover_template_as_action = self.cover_template_actions_menu.addAction("另存为") self.save_cover_template_as_action.setObjectName("saveCoverTemplateAsAction") self.rename_cover_template_action = self.cover_template_actions_menu.addAction("重命名") self.rename_cover_template_action.setObjectName("renameCoverTemplateAction") self.delete_cover_template_action = self.cover_template_actions_menu.addAction("删除") self.delete_cover_template_action.setObjectName("deleteCoverTemplateAction") self.cover_template_actions_button.setMenu(self.cover_template_actions_menu) self.insert_title_button = QPushButton("插入标题") self.preview_prompt_button = QPushButton("预览") self.generate_button = QPushButton("开始生成") self.stop_generate_button = QPushButton("停止") self.reset_generate_button = QPushButton("重置生成结果") self.reset_generate_button.setObjectName("resetGenerateButton") self.stop_generate_button.setEnabled(False) self.generate_mode_combo = QComboBox() self.generate_mode_combo.setObjectName("generateModeCombo") self.generate_mode_combo.setToolTip("选择本轮生成标题、封面，或同时生成标题和封面") for label, value in self.GENERATE_MODE_ITEMS: self.generate_mode_combo.addItem(label, value) self._set_combo_by_data(self.generate_mode_combo, appconfig.ai_generate_mode(self.config)) self.progress_label = QLabel("进度：标题0/0 · 图片0/0 · 失败0") self.title_progress_label = QLabel("标题 0/0") self.title_progress_label.setObjectName("generateTitleProgressLabel") self.title_progress_bar = QProgressBar() self.title_progress_bar.setObjectName("generateTitleProgressBar") self.title_progress_bar.setTextVisible(False) self.title_progress_bar.setRange(0, 1) self.title_progress_bar.setValue(0) self.cover_progress_label = QLabel("图片 0/0") self.cover_progress_label.setObjectName("generateCoverProgressLabel") self.cover_progress_bar = _SegmentedProgressBar() self.cover_progress_bar.setObjectName("generateCoverProgressBar") self.cover_progress_bar.set_segments(0, 0, 0) self.failed_progress_label = QLabel("失败 0") self.failed_progress_label.setObjectName("generateFailedProgressLabel") self.failed_progress_label.setVisible(False) self.cmhub_balance_label = QLabel("默认网关余额：未获取") self.cmhub_balance_label.setObjectName("generateCmhubBalanceLabel") self.cmhub_balance_label.setVisible(False) self.title_elapsed_label = QLabel("生标题用时 0 秒") self.title_elapsed_label.setObjectName("generateTitleElapsedLabel") self.cover_elapsed_label = QLabel("生图用时 0 秒") self.cover_elapsed_label.setObjectName("generateCoverElapsedLabel") elapsed_width = max( self.title_elapsed_label.fontMetrics().horizontalAdvance("生标题用时 9999 秒"), self.cover_elapsed_label.fontMetrics().horizontalAdvance("生图用时 9999 秒"), ) + 12 self.title_elapsed_label.setFixedWidth(elapsed_width) self.cover_elapsed_label.setFixedWidth(elapsed_width) self._generate_elapsed_timer = QTimer(self) self._generate_elapsed_timer.setInterval(1000) self._generate_elapsed_timer.timeout.connect(self._refresh_elapsed_labels) left_panel = QWidget() left_layout = QVBoxLayout(left_panel) left_layout.setContentsMargins(0, 0, 12, 0) left_layout.addWidget(QLabel("标题提示词")) title_template_layout = QHBoxLayout() title_template_layout.addWidget(self.title_template_combo, 1) title_template_layout.addWidget(self.new_title_template_button) title_template_layout.addWidget(self.save_title_template_button) title_template_layout.addWidget(self.title_template_actions_button) left_layout.addLayout(title_template_layout) left_layout.addWidget(self.title_prompt_edit, 2) title_action_layout = QHBoxLayout() title_action_layout.addWidget(self.insert_old_title_button) title_action_layout.addWidget(self.save_title_button) title_action_layout.addStretch(1) left_layout.addLayout(title_action_layout) left_layout.addWidget(QLabel("封面提示词")) cover_template_layout = QHBoxLayout() cover_template_layout.addWidget(self.cover_template_combo, 1) cover_template_layout.addWidget(self.new_cover_template_button) cover_template_layout.addWidget(self.save_cover_template_button) cover_template_layout.addWidget(self.cover_template_actions_button) left_layout.addLayout(cover_template_layout) left_layout.addWidget(self.cover_prompt_edit, 2) cover_action_layout = QHBoxLayout() cover_action_layout.addWidget(self.insert_title_button) cover_action_layout.addWidget(self.preview_prompt_button) left_layout.addLayout(cover_action_layout) self.batch_filter = QComboBox() self.batch_filter.setObjectName("batchFilter") self.shop_filter = QComboBox() self.shop_filter.setObjectName("shopFilter") self.item_filter = QLineEdit() self.item_filter.setObjectName("generateItemFilter") self.item_filter.setPlaceholderText("商品ID") self.status_filter = QComboBox() self.status_filter.setObjectName("statusFilter") for label, value in self.STATUS_FILTERS: self.status_filter.addItem(label, value) self.product_status_filter = QComboBox() self.product_status_filter.setObjectName("generateProductStatusFilter") for label, value in product_status.PRODUCT_STATUS_FILTER_ITEMS: self.product_status_filter.addItem(label, value) self.product_status_filter.setCurrentIndex( self.product_status_filter.findData(product_status.STATUS_NORMAL) ) self.refresh_button = QPushButton("刷新") self.open_image_dir_button = QPushButton("打开图片文件夹") self.open_image_dir_button.setObjectName("openImageDirButton") self.open_image_dir_button.setToolTip( "选中某行→打开该账号图片文件夹；未选中→打开当前批次文件夹；全部批次→打开图片根目录" ) filter_layout = QHBoxLayout() filter_layout.addWidget(QLabel("批次")) filter_layout.addWidget(self.batch_filter, 2) filter_layout.addWidget(QLabel("店铺")) filter_layout.addWidget(self.shop_filter, 1) filter_layout.addWidget(QLabel("商品ID")) filter_layout.addWidget(self.item_filter, 1) filter_layout.addWidget(QLabel("商品状态")) filter_layout.addWidget(self.product_status_filter, 1) filter_layout.addWidget(QLabel("生成状态")) filter_layout.addWidget(self.status_filter, 1) filter_layout.addWidget(self.refresh_button) filter_layout.addWidget(self.open_image_dir_button) self.summary_label = QLabel("任务 0 条") self.batch_progress_label = _build_batch_progress_overview("generateBatchProgressOverview") ( self.empty_state_card, self.empty_state_label, self.empty_state_button, ) = _build_empty_state_card("generateEmptyStateCard") if self.open_accounts_callback is not None: self.empty_state_button.clicked.connect(self.open_accounts_callback) self.task_table = _GenerateTaskTableView() self.model = GenerateTaskTableModel(self.task_table, db_path=self.db_path, status_callback=self._set_status) self.model.set_generate_mode(self._current_generate_mode()) self.task_table.setModel(self.model) self.task_table.setSelectionBehavior(QAbstractItemView.SelectRows) self.task_table.setSelectionMode(QAbstractItemView.ExtendedSelection) self.task_table.setEditTriggers(QAbstractItemView.DoubleClicked | QAbstractItemView.EditKeyPressed) self.task_table.setContextMenuPolicy(Qt.CustomContextMenu) self.task_table.horizontalHeader().setSectionResizeMode(QHeaderView.Interactive) self.task_table.horizontalHeader().setStretchLastSection(False) self.task_table.verticalHeader().setVisible(False) self.run_log_view = QPlainTextEdit() self.run_log_view.setObjectName("generateRunLogView") self.run_log_view.setReadOnly(True) self.run_log_view.setMaximumHeight(128) self.run_log_view.setPlaceholderText("AI生成运行日志") right_panel = QWidget() right_layout = QVBoxLayout(right_panel) right_layout.setContentsMargins(12, 0, 0, 0) right_layout.addLayout(filter_layout) right_layout.addWidget(self.summary_label) right_layout.addWidget(self.batch_progress_label) right_layout.addWidget(self.empty_state_card) right_layout.addWidget(self.task_table, 1) right_layout.addWidget(QLabel("AI生成运行日志")) right_layout.addWidget(self.run_log_view) self.splitter = QSplitter(Qt.Horizontal) self.splitter.addWidget(left_panel) self.splitter.addWidget(right_panel) self.splitter.setStretchFactor(0, 1) self.splitter.setStretchFactor(1, 3) self.splitter.setSizes([280, 860]) self.splitter.splitterMoved.connect(lambda *_: self._apply_task_table_column_widths()) title_progress_layout = QHBoxLayout() title_progress_layout.addWidget(self.title_progress_label) title_progress_layout.addWidget(self.title_progress_bar, 1) title_progress_layout.addWidget(self.title_elapsed_label) cover_progress_layout = QHBoxLayout() cover_progress_layout.addWidget(self.cover_progress_label) cover_progress_layout.addWidget(self.cover_progress_bar, 1) cover_progress_layout.addWidget(self.cover_elapsed_label) progress_layout = QVBoxLayout() progress_layout.addLayout(title_progress_layout) progress_layout.addLayout(cover_progress_layout) button_layout = QHBoxLayout() button_layout.addStretch(1) button_layout.addWidget(self.generate_button) button_layout.addWidget(self.stop_generate_button) button_layout.addWidget(self.reset_generate_button) bottom_layout = QHBoxLayout() bottom_layout.addWidget(QLabel("生成内容")) bottom_layout.addWidget(self.generate_mode_combo) bottom_layout.addLayout(progress_layout, 1) bottom_layout.addLayout(button_layout) layout = QVBoxLayout(self) layout.setContentsMargins(18, 18, 18, 18) layout.addWidget(self.splitter, 1) layout.addLayout(bottom_layout) self.batch_filter.currentIndexChanged.connect(self.refresh_tasks) self.shop_filter.currentIndexChanged.connect(self.refresh_tasks) self.item_filter.textChanged.connect(self.refresh_tasks) self.product_status_filter.currentIndexChanged.connect(self.refresh_tasks) self.status_filter.currentIndexChanged.connect(self.refresh_tasks) self.refresh_button.clicked.connect(self.refresh_tasks) self.open_image_dir_button.clicked.connect(self.open_image_directory) self.title_template_combo.currentIndexChanged.connect(self.load_selected_title_template) self.new_title_template_button.clicked.connect(self.new_title_template) self.save_title_template_button.clicked.connect(self.save_title_template) self.save_title_template_as_action.triggered.connect(self.save_title_template_as) self.rename_title_template_action.triggered.connect(self.rename_title_template) self.delete_title_template_action.triggered.connect(self.delete_title_template) self.save_title_button.clicked.connect(self.save_title_prompt) self.insert_old_title_button.clicked.connect(self.insert_old_title_placeholder) self.cover_template_combo.currentIndexChanged.connect(self.load_selected_cover_template) self.new_cover_template_button.clicked.connect(self.new_cover_template) self.save_cover_template_button.clicked.connect(self.save_cover_template) self.save_cover_template_as_action.triggered.connect(self.save_cover_template_as) self.rename_cover_template_action.triggered.connect(self.rename_cover_template) self.delete_cover_template_action.triggered.connect(self.delete_cover_template) self.insert_title_button.clicked.connect(self.insert_title_placeholder) self.preview_prompt_button.clicked.connect(self.preview_cover_prompt) self.generate_mode_combo.currentIndexChanged.connect(self._on_generate_mode_changed) self.generate_button.clicked.connect(self.start_generate) self.stop_generate_button.clicked.connect(self.stop_generate) self.reset_generate_button.clicked.connect(self.reset_generated_result) self.task_table.leftDoubleClicked.connect(self.show_task_images) self.task_table.customContextMenuRequested.connect(self.show_task_context_menu) self.refresh_title_templates() self.refresh_cover_templates() self.refresh_tasks() self._show_current_run_log_empty() self._reset_generation_elapsed() self._update_cmhub_balance_label({}) def resizeEvent(self, event): super().resizeEvent(event) self._apply_task_table_column_widths() def _set_status(self, message, level=None): _emit_status(self.status_callback, message, level=level) def _on_generate_log(self, message): self._append_generate_log(message) self._set_status(message) def _append_generate_log(self, message): self.run_log_view.appendPlainText(str(message)) scroll_bar = self.run_log_view.verticalScrollBar() scroll_bar.setValue(scroll_bar.maximum()) def _show_current_run_log_empty(self): self.run_log_view.setPlainText("本轮日志会在开始运行后显示") def _reset_generate_run_log(self, tasks, generate_mode): self.run_log_view.clear() mode_text = self._generate_mode_label(generate_mode) self._append_generate_log( f"本轮AI生成开始：任务 {len(tasks)} 条，生成内容：{mode_text}" ) def _reset_generation_elapsed(self): self._generate_elapsed_timer.stop() self._title_elapsed_start = None self._cover_elapsed_start = None self._title_elapsed_seconds = 0 self._cover_elapsed_seconds = 0 self._refresh_elapsed_labels() def _start_generation_elapsed(self, title_total, cover_total): self._reset_generation_elapsed() now = time.monotonic() if int(title_total or 0) > 0: self._title_elapsed_start = now elif int(cover_total or 0) > 0: self._cover_elapsed_start = now self._refresh_elapsed_labels() self._update_elapsed_timer_state() def _elapsed_seconds(self, start, frozen_seconds): if start is None: return max(0, int(frozen_seconds or 0)) return max(0, int(time.monotonic() - start)) def _refresh_elapsed_labels(self): title_seconds = self._elapsed_seconds( self._title_elapsed_start, self._title_elapsed_seconds, ) cover_seconds = self._elapsed_seconds( self._cover_elapsed_start, self._cover_elapsed_seconds, ) self.title_elapsed_label.setText(f"生标题用时 {title_seconds} 秒") self.cover_elapsed_label.setText(f"生图用时 {cover_seconds} 秒") def _update_elapsed_timer_state(self): if self._title_elapsed_start is not None or self._cover_elapsed_start is not None: if not self._generate_elapsed_timer.isActive(): self._generate_elapsed_timer.start() return self._generate_elapsed_timer.stop() def _freeze_title_elapsed(self): if self._title_elapsed_start is None: return self._title_elapsed_seconds = self._elapsed_seconds( self._title_elapsed_start, self._title_elapsed_seconds, ) self._title_elapsed_start = None def _freeze_cover_elapsed(self): if self._cover_elapsed_start is None: return self._cover_elapsed_seconds = self._elapsed_seconds( self._cover_elapsed_start, self._cover_elapsed_seconds, ) self._cover_elapsed_start = None def _sync_generation_elapsed(self, payload): title_total = max(0, int(payload.get("title_total", 0) or 0)) title_done = max(0, int(payload.get("title_done", 0) or 0)) cover_total = self._cover_total_for_progress(payload, int(payload.get("total", 0) or 0)) cover_done = max(0, int(payload.get("cover_done", 0) or 0)) if title_total > 0 and title_done >= title_total: self._freeze_title_elapsed() if ( cover_total > 0 and cover_done < cover_total and self._cover_elapsed_start is None and self._cover_elapsed_seconds == 0 and (title_total == 0 or title_done >= title_total) ): self._cover_elapsed_start = time.monotonic() if cover_total > 0 and cover_done >= cover_total: self._freeze_cover_elapsed() self._refresh_elapsed_labels() self._update_elapsed_timer_state() def _finish_generation_elapsed(self, payload=None): self._freeze_title_elapsed() self._freeze_cover_elapsed() self._refresh_elapsed_labels() self._update_elapsed_timer_state() def _on_generate_mode_changed(self, index=None): previous = appconfig.ai_generate_mode(self.config) current = self._current_generate_mode() if self._save_generate_mode_setting(show_status=True): self.model.set_generate_mode(current) self.refresh_tasks() return self.generate_mode_combo.blockSignals(True) self._set_combo_by_data(self.generate_mode_combo, previous) self.generate_mode_combo.blockSignals(False) self.model.set_generate_mode(previous) def _apply_task_table_column_widths(self): if self.model.columnCount() != len(self.TABLE_COLUMN_WEIGHTS): return width = max(0, int(self.task_table.viewport().width())) if width <= 0: return total_weight = sum(self.TABLE_COLUMN_WEIGHTS) used = 0 last_column = len(self.TABLE_COLUMN_WEIGHTS) - 1 header = self.task_table.horizontalHeader() for column, weight in enumerate(self.TABLE_COLUMN_WEIGHTS): if column == last_column: section_width = max(1, width - used) else: section_width = max(1, int(width * weight / total_weight)) used += section_width header.resizeSection(column, section_width) def _save_generate_mode_setting(self, show_status=True): generate_mode = self._current_generate_mode() generate_cover = appconfig.generate_mode_includes_cover(generate_mode) ai_settings = appconfig.ai_config(self.config) ai_settings["generate_mode"] = generate_mode ai_settings["generate_cover"] = generate_cover payload = { key: value for key, value in self.config.items() if key not in {"config_path", "ai_models_path", "cmhub_config_path", "data_dir"} } payload["ai"] = ai_settings try: saved = appconfig.save_config(payload, path=self.config_path) except Exception as exc: self._set_status(f"生成内容设置保存失败：{exc}") return False internal = { key: value for key, value in self.config.items() if key in {"config_path", "ai_models_path", "cmhub_config_path", "data_dir"} } self.config.clear() self.config.update(saved) self.config.update(internal) if self.config_path != appconfig.CONFIG_PATH: self.config["config_path"] = self.config_path if show_status: self._set_status(f"AI生成设置已保存：{self._generate_mode_label(generate_mode)}") return True def _current_generate_mode(self): return appconfig.normalize_generate_mode( self.generate_mode_combo.currentData(), generate_cover=False, ) def _current_generate_mode_includes_cover(self): return appconfig.generate_mode_includes_cover(self._current_generate_mode()) def _generate_mode_label(self, mode): mode = appconfig.normalize_generate_mode(mode) return { "title": "只生成标题", "cover": "只生成封面", "title_cover": "生成标题和封面", }.get(mode, "只生成标题") def _set_combo_by_data(self, combo, value): index = combo.findData(value) combo.setCurrentIndex(index if index >= 0 else 0) def _load_latest_generate_run_log(self): try: logs = db.list_run_logs(limit=1, run_type="generate", path=self.db_path) if not logs: return events = db.list_run_log_events(logs[0].id, limit=40, path=self.db_path) except Exception: return lines = [ f"{event.created_at} [{event.level}] {event.message}" for event in reversed(events) ] self.run_log_view.setPlainText("\n".join(lines)) scroll_bar = self.run_log_view.verticalScrollBar() scroll_bar.setValue(scroll_bar.maximum()) def save_title_prompt(self, checked=False): try: prompts.save_title_prompt( self.title_prompt_edit.toPlainText(), self.title_prompt_path, ) except Exception as exc: self._show_prompt_error(exc) return self._set_status("标题提示词已保存") def refresh_title_templates(self, selected=None): try: template_names = prompts.list_title_templates(self.title_templates_dir) except Exception as exc: template_names = [] self._show_prompt_error(exc) current = selected if selected is not None else self.current_title_template self.title_template_combo.blockSignals(True) self.title_template_combo.clear() self.title_template_combo.addItem("选择模板", None) for name in template_names: self.title_template_combo.addItem(name, name) index = self.title_template_combo.findData(current) if index >= 0: self.title_template_combo.setCurrentIndex(index) self.current_title_template = current else: self.title_template_combo.setCurrentIndex(0) self.current_title_template = None self.title_template_combo.blockSignals(False) def load_selected_title_template(self, index=None): name = self.title_template_combo.currentData() self.current_title_template = name if name is None: return try: self.title_prompt_edit.setPlainText( prompts.load_title_template(name, self.title_templates_dir) ) except Exception as exc: self._show_prompt_error(exc) def new_title_template(self, checked=False): name = self._ask_template_name("新建标题提示词模板") if not name: return try: prompts.save_title_template(name, "", self.title_templates_dir) except Exception as exc: self._show_prompt_error(exc) return self.refresh_title_templates(selected=name) self.title_prompt_edit.setPlainText("") self._set_status(f"标题提示词模板已新建：{name}") def save_title_template(self, checked=False): name = self.current_title_template if name is None: self.save_title_template_as() return try: prompts.save_title_template( name, self.title_prompt_edit.toPlainText(), self.title_templates_dir, ) except Exception as exc: self._show_prompt_error(exc) return self._set_status(f"标题提示词模板已保存：{name}") def save_title_template_as(self, checked=False): name = self._ask_template_name("另存标题提示词模板") if not name: return try: prompts.save_title_template( name, self.title_prompt_edit.toPlainText(), self.title_templates_dir, ) except Exception as exc: self._show_prompt_error(exc) return self.refresh_title_templates(selected=name) self._set_status(f"标题提示词模板已另存为：{name}") def rename_title_template(self, checked=False): old_name = self.current_title_template if old_name is None: self._set_status("没有可重命名的标题提示词模板") return new_name = self._ask_template_name("重命名标题提示词模板", text=old_name) if not new_name or new_name == old_name: return try: prompts.rename_title_template(old_name, new_name, self.title_templates_dir) except Exception as exc: self._show_prompt_error(exc) return self.refresh_title_templates(selected=new_name) self._set_status(f"标题提示词模板已重命名：{new_name}") def delete_title_template(self, checked=False): name = self.current_title_template if name is None: self._set_status("没有可删除的标题提示词模板") return choice = QMessageBox.question( self, "删除标题提示词模板", f"确定删除「{name}」吗？", ) if choice != QMessageBox.Yes: return try: prompts.delete_title_template(name, self.title_templates_dir) except Exception as exc: self._show_prompt_error(exc) return self.refresh_title_templates() self._set_status(f"标题提示词模板已删除：{name}") def refresh_cover_templates(self, selected=None): try: template_names = prompts.list_cover_templates(self.cover_prompts_dir) except Exception as exc: template_names = [] self._show_prompt_error(exc) current = selected if selected is not None else self.current_cover_template self.cover_template_combo.blockSignals(True) self.cover_template_combo.clear() if template_names: for name in template_names: self.cover_template_combo.addItem(name, name) index = self.cover_template_combo.findData(current) self.cover_template_combo.setCurrentIndex(index if index >= 0 else 0) else: self.cover_template_combo.addItem("默认", None) self.cover_template_combo.setCurrentIndex(0) self.cover_template_combo.blockSignals(False) self.load_selected_cover_template() def load_selected_cover_template(self, index=None): name = self.cover_template_combo.currentData() self.current_cover_template = name if name is None: self.cover_prompt_edit.setPlainText("") return try: self.cover_prompt_edit.setPlainText( prompts.load_cover_template(name, self.cover_prompts_dir) ) except Exception as exc: self.cover_prompt_edit.setPlainText("") self._show_prompt_error(exc) def new_cover_template(self, checked=False): name = self._ask_template_name("新建封面提示词模板") if not name: return try: prompts.save_cover_template(name, "", self.cover_prompts_dir) except Exception as exc: self._show_prompt_error(exc) return self.refresh_cover_templates(selected=name) self._set_status(f"封面提示词模板已新建：{name}") def save_cover_template(self, checked=False): name = self.current_cover_template if name is None: self.save_cover_template_as() return try: prompts.save_cover_template( name, self.cover_prompt_edit.toPlainText(), self.cover_prompts_dir, ) except Exception as exc: self._show_prompt_error(exc) return self._set_status(f"封面提示词模板已保存：{name}") def save_cover_template_as(self, checked=False): name = self._ask_template_name("另存封面提示词模板") if not name: return try: prompts.save_cover_template( name, self.cover_prompt_edit.toPlainText(), self.cover_prompts_dir, ) except Exception as exc: self._show_prompt_error(exc) return self.refresh_cover_templates(selected=name) self._set_status(f"封面提示词模板已另存为：{name}") def rename_cover_template(self, checked=False): old_name = self.current_cover_template if old_name is None: self._set_status("没有可重命名的封面提示词模板") return new_name = self._ask_template_name("重命名封面提示词模板", text=old_name) if not new_name or new_name == old_name: return try: prompts.rename_cover_template(old_name, new_name, self.cover_prompts_dir) except Exception as exc: self._show_prompt_error(exc) return self.refresh_cover_templates(selected=new_name) self._set_status(f"封面提示词模板已重命名：{new_name}") def delete_cover_template(self, checked=False): name = self.current_cover_template if name is None: self._set_status("没有可删除的封面提示词模板") return choice = QMessageBox.question( self, "删除封面提示词模板", f"确定删除「{name}」吗？", ) if choice != QMessageBox.Yes: return try: prompts.delete_cover_template(name, self.cover_prompts_dir) except Exception as exc: self._show_prompt_error(exc) return self.refresh_cover_templates() self._set_status(f"封面提示词模板已删除：{name}") def insert_title_placeholder(self, checked=False): self.cover_prompt_edit.insertPlainText("{新标题}") def insert_old_title_placeholder(self, checked=False): self.title_prompt_edit.insertPlainText("{旧标题}") def preview_cover_prompt(self, checked=False): task = self._selected_task() if task is None: self._set_status("没有可预览的任务") return rendered = prompts.render_prompt( self.cover_prompt_edit.toPlainText(), self._prompt_context(task), ) QMessageBox.information(self, "封面提示词预览", rendered) self._set_status("封面提示词预览已生成") def start_generate(self, checked=False): if self.generate_thread is not None: self._set_status("AI 生成正在进行...") return if self.subscription_preflight_callback is not None and not self.subscription_preflight_callback("开始 AI 生成"): return if not self._save_generate_mode_setting(show_status=False): return generate_mode = self._current_generate_mode() if not self._is_cmhub_backend(): try: ai.validate_direct_generation_config( self.config, generate_mode, models_path=( self.config.get("ai_models_path") or appconfig.ai_models_config_path(self.config) ), ) except Exception as exc: message = "自定义网关配置不完整：%s。请到⑤设置补齐本轮所需模型的地址、模型ID、API Key、接口类型和启用状态。" % str(exc) QMessageBox.warning(self, "无法开始生成", message) self._set_status("自定义网关配置不完整，请到⑤设置补齐", "warning") return generate_cover = appconfig.generate_mode_includes_cover(generate_mode) base_candidates = self._generation_candidates(generate_mode) if not base_candidates: if generate_mode == "cover": self._set_status("当前筛选结果没有可生成的封面；请确认商品已完成采集且尚未生成新封面。") else: self._set_status("当前筛选结果没有可生成的缺失内容；请先在①导入采集完成采集，或调整生成内容") return preview_plan = product_status.build_generation_plan( base_candidates, generate_mode, ) generation_scope = self._choose_generation_scope(preview_plan) if generation_scope is None: self._set_status("已取消 AI 生成") return current_plan = product_status.build_generation_plan( self._generation_candidates(generate_mode), generate_mode, generation_scope, ) if current_plan["fingerprint"] != preview_plan["fingerprint"]: self._set_status("当前任务数据已变化，请重新开始生成") return tasks = current_plan["execution_tasks"] if not tasks: self._set_status("当前筛选结果没有状态正常的可生成任务；请先完成采集或选择生成所有状态的商品") return component_totals = ai.generation_component_totals( tasks, generate_mode=generate_mode, ) prompt_values = { "title": self.title_prompt_edit.toPlainText(), "cover": self.cover_prompt_edit.toPlainText(), } worker = GenerateWorker( tasks, prompt_values, db_path=self.db_path, config=self.config, diagnostic_log_dir=diagnostics.DEFAULT_LOG_DIR, generation_scope=generation_scope, product_status_counts=current_plan["status_counts"], status_scope_excluded=current_plan["scope_excluded"], generation_plan_fingerprint=current_plan["fingerprint"], ) worker.progress.connect(self._on_generate_progress) worker.row_updated.connect(self._on_generate_row_updated) worker.log.connect(self._on_generate_log) worker.failed.connect(self._on_generate_failed) worker.finished.connect(self._on_generate_finished) worker.cancelled.connect(self._on_generate_cancelled) self._reset_generate_run_log(tasks, generate_mode) self._reset_cmhub_balance_label() self._start_generation_elapsed( component_totals["title_total"], component_totals["cover_total"], ) thread = run_worker(worker, thread_name="GenerateWorker", start=False) thread.finished.connect(lambda: self._forget_generate_thread(thread)) self.generate_worker = worker self.generate_thread = thread self._set_generate_running(True) self._update_generate_progress( { "total": len(tasks), "title_total": component_totals["title_total"], "title_done": 0, "cover_done": 0, "cover_total": component_totals["cover_total"], "generated_done": 0, "failed": 0, "generate_cover": generate_cover, "generate_mode": generate_mode, } ) if self._is_cmhub_backend(): self._set_status(f"开始 AI 生成：{len(tasks)} 条") else: self._set_status( "开始 AI 生成：%d 条；自定义网关（不计点数，费用由服务商收取）" % len(tasks) ) thread.start() def _generation_candidates(self, generate_mode): return [ task for task in self.model.tasks if ai.is_generatable_task(task, generate_mode=generate_mode) ] def _choose_generation_scope(self, plan): status_counts = dict(plan.get("status_counts") or {}) normal_count = status_counts.get(product_status.STATUS_NORMAL, 0) unlisted_count = status_counts.get(product_status.STATUS_UNLISTED, 0) reviewing_count = status_counts.get(product_status.STATUS_REVIEWING, 0) unknown_count = status_counts.get(product_status.STATUS_UNKNOWN, 0) total = len(plan.get("base_candidates") or []) non_normal_count = total - normal_count box = ProductStatusScopeDialog( title="选择生成范围", text="请选择本轮要生成的商品范围。", informative_text=( "真实候选共{total}条：正常{normal}，未上架{unlisted}，审核中{reviewing}，状态未知{unknown}。\n" "生成架上商品仅将检测结果为正常的商品交给 AI；" "生成全部商品还会处理未上架、审核中和状态未知商品，可能额外消耗点数。" "状态未知商品请优先回到①重新采集确认。".format( total=total, normal=normal_count, unlisted=unlisted_count, reviewing=reviewing_count, unknown=unknown_count, ) ), normal_text="生成架上商品", all_text="生成全部商品", normal_value=product_status.SCOPE_NORMAL_ONLY, all_value=product_status.SCOPE_ALL, normal_object_name="generateNormalOnlyButton", all_object_name="generateAllStatusesButton", cancel_object_name="generateScopeCancelButton", all_enabled=non_normal_count > 0, parent=self, ) box.exec() return box.choice() def stop_generate(self, checked=False): if self.generate_worker is not None: self.generate_worker.cancel() self._append_generate_log("[停止] 已收到停止请求，当前正在运行的任务结束后停止") self._set_status("正在停止 AI 生成...") def reset_generated_result(self, checked=False): if self.generate_thread is not None: self._set_status("AI 生成正在进行，不能重置") return scope, tasks = self._reset_scope_tasks() if not tasks: self._set_status("当前没有可重置生成结果的任务") return if not self._resettable_tasks(tasks, reset_title=True, reset_cover=True): self._set_status("当前范围没有可重置的生成结果") return mode = self._ask_reset_generated_mode(tasks, scope) if mode is None: self._set_status("已取消重置生成结果") return reset_title, reset_cover = self._reset_mode_flags(mode) target_tasks = self._resettable_tasks( tasks, reset_title=reset_title, reset_cover=reset_cover, ) if not target_tasks: self._set_status(f"当前范围没有可重置的{self.RESET_MODE_LABELS[mode]}") return label = self.RESET_MODE_LABELS[mode] try: for task in target_tasks: db.reset_generated( task.id, reset_title=reset_title, reset_cover=reset_cover, path=self.db_path, ) run_id = self._write_generated_reset_run_log( target_tasks, scope, mode, reset_title=reset_title, reset_cover=reset_cover, ) except Exception as exc: QMessageBox.warning(self, "重置生成结果", str(exc)) self._set_status(f"重置生成结果失败：{exc}") return message = ( "action=reset_generated step=db_write result=success " f"detail=重置{label} scope={scope} total={len(target_tasks)}" ) self.refresh_tasks() self._append_generate_log(message) self._set_status( f"已重置生成结果：{len(target_tasks)} 条，内容：{label}，run_id={run_id}" ) def show_task_context_menu(self, position): index = self.task_table.indexAt(position) if not index.isValid(): return self.task_table.selectRow(index.row()) self.task_table.setCurrentIndex(index) task = self.model.task_at(index.row()) if task is None: return menu = QMenu(self.task_table) delete_action = menu.addAction("删除本条记录") delete_action.setEnabled( self.generate_thread is None and getattr(task, "status", "") != "running" ) delete_action.triggered.connect(self.delete_selected_task) menu.exec(self.task_table.viewport().mapToGlobal(position)) def delete_selected_task(self, checked=False): if self.generate_thread is not None: self._set_status("AI 生成正在进行，不能删除记录", level="warning") return task = self._selected_task() if task is None: self._set_status("请选择要删除的记录", level="warning") return if getattr(task, "status", "") == "running": self._set_status("任务正在处理，不能删除", level="warning") return batch = db.get_batch(task.batch_id, path=self.db_path) batch_text = self._batch_label(batch) if batch is not None else task.batch_id lines = [ "确定要删除本条本地记录吗？", f"商品ID：{task.item_id}", f"店铺：{self.model.account_name_for(task)}（{task.alias}）", f"所属批次：{batch_text}", ] if int(getattr(task, "committed", 0) or 0) == 1: lines.extend( [ "", "该商品已提交过蝦皮。本次仅删除本地记录，不会回滚蝦皮线上商品。", ] ) lines.extend( [ "", "删除后，该记录不会出现在①、②、③的列表、筛选、采集、生成、更新或回写范围。", "不会删除原 Excel、本地图片或账号 Chrome 数据。", ] ) answer = QMessageBox.question( self, "删除本条记录", "\n".join(lines), QMessageBox.Yes | QMessageBox.No, QMessageBox.No, ) if answer != QMessageBox.Yes: self._set_status("已取消删除本条记录") return try: result = db.delete_task( task.id, reason="用户在AI生成页删除单条记录", path=self.db_path, ) except Exception as exc: QMessageBox.warning(self, "删除本条记录", str(exc)) self._set_status(f"删除本条记录失败：{exc}", level="danger") return self.refresh_tasks() if self.refresh_workflow_callback is not None: self.refresh_workflow_callback() message = f"已删除本地记录：商品 {result['item_id']}" self._set_status(message, level="success") QMessageBox.information(self, "删除本条记录", message) def _reset_scope_tasks(self): selection = self.task_table.selectionModel() selected_rows = [] if selection is not None: selected_rows = sorted({index.row() for index in selection.selectedRows()}) if selected_rows: tasks = [self.model.task_at(row) for row in selected_rows] return "selected", [task for task in tasks if task is not None] return "filtered", list(self.model.tasks) def _ask_reset_generated_mode(self, tasks, scope): scope_text = "选中任务" if scope == "selected" else "当前筛选结果" title_count = sum(1 for task in tasks if getattr(task, "new_title", None)) cover_count = sum(1 for task in tasks if getattr(task, "new_cover_path", None)) committed_count = sum(1 for task in tasks if int(getattr(task, "committed", 0) or 0)) running_count = sum(1 for task in tasks if getattr(task, "status", None) == "running") sample_ids = [str(getattr(task, "item_id", "") or "") for task in tasks[:5]] sample_ids = [item_id for item_id in sample_ids if item_id] sample_text = "、".join(sample_ids) if len(tasks) > 5: sample_text += f" 等 {len(tasks)} 条" lines = [ "请选择要重置的本地生成内容。", "", f"作用范围：{scope_text} {len(tasks)} 条", f"包含新标题：{title_count} 条", f"包含新封面：{cover_count} 条", ] if sample_text: lines.append(f"商品ID：{sample_text}") if running_count: lines.append(f"运行中任务会跳过：{running_count} 条") lines.extend([ "", "重置标题：只清空新标题，保留新封面。", "重置封面：只清空新封面，保留新标题和手动微调标题。", "重置全部：清空新标题和新封面。", "", "不会触碰蝦皮，也不会自动回写 Excel；后续重新生成会消耗 cmhub 点数。", "默认不删除本地新封面文件。", ]) if committed_count: lines.extend([ "", f"注意：其中 {committed_count} 条已经提交过线上。本地重置不会回滚蝦皮，重生成后再更新会再次提交线上。", ]) box = QMessageBox(self) box.setWindowTitle("重置生成结果") box.setText("\n".join(lines)) title_button = box.addButton("重置标题", QMessageBox.AcceptRole) cover_button = box.addButton("重置封面", QMessageBox.AcceptRole) all_button = box.addButton("重置全部", QMessageBox.DestructiveRole) cancel_button = box.addButton("取消", QMessageBox.RejectRole) box.setDefaultButton(cancel_button) box.exec() clicked = box.clickedButton() if clicked is title_button: return "title" if clicked is cover_button: return "cover" if clicked is all_button: return "all" return None def _reset_mode_flags(self, mode): if mode == "title": return True, False if mode == "cover": return False, True return True, True def _resettable_tasks(self, tasks, reset_title=True, reset_cover=True): resettable = [] for task in tasks: if getattr(task, "status", None) == "running": continue has_title = bool(getattr(task, "new_title", None)) has_cover = bool(getattr(task, "new_cover_path", None)) if reset_title and has_title: resettable.append(task) continue if reset_cover and has_cover: resettable.append(task) return resettable def _write_generated_reset_run_log( self, tasks, scope, mode, reset_title=True, reset_cover=True, ): task_ids = [getattr(task, "id", None) for task in tasks] committed_count = sum(1 for task in tasks if int(getattr(task, "committed", 0) or 0)) options = { "action": "reset_generated", "scope": scope, "mode": mode, "reset_title": bool(reset_title), "reset_cover": bool(reset_cover), "task_ids": task_ids, "committed_count": committed_count, } run_id = db.create_run_log( "reset", total=len(tasks), options=options, path=self.db_path, ) label = self.RESET_MODE_LABELS[mode] for task in tasks: message = ( "action=reset_generated step=db_write result=success " f"detail=重置{label} task_id={getattr(task, 'id', None)}" ) db.add_run_log_event( run_id, message, task_id=getattr(task, "id", None), alias=getattr(task, "alias", None), item_id=getattr(task, "item_id", None), path=self.db_path, ) db.finish_run_log( run_id, status="done", done=len(tasks), success_count=len(tasks), failed_count=0, summary_json=options, path=self.db_path, ) return run_id def show_task_images(self, index): if index.isValid() and index.column() == 3: return row = index.row() if index.isValid() else self.task_table.currentIndex().row() task = self.model.task_at(row) if row >= 0 else self._selected_task() if task is None: self._set_status("没有可预览的任务") return account = self.model.account_by_alias.get(str(getattr(task, "alias", "") or "").strip()) dialog = CoverGalleryDialog( task, image_root=appconfig.image_dir(self.config), db_path=self.db_path, account=account, parent=self, config=self.config, bulk_running_checker=lambda: self.generate_thread is not None, gallery_updated_callback=self._on_cover_gallery_updated, visible_tasks=list(self.model.tasks), task_index=row if row >= 0 else 0, account_by_alias=self.model.account_by_alias, ) if dialog.exec() == QDialog.Accepted: self.refresh_tasks() if dialog.changed: self._set_status(f"已保存当前新封面：{task.item_id}") else: self._set_status(f"当前新封面未变更：{task.item_id}") def _set_generate_running(self, running): self.generate_running = bool(running) self.generate_button.setEnabled(not running) self.stop_generate_button.setEnabled(running) self.reset_generate_button.setEnabled(not running) self.refresh_button.setEnabled(not running) self.batch_filter.setEnabled(not running) self.shop_filter.setEnabled(not running) self.item_filter.setEnabled(not running) self.product_status_filter.setEnabled(not running) self.status_filter.setEnabled(not running) self.title_template_combo.setEnabled(not running) self.new_title_template_button.setEnabled(not running) self.save_title_template_button.setEnabled(not running) self.title_template_actions_button.setEnabled(not running) self.save_title_template_as_action.setEnabled(not running) self.rename_title_template_action.setEnabled(not running) self.delete_title_template_action.setEnabled(not running) self.save_title_button.setEnabled(not running) self.cover_template_combo.setEnabled(not running) self.new_cover_template_button.setEnabled(not running) self.save_cover_template_button.setEnabled(not running) self.cover_template_actions_button.setEnabled(not running) self.save_cover_template_as_action.setEnabled(not running) self.rename_cover_template_action.setEnabled(not running) self.delete_cover_template_action.setEnabled(not running) self.generate_mode_combo.setEnabled(not running) def _on_cover_gallery_updated(self, task): self.refresh_tasks() def _forget_generate_thread(self, thread): if self.generate_thread is thread: self.generate_thread = None self.generate_worker = None def _on_generate_progress(self, payload): self._update_generate_progress(payload) self._sync_generation_elapsed(payload) self._update_cmhub_balance_label(payload) self._set_status("生成进度：" + self._generate_progress_text(payload)) def _on_generate_row_updated(self, task_id, fields): self.refresh_tasks() def _on_generate_failed(self, task_id, error): self._set_status(f"AI 生成失败：{error}") def _on_generate_finished(self, payload): self._set_generate_running(False) self.refresh_tasks() self._update_generate_progress(payload) self._finish_generation_elapsed(payload) self._update_cmhub_balance_label(payload) billing_error = payload.get("billing_error") if billing_error: self._show_billing_error(billing_error) return if payload.get("error"): self._set_status(f"AI 生成失败：{payload.get('error')}") self._show_generate_completion_message(payload, error=True) return self._set_status("AI 生成完成：" + self._generate_progress_text(payload)) self._show_generate_completion_message(payload) def _on_generate_cancelled(self, payload): self._set_generate_running(False) self.refresh_tasks() self._update_generate_progress(payload) self._finish_generation_elapsed(payload) self._update_cmhub_balance_label(payload) billing_error = payload.get("billing_error") if billing_error: self._show_billing_error(billing_error) return self._set_status("AI 生成已停止：" + self._generate_progress_text(payload)) self._show_generate_completion_message(payload, cancelled=True) def _update_generate_progress(self, payload): total = max(0, int(payload.get("total", 0) or 0)) title_total = max(0, int(payload.get("title_total", total) or 0)) title_done = max(0, int(payload.get("title_done", 0) or 0)) cover_done = max(0, int(payload.get("cover_done", 0) or 0)) cover_total = self._cover_total_for_progress(payload, total) failed = max(0, int(payload.get("failed", 0) or 0)) cover_failed = self._cover_failed_for_progress(payload, cover_done, cover_total) self.progress_label.setText("进度：" + self._generate_progress_text(payload)) self.title_progress_label.setText(f"标题 {title_done}/{title_total}") cover_failed_text = f"，失败 {cover_failed}" if cover_failed else "" self.cover_progress_label.setText(f"图片 {cover_done}/{cover_total}{cover_failed_text}") self.failed_progress_label.setText(f"失败 {failed}") self._set_progress_bar(self.title_progress_bar, title_done, title_total) self.cover_progress_bar.set_segments(cover_done, cover_failed, cover_total) def _reset_cmhub_balance_label(self): self.cmhub_balance_label.setVisible(False) if self._is_cmhub_backend(): self.cmhub_balance_label.setText("默认网关余额：生成后刷新") else: self.cmhub_balance_label.setText("") def _update_cmhub_balance_label(self, payload): self.cmhub_balance_label.setVisible(False) if not self._is_cmhub_backend(): return balance = payload.get("points_balance") if isinstance(payload, dict) else None if balance is None: if not self.cmhub_balance_label.text(): self.cmhub_balance_label.setText("默认网关余额：未获取") return self.cmhub_balance_label.setText(f"默认网关余额：{balance}") def refresh_gateway_state(self): """Clear source-specific feedback after ⑤ saves a new gateway source.""" self._reset_cmhub_balance_label() def _is_cmhub_backend(self): try: return appconfig.ai_config(self.config).get("backend") == "cmhub" except Exception: return False def _show_billing_error(self, billing_error): message = billing_error.get("message") or "点数不足，请先充值。" QMessageBox.warning(self, "AI生成", message) self._set_status("AI 生成已中止：" + message) def _show_generate_completion_message(self, payload, cancelled=False, error=False): failed = max(0, int(payload.get("failed", 0) or 0)) title_total = max(0, int(payload.get("title_total", payload.get("total", 0)) or 0)) title_done = max(0, int(payload.get("title_done", 0) or 0)) cover_total = self._cover_total_for_progress(payload, max(0, int(payload.get("total", 0) or 0))) cover_done = max(0, int(payload.get("cover_done", 0) or 0)) elapsed_seconds = self._generation_elapsed_total_seconds() if cancelled: title = "AI生成已停止" message = ( f"本轮AI生成已停止：标题 {title_done}/{title_total}，图片 {cover_done}/{cover_total}，" f"失败 {failed}，用时 {elapsed_seconds} 秒。未完成任务可再次点击开始生成继续处理。" ) QMessageBox.warning(self, title, message) return if error: title = "AI生成失败" message = ( f"本轮AI生成失败：标题 {title_done}/{title_total}，图片 {cover_done}/{cover_total}，" f"失败 {failed}，用时 {elapsed_seconds} 秒。失败原因请查看下方运行日志。" ) QMessageBox.warning(self, title, message) return if failed: title = "AI生成完成，有失败任务" message = ( f"本轮AI生成完成：标题 {title_done}/{title_total}，图片 {cover_done}/{cover_total}，" f"失败 {failed}，用时 {elapsed_seconds} 秒。失败明细请查看下方运行日志，可稍后重新生成失败任务。" ) QMessageBox.warning(self, title, message) return title = "AI生成完成" message = ( f"本轮AI生成完成：标题 {title_done}/{title_total}，图片 {cover_done}/{cover_total}，" f"失败 0，用时 {elapsed_seconds} 秒。" ) QMessageBox.information(self, title, message) def _set_progress_bar(self, bar, done, total): maximum = max(1, int(total or 0)) value = min(max(0, int(done or 0)), maximum) bar.setRange(0, maximum) bar.setValue(value) def _cover_total_for_progress(self, payload, total): cover_total = payload.get("cover_total") if cover_total is None: cover_total = total if payload.get("generate_cover", True) else 0 return max(0, int(cover_total or 0)) def _cover_failed_for_progress(self, payload, cover_done, cover_total): if cover_total <= 0: return 0 failed = max(0, int(payload.get("failed", 0) or 0)) remaining = max(0, int(cover_total or 0) - max(0, int(cover_done or 0))) return min(failed, remaining) def _generation_elapsed_total_seconds(self): return max(0, int(self._title_elapsed_seconds or 0)) + max(0, int(self._cover_elapsed_seconds or 0)) def _generate_progress_text(self, payload): total = max(0, int(payload.get("total", 0) or 0)) title_total = max(0, int(payload.get("title_total", total) or 0)) cover_total = self._cover_total_for_progress(payload, total) return "标题{title}/{title_total} · 图片{cover}/{cover_total} · 失败{failed}".format( title=payload.get("title_done", 0), cover=payload.get("cover_done", 0), cover_total=cover_total, title_total=title_total, failed=payload.get("failed", 0), ) def open_image_directory(self): target_dir, missing_message = self._resolve_image_directory_target() if not target_dir: message = missing_message or "无法定位图片文件夹" QMessageBox.warning(self, "打开图片文件夹", message) self._set_status(message, level="warning") return if not os.path.isdir(target_dir): message = missing_message or "该批次/账号还没有图片" QMessageBox.warning(self, "打开图片文件夹", message) self._set_status(message, level="warning") return try: opened_dir = file_manager.open_in_file_manager(target_dir) except Exception as exc: message = f"打开图片文件夹失败：{exc}" QMessageBox.warning(self, "打开图片文件夹", message) self._set_status(message, level="danger") return self._set_status(f"已打开图片文件夹：{opened_dir}", level="success") def _resolve_image_directory_target(self): image_root = appconfig.image_dir(self.config) row = self._selected_row_for_image_directory() if row is not None: task = self.model.task_at(row) if task is None: return None, "没有选中可打开图片文件夹的任务" existing_dir = self._existing_cover_parent_dir(task) if existing_dir: return existing_dir, "该账号还没有图片" account = self._account_for_task(task) if account is None: return None, "选中任务没有匹配账号，无法定位图片文件夹" target = os.path.dirname( image_paths.task_image_path( image_root, task, account=account, suffix="new", ext=".jpg", ) ) return target, "该账号还没有图片" selected_batch = self.batch_filter.currentData() if selected_batch not in (None, "", "all"): return os.path.abspath(os.path.join(image_root, str(selected_batch))), "该批次还没有图片" return os.path.abspath(image_root), "图片根目录不存在，请先采集或生成图片" def _selected_row_for_image_directory(self): selection = self.task_table.selectionModel() selected_rows = [] if selection is not None: selected_rows = sorted({index.row() for index in selection.selectedRows()}) if not selected_rows: return None current = self.task_table.currentIndex() if current.isValid() and current.row() in selected_rows: return current.row() return selected_rows[0] def _existing_cover_parent_dir(self, task): for attr in ("new_cover_path", "old_cover_path"): value = str(getattr(task, attr, "") or "").strip() if not value: continue path = os.path.abspath(value) if os.path.isfile(path): return os.path.dirname(path) return None def _account_for_task(self, task): alias = str(getattr(task, "alias", "") or "").strip() if not alias: return None return self.model.account_by_alias.get(alias) def _selected_task(self): index = self.task_table.currentIndex() if index.isValid(): return self.model.task_at(index.row()) if self.model.rowCount() > 0: return self.model.task_at(0) return None def _prompt_context(self, task): new_title = task.new_title if self._current_generate_mode() == "cover": new_title = ai.cover_title_context(task) return { "old_title": task.old_title, "new_title": new_title, "item_id": task.item_id, "account_name": self.model.account_name_for(task), "alias": task.alias, } def _ask_template_name(self, title, text=""): value, ok = QInputDialog.getText( self, title, "模板名", QLineEdit.Normal, text, ) if not ok: return None return str(value).strip() def _show_prompt_error(self, error): message = str(error) QMessageBox.warning(self, "提示词管理", message) self._set_status(message) def refresh_tasks(self, checked=False): try: db.init_db(self.db_path) batches = db.list_batches(path=self.db_path) accounts_rows = accounts.list_accounts(path=self.db_path, config=self.config) selected_batch = self.batch_filter.currentData() selected_shop = self.shop_filter.currentData() selected_status = self.status_filter.currentData() or "all" selected_product_status = self.product_status_filter.currentData() or "all" item_query = self.item_filter.text().strip() self._populate_batch_filter(batches, selected_batch) selected_batch = self.batch_filter.currentData() batch_tasks = db.list_tasks(batch_id=selected_batch, path=self.db_path) self._populate_shop_filter(batch_tasks, accounts_rows, selected_shop) selected_shop = self.shop_filter.currentData() filtered_tasks = [ task for task in batch_tasks if self._matches_shop(task, selected_shop) and self._matches_item(task, item_query) and self._matches_product_status(task, selected_product_status) and self._matches_status(task, selected_status) ] except Exception as exc: self.model.set_tasks([], []) self.summary_label.setText("任务读取失败") _set_batch_progress_overview(self.batch_progress_label, []) _set_empty_state(self.empty_state_card, self.empty_state_label, self.empty_state_button) self._set_status(f"AI 生成任务读取失败：{exc}") return self.model.set_generate_mode(self._current_generate_mode()) self.model.set_tasks(filtered_tasks, accounts_rows) self._apply_task_table_column_widths() self.summary_label.setText( f"任务 {len(filtered_tasks)}/{len(batch_tasks)} 条" ) _set_batch_progress_overview(self.batch_progress_label, batch_tasks) self._update_empty_state(batch_tasks, filtered_tasks, accounts_rows) def _update_empty_state(self, batch_tasks, filtered_tasks, account_rows): if not account_rows: _set_empty_state( self.empty_state_card, self.empty_state_label, self.empty_state_button, "第一步：前往『账号管理』配置并登录账号，再回到②生成标题和封面。", self.open_accounts_callback is not None, ) return if not batch_tasks: _set_empty_state( self.empty_state_card, self.empty_state_label, self.empty_state_button, "还没有可生成任务。请先在①导入采集完成旧标题和旧封面采集。", ) return if not filtered_tasks: _set_empty_state( self.empty_state_card, self.empty_state_label, self.empty_state_button, "当前筛选没有匹配的生成任务，请调整批次、店铺、商品ID或状态筛选。", ) return _set_empty_state(self.empty_state_card, self.empty_state_label, self.empty_state_button) def _populate_batch_filter(self, batches, selected_batch): had_previous_items = self.batch_filter.count() > 0 self.batch_filter.blockSignals(True) self.batch_filter.clear() self.batch_filter.addItem("全部批次", None) for batch in batches: self.batch_filter.addItem(self._batch_label(batch), batch.id) self.batch_filter.setCurrentIndex( _batch_filter_current_index( self.batch_filter, batches, selected_batch, had_previous_items, ) ) self.batch_filter.blockSignals(False) def _populate_shop_filter(self, tasks, account_rows, selected_shop): account_by_alias = { str(account.alias).strip(): account for account in account_rows if str(account.alias).strip() } aliases = [] for task in tasks: alias = str(task.alias).strip() if alias and alias not in aliases: aliases.append(alias) previous = selected_shop if selected_shop in aliases else None self.shop_filter.blockSignals(True) self.shop_filter.clear() self.shop_filter.addItem("全部店铺", None) for alias in sorted(aliases, key=lambda value: self._shop_label(value, account_by_alias)): self.shop_filter.addItem(self._shop_label(alias, account_by_alias), alias) index = self.shop_filter.findData(previous) self.shop_filter.setCurrentIndex(index if index >= 0 else 0) self.shop_filter.blockSignals(False) def _batch_label(self, batch): source_files = batch.source_files first_file = os.path.basename(source_files[0]) if source_files else batch.id return f"{batch.created_at} · {first_file}" def _shop_label(self, alias, account_by_alias): account = account_by_alias.get(alias) if account is not None: return f"{account.account_name} ({alias})" return alias def _matches_shop(self, task, selected_shop): return selected_shop is None or str(task.alias).strip() == selected_shop def _matches_item(self, task, item_query): if not item_query: return True return item_query in str(getattr(task, "item_id", "")) def _matches_product_status(self, task, selected_product_status): if selected_product_status in (None, "all"): return True return product_status.raw_status(getattr(task, "product_status", None)) == selected_product_status def _matches_status(self, task, selected_status): if selected_status in (None, "all"): return True if selected_status == "to_generate": generate_mode = self._current_generate_mode() return ai.is_generatable_task(task, generate_mode=generate_mode) and task.status != "failed" if selected_status == "generated": return task.stage == "generated" if selected_status == "applied": return task.stage == "applied" if selected_status == "failed": return task.status == "failed" if selected_status == "skipped": return task.status == "skipped" return True