chengma
91683abe7d
- 在SettingsTab增加标题/图片默认模型角色下拉并按text/image过滤 - 增加并发、重试、分辨率、返回超时展示和jpg质量设置 - 增加Chrome路径、账号数据根目录、图片目录、DB路径和端口配置 - 保存到config.json并校验端口范围，避免写入测试注入路径 - 补充GUI测试并同步任务、api、routes、current-state和progress
3393 lines
140 KiB
Python
3393 lines
140 KiB
Python
"""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
|
||
|
||
|
||
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
|
||
answer = QMessageBox.question(
|
||
self,
|
||
"确认开始更新",
|
||
self._confirmation_message(tasks),
|
||
QMessageBox.Yes | QMessageBox.No,
|
||
QMessageBox.No,
|
||
)
|
||
if answer != QMessageBox.Yes:
|
||
self._set_status("已取消开始更新")
|
||
return
|
||
worker = ApplyWorker(tasks, db_path=self.db_path, config=self.config)
|
||
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):
|
||
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"
|
||
"确认后后续执行会打开商品编辑页、替换标题和封面，并点击「更新」提交线上。"
|
||
)
|
||
|
||
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):
|
||
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 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)
|
||
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(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("保存设置")
|
||
|
||
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)
|
||
|
||
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.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
|
||
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,
|
||
}
|
||
)
|
||
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))
|
||
)
|
||
self._update_response_timeout_label()
|
||
|
||
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 _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
|
||
try:
|
||
accounts.create_account(
|
||
path=self.db_path,
|
||
config=self.config,
|
||
**dialog.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
|
||
try:
|
||
updated = accounts.update_account(
|
||
account.alias,
|
||
path=self.db_path,
|
||
config=self.config,
|
||
**dialog.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}")
|
||
|
||
|
||
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()
|