"""PySide6 GUI entry point.""" from __future__ import annotations import os import sys try: from PySide6.QtCore import QAbstractTableModel, QModelIndex, Qt from PySide6.QtGui import QPixmap from PySide6.QtWidgets import ( QAbstractItemView, QApplication, QCheckBox, QComboBox, QDialog, QDialogButtonBox, QFileDialog, QFormLayout, QHBoxLayout, QHeaderView, QInputDialog, QLabel, QLineEdit, QMainWindow, QMessageBox, QPlainTextEdit, QPushButton, QScrollArea, QSplitter, QTableView, QSpinBox, QTableWidget, QTableWidgetItem, QTabWidget, QVBoxLayout, QWidget, ) QT_IMPORT_ERROR = None except ModuleNotFoundError as exc: QApplication = None QMainWindow = object QTabWidget = None QVBoxLayout = None QWidget = object QT_IMPORT_ERROR = exc TAB_TITLES = [ "① 导入采集", "② AI生成", "③ 更新shopee", "④ 账号管理", "⑤ 设置", ] TAB_STYLE = """ QTabWidget::pane { border-top: 1px solid #c9d1d9; } QTabBar::tab { min-width: 128px; min-height: 34px; padding: 8px 18px; margin-right: 8px; border: 1px solid #c9d1d9; border-bottom-color: #b8c0ca; background: #f4f6f8; color: #24292f; } QTabBar::tab:selected { background: #ffffff; border-color: #687785; border-bottom-color: #ffffff; font-weight: 600; } QTabBar::tab:hover:!selected { background: #eaf2ff; } """ if QT_IMPORT_ERROR is None: from . import accounts, ai, appconfig, chrome, db, editor, excel, prompts from . import config as account_config PLAINTEXT_SECRET_TITLE = "本地明文保存提示" PLAINTEXT_API_KEY_WARNING = ( "API Key 会以本地明文保存到 config/ai_models.json，仅供本机调用 AI 使用。" "该文件已 gitignore，UI 打码显示，日志/导出不记录明文。" ) PLAINTEXT_PASSWORD_WARNING = ( "密码会以本地明文保存到本地 SQLite，仅供人工参考，不会自动登录/自动填。" "数据库文件已 gitignore，请勿提交或分享。" ) def _database_path(db_path=None, config=None) -> str: return db_path or appconfig.db_path(config) class TaskTableModel(QAbstractTableModel): """Table model for task rows shared by workflow tabs.""" HEADERS = ["账号", "别名", "商品ID", "阶段"] STAGE_TEXT = { "imported": "待采集", "collected": "已采集", "generated": "已生成", "applied": "已更新", } STATUS_TEXT = { "running": "处理中", "failed": "失败", "skipped": "略过", "cancelled": "已取消", } def __init__(self, parent=None): super().__init__(parent) self.tasks = [] self.all_tasks = [] self.account_by_alias = {} self.filter_mode = "all" def set_tasks(self, tasks, accounts): self.beginResetModel() self.all_tasks = list(tasks) self.account_by_alias = { str(account.alias).strip(): account for account in accounts if str(account.alias).strip() } self.tasks = self._filtered_tasks() self.endResetModel() def set_filter_mode(self, mode): self.beginResetModel() self.filter_mode = mode if mode in {"all", "unmatched"} else "all" self.tasks = self._filtered_tasks() self.endResetModel() def _filtered_tasks(self): if self.filter_mode == "unmatched": return [task for task in self.all_tasks if self.is_unmatched(task)] return list(self.all_tasks) def rowCount(self, parent=QModelIndex()): return 0 if parent.isValid() else len(self.tasks) def columnCount(self, parent=QModelIndex()): return 0 if parent.isValid() else len(self.HEADERS) def headerData(self, section, orientation, role=Qt.DisplayRole): if role != Qt.DisplayRole: return None if orientation == Qt.Horizontal and 0 <= section < len(self.HEADERS): return self.HEADERS[section] return section + 1 if orientation == Qt.Vertical else None def data(self, index, role=Qt.DisplayRole): if not index.isValid(): return None task = self.tasks[index.row()] if role == Qt.DisplayRole: return self._display_value(task, index.column()) if role == Qt.ToolTipRole and self.is_unmatched(task): return "别名未匹配账号，采集时将略过" return None def flags(self, index): if not index.isValid(): return Qt.NoItemFlags return Qt.ItemIsEnabled | Qt.ItemIsSelectable def task_at(self, row): if row < 0 or row >= len(self.tasks): return None return self.tasks[row] def is_unmatched(self, task) -> bool: return str(task.alias).strip() not in self.account_by_alias def unmatched_count(self) -> int: return sum(1 for task in self.all_tasks if self.is_unmatched(task)) def _account_name(self, task) -> str: account = self.account_by_alias.get(str(task.alias).strip()) if account is not None: return account.account_name return task.account_name or "" def _stage_text(self, task) -> str: if self.is_unmatched(task): return "略过" if task.status in self.STATUS_TEXT and task.status != "pending": return self.STATUS_TEXT[task.status] return self.STAGE_TEXT.get(task.stage, task.stage) def _display_value(self, task, column): values = [ self._account_name(task), task.alias, task.item_id, self._stage_text(task), ] return values[column] if 0 <= column < len(values) else None class GenerateTaskTableModel(QAbstractTableModel): """Table model for Tab 2 generation candidates.""" HEADERS = ["店铺", "商品ID", "旧标题", "新标题", "状态"] STATUS_TEXT = { "running": "处理中", "failed": "失败", "skipped": "略过", "cancelled": "已取消", } STAGE_TEXT = { "imported": "未采集", "collected": "待生成", "generated": "已生成", "applied": "已更新", } def __init__(self, parent=None): super().__init__(parent) self.tasks = [] self.account_by_alias = {} def set_tasks(self, tasks, accounts): self.beginResetModel() self.tasks = list(tasks) self.account_by_alias = { str(account.alias).strip(): account for account in accounts if str(account.alias).strip() } self.endResetModel() def rowCount(self, parent=QModelIndex()): return 0 if parent.isValid() else len(self.tasks) def columnCount(self, parent=QModelIndex()): return 0 if parent.isValid() else len(self.HEADERS) def headerData(self, section, orientation, role=Qt.DisplayRole): if role != Qt.DisplayRole: return None if orientation == Qt.Horizontal and 0 <= section < len(self.HEADERS): return self.HEADERS[section] return section + 1 if orientation == Qt.Vertical else None def data(self, index, role=Qt.DisplayRole): if not index.isValid(): return None task = self.tasks[index.row()] if role == Qt.DisplayRole: return self._display_value(task, index.column()) if role == Qt.ToolTipRole and task.last_error: return task.last_error return None def flags(self, index): if not index.isValid(): return Qt.NoItemFlags return Qt.ItemIsEnabled | Qt.ItemIsSelectable def task_at(self, row): if row < 0 or row >= len(self.tasks): return None return self.tasks[row] def account_name_for(self, task): return self._account_name(task) def _account_name(self, task): account = self.account_by_alias.get(str(task.alias).strip()) if account is not None: return account.account_name return task.account_name or task.alias or "" def _status_text(self, task): if task.status in self.STATUS_TEXT and task.status != "pending": return self.STATUS_TEXT[task.status] return self.STAGE_TEXT.get(task.stage, task.stage) def _display_value(self, task, column): values = [ self._account_name(task), task.item_id, task.old_title or "", task.new_title or "", self._status_text(task), ] return values[column] if 0 <= column < len(values) else None class ApplyTaskTableModel(QAbstractTableModel): """Table model for Tab 3 update candidates.""" HEADERS = ["店铺", "商品ID", "新标题", "新封面", "阶段", "结果"] STATUS_TEXT = { "running": "处理中", "failed": "失败", "skipped": "略过", "cancelled": "已取消", "pending": "待更新", "success": "成功", } def __init__(self, parent=None): super().__init__(parent) self.tasks = [] self.account_by_alias = {} def set_tasks(self, tasks, accounts): self.beginResetModel() self.tasks = list(tasks) self.account_by_alias = { str(account.alias).strip(): account for account in accounts if str(account.alias).strip() } self.endResetModel() def rowCount(self, parent=QModelIndex()): return 0 if parent.isValid() else len(self.tasks) def columnCount(self, parent=QModelIndex()): return 0 if parent.isValid() else len(self.HEADERS) def headerData(self, section, orientation, role=Qt.DisplayRole): if role != Qt.DisplayRole: return None if orientation == Qt.Horizontal and 0 <= section < len(self.HEADERS): return self.HEADERS[section] return section + 1 if orientation == Qt.Vertical else None def data(self, index, role=Qt.DisplayRole): if not index.isValid(): return None task = self.tasks[index.row()] if role == Qt.DisplayRole: return self._display_value(task, index.column()) if role == Qt.ToolTipRole and task.last_error: return task.last_error return None def flags(self, index): if not index.isValid(): return Qt.NoItemFlags return Qt.ItemIsEnabled | Qt.ItemIsSelectable def task_at(self, row): if row < 0 or row >= len(self.tasks): return None return self.tasks[row] def account_name_for(self, task): account = self.account_by_alias.get(str(task.alias).strip()) if account is not None: return account.account_name return task.account_name or task.alias or "" def _display_value(self, task, column): values = [ self.account_name_for(task), task.item_id, task.new_title or "", os.path.basename(task.new_cover_path or ""), self._stage_text(task), self._result_text(task), ] return values[column] if 0 <= column < len(values) else None def _stage_text(self, task): if task.stage == "generated": return "待更新" if task.stage == "applied": return "已更新" return task.stage def _result_text(self, task): if task.status == "success" and task.stage == "generated": return "待更新" return self.STATUS_TEXT.get(task.status, task.status) class GenerateTab(QWidget): """Tab 2: prompt area plus generation task filters/list.""" STATUS_FILTERS = [ ("全部状态", "all"), ("待生成", "to_generate"), ("已生成", "generated"), ("失败", "failed"), ("略过", "skipped"), ("已更新", "applied"), ] def __init__( self, parent=None, db_path=None, config=None, status_callback=None, title_prompt_path=None, cover_prompts_dir=None, ): super().__init__(parent) self.config = appconfig.load_config() if config is None else config self.db_path = _database_path(db_path, self.config) self.status_callback = status_callback self.title_prompt_path = title_prompt_path or prompts.TITLE_PROMPT_PATH self.cover_prompts_dir = cover_prompts_dir or prompts.COVER_PROMPTS_DIR self.current_cover_template = None self.generate_worker = None self.generate_thread = None 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.save_title_button = QPushButton("保存标题提示词") 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_as_button = QPushButton("另存为") self.rename_cover_template_button = QPushButton("重命名") self.delete_cover_template_button = QPushButton("删除") self.insert_title_button = QPushButton("插入标题") self.preview_prompt_button = QPushButton("预览") self.generate_button = QPushButton("开始生成") self.stop_generate_button = QPushButton("停止") self.stop_generate_button.setEnabled(False) self.progress_label = QLabel("进度：标题0/0 · 封面0/0 · 失败0") left_panel = QWidget() left_layout = QVBoxLayout(left_panel) left_layout.setContentsMargins(0, 0, 12, 0) left_layout.addWidget(QLabel("标题提示词")) left_layout.addWidget(self.title_prompt_edit, 1) left_layout.addWidget(self.save_title_button) left_layout.addWidget(QLabel("封面提示词")) left_layout.addWidget(self.cover_template_combo) cover_template_layout = QHBoxLayout() cover_template_layout.addWidget(self.new_cover_template_button) cover_template_layout.addWidget(self.save_cover_template_button) cover_template_layout.addWidget(self.save_cover_template_as_button) cover_template_layout.addWidget(self.rename_cover_template_button) cover_template_layout.addWidget(self.delete_cover_template_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.status_filter = QComboBox() self.status_filter.setObjectName("statusFilter") for label, value in self.STATUS_FILTERS: self.status_filter.addItem(label, value) self.refresh_button = QPushButton("刷新") 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("状态")) filter_layout.addWidget(self.status_filter, 1) filter_layout.addWidget(self.refresh_button) self.summary_label = QLabel("任务 0 条") self.task_table = QTableView() self.model = GenerateTaskTableModel(self.task_table) self.task_table.setModel(self.model) self.task_table.setSelectionBehavior(QAbstractItemView.SelectRows) self.task_table.setSelectionMode(QAbstractItemView.SingleSelection) self.task_table.setEditTriggers(QAbstractItemView.NoEditTriggers) self.task_table.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch) self.task_table.verticalHeader().setVisible(False) 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.task_table, 1) 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]) bottom_layout = QHBoxLayout() bottom_layout.addWidget(self.progress_label) bottom_layout.addStretch(1) bottom_layout.addWidget(self.generate_button) bottom_layout.addWidget(self.stop_generate_button) 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.status_filter.currentIndexChanged.connect(self.refresh_tasks) self.refresh_button.clicked.connect(self.refresh_tasks) self.save_title_button.clicked.connect(self.save_title_prompt) 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_button.clicked.connect(self.save_cover_template_as) self.rename_cover_template_button.clicked.connect(self.rename_cover_template) self.delete_cover_template_button.clicked.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_button.clicked.connect(self.start_generate) self.stop_generate_button.clicked.connect(self.stop_generate) self.task_table.doubleClicked.connect(self.show_task_images) self.refresh_cover_templates() self.refresh_tasks() def _set_status(self, message): if self.status_callback is not None: self.status_callback(message) 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_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 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 tasks = [ task for task in self.model.tasks if getattr(task, "stage", None) == "collected" ] if not tasks: self._set_status("当前筛选结果没有可生成任务") return 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, ) worker.progress.connect(self._on_generate_progress) worker.row_updated.connect(self._on_generate_row_updated) worker.log.connect(self._set_status) worker.failed.connect(self._on_generate_failed) worker.finished.connect(self._on_generate_finished) worker.cancelled.connect(self._on_generate_cancelled) 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_done": 0, "cover_done": 0, "failed": 0} ) self._set_status(f"开始 AI 生成：{len(tasks)} 条") thread.start() def stop_generate(self, checked=False): if self.generate_worker is not None: self.generate_worker.cancel() self._set_status("正在停止 AI 生成...") def show_task_images(self, index): task = self.model.task_at(index.row()) if index.isValid() else self._selected_task() if task is None: self._set_status("没有可预览的任务") return dialog = QDialog(self) dialog.setWindowTitle(f"封面对照：{task.item_id}") layout = QVBoxLayout(dialog) images_layout = QHBoxLayout() images_layout.addWidget(self._image_panel("旧封面", task.old_cover_path)) images_layout.addWidget(self._image_panel("新封面", task.new_cover_path)) layout.addLayout(images_layout) buttons = QDialogButtonBox(QDialogButtonBox.Close) buttons.rejected.connect(dialog.reject) layout.addWidget(buttons) dialog.resize(720, 420) dialog.exec() def _image_panel(self, title, path): panel = QWidget() layout = QVBoxLayout(panel) layout.addWidget(QLabel(title)) image_label = QLabel() image_label.setAlignment(Qt.AlignCenter) image_label.setMinimumSize(260, 260) image_label.setWordWrap(True) if path and os.path.exists(str(path)): pixmap = QPixmap(str(path)) if not pixmap.isNull(): image_label.setPixmap( pixmap.scaled( 260, 260, Qt.KeepAspectRatio, Qt.SmoothTransformation, ) ) else: image_label.setText(f"图片无法读取\n{path}") else: image_label.setText(f"无图片\n{path or ''}".strip()) layout.addWidget(image_label, 1) return panel def _set_generate_running(self, running): self.generate_button.setEnabled(not running) self.stop_generate_button.setEnabled(running) self.refresh_button.setEnabled(not running) self.batch_filter.setEnabled(not running) self.shop_filter.setEnabled(not running) self.status_filter.setEnabled(not running) self.save_title_button.setEnabled(not running) self.save_cover_template_button.setEnabled(not running) self.save_cover_template_as_button.setEnabled(not running) self.rename_cover_template_button.setEnabled(not running) self.delete_cover_template_button.setEnabled(not running) 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._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) if payload.get("error"): self._set_status(f"AI 生成失败：{payload.get('error')}") return self._set_status("AI 生成完成：" + self._generate_progress_text(payload)) def _on_generate_cancelled(self, payload): self._set_generate_running(False) self.refresh_tasks() self._update_generate_progress(payload) self._set_status("AI 生成已停止：" + self._generate_progress_text(payload)) def _update_generate_progress(self, payload): self.progress_label.setText("进度：" + self._generate_progress_text(payload)) def _generate_progress_text(self, payload): return "标题{title}/{total} · 封面{cover}/{total} · 失败{failed}".format( title=payload.get("title_done", 0), cover=payload.get("cover_done", 0), total=payload.get("total", 0), failed=payload.get("failed", 0), ) 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): return { "old_title": task.old_title, "new_title": task.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" 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_status(task, selected_status) ] except Exception as exc: self.model.set_tasks([], []) self.summary_label.setText("任务读取失败") self._set_status(f"AI 生成任务读取失败：{exc}") return self.model.set_tasks(filtered_tasks, accounts_rows) self.summary_label.setText( f"任务 {len(filtered_tasks)}/{len(batch_tasks)} 条" ) def _populate_batch_filter(self, batches, selected_batch): previous = selected_batch if selected_batch in {batch.id for batch in batches} else None 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) index = self.batch_filter.findData(previous) self.batch_filter.setCurrentIndex(index if index >= 0 else 0) 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_status(self, task, selected_status): if selected_status in (None, "all"): return True if selected_status == "to_generate": return task.stage == "collected" and task.status in {"success", "pending"} 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 class ApplyTab(QWidget): """Tab 3: list generated tasks and confirm the update scope.""" STATUS_FILTERS = [ ("已生成", "generated"), ("失败", "failed"), ("已更新", "applied"), ("略过", "skipped"), ("全部状态", "all"), ] def __init__( self, parent=None, db_path=None, config=None, status_callback=None, open_accounts_callback=None, ): super().__init__(parent) self.config = appconfig.load_config() if config is None else config self.db_path = _database_path(db_path, self.config) self.status_callback = status_callback self.open_accounts_callback = open_accounts_callback self.apply_worker = None self.apply_thread = None self.result_write_back_worker = None self.result_write_back_thread = None self.last_apply_summary = None self.batch_filter = QComboBox() self.batch_filter.setObjectName("applyBatchFilter") self.shop_filter = QComboBox() self.shop_filter.setObjectName("applyShopFilter") self.status_filter = QComboBox() self.status_filter.setObjectName("applyStatusFilter") for label, value in self.STATUS_FILTERS: self.status_filter.addItem(label, value) self.refresh_button = QPushButton("刷新") 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("状态")) filter_layout.addWidget(self.status_filter, 1) filter_layout.addWidget(self.refresh_button) self.summary_label = QLabel("任务 0 条") self.risk_label = QLabel("点击「开始更新」后会先确认当前筛选范围；确认后才允许后续任务提交线上。") self.task_table = QTableView() self.model = ApplyTaskTableModel(self.task_table) self.task_table.setModel(self.model) self.task_table.setSelectionBehavior(QAbstractItemView.SelectRows) self.task_table.setSelectionMode(QAbstractItemView.SingleSelection) self.task_table.setEditTriggers(QAbstractItemView.NoEditTriggers) self.task_table.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch) self.task_table.verticalHeader().setVisible(False) self.start_update_button = QPushButton("开始更新") self.stop_update_button = QPushButton("停止") self.write_back_button = QPushButton("回写结果到 Excel") self.stop_update_button.setEnabled(False) self.write_back_button.setEnabled(False) action_layout = QHBoxLayout() action_layout.addWidget(self.start_update_button) action_layout.addWidget(self.stop_update_button) action_layout.addStretch(1) action_layout.addWidget(self.write_back_button) layout = QVBoxLayout(self) layout.setContentsMargins(18, 18, 18, 18) layout.addLayout(filter_layout) layout.addWidget(self.risk_label) layout.addWidget(self.summary_label) layout.addWidget(self.task_table, 1) layout.addLayout(action_layout) self.batch_filter.currentIndexChanged.connect(self.refresh_tasks) self.shop_filter.currentIndexChanged.connect(self.refresh_tasks) self.status_filter.currentIndexChanged.connect(self.refresh_tasks) self.refresh_button.clicked.connect(self.refresh_tasks) self.start_update_button.clicked.connect(self.start_update) self.stop_update_button.clicked.connect(self.stop_update) self.write_back_button.clicked.connect(self.write_back_results) self.refresh_tasks() def _set_status(self, message): if self.status_callback is not None: self.status_callback(message) def refresh_tasks(self, checked=False): try: db.init_db(self.db_path) batches = db.list_batches(path=self.db_path) account_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 "generated" self._populate_batch_filter(batches, selected_batch) selected_batch = self.batch_filter.currentData() batch_tasks = [ task for task in db.list_tasks(batch_id=selected_batch, path=self.db_path) if self._is_update_task(task) ] self._populate_shop_filter(batch_tasks, account_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_status(task, selected_status) ] except Exception as exc: self.model.set_tasks([], []) self.summary_label.setText("任务读取失败") self._set_status(f"更新任务读取失败：{exc}") return self.model.set_tasks(filtered_tasks, account_rows) self.summary_label.setText( f"任务 {len(filtered_tasks)}/{len(batch_tasks)} 条" ) self._update_write_back_button() def start_update(self, checked=False): if self.apply_thread is not None: self._set_status("更新正在进行...") return tasks = [ task for task in self.model.tasks if self._is_actionable_task(task) ] if not tasks: self._set_status("当前筛选结果没有可更新任务") return safety_error = self._update_safety_error(tasks) if safety_error: QMessageBox.warning(self, "更新安全开关", safety_error) self._set_status(safety_error.replace("\n", " ")) return answer = QMessageBox.question( self, "确认开始更新", self._confirmation_message(tasks), QMessageBox.Yes | QMessageBox.No, QMessageBox.No, ) if answer != QMessageBox.Yes: self._set_status("已取消开始更新") return update_cfg = self._shopee_update_config() worker = ApplyWorker( tasks, db_path=self.db_path, config=self.config, close_success_tab=bool(update_cfg.get("close_success_tab", False)), ) worker.progress.connect(self._on_apply_progress) worker.row_updated.connect(self._on_apply_row_updated) worker.log.connect(self._set_status) worker.failed.connect(self._on_apply_failed) worker.finished.connect(self._on_apply_finished) worker.cancelled.connect(self._on_apply_cancelled) thread = run_worker(worker, thread_name="ApplyWorker", start=False) thread.finished.connect(lambda: self._forget_apply_thread(thread)) self.apply_worker = worker self.apply_thread = thread self._set_apply_running(True) self._set_status(f"开始更新：{len(tasks)} 条") thread.start() def stop_update(self, checked=False): if self.apply_worker is not None: self.apply_worker.cancel() self._set_status("正在停止更新...") def write_back_results(self, checked=False): batch_ids = self._active_batch_ids() if not batch_ids: self._set_status("没有可回写结果的批次") return self._start_result_write_back(batch_ids, auto=False) def _is_actionable_task(self, task): return ( getattr(task, "stage", None) == "generated" and getattr(task, "status", None) in {"success", "pending", "failed"} and bool(getattr(task, "new_title", None) or getattr(task, "new_cover_path", None)) ) def _populate_batch_filter(self, batches, selected_batch): previous = selected_batch if selected_batch in {batch.id for batch in batches} else None 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) index = self.batch_filter.findData(previous) self.batch_filter.setCurrentIndex(index if index >= 0 else 0) 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 _status_label(self): return self.status_filter.currentText() or "已生成" def _batch_filter_label(self): return self.batch_filter.currentText() or "全部批次" def _shop_filter_label(self): return self.shop_filter.currentText() or "全部店铺" def _is_update_task(self, task): if task.stage in {"generated", "applied"}: return True return bool((task.new_title or task.new_cover_path) and task.status in {"failed", "skipped"}) def _matches_shop(self, task, selected_shop): return selected_shop is None or str(task.alias).strip() == selected_shop def _matches_status(self, task, selected_status): if selected_status in (None, "all"): return True if selected_status == "generated": return task.stage == "generated" and task.status in {"success", "pending"} if selected_status == "failed": return task.status == "failed" if selected_status == "applied": return task.stage == "applied" if selected_status == "skipped": return task.status == "skipped" return True def _confirmation_message(self, tasks): update_cfg = self._shopee_update_config() cover_text = "允许" if update_cfg.get("allow_cover_update") else "不允许" close_text = "是" if update_cfg.get("close_success_tab") else "否" return ( "即将按当前筛选结果开始更新 Shopee 线上商品。\n\n" f"批次：{self._batch_filter_label()}\n" f"店铺：{self._shop_filter_label()}\n" f"状态：{self._status_label()}\n" f"任务数：{len(tasks)}\n\n" "安全设置：" f"测试商品ID={update_cfg.get('test_item_id') or '未配置'}，" f"封面更新={cover_text}，" f"最大条数={update_cfg.get('max_items_per_run', 1)}，" f"成功后关闭新页={close_text}\n\n" "确认后后续执行会打开商品编辑页、替换标题/允许时替换封面，并点击「更新」提交线上。" ) def _update_safety_error(self, tasks): update_cfg = self._shopee_update_config() if not update_cfg.get("allow_real_submit", False): return "设置未开启「允许真实提交线上商品」，已阻止本次更新。" max_items = max(1, int(update_cfg.get("max_items_per_run", 1) or 1)) if len(tasks) > max_items: return f"当前筛选结果有 {len(tasks)} 条，超过单次最大更新条数 {max_items}。" test_item_id = str(update_cfg.get("test_item_id", "")).strip() if not test_item_id: return "未配置测试商品ID，已阻止真实提交。" mismatched = [ str(getattr(task, "item_id", "")) for task in tasks if str(getattr(task, "item_id", "")) != test_item_id ] if mismatched: shown = "、".join(mismatched[:5]) return f"当前任务包含非测试商品ID：{shown}。只允许更新测试商品 {test_item_id}。" if not update_cfg.get("allow_cover_update", False): cover_tasks = [ str(getattr(task, "item_id", "")) for task in tasks if getattr(task, "new_cover_path", None) ] if cover_tasks: return "设置未开启「允许更新封面」，当前任务包含新封面路径，已阻止本次更新。" return None def _shopee_update_config(self): defaults = appconfig.default_config().get("shopee_update", {}) loaded = self.config.get("shopee_update", {}) if not isinstance(loaded, dict): loaded = {} merged = dict(defaults) merged.update(loaded) return merged def _set_apply_running(self, running): self.start_update_button.setEnabled(not running) self.stop_update_button.setEnabled(running) self.refresh_button.setEnabled(not running) self.batch_filter.setEnabled(not running) self.shop_filter.setEnabled(not running) self.status_filter.setEnabled(not running) self._update_write_back_button() def _set_result_write_back_running(self, running): self.start_update_button.setEnabled(not running) self.refresh_button.setEnabled(not running) self.batch_filter.setEnabled(not running) self.shop_filter.setEnabled(not running) self.status_filter.setEnabled(not running) self.write_back_button.setEnabled(False if running else bool(self._active_batch_ids())) def _forget_apply_thread(self, thread): if self.apply_thread is thread: self.apply_thread = None self.apply_worker = None def _forget_result_write_back_thread(self, thread): if self.result_write_back_thread is thread: self.result_write_back_thread = None self.result_write_back_worker = None self._update_write_back_button() def _on_apply_progress(self, payload): self._set_status("更新进度：" + self._apply_progress_text(payload)) def _on_apply_row_updated(self, task_id, fields): self.refresh_tasks() def _on_apply_failed(self, task_id, error): self._set_status(f"任务 {task_id} 更新失败：{error}") def _on_apply_finished(self, payload): self._set_apply_running(False) self.refresh_tasks() if payload.get("blocked"): self._show_apply_blocked(payload) return self.last_apply_summary = dict(payload) message = "更新完成：" + self._apply_progress_text(payload) batch_ids = payload.get("batch_ids") or self._active_batch_ids() if payload.get("done", 0) > 0 and batch_ids: if self._start_result_write_back( batch_ids, auto=True, apply_summary=payload, ): self._set_status(f"{message}，正在自动回写结果到 Excel...") return self._set_status(message) self._show_apply_summary(payload) def _on_apply_cancelled(self, payload): self._set_apply_running(False) self.refresh_tasks() self._set_status("更新已停止：" + self._apply_progress_text(payload)) def _apply_progress_text(self, payload): return "完成{done}/{total}，成功{applied}，略过{skipped}，失败{failed}".format( done=payload.get("done", 0), total=payload.get("total", 0), applied=payload.get("applied", 0), skipped=payload.get("skipped", 0), failed=payload.get("failed", 0), ) def _show_apply_blocked(self, payload): lines = ["更新前检查未通过。"] if payload.get("no_accounts"): lines.append("当前没有配置账号。") not_running = payload.get("not_running") or [] if not_running: lines.append( "以下账号 Chrome 未启动或调试端口不可访问：" + "、".join(self._account_label(item) for item in not_running) ) logged_out = payload.get("logged_out") or [] if logged_out: lines.append( "以下账号未登录 Shopee：" + "、".join(self._account_label(item) for item in logged_out) ) self._show_account_guide("\n".join(lines)) def _show_account_guide(self, message): full_message = ( f"{message}\n\n" "请先到「④ 账号管理」配置账号、启动对应账号 Chrome，并确认已人工登录 Shopee。" ) QMessageBox.warning(self, "账号未就绪", full_message) self._set_status(full_message.replace("\n", " ")) if self.open_accounts_callback is not None: self.open_accounts_callback() def _account_label(self, item): if isinstance(item, dict): name = item.get("account_name") or item.get("alias") or "" alias = item.get("alias") or "" reason = item.get("reason") else: name = getattr(item, "account_name", "") or getattr(item, "alias", "") alias = getattr(item, "alias", "") reason = getattr(item, "reason", None) label = f"{name}({alias})" if alias and name != alias else (name or alias) return f"{label}: {reason}" if reason else label def _active_batch_ids(self): selected_batch = self.batch_filter.currentData() if selected_batch: return [selected_batch] batch_ids = [] for task in self.model.tasks: batch_id = getattr(task, "batch_id", None) if batch_id and batch_id not in batch_ids: batch_ids.append(batch_id) return batch_ids def _update_write_back_button(self): if getattr(self, "write_back_button", None) is None: return enabled = ( self.apply_thread is None and self.result_write_back_thread is None and bool(self._active_batch_ids()) ) self.write_back_button.setEnabled(enabled) def _start_result_write_back(self, batch_ids, auto=False, apply_summary=None): if self.result_write_back_thread is not None: self._set_status("Excel 结果回写正在进行...") return False worker = WriteBackWorker(batch_ids, db_path=self.db_path, mode="results") worker.failed.connect( lambda task_id, error, auto=auto, apply_summary=apply_summary: self._on_result_write_back_failed( task_id, error, auto=auto, apply_summary=apply_summary, ) ) worker.finished.connect( lambda payload, auto=auto, apply_summary=apply_summary: self._on_result_write_back_finished( payload, auto=auto, apply_summary=apply_summary, ) ) thread = run_worker(worker, thread_name="ResultWriteBackWorker", start=False) thread.finished.connect(lambda: self._forget_result_write_back_thread(thread)) self.result_write_back_worker = worker self.result_write_back_thread = thread self._set_result_write_back_running(True) self._set_status("正在自动回写更新结果到 Excel..." if auto else "正在回写更新结果到 Excel...") thread.start() return True def _on_result_write_back_failed(self, task_id, error, auto=False, apply_summary=None): message = f"Excel {'自动' if auto else ''}回写更新结果失败：{error}" if "被占用" in str(error): message += "\n请关闭原 Excel 后点击「回写结果到 Excel」手动重试；SQLite 已保留更新结果。" if auto and apply_summary: message = self._apply_summary_message(apply_summary, error=message) QMessageBox.warning(self, "回写结果到 Excel", message) self._set_status(message.replace("\n", " ")) def _on_result_write_back_finished(self, payload, auto=False, apply_summary=None): self._set_result_write_back_running(False) self.refresh_tasks() if payload.get("ok") is False: error = payload.get("error") or "未知错误" retry_hint = "，可点击「回写结果到 Excel」手动重试" if auto else "" self._set_status(f"Excel {'自动' if auto else ''}回写更新结果失败：{error}{retry_hint}") return self._set_status( "Excel {prefix}回写更新结果完成：文件{files}，行{rows}".format( prefix="自动" if auto else "", files=payload.get("files", 0), rows=payload.get("rows", 0), ) ) if auto and apply_summary: self._show_apply_summary(apply_summary, write_back_payload=payload) elif not auto: QMessageBox.information( self, "回写结果到 Excel", "结果回写完成：文件{files}，行{rows}".format( files=payload.get("files", 0), rows=payload.get("rows", 0), ), ) def _show_apply_summary(self, apply_summary, write_back_payload=None): QMessageBox.information( self, "更新完成", self._apply_summary_message(apply_summary, write_back_payload), ) def _apply_summary_message(self, apply_summary, write_back_payload=None, error=None): lines = [ "更新完成。", "成功：{applied}，失败：{failed}，略过：{skipped}".format( applied=apply_summary.get("applied", 0), failed=apply_summary.get("failed", 0), skipped=apply_summary.get("skipped", 0), ), ] if write_back_payload: lines.append( "Excel 回写：文件{files}，行{rows}".format( files=write_back_payload.get("files", 0), rows=write_back_payload.get("rows", 0), ) ) if error: lines.append(str(error)) return "\n".join(lines) class CollectTab(QWidget): """Tab 1: import Excel files and list imported tasks.""" def __init__( self, parent=None, db_path=None, config=None, status_callback=None, open_accounts_callback=None, ): super().__init__(parent) self.config = appconfig.load_config() if config is None else config self.db_path = _database_path(db_path, self.config) self.status_callback = status_callback self.open_accounts_callback = open_accounts_callback self.current_batch_id = None self.has_import_result = False self.last_import_stats = None self.collect_worker = None self.collect_thread = None self.write_back_worker = None self.write_back_thread = None self.import_button = QPushButton("导入 Excel...") self.refresh_button = QPushButton("刷新") self.collect_button = QPushButton("采集旧标题/旧封面") self.stop_collect_button = QPushButton("停止") self.write_back_button = QPushButton("回写旧数据到 Excel") self.stop_collect_button.setEnabled(False) toolbar = QHBoxLayout() toolbar.addWidget(self.import_button) toolbar.addWidget(self.refresh_button) toolbar.addWidget(self.collect_button) toolbar.addWidget(self.stop_collect_button) toolbar.addWidget(self.write_back_button) toolbar.addStretch(1) self.summary_label = QLabel("未导入任务") self.match_detail_label = QLabel("") self.show_all_button = QPushButton("全部") self.show_unmatched_button = QPushButton("未匹配(0)") summary_layout = QHBoxLayout() summary_layout.addWidget(self.summary_label) summary_layout.addStretch(1) summary_layout.addWidget(self.show_all_button) summary_layout.addWidget(self.show_unmatched_button) self.table = QTableView() self.model = TaskTableModel(self.table) self.table.setModel(self.model) self.table.setSelectionBehavior(QAbstractItemView.SelectRows) self.table.setSelectionMode(QAbstractItemView.SingleSelection) self.table.setEditTriggers(QAbstractItemView.NoEditTriggers) self.table.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch) self.table.verticalHeader().setVisible(False) self.empty_label = QLabel("") layout = QVBoxLayout(self) layout.setContentsMargins(18, 18, 18, 18) layout.addLayout(toolbar) layout.addLayout(summary_layout) layout.addWidget(self.match_detail_label) layout.addWidget(self.table, 1) layout.addWidget(self.empty_label) self.import_button.clicked.connect(self.import_excel) self.refresh_button.clicked.connect(self.refresh_tasks) self.collect_button.clicked.connect(self.collect_old_data) self.stop_collect_button.clicked.connect(self.stop_collect) self.write_back_button.clicked.connect(self.write_back_old_data) self.show_all_button.clicked.connect(self.show_all_tasks) self.show_unmatched_button.clicked.connect(self.show_unmatched_tasks) self.refresh_tasks() def _set_status(self, message): if self.status_callback is not None: self.status_callback(message) def _show_error(self, message): QMessageBox.warning(self, "导入采集", str(message)) self._set_status(str(message)) def _show_account_guide(self, message): full_message = ( f"{message}\n\n" "请先到「④ 账号管理」配置账号、启动对应账号 Chrome，并确认已人工登录 Shopee。" ) QMessageBox.warning(self, "账号未就绪", full_message) self._set_status(full_message.replace("\n", " ")) if self.open_accounts_callback is not None: self.open_accounts_callback() def _choose_excel_files(self): files, _selected_filter = QFileDialog.getOpenFileNames( self, "选择 Excel 文件", "", "Excel 文件 (*.xlsx *.xlsm)", ) return files def import_excel(self, checked=False): file_paths = self._choose_excel_files() if not file_paths: return try: result = excel.import_tasks(file_paths, path=self.db_path) except Exception as exc: self._show_error(exc) return self.has_import_result = True self.last_import_stats = result.get("stats") or {} self.current_batch_id = result.get("batch_id") self.refresh_tasks() self._set_status( "导入完成：有效{valid}，无效{invalid}，入库{inserted}，未匹配{unmatched}".format( valid=self.last_import_stats.get("valid", 0), invalid=self.last_import_stats.get("invalid", 0), inserted=self.last_import_stats.get("inserted", 0), unmatched=self.model.unmatched_count(), ) ) def refresh_tasks(self, checked=False): try: db.init_db(self.db_path) if self.current_batch_id is None and self.has_import_result: task_rows = [] else: task_rows = db.list_tasks(batch_id=self.current_batch_id, path=self.db_path) account_rows = accounts.list_accounts(path=self.db_path, config=self.config) except Exception as exc: self.model.set_tasks([], []) self.empty_label.setText("任务读取失败") self._set_status(f"任务读取失败：{exc}") return self.model.set_tasks(task_rows, account_rows) self._update_summary(task_rows, account_rows) self._update_empty_label(len(task_rows)) def collect_old_data(self, checked=False): tasks = list(self.model.all_tasks) if not tasks: self._set_status("没有可采集任务") return worker = CollectWorker(tasks, db_path=self.db_path, config=self.config) worker.progress.connect(self._on_collect_progress) worker.row_updated.connect(self._on_collect_row_updated) worker.log.connect(self._set_status) worker.failed.connect(self._on_collect_failed) worker.finished.connect(self._on_collect_finished) worker.cancelled.connect(self._on_collect_cancelled) thread = run_worker(worker, thread_name="CollectWorker", start=False) thread.finished.connect(lambda: self._forget_collect_thread(thread)) self.collect_worker = worker self.collect_thread = thread self._set_collect_running(True) thread.start() def stop_collect(self, checked=False): if self.collect_worker is not None: self.collect_worker.cancel() self._set_status("正在停止采集...") def write_back_old_data(self, checked=False): batch_id = self._active_batch_id() if not batch_id: self._set_status("没有可回写批次") return self._start_write_back(batch_id) def _start_write_back(self, batch_id, auto=False): if self.write_back_thread is not None: self._set_status("Excel 回写正在进行...") return False worker = WriteBackWorker(batch_id, db_path=self.db_path) worker.failed.connect( lambda task_id, error, auto=auto: self._on_write_back_failed( task_id, error, auto=auto, ) ) worker.finished.connect( lambda payload, auto=auto: self._on_write_back_finished( payload, auto=auto, ) ) thread = run_worker(worker, thread_name="WriteBackWorker", start=False) thread.finished.connect(lambda: self._forget_write_back_thread(thread)) self.write_back_worker = worker self.write_back_thread = thread self._set_write_back_running(True) self._set_status("正在自动回写旧数据到 Excel..." if auto else "正在回写旧数据到 Excel...") thread.start() return True def _active_batch_id(self): if self.current_batch_id: return self.current_batch_id batch_ids = { task.batch_id for task in self.model.all_tasks if getattr(task, "batch_id", None) } if len(batch_ids) == 1: return next(iter(batch_ids)) return None def _set_collect_running(self, running): self.import_button.setEnabled(not running) self.refresh_button.setEnabled(not running) self.collect_button.setEnabled(not running) self.write_back_button.setEnabled(not running) self.stop_collect_button.setEnabled(running) def _set_write_back_running(self, running): self.import_button.setEnabled(not running) self.refresh_button.setEnabled(not running) self.collect_button.setEnabled(not running) self.write_back_button.setEnabled(not running) def _forget_collect_thread(self, thread): if self.collect_thread is thread: self.collect_thread = None self.collect_worker = None def _forget_write_back_thread(self, thread): if self.write_back_thread is thread: self.write_back_thread = None self.write_back_worker = None def _on_collect_progress(self, payload): self._set_status( "采集进度：{done}/{total}，成功{collected}，略过{skipped}，失败{failed}".format( done=payload.get("done", 0), total=payload.get("total", 0), collected=payload.get("collected", 0), skipped=payload.get("skipped", 0), failed=payload.get("failed", 0), ) ) def _on_collect_row_updated(self, task_id, fields): self.refresh_tasks() def _on_collect_failed(self, task_id, error): self._set_status(f"任务 {task_id} 采集失败：{error}") def _on_collect_finished(self, payload): self._set_collect_running(False) self.refresh_tasks() if payload.get("blocked"): self._show_collect_blocked(payload) return message = "采集完成：成功{collected}，略过{skipped}，失败{failed}".format( collected=payload.get("collected", 0), skipped=payload.get("skipped", 0), failed=payload.get("failed", 0), ) if payload.get("collected", 0) > 0: batch_id = self._active_batch_id() if batch_id and self._start_write_back(batch_id, auto=True): self._set_status(f"{message}，正在自动回写 Excel...") return if not batch_id: self._set_status(f"{message}，但没有可回写批次") return self._set_status(f"{message}，Excel 回写已在进行") return self._set_status(message) def _show_collect_blocked(self, payload): lines = ["采集前检查未通过。"] if payload.get("no_accounts"): lines.append("当前没有配置账号。") not_running = payload.get("not_running") or [] if not_running: lines.append( "以下账号 Chrome 未启动或调试端口不可访问：" + "、".join(self._account_label(item) for item in not_running) ) logged_out = payload.get("logged_out") or [] if logged_out: lines.append( "以下账号未登录 Shopee：" + "、".join(self._account_label(item) for item in logged_out) ) self._show_account_guide("\n".join(lines)) def _account_label(self, item): if isinstance(item, dict): name = item.get("account_name") or item.get("alias") or "" alias = item.get("alias") or "" reason = item.get("reason") else: name = getattr(item, "account_name", "") or getattr(item, "alias", "") alias = getattr(item, "alias", "") reason = getattr(item, "reason", None) label = f"{name}({alias})" if alias and name != alias else (name or alias) return f"{label}: {reason}" if reason else label def _on_collect_cancelled(self, payload): self._set_collect_running(False) self.refresh_tasks() self._set_status( "采集已停止：完成{done}/{total}".format( done=payload.get("done", 0), total=payload.get("total", 0), ) ) def _on_write_back_failed(self, task_id, error, auto=False): message = f"Excel {'自动' if auto else ''}回写失败：{error}" if "被占用" in str(error): if auto: message += "\n请关闭原 Excel 后点击「回写旧数据到 Excel」手动重试；SQLite 已保留采集结果，也可另存副本。" else: message += "\n请关闭原 Excel 后重试；SQLite 已保留采集结果，也可另存副本。" QMessageBox.warning(self, "回写旧数据", message) self._set_status(message.replace("\n", " ")) def _on_write_back_finished(self, payload, auto=False): self._set_write_back_running(False) if payload.get("ok") is False: error = payload.get("error") or "未知错误" retry_hint = "，可点击「回写旧数据到 Excel」手动重试" if auto else "" self._set_status(f"Excel {'自动' if auto else ''}回写失败：{error}{retry_hint}") return self.refresh_tasks() self._set_status( "Excel {prefix}回写完成：文件{files}，行{rows}".format( prefix="自动" if auto else "", files=payload.get("files", 0), rows=payload.get("rows", 0), ) ) def show_all_tasks(self, checked=False): self.model.set_filter_mode("all") self._update_empty_label(len(self.model.all_tasks)) def show_unmatched_tasks(self, checked=False): self.model.set_filter_mode("unmatched") self._update_empty_label(len(self.model.all_tasks)) def _update_summary(self, task_rows, account_rows): stats = self.last_import_stats or {} unmatched = self._unmatched_count(task_rows, account_rows) matched = len(task_rows) - unmatched files = stats.get("files", 0 if not task_rows else 1) total = stats.get("total", len(task_rows)) valid = stats.get("valid", len(task_rows)) invalid = stats.get("invalid", 0) self.summary_label.setText( f"{files} 文件 · {total} 行 · 有效{valid}/无效{invalid} · 匹配{matched} · 未匹配{unmatched}" ) self.match_detail_label.setText(self._match_detail(task_rows, account_rows)) self.show_unmatched_button.setText(f"未匹配({unmatched})") self.show_unmatched_button.setEnabled(unmatched > 0) if unmatched == 0 and self.model.filter_mode == "unmatched": self.model.set_filter_mode("all") def _match_detail(self, task_rows, account_rows): account_by_alias = { str(account.alias).strip(): account for account in account_rows if str(account.alias).strip() } counts = {} for task in task_rows: account = account_by_alias.get(str(task.alias).strip()) if account is None: continue name = account.account_name or account.alias counts[name] = counts.get(name, 0) + 1 if not counts: return "匹配明细：无" parts = [f"{name}{count}" for name, count in sorted(counts.items())] return "匹配明细：" + " · ".join(parts) def _unmatched_count(self, task_rows, account_rows): aliases = { str(account.alias).strip() for account in account_rows if str(account.alias).strip() } return sum(1 for task in task_rows if str(task.alias).strip() not in aliases) def _update_empty_label(self, total_rows): if total_rows == 0: self.empty_label.setText("暂无任务") return if self.model.rowCount() == 0 and self.model.filter_mode == "unmatched": self.empty_label.setText("没有未匹配任务") return unmatched = self.model.unmatched_count() self.empty_label.setText( "" if unmatched == 0 else f"{unmatched} 条任务别名未匹配账号，阶段显示为“略过”" ) class AccountDialog(QDialog): """Dialog for adding or editing one account.""" def __init__(self, parent=None, account=None, default_port=9222, config=None): super().__init__(parent) self._account = account self._config = config self.setWindowTitle("编辑账号" if account else "新增账号") self.account_name_edit = QLineEdit() self.alias_edit = QLineEdit() self.region_host_edit = QLineEdit(accounts.DEFAULT_REGION_HOST) self.debug_port_spin = QSpinBox() self.debug_port_spin.setRange(1, 65535) self.debug_port_spin.setValue(int(default_port)) self.password_edit = QLineEdit() self.password_edit.setEchoMode(QLineEdit.Password) self.note_edit = QPlainTextEdit() self.note_edit.setMaximumHeight(76) self.slug_edit = QLineEdit() self.slug_edit.setReadOnly(True) self.user_data_dir_edit = QLineEdit() self.user_data_dir_edit.setReadOnly(True) if account is not None: self.account_name_edit.setText(account.account_name) self.alias_edit.setText(account.alias) self.region_host_edit.setText(account.region_host) self.debug_port_spin.setValue(int(account.debug_port)) self.password_edit.setText(account.password or "") self.note_edit.setPlainText(account.note or "") self.slug_edit.setText(account.slug) self.user_data_dir_edit.setText(account.user_data_dir) form = QFormLayout() form.addRow("账号名", self.account_name_edit) form.addRow("别名", self.alias_edit) form.addRow("地区", self.region_host_edit) form.addRow("调试端口", self.debug_port_spin) form.addRow("密码", self.password_edit) form.addRow("备注", self.note_edit) form.addRow("slug", self.slug_edit) form.addRow("数据目录", self.user_data_dir_edit) buttons = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel) buttons.accepted.connect(self.accept) buttons.rejected.connect(self.reject) layout = QVBoxLayout(self) layout.addLayout(form) layout.addWidget(buttons) self.alias_edit.textChanged.connect(self._update_path_preview) self._update_path_preview() def _update_path_preview(self): alias = self.alias_edit.text().strip() if not alias: self.slug_edit.clear() self.user_data_dir_edit.clear() return try: slug = account_config.make_slug(alias) self.slug_edit.setText(slug) if self._account is not None and alias == self._account.alias: self.user_data_dir_edit.setText(self._account.user_data_dir) else: self.user_data_dir_edit.setText( accounts.preview_user_data_dir(alias, config=self._config) ) except Exception: self.slug_edit.clear() self.user_data_dir_edit.clear() def values(self): return { "account_name": self.account_name_edit.text().strip(), "alias": self.alias_edit.text().strip(), "region_host": self.region_host_edit.text().strip(), "debug_port": self.debug_port_spin.value(), "password": self.password_edit.text(), "note": self.note_edit.toPlainText().strip(), } from .workers import BaseWorker, run_worker class GenerateWorker(BaseWorker): """Generate titles and covers for collected tasks.""" def __init__(self, tasks, prompt_values, db_path=None, config=None): super().__init__() self.tasks = list(tasks) self.prompt_values = dict(prompt_values or {}) self.db_path = db_path self.config = config def execute(self): account_rows = accounts.list_accounts(path=self.db_path, config=self.config) account_by_alias = { str(account.alias).strip(): account for account in account_rows if str(account.alias).strip() } return ai.generate_batch( self.tasks, self.prompt_values, ai_cfg={ "config": self.config, "db_path": self.db_path, "image_dir": appconfig.image_dir(self.config), "account_by_alias": account_by_alias, "on_task_update": self._emit_row_update, }, on_progress=self.progress.emit, should_stop=self.should_cancel, ) def _emit_row_update(self, task_id, fields): self.row_updated.emit(int(task_id), dict(fields or {})) class ApplyWorker(BaseWorker): """Apply generated title/cover changes to Shopee one task at a time.""" def __init__( self, tasks, db_path=None, config=None, preflight=True, close_success_tab=False, ): super().__init__() self.tasks = list(tasks) self.db_path = db_path self.config = config self.preflight = preflight self.close_success_tab = close_success_tab def execute(self): account_rows = accounts.list_accounts(path=self.db_path, config=self.config) account_by_alias = { str(account.alias).strip(): account for account in account_rows if str(account.alias).strip() } eligible = [task for task in self.tasks if self._is_actionable_task(task)] batch_ids = self._batch_ids(eligible) total = len(eligible) applied = 0 skipped = 0 failed = 0 done = 0 if self.preflight: blocked = self._preflight_block(eligible, account_rows, account_by_alias) if blocked: blocked.update( { "ok": False, "blocked": True, "total": total, "done": 0, "applied": 0, "skipped": 0, "failed": 0, "batch_ids": batch_ids, } ) return blocked for task in eligible: if self.should_cancel(): break account = account_by_alias.get(str(task.alias).strip()) if account is None: skipped += 1 done += 1 reason = "别名未匹配账号" db.mark_skipped(task.id, reason, path=self.db_path) self.row_updated.emit(task.id, {"status": "skipped", "last_error": reason}) self._emit_progress(done, total, applied, skipped, failed) continue try: db.mark_running(task.id, "apply", path=self.db_path) self.row_updated.emit(task.id, {"status": "running", "last_error": None}) result = editor.apply_task( account, task, close_success_tab=self.close_success_tab, ) committed = bool(result.get("committed")) and not result.get("error") error = result.get("error") if committed: db.set_applied(task.id, True, path=self.db_path) applied += 1 self.row_updated.emit( task.id, { "stage": "applied", "status": "success", "committed": 1, "last_error": None, }, ) else: failed += 1 error = error or "更新未提交" db.set_applied(task.id, False, error, path=self.db_path) self.failed.emit(task.id, str(error)) self.row_updated.emit( task.id, {"status": "failed", "last_error": str(error), "committed": 0}, ) except Exception as exc: failed += 1 error = str(exc) or exc.__class__.__name__ db.set_applied(task.id, False, error, path=self.db_path) self.failed.emit(task.id, error) self.row_updated.emit( task.id, {"status": "failed", "last_error": error, "committed": 0}, ) finally: done += 1 self._emit_progress(done, total, applied, skipped, failed) return { "ok": failed == 0, "total": total, "done": done, "applied": applied, "skipped": skipped, "failed": failed, "batch_ids": batch_ids, } def _is_actionable_task(self, task): return ( getattr(task, "stage", None) == "generated" and getattr(task, "status", None) in {"success", "pending", "failed"} and bool(getattr(task, "new_title", None) or getattr(task, "new_cover_path", None)) ) def _preflight_block(self, eligible, account_rows, account_by_alias): if not account_rows: return { "reason": "NO_ACCOUNTS", "no_accounts": True, } required_accounts = [] seen_aliases = set() for task in eligible: alias = str(task.alias).strip() account = account_by_alias.get(alias) if account is not None and alias not in seen_aliases: required_accounts.append(account) seen_aliases.add(alias) not_running = [] logged_out = [] for account in required_accounts: if not chrome.is_running(account.debug_port): not_running.append(self._account_payload(account, "CDP 端口未响应")) continue status = self._login_status(account) if not status.get("logged_in"): logged_out.append( self._account_payload(account, self._login_skip_reason(status)) ) if not_running or logged_out: return { "reason": "ACCOUNT_NOT_READY", "not_running": not_running, "logged_out": logged_out, } return None def _account_payload(self, account, reason=None): payload = { "account_name": account.account_name, "alias": account.alias, "debug_port": account.debug_port, } if reason: payload["reason"] = reason return payload def _emit_progress(self, done, total, applied, skipped, failed): self.progress.emit( { "done": done, "total": total, "applied": applied, "skipped": skipped, "failed": failed, } ) def _login_status(self, account): try: return accounts.detect_login(account, path=self.db_path, config=self.config) except Exception as exc: return { "logged_in": False, "reason": f"LOGIN_CHECK_FAILED: {exc}", } def _login_skip_reason(self, status): reason = status.get("reason") return f"账号未登录: {reason}" if reason else "账号未登录" def _batch_ids(self, tasks): batch_ids = [] for task in tasks: batch_id = getattr(task, "batch_id", None) if batch_id and batch_id not in batch_ids: batch_ids.append(batch_id) return batch_ids class CollectWorker(BaseWorker): """Collect old title and cover for imported tasks.""" def __init__(self, tasks, db_path=None, config=None, preflight=True): super().__init__() self.tasks = list(tasks) self.db_path = db_path self.config = config self.preflight = preflight def execute(self): account_rows = accounts.list_accounts(path=self.db_path, config=self.config) account_by_alias = { str(account.alias).strip(): account for account in account_rows if str(account.alias).strip() } eligible = [ task for task in self.tasks if getattr(task, "stage", None) == "imported" ] total = len(eligible) collected = 0 skipped = 0 failed = 0 done = 0 if self.preflight: blocked = self._preflight_block(eligible, account_rows, account_by_alias) if blocked: blocked.update( { "ok": False, "blocked": True, "total": total, "done": 0, "collected": 0, "skipped": 0, "failed": 0, } ) return blocked for task in eligible: if self.should_cancel(): break account = account_by_alias.get(str(task.alias).strip()) if account is None: skipped += 1 done += 1 reason = "别名未匹配账号" db.mark_skipped(task.id, reason, path=self.db_path) self.row_updated.emit(task.id, {"status": "skipped", "last_error": reason}) self._emit_progress(done, total, collected, skipped, failed) continue status = self._login_status(account) if not status.get("logged_in"): skipped += 1 done += 1 reason = self._login_skip_reason(status) db.mark_skipped(task.id, reason, path=self.db_path) self.row_updated.emit(task.id, {"status": "skipped", "last_error": reason}) self._emit_progress(done, total, collected, skipped, failed) continue try: db.mark_running(task.id, "collect", path=self.db_path) self.row_updated.emit(task.id, {"status": "running"}) result = editor.collect( account, { "item_id": task.item_id, "old_cover_path": self._old_cover_path(account, task), }, ) db.set_collected( task.id, result.get("old_title", ""), result.get("old_cover_path", ""), path=self.db_path, ) collected += 1 self.row_updated.emit( task.id, { "stage": "collected", "status": "success", "old_title": result.get("old_title", ""), "old_cover_path": result.get("old_cover_path", ""), }, ) except Exception as exc: failed += 1 error = str(exc) or exc.__class__.__name__ db.mark_failed(task.id, "collect", error, path=self.db_path) self.failed.emit(task.id, error) self.row_updated.emit(task.id, {"status": "failed", "last_error": error}) finally: done += 1 self._emit_progress(done, total, collected, skipped, failed) return { "ok": failed == 0, "total": total, "done": done, "collected": collected, "skipped": skipped, "failed": failed, } def _preflight_block(self, eligible, account_rows, account_by_alias): if not account_rows: return { "reason": "NO_ACCOUNTS", "no_accounts": True, } required_accounts = [] seen_aliases = set() for task in eligible: alias = str(task.alias).strip() account = account_by_alias.get(alias) if account is not None and alias not in seen_aliases: required_accounts.append(account) seen_aliases.add(alias) not_running = [] logged_out = [] for account in required_accounts: if not chrome.is_running(account.debug_port): not_running.append(self._account_payload(account, "CDP 端口未响应")) continue status = self._login_status(account) if not status.get("logged_in"): logged_out.append( self._account_payload(account, self._login_skip_reason(status)) ) if not_running or logged_out: return { "reason": "ACCOUNT_NOT_READY", "not_running": not_running, "logged_out": logged_out, } return None def _account_payload(self, account, reason=None): payload = { "account_name": account.account_name, "alias": account.alias, "debug_port": account.debug_port, } if reason: payload["reason"] = reason return payload def _emit_progress(self, done, total, collected, skipped, failed): self.progress.emit( { "done": done, "total": total, "collected": collected, "skipped": skipped, "failed": failed, } ) def _login_status(self, account): try: return accounts.detect_login(account, path=self.db_path, config=self.config) except Exception as exc: return { "logged_in": False, "reason": f"LOGIN_CHECK_FAILED: {exc}", } def _login_skip_reason(self, status): reason = status.get("reason") return f"账号未登录: {reason}" if reason else "账号未登录" def _old_cover_path(self, account, task): image_root = appconfig.image_dir(self.config) return os.path.abspath( os.path.join( image_root, account.slug, f"{task.item_id}_old.jpg", ) ) class WriteBackWorker(BaseWorker): """Write Excel fields back in a background thread.""" def __init__(self, batch_id, db_path=None, excel_path=None, mode="old"): super().__init__() self.batch_id = batch_id self.db_path = db_path self.excel_path = excel_path self.mode = mode def execute(self): results = [ self._write_one(batch_id) for batch_id in self._batch_ids() ] result = results[0] if len(results) == 1 else self._combined_result(results) self.progress.emit( { "done": result.get("rows", 0), "total": result.get("rows", 0), "files": result.get("files", 0), } ) return result def _batch_ids(self): if isinstance(self.batch_id, (list, tuple, set)): return list(self.batch_id) return [self.batch_id] def _write_one(self, batch_id): if self.mode == "results": return excel.write_back_results( batch_id, excel_path=self.excel_path, path=self.db_path, ) return excel.write_back( batch_id, excel_path=self.excel_path, path=self.db_path, ) def _combined_result(self, results): written_files = [] for result in results: for file_path in result.get("written_files", []): if file_path not in written_files: written_files.append(file_path) return { "ok": all(result.get("ok", False) for result in results), "batch_id": [result.get("batch_id") for result in results], "files": sum(result.get("files", 0) for result in results), "rows": sum(result.get("rows", 0) for result in results), "written_files": written_files, } class AccountLoginCheckWorker(BaseWorker): def __init__(self, account, db_path=None, config=None, timeout=8): super().__init__() self.account = account self.db_path = db_path self.config = config self.timeout = timeout def execute(self): status = accounts.detect_login( self.account, timeout=self.timeout, path=self.db_path, config=self.config, ) self.row_updated.emit(self.account.id, status) return {"alias": self.account.alias, "status": status} class AIModelTestWorker(BaseWorker): """Test one AI model connection without blocking the GUI thread.""" def __init__(self, model_name, ai_models_path=None): super().__init__() self.model_name = model_name self.ai_models_path = ai_models_path or appconfig.AI_MODELS_PATH def execute(self): result = appconfig.test_ai_model(self.model_name, path=self.ai_models_path) payload = dict(appconfig.sanitize_for_log(result or {})) payload["name"] = self.model_name return payload class SettingsTab(QWidget): """Tab 5: AI model definitions stored in config/ai_models.json.""" CATEGORY_ITEMS = [("文本", "text"), ("图像", "image")] API_TYPE_ITEMS = [("chat", "chat"), ("images_edits", "images_edits"), ("auto", "auto")] RESOLUTION_ITEMS = ["512", "1k", "2k", "4k"] def __init__( self, parent=None, config=None, config_path=None, ai_models_path=None, status_callback=None, ): super().__init__(parent) self.config = appconfig.load_config() if config is None else config self.config_path = ( config_path or self.config.get("config_path") or appconfig.CONFIG_PATH ) self.ai_models_path = ( ai_models_path or self.config.get("ai_models_path") or appconfig.AI_MODELS_PATH ) self.status_callback = status_callback self.models = [] self.current_model_name = None self.test_worker = None self.test_thread = None self.model_combo = QComboBox() self.model_combo.setObjectName("aiModelCombo") self.add_model_button = QPushButton("新增") self.delete_model_button = QPushButton("删除") left_panel = QWidget() left_layout = QVBoxLayout(left_panel) left_layout.setContentsMargins(0, 0, 12, 0) left_layout.addWidget(QLabel("AI 模型")) left_layout.addWidget(self.model_combo) left_toolbar = QHBoxLayout() left_toolbar.addWidget(self.add_model_button) left_toolbar.addWidget(self.delete_model_button) left_toolbar.addStretch(1) left_layout.addLayout(left_toolbar) left_layout.addStretch(1) self.enabled_checkbox = QCheckBox("启用") self.name_edit = QLineEdit() self.name_edit.setObjectName("modelNameEdit") self.category_combo = QComboBox() self.category_combo.setObjectName("modelCategoryCombo") for label, value in self.CATEGORY_ITEMS: self.category_combo.addItem(label, value) self.api_type_combo = QComboBox() self.api_type_combo.setObjectName("modelApiTypeCombo") for label, value in self.API_TYPE_ITEMS: self.api_type_combo.addItem(label, value) self.model_id_edit = QLineEdit() self.model_id_edit.setObjectName("modelIdEdit") self.url_edit = QLineEdit() self.url_edit.setObjectName("modelUrlEdit") self.api_key_edit = QLineEdit() self.api_key_edit.setObjectName("modelApiKeyEdit") self.api_key_edit.setEchoMode(QLineEdit.Password) self.connect_timeout_spin = QSpinBox() self.connect_timeout_spin.setObjectName("connectTimeoutSpin") self.connect_timeout_spin.setRange(1, 3600) self.connect_timeout_spin.setValue(30) self.save_model_button = QPushButton("保存") self.test_connection_button = QPushButton("测试连接") self.test_result_label = QLabel("") self.test_result_label.setWordWrap(True) self.default_text_model_combo = QComboBox() self.default_text_model_combo.setObjectName("defaultTextModelCombo") self.default_image_model_combo = QComboBox() self.default_image_model_combo.setObjectName("defaultImageModelCombo") self.title_concurrency_spin = QSpinBox() self.title_concurrency_spin.setObjectName("titleConcurrencySpin") self.title_concurrency_spin.setRange(1, 64) self.image_concurrency_spin = QSpinBox() self.image_concurrency_spin.setObjectName("imageConcurrencySpin") self.image_concurrency_spin.setRange(1, 64) self.retry_spin = QSpinBox() self.retry_spin.setObjectName("retrySpin") self.retry_spin.setRange(0, 20) self.resolution_combo = QComboBox() self.resolution_combo.setObjectName("resolutionCombo") for resolution in self.RESOLUTION_ITEMS: self.resolution_combo.addItem(resolution, resolution) self.response_timeout_label = QLabel("") self.jpg_quality_spin = QSpinBox() self.jpg_quality_spin.setObjectName("jpgQualitySpin") self.jpg_quality_spin.setRange(1, 100) self.chrome_path_edit = QLineEdit() self.chrome_path_edit.setObjectName("chromePathEdit") self.user_data_root_edit = QLineEdit() self.user_data_root_edit.setObjectName("userDataRootEdit") self.image_dir_edit = QLineEdit() self.image_dir_edit.setObjectName("imageDirEdit") self.db_path_edit = QLineEdit() self.db_path_edit.setObjectName("dbPathEdit") self.default_debug_port_spin = QSpinBox() self.default_debug_port_spin.setObjectName("defaultDebugPortSpin") self.default_debug_port_spin.setRange(1, 65535) self.debug_port_start_spin = QSpinBox() self.debug_port_start_spin.setObjectName("debugPortStartSpin") self.debug_port_start_spin.setRange(1, 65535) self.debug_port_end_spin = QSpinBox() self.debug_port_end_spin.setObjectName("debugPortEndSpin") self.debug_port_end_spin.setRange(1, 65535) self.cdp_ready_timeout_spin = QSpinBox() self.cdp_ready_timeout_spin.setObjectName("cdpReadyTimeoutSpin") self.cdp_ready_timeout_spin.setRange(1, 3600) self.save_config_button = QPushButton("保存设置") self.test_item_id_edit = QLineEdit() self.test_item_id_edit.setObjectName("testItemIdEdit") self.allow_real_submit_checkbox = QCheckBox("允许真实提交线上商品") self.allow_real_submit_checkbox.setObjectName("allowRealSubmitCheckbox") self.allow_cover_update_checkbox = QCheckBox("允许更新封面") self.allow_cover_update_checkbox.setObjectName("allowCoverUpdateCheckbox") self.max_items_per_run_spin = QSpinBox() self.max_items_per_run_spin.setObjectName("maxItemsPerRunSpin") self.max_items_per_run_spin.setRange(1, 9999) self.close_success_tab_checkbox = QCheckBox("成功后关闭本次新开编辑页") self.close_success_tab_checkbox.setObjectName("closeSuccessTabCheckbox") form = QFormLayout() form.addRow("", self.enabled_checkbox) form.addRow("服务商名", self.name_edit) form.addRow("类别", self.category_combo) form.addRow("api_type", self.api_type_combo) form.addRow("模型ID", self.model_id_edit) form.addRow("网址", self.url_edit) form.addRow("密钥", self.api_key_edit) form.addRow("连接超时(秒)", self.connect_timeout_spin) action_layout = QHBoxLayout() action_layout.addWidget(self.save_model_button) action_layout.addWidget(self.test_connection_button) action_layout.addStretch(1) ai_form = QFormLayout() ai_form.addRow("标题大模型", self.default_text_model_combo) ai_form.addRow("图片大模型", self.default_image_model_combo) ai_form.addRow("标题并发数", self.title_concurrency_spin) ai_form.addRow("图片并发数", self.image_concurrency_spin) ai_form.addRow("失败重试次数", self.retry_spin) ai_form.addRow("分辨率", self.resolution_combo) ai_form.addRow("返回超时", self.response_timeout_label) ai_form.addRow("jpg质量", self.jpg_quality_spin) port_range_layout = QHBoxLayout() port_range_layout.addWidget(self.debug_port_start_spin) port_range_layout.addWidget(QLabel("到")) port_range_layout.addWidget(self.debug_port_end_spin) path_form = QFormLayout() path_form.addRow("Chrome路径", self.chrome_path_edit) path_form.addRow("账号数据根目录", self.user_data_root_edit) path_form.addRow("图片目录", self.image_dir_edit) path_form.addRow("DB路径", self.db_path_edit) path_form.addRow("默认调试端口", self.default_debug_port_spin) path_form.addRow("调试端口范围", port_range_layout) path_form.addRow("CDP就绪超时(秒)", self.cdp_ready_timeout_spin) update_form = QFormLayout() update_form.addRow("测试商品ID", self.test_item_id_edit) update_form.addRow("", self.allow_real_submit_checkbox) update_form.addRow("", self.allow_cover_update_checkbox) update_form.addRow("单次最大更新条数", self.max_items_per_run_spin) update_form.addRow("", self.close_success_tab_checkbox) right_panel = QWidget() right_layout = QVBoxLayout(right_panel) right_layout.setContentsMargins(12, 0, 0, 0) right_layout.addWidget(QLabel("模型详情")) right_layout.addLayout(form) right_layout.addLayout(action_layout) right_layout.addWidget(self.test_result_label) right_layout.addSpacing(18) right_layout.addWidget(QLabel("角色与生成参数")) right_layout.addLayout(ai_form) right_layout.addSpacing(18) right_layout.addWidget(QLabel("路径与端口")) right_layout.addLayout(path_form) right_layout.addSpacing(18) right_layout.addWidget(QLabel("Shopee 更新安全")) right_layout.addLayout(update_form) right_layout.addWidget(self.save_config_button) right_layout.addStretch(1) right_scroll = QScrollArea() right_scroll.setWidgetResizable(True) right_scroll.setWidget(right_panel) self.splitter = QSplitter(Qt.Horizontal) self.splitter.addWidget(left_panel) self.splitter.addWidget(right_scroll) self.splitter.setStretchFactor(0, 1) self.splitter.setStretchFactor(1, 3) self.splitter.setSizes([280, 860]) layout = QVBoxLayout(self) layout.setContentsMargins(18, 18, 18, 18) layout.addWidget(self.splitter, 1) self.model_combo.currentIndexChanged.connect(self.load_selected_model) self.add_model_button.clicked.connect(self.add_model) self.delete_model_button.clicked.connect(self.delete_model) self.save_model_button.clicked.connect(self.save_model) self.test_connection_button.clicked.connect(self.test_connection) self.resolution_combo.currentIndexChanged.connect( self._update_response_timeout_label ) self.save_config_button.clicked.connect(self.save_app_settings) self.refresh_models() self._populate_app_settings() def _set_status(self, message): if self.status_callback is not None: self.status_callback(message) def refresh_models(self, selected=None): try: self.models = appconfig.list_ai_models( path=self.ai_models_path, reveal_api_key=True, ) except Exception as exc: self.models = [] self.current_model_name = None self._show_error(exc) current = selected or self.current_model_name self.model_combo.blockSignals(True) self.model_combo.clear() for model in self.models: label = f"{model['name']} · {self._category_label(model['category'])}" if not model.get("enabled", True): label += " · 已停用" self.model_combo.addItem(label, model["name"]) index = self.model_combo.findData(current) self.model_combo.setCurrentIndex(index if index >= 0 else (0 if self.models else -1)) self.model_combo.blockSignals(False) self.load_selected_model() if hasattr(self, "default_text_model_combo"): self._populate_role_model_combos() def load_selected_model(self, index=None): name = self.model_combo.currentData() model = self._model_by_name(name) self.current_model_name = model["name"] if model else None self._populate_form(model) self._update_button_state() def add_model(self, checked=False): name = self._unique_model_name("新文本模型") model = { "name": name, "category": "text", "enabled": True, "url": "", "model": "", "api_key": "", "api_type": "chat", "connect_timeout_seconds": 30, "timeout_seconds": 0, "extra_body": {}, } try: appconfig.add_ai_model(model, path=self.ai_models_path) except Exception as exc: self._show_error(exc) return self.refresh_models(selected=name) self._set_status(f"AI 模型已新增：{name}") def save_model(self, checked=False): model = self._form_values() if model is None: return current = self._current_model() if self._should_warn_plaintext_api_key(model, current): self._show_plaintext_api_key_warning() try: if self.current_model_name is None: appconfig.add_ai_model(model, path=self.ai_models_path) else: appconfig.update_ai_model( self.current_model_name, path=self.ai_models_path, **model, ) except Exception as exc: self._show_error(exc) return self.refresh_models(selected=model["name"]) self._set_status(f"AI 模型已保存：{model['name']}") def delete_model(self, checked=False): model = self._current_model() if model is None: return if not self._can_delete_model(model): self._set_status("每个类别至少保留一个模型，当前模型不能删除") return answer = QMessageBox.question( self, "删除 AI 模型", f"确认删除模型「{model['name']}」？", QMessageBox.Yes | QMessageBox.No, QMessageBox.No, ) if answer != QMessageBox.Yes: return try: appconfig.delete_ai_model(model["name"], path=self.ai_models_path) except Exception as exc: self._show_error(exc) return self.refresh_models() self._set_status(f"AI 模型已删除：{model['name']}") def test_connection(self, checked=False): if self.test_thread is not None: self._set_status("模型连接测试正在进行...") return model = self._current_model() if model is None: return if self.name_edit.text().strip() != model["name"]: self._set_status("请先保存模型名称变更后再测试连接") return worker = AIModelTestWorker(model["name"], ai_models_path=self.ai_models_path) worker.finished.connect(self._on_test_finished) worker.failed.connect(self._on_test_failed) thread = run_worker(worker, thread_name="AIModelTestWorker", start=False) thread.finished.connect(lambda: self._forget_test_thread(thread)) self.test_worker = worker self.test_thread = thread self._set_test_running(True) self.test_result_label.setText("正在测试连接...") self._set_status(f"正在测试 AI 模型连接：{model['name']}") thread.start() def save_app_settings(self, checked=False): settings = self._app_settings_values() if settings is None: return try: saved = appconfig.save_config(settings, path=self.config_path) except Exception as exc: self._show_error(exc) return self._replace_config(saved) self._populate_app_settings() self._set_status("设置已保存") def _app_settings_values(self): start_port = self.debug_port_start_spin.value() end_port = self.debug_port_end_spin.value() default_port = self.default_debug_port_spin.value() if start_port > end_port: self._show_error("调试端口范围起始值不能大于结束值") return None if not (start_port <= default_port <= end_port): self._show_error("默认调试端口必须在调试端口范围内") return None text_model = self.default_text_model_combo.currentData() image_model = self.default_image_model_combo.currentData() if not text_model or not image_model: self._show_error("标题大模型和图片大模型不能为空") return None ai_cfg = appconfig.ai_config(self.config) ai_cfg.update( { "default_text_model": text_model, "default_image_model": image_model, "title_concurrency": self.title_concurrency_spin.value(), "image_concurrency": self.image_concurrency_spin.value(), "retry": self.retry_spin.value(), "jpg_quality": self.jpg_quality_spin.value(), "resolution": self.resolution_combo.currentData() or "1k", "resolution_timeouts": dict(ai_cfg.get("resolution_timeouts", {})), } ) settings = { key: value for key, value in self.config.items() if key not in {"config_path", "ai_models_path"} } settings.update( { "chrome_path": self.chrome_path_edit.text().strip(), "user_data_root": self.user_data_root_edit.text().strip(), "image_dir": self.image_dir_edit.text().strip(), "db_path": self.db_path_edit.text().strip(), "default_debug_port": default_port, "debug_port_range": [start_port, end_port], "cdp_ready_timeout": self.cdp_ready_timeout_spin.value(), "ai": ai_cfg, "shopee_update": { "test_item_id": self.test_item_id_edit.text().strip(), "allow_real_submit": self.allow_real_submit_checkbox.isChecked(), "allow_cover_update": self.allow_cover_update_checkbox.isChecked(), "max_items_per_run": self.max_items_per_run_spin.value(), "close_success_tab": self.close_success_tab_checkbox.isChecked(), }, } ) return settings def _replace_config(self, saved): internal = {} if self.config_path != appconfig.CONFIG_PATH: internal["config_path"] = self.config_path if self.ai_models_path != appconfig.AI_MODELS_PATH: internal["ai_models_path"] = self.ai_models_path self.config.clear() self.config.update(saved) self.config.update(internal) def _populate_app_settings(self): self._populate_role_model_combos() ai_cfg = appconfig.ai_config(self.config) self._set_combo_by_data( self.default_text_model_combo, ai_cfg.get("default_text_model", ""), ) self._set_combo_by_data( self.default_image_model_combo, ai_cfg.get("default_image_model", ""), ) self.title_concurrency_spin.setValue( int(ai_cfg.get("title_concurrency", 4) or 4) ) self.image_concurrency_spin.setValue( int(ai_cfg.get("image_concurrency", 4) or 4) ) self.retry_spin.setValue(int(ai_cfg.get("retry", 2) or 0)) self._set_combo_by_data( self.resolution_combo, str(ai_cfg.get("resolution", "1k")), ) self.jpg_quality_spin.setValue(int(ai_cfg.get("jpg_quality", 90) or 90)) self.chrome_path_edit.setText(appconfig.chrome_path(self.config)) self.user_data_root_edit.setText(appconfig.user_data_root(self.config)) self.image_dir_edit.setText(appconfig.image_dir(self.config)) self.db_path_edit.setText(appconfig.db_path(self.config)) self.default_debug_port_spin.setValue( int(appconfig.default_debug_port(self.config)) ) start_port, end_port = appconfig.debug_port_range(self.config) self.debug_port_start_spin.setValue(int(start_port)) self.debug_port_end_spin.setValue(int(end_port)) self.cdp_ready_timeout_spin.setValue( int(appconfig.cdp_ready_timeout(self.config)) ) update_cfg = self._shopee_update_config() self.test_item_id_edit.setText(str(update_cfg.get("test_item_id", ""))) self.allow_real_submit_checkbox.setChecked( bool(update_cfg.get("allow_real_submit", False)) ) self.allow_cover_update_checkbox.setChecked( bool(update_cfg.get("allow_cover_update", False)) ) self.max_items_per_run_spin.setValue( max(1, int(update_cfg.get("max_items_per_run", 1) or 1)) ) self.close_success_tab_checkbox.setChecked( bool(update_cfg.get("close_success_tab", False)) ) self._update_response_timeout_label() def _shopee_update_config(self): defaults = appconfig.default_config().get("shopee_update", {}) loaded = self.config.get("shopee_update", {}) if not isinstance(loaded, dict): loaded = {} merged = dict(defaults) merged.update(loaded) return merged def _populate_role_model_combos(self): ai_cfg = appconfig.ai_config(self.config) self._populate_role_combo( self.default_text_model_combo, "text", ai_cfg.get("default_text_model"), ) self._populate_role_combo( self.default_image_model_combo, "image", ai_cfg.get("default_image_model"), ) def _populate_role_combo(self, combo, category, selected): combo.blockSignals(True) combo.clear() for model in self.models: if model.get("category") == category and model.get("enabled", True): combo.addItem(model.get("name", ""), model.get("name", "")) if combo.count() == 0: combo.addItem("无可用模型", None) index = combo.findData(selected) combo.setCurrentIndex(index if index >= 0 else 0) combo.blockSignals(False) def _update_response_timeout_label(self, index=None): ai_cfg = appconfig.ai_config(self.config) resolution = self.resolution_combo.currentData() or ai_cfg.get("resolution", "1k") timeouts = ai_cfg.get("resolution_timeouts", {}) timeout = timeouts.get(str(resolution)) if timeout is None: self.response_timeout_label.setText("未配置") return self.response_timeout_label.setText(f"{int(timeout)} 秒") def _form_values(self): current = self._current_model() or {} name = self.name_edit.text().strip() if not name: self._show_error("AI 模型服务商名不能为空") return None extra_body = current.get("extra_body", {}) if not isinstance(extra_body, dict): extra_body = {} return { "name": name, "category": self.category_combo.currentData() or "text", "enabled": self.enabled_checkbox.isChecked(), "url": self.url_edit.text().strip(), "model": self.model_id_edit.text().strip(), "api_key": self.api_key_edit.text(), "api_type": self.api_type_combo.currentData() or "auto", "connect_timeout_seconds": self.connect_timeout_spin.value(), "timeout_seconds": int(current.get("timeout_seconds", 0) or 0), "extra_body": dict(extra_body), } def _populate_form(self, model): widgets = [ self.enabled_checkbox, self.name_edit, self.category_combo, self.api_type_combo, self.model_id_edit, self.url_edit, self.api_key_edit, self.connect_timeout_spin, ] for widget in widgets: widget.blockSignals(True) if model is None: self.enabled_checkbox.setChecked(False) self.name_edit.clear() self.category_combo.setCurrentIndex(0) self.api_type_combo.setCurrentIndex(0) self.model_id_edit.clear() self.url_edit.clear() self.api_key_edit.clear() self.connect_timeout_spin.setValue(30) else: self.enabled_checkbox.setChecked(bool(model.get("enabled", True))) self.name_edit.setText(model.get("name", "")) self._set_combo_by_data(self.category_combo, model.get("category", "text")) self._set_combo_by_data(self.api_type_combo, model.get("api_type", "auto")) self.model_id_edit.setText(model.get("model", "")) self.url_edit.setText(model.get("url", "")) self.api_key_edit.setText(model.get("api_key", "")) self.connect_timeout_spin.setValue( int(model.get("connect_timeout_seconds", 30) or 30) ) for widget in widgets: widget.blockSignals(False) def _set_combo_by_data(self, combo, value): index = combo.findData(value) combo.setCurrentIndex(index if index >= 0 else 0) def _update_button_state(self): has_model = self._current_model() is not None testing = self.test_thread is not None for widget in ( self.enabled_checkbox, self.name_edit, self.category_combo, self.api_type_combo, self.model_id_edit, self.url_edit, self.api_key_edit, self.connect_timeout_spin, self.save_model_button, ): widget.setEnabled(has_model and not testing) self.add_model_button.setEnabled(not testing) self.delete_model_button.setEnabled( has_model and not testing and self._can_delete_model(self._current_model()) ) self.test_connection_button.setEnabled(has_model and not testing) def _set_test_running(self, running): self._update_button_state() self.test_connection_button.setEnabled( not running and self._current_model() is not None ) def _forget_test_thread(self, thread): if self.test_thread is thread: self.test_thread = None self.test_worker = None self._set_test_running(False) def _on_test_finished(self, payload): if payload.get("ok"): status = payload.get("status") suffix = f"（HTTP {status}）" if status else "" message = f"测试连接成功：{payload.get('name')}{suffix}" else: error = payload.get("error") or "连接失败" status = payload.get("status") status_text = f"HTTP {status}，" if status else "" message = f"测试连接失败：{status_text}{error}" self.test_result_label.setText(message) self._set_status(message) def _on_test_failed(self, _task_id, error): message = f"测试连接失败：{error}" self.test_result_label.setText(message) self._set_status(message) def _show_error(self, error): message = str(error) QMessageBox.warning(self, "设置", message) self._set_status(message) def _should_warn_plaintext_api_key(self, model, current): new_key = str((model or {}).get("api_key") or "") current_key = str((current or {}).get("api_key") or "") return bool(new_key) and new_key != current_key def _show_plaintext_api_key_warning(self): QMessageBox.warning( self, PLAINTEXT_SECRET_TITLE, PLAINTEXT_API_KEY_WARNING, ) def _current_model(self): return self._model_by_name(self.current_model_name) def _model_by_name(self, name): for model in self.models: if model.get("name") == name: return model return None def _unique_model_name(self, base): names = {model.get("name") for model in self.models} if base not in names: return base counter = 2 while f"{base} {counter}" in names: counter += 1 return f"{base} {counter}" def _can_delete_model(self, model): if model is None: return False category = model.get("category") return sum(1 for item in self.models if item.get("category") == category) > 1 def _category_label(self, category): return {"text": "文本", "image": "图像"}.get(category, category) class AccountsTab(QWidget): COLUMNS = ["账号名", "别名", "地区", "端口", "登录状态", "备注"] def __init__(self, parent=None, db_path=None, config=None, status_callback=None): super().__init__(parent) self.db_path = db_path self.config = appconfig.load_config() if config is None else config self.status_callback = status_callback self.account_rows = [] self.login_statuses = {} self.threads = [] self.table = QTableWidget(0, len(self.COLUMNS)) self.table.setHorizontalHeaderLabels(self.COLUMNS) self.table.setSelectionBehavior(QAbstractItemView.SelectRows) self.table.setSelectionMode(QAbstractItemView.SingleSelection) self.table.setEditTriggers(QAbstractItemView.NoEditTriggers) self.table.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch) self.table.verticalHeader().setVisible(False) self.add_button = QPushButton("新增") self.edit_button = QPushButton("编辑") self.delete_button = QPushButton("删除") self.launch_button = QPushButton("启动登录") self.check_button = QPushButton("检测登录") self.shortcut_button = QPushButton("快捷方式") toolbar = QHBoxLayout() for button in ( self.add_button, self.edit_button, self.delete_button, self.launch_button, self.check_button, self.shortcut_button, ): toolbar.addWidget(button) toolbar.addStretch(1) self.empty_label = QLabel("") layout = QVBoxLayout(self) layout.setContentsMargins(18, 18, 18, 18) layout.addLayout(toolbar) layout.addWidget(self.table, 1) layout.addWidget(self.empty_label) self.add_button.clicked.connect(self.add_account) self.edit_button.clicked.connect(self.edit_account) self.delete_button.clicked.connect(self.delete_account) self.launch_button.clicked.connect(self.launch_login) self.check_button.clicked.connect(self.check_login) self.shortcut_button.clicked.connect(self.create_shortcut) self.table.itemSelectionChanged.connect(self._update_button_state) self.table.doubleClicked.connect(self.edit_account) self.refresh_accounts() def _set_status(self, message): if self.status_callback is not None: self.status_callback(message) def _selected_account(self): selected = self.table.selectionModel().selectedRows() if not selected: return None row = selected[0].row() if row < 0 or row >= len(self.account_rows): return None return self.account_rows[row] def _update_button_state(self): has_selection = self._selected_account() is not None for button in ( self.edit_button, self.delete_button, self.launch_button, self.check_button, self.shortcut_button, ): button.setEnabled(has_selection) def refresh_accounts(self): try: self.account_rows = accounts.list_accounts( path=self.db_path, config=self.config, ) except Exception as exc: self.account_rows = [] self._set_status(f"账号读取失败：{exc}") self.table.setRowCount(len(self.account_rows)) for row, account in enumerate(self.account_rows): status = self.login_statuses.get(account.alias, "未知") values = [ account.account_name, account.alias, account.region_host, str(account.debug_port), status, account.note or "", ] for column, value in enumerate(values): item = QTableWidgetItem(value) self.table.setItem(row, column, item) self.empty_label.setText("" if self.account_rows else "暂无账号") self._update_button_state() def _show_error(self, message): QMessageBox.warning(self, "账号管理", str(message)) self._set_status(str(message)) def add_account(self, checked=False): try: default_port = accounts.next_debug_port( path=self.db_path, config=self.config, ) except Exception: default_port = appconfig.default_debug_port(self.config) dialog = AccountDialog(self, default_port=default_port, config=self.config) if dialog.exec() != QDialog.Accepted: return values = dialog.values() if self._should_warn_plaintext_password(values): self._show_plaintext_password_warning() try: accounts.create_account( path=self.db_path, config=self.config, **values, ) except Exception as exc: self._show_error(exc) return self.refresh_accounts() self._set_status("账号已新增") def edit_account(self, checked=False): account = self._selected_account() if account is None: return dialog = AccountDialog( self, account=account, default_port=account.debug_port, config=self.config, ) if dialog.exec() != QDialog.Accepted: return values = dialog.values() if self._should_warn_plaintext_password(values, account): self._show_plaintext_password_warning() try: updated = accounts.update_account( account.alias, path=self.db_path, config=self.config, **values, ) except Exception as exc: self._show_error(exc) return if updated.alias != account.alias: self.login_statuses.pop(account.alias, None) self.refresh_accounts() self._set_status("账号已保存") def delete_account(self, checked=False): account = self._selected_account() if account is None: return answer = QMessageBox.question( self, "删除账号", f"确认删除账号「{account.alias}」？", QMessageBox.Yes | QMessageBox.No, QMessageBox.No, ) if answer != QMessageBox.Yes: return try: accounts.delete_account(account.alias, path=self.db_path, config=self.config) except Exception as exc: self._show_error(exc) return self.login_statuses.pop(account.alias, None) self.refresh_accounts() self._set_status("账号已删除") def launch_login(self, checked=False): account = self._selected_account() if account is None: return try: accounts.launch_for_login(account, config=self.config) except Exception as exc: self._show_error(exc) return self.login_statuses[account.alias] = "已启动" self.refresh_accounts() self._set_status("Chrome 已启动，请人工登录") def check_login(self, checked=False): account = self._selected_account() if account is None: return self.login_statuses[account.alias] = "检测中" self.refresh_accounts() worker = AccountLoginCheckWorker( account, db_path=self.db_path, config=self.config, ) worker.finished.connect(self._on_login_check_finished) worker.failed.connect( lambda _task_id, error, alias=account.alias: self._on_login_check_failed( alias, error, ) ) thread = run_worker(worker, start=False) thread.finished.connect(lambda: self._forget_thread(thread)) self.threads.append(thread) thread.start() def _forget_thread(self, thread): if thread in self.threads: self.threads.remove(thread) def _on_login_check_finished(self, payload): if payload.get("ok") is False and not payload.get("alias"): return alias = payload.get("alias") status = payload.get("status") or {} if alias: self.login_statuses[alias] = accounts.login_status_text(status) self.refresh_accounts() self._set_status("登录状态已刷新") def _on_login_check_failed(self, alias, error): self.login_statuses[alias] = "检测失败" self.refresh_accounts() self._set_status(f"登录检测失败：{error}") def create_shortcut(self, checked=False): account = self._selected_account() if account is None: return try: shortcut_path = accounts.create_shortcut(account, config=self.config) except Exception as exc: self._show_error(exc) return self._set_status(f"快捷方式已生成：{shortcut_path}") QMessageBox.information(self, "账号管理", f"快捷方式已生成：\n{shortcut_path}") def _should_warn_plaintext_password(self, values, account=None): new_password = str((values or {}).get("password") or "") current_password = str(getattr(account, "password", None) or "") return bool(new_password) and new_password != current_password def _show_plaintext_password_warning(self): QMessageBox.warning( self, PLAINTEXT_SECRET_TITLE, PLAINTEXT_PASSWORD_WARNING, ) class MainWindow(QMainWindow): """Main application window with the fixed five-tab workflow.""" def __init__(self, db_path=None, config=None, config_path=None, ai_models_path=None): super().__init__() self.config = appconfig.load_config() if config is None else config self.db_path = _database_path(db_path, self.config) self.config_path = ( config_path or self.config.get("config_path") or appconfig.CONFIG_PATH ) self.ai_models_path = ( ai_models_path or self.config.get("ai_models_path") or appconfig.AI_MODELS_PATH ) self.setWindowTitle("cmshopee") self.resize(1180, 760) self.tabs = QTabWidget() self.tabs.setObjectName("mainTabs") self.tabs.setStyleSheet(TAB_STYLE) self.tabs.currentChanged.connect(self._on_tab_changed) for title in TAB_TITLES: self.tabs.addTab(self._build_tab(title), title) self.setCentralWidget(self.tabs) self.statusBar().showMessage("就绪") def _build_tab(self, title): if title == "① 导入采集": return CollectTab( db_path=self.db_path, config=self.config, status_callback=self.statusBar().showMessage, open_accounts_callback=lambda: self.open_accounts_tab(), ) if title == "② AI生成": return GenerateTab( db_path=self.db_path, config=self.config, status_callback=self.statusBar().showMessage, ) if title == "③ 更新shopee": return ApplyTab( db_path=self.db_path, config=self.config, status_callback=self.statusBar().showMessage, open_accounts_callback=lambda: self.open_accounts_tab(), ) if title == "④ 账号管理": return AccountsTab( db_path=self.db_path, config=self.config, status_callback=self.statusBar().showMessage, ) return SettingsTab( config=self.config, config_path=self.config_path, ai_models_path=self.ai_models_path, status_callback=self.statusBar().showMessage, ) def _on_tab_changed(self, index): self.statusBar().showMessage(f"当前：{self.tabs.tabText(index)}") def open_accounts_tab(self): self.tabs.setCurrentIndex(TAB_TITLES.index("④ 账号管理")) else: class MainWindow(QMainWindow): def __init__(self): raise RuntimeError("PySide6 未安装，无法启动 GUI") def _ensure_offscreen_for_headless_tests(): if "PYTEST_CURRENT_TEST" in os.environ and "QT_QPA_PLATFORM" not in os.environ: os.environ["QT_QPA_PLATFORM"] = "offscreen" def main() -> int: if QT_IMPORT_ERROR is not None: print("cmshopee GUI 无法启动：当前 Python 环境未安装 PySide6。") return 1 _ensure_offscreen_for_headless_tests() app = QApplication.instance() or QApplication(sys.argv) window = MainWindow() window.show() return app.exec()