Files
cmshoppe/app/gui/tabs/settings.py
T

1148 lines
48 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Tab 5: settings UI."""
from __future__ import annotations
from ..widgets import *
from ..workers import AIModelTestWorker as _RealAIModelTestWorker
from ..workers import CMHubSettingsWorker as _RealCMHubSettingsWorker
PLAINTEXT_CMHUB_API_KEY_WARNING = (
"cmhub API Key 会以本地明文保存到 config/cmhub.json,仅供本机调用 cmhub 网关使用。"
"该文件已 gitignore,UI 打码显示,日志/导出不记录明文。"
)
def AIModelTestWorker(*args, **kwargs):
return _call_package_attr("AIModelTestWorker", _RealAIModelTestWorker, *args, **kwargs)
def CMHubSettingsWorker(*args, **kwargs):
return _call_package_attr("CMHubSettingsWorker", _RealCMHubSettingsWorker, *args, **kwargs)
class SettingsTab(QWidget):
"""Tab 5: AI model definitions stored in config/ai_models.json."""
BACKEND_ITEMS = [("直连模型", "direct"), ("cmhub 网关", "cmhub")]
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.cmhub_config_path = (
self.config.get("cmhub_config_path")
or self._default_cmhub_config_path(self.config_path)
)
self.status_callback = status_callback
self.models = []
self.current_model_name = None
self.test_worker = None
self.test_thread = None
self.cmhub_models = []
self.cmhub_worker = None
self.cmhub_thread = None
self._loaded_cmhub_api_key = ""
self._cmhub_auto_refresh_done = False
self._compat_test_item_id = ""
self.backend_combo = QComboBox()
self.backend_combo.setObjectName("aiBackendCombo")
for label, value in self.BACKEND_ITEMS:
self.backend_combo.addItem(label, value)
self.backend_combo.setVisible(False)
self.cmhub_base_url_edit = QLineEdit()
self.cmhub_base_url_edit.setObjectName("cmhubBaseUrlEdit")
self.cmhub_base_url_edit.setPlaceholderText("https://<cmhub>")
self.cmhub_api_key_edit = QLineEdit()
self.cmhub_api_key_edit.setObjectName("cmhubApiKeyEdit")
self.cmhub_api_key_edit.setEchoMode(QLineEdit.Password)
self.cmhub_api_key_edit.setPlaceholderText("从 cmhub 网页端复制 API Key")
self.cmhub_title_alias_combo = QComboBox()
self.cmhub_title_alias_combo.setObjectName("cmhubTitleAliasCombo")
self.cmhub_image_alias_combo = QComboBox()
self.cmhub_image_alias_combo.setObjectName("cmhubImageAliasCombo")
self.cmhub_connect_timeout_spin = QSpinBox()
self.cmhub_connect_timeout_spin.setObjectName("cmhubConnectTimeoutSpin")
self.cmhub_connect_timeout_spin.setRange(1, 3600)
self.cmhub_connect_timeout_spin.setValue(10)
self.cmhub_check_balance_checkbox = QCheckBox("批量生成前检查余额")
self.cmhub_check_balance_checkbox.setObjectName("cmhubCheckBalanceCheckbox")
self.cmhub_refresh_button = QPushButton("刷新别名")
self.cmhub_refresh_button.setObjectName("cmhubRefreshButton")
self.cmhub_test_button = QPushButton("测试连接/查余额")
self.cmhub_test_button.setObjectName("cmhubTestButton")
self.cmhub_result_label = QLabel("")
self.cmhub_result_label.setObjectName("cmhubResultLabel")
self.cmhub_result_label.setWordWrap(True)
self.cmhub_key_hint_label = QLabel("API Key 仅在 cmhub 网页端创建时显示一次;复制到此处后会本地明文保存并打码显示。")
self.cmhub_key_hint_label.setObjectName("cmhubKeyHintLabel")
self.cmhub_key_hint_label.setWordWrap(True)
self.model_combo = QComboBox()
self.model_combo.setObjectName("aiModelCombo")
self.add_model_button = QPushButton("新增")
self.delete_model_button = QPushButton("删除")
self.enabled_checkbox = QCheckBox("启用")
self.name_edit = QLineEdit()
self.name_edit.setObjectName("modelNameEdit")
self.category_combo = QComboBox()
self.category_combo.setObjectName("modelCategoryCombo")
for label, value in self.CATEGORY_ITEMS:
self.category_combo.addItem(label, value)
self.api_type_combo = QComboBox()
self.api_type_combo.setObjectName("modelApiTypeCombo")
for label, value in self.API_TYPE_ITEMS:
self.api_type_combo.addItem(label, value)
self.model_id_edit = QLineEdit()
self.model_id_edit.setObjectName("modelIdEdit")
self.url_edit = QLineEdit()
self.url_edit.setObjectName("modelUrlEdit")
self.api_key_edit = QLineEdit()
self.api_key_edit.setObjectName("modelApiKeyEdit")
self.api_key_edit.setEchoMode(QLineEdit.Password)
self.connect_timeout_spin = QSpinBox()
self.connect_timeout_spin.setObjectName("connectTimeoutSpin")
self.connect_timeout_spin.setRange(1, 3600)
self.connect_timeout_spin.setValue(30)
self.save_model_button = QPushButton("保存")
self.test_connection_button = QPushButton("测试连接")
self.test_result_label = QLabel("")
self.test_result_label.setWordWrap(True)
self.default_text_model_combo = QComboBox()
self.default_text_model_combo.setObjectName("defaultTextModelCombo")
self.default_image_model_combo = QComboBox()
self.default_image_model_combo.setObjectName("defaultImageModelCombo")
self.title_concurrency_spin = QSpinBox()
self.title_concurrency_spin.setObjectName("titleConcurrencySpin")
self.title_concurrency_spin.setRange(1, 64)
self.image_concurrency_spin = QSpinBox()
self.image_concurrency_spin.setObjectName("imageConcurrencySpin")
self.image_concurrency_spin.setRange(1, 64)
self.retry_spin = QSpinBox()
self.retry_spin.setObjectName("retrySpin")
self.retry_spin.setRange(0, 20)
self.resolution_combo = QComboBox()
self.resolution_combo.setObjectName("resolutionCombo")
for resolution in self.RESOLUTION_ITEMS:
self.resolution_combo.addItem(resolution, resolution)
self.response_timeout_label = QLabel("")
self.jpg_quality_spin = QSpinBox()
self.jpg_quality_spin.setObjectName("jpgQualitySpin")
self.jpg_quality_spin.setRange(1, 100)
self.chrome_path_edit = QLineEdit()
self.chrome_path_edit.setObjectName("chromePathEdit")
self.user_data_root_edit = QLineEdit()
self.user_data_root_edit.setObjectName("userDataRootEdit")
self.image_dir_edit = QLineEdit()
self.image_dir_edit.setObjectName("imageDirEdit")
self.db_path_edit = QLineEdit()
self.db_path_edit.setObjectName("dbPathEdit")
self.default_debug_port_spin = QSpinBox()
self.default_debug_port_spin.setObjectName("defaultDebugPortSpin")
self.default_debug_port_spin.setRange(1, 65535)
self.debug_port_start_spin = QSpinBox()
self.debug_port_start_spin.setObjectName("debugPortStartSpin")
self.debug_port_start_spin.setRange(1, 65535)
self.debug_port_end_spin = QSpinBox()
self.debug_port_end_spin.setObjectName("debugPortEndSpin")
self.debug_port_end_spin.setRange(1, 65535)
self.cdp_ready_timeout_spin = QSpinBox()
self.cdp_ready_timeout_spin.setObjectName("cdpReadyTimeoutSpin")
self.cdp_ready_timeout_spin.setRange(1, 3600)
self.save_config_button = QPushButton("保存设置")
self.allow_real_submit_checkbox = QCheckBox("允许真实提交线上商品")
self.allow_real_submit_checkbox.setObjectName("allowRealSubmitCheckbox")
self.allow_cover_update_checkbox = QCheckBox("允许更新封面")
self.allow_cover_update_checkbox.setObjectName("allowCoverUpdateCheckbox")
self.max_items_per_run_spin = QSpinBox()
self.max_items_per_run_spin.setObjectName("maxItemsPerRunSpin")
self.max_items_per_run_spin.setRange(1, 9999)
self.max_items_per_run_spin.setToolTip("作为每批最大更新条数;正式更新会分批处理当前筛选全部可更新记录。")
self.close_success_tab_checkbox = QCheckBox("成功后关闭本次新开编辑页")
self.close_success_tab_checkbox.setObjectName("closeSuccessTabCheckbox")
self.parallel_accounts_checkbox = QCheckBox("多账号并行更新")
self.parallel_accounts_checkbox.setObjectName("parallelAccountsCheckbox")
self.max_parallel_accounts_spin = QSpinBox()
self.max_parallel_accounts_spin.setObjectName("maxParallelAccountsSpin")
self.max_parallel_accounts_spin.setRange(1, 16)
self.max_parallel_accounts_label = QLabel("最大并行账号数")
self.parallel_accounts_group = QWidget()
self.parallel_accounts_group.setObjectName("parallelAccountsGroup")
parallel_accounts_layout = QHBoxLayout(self.parallel_accounts_group)
parallel_accounts_layout.setContentsMargins(0, 0, 0, 0)
parallel_accounts_layout.setSpacing(12)
parallel_accounts_layout.addWidget(self.parallel_accounts_checkbox)
parallel_accounts_layout.addWidget(self.max_parallel_accounts_label)
parallel_accounts_layout.addWidget(self.max_parallel_accounts_spin)
parallel_accounts_layout.addStretch(1)
model_picker_layout = QHBoxLayout()
model_picker_layout.addWidget(self.model_combo, 1)
model_picker_layout.addWidget(self.add_model_button)
model_picker_layout.addWidget(self.delete_model_button)
action_layout = QHBoxLayout()
action_layout.addWidget(self.save_model_button)
action_layout.addWidget(self.test_connection_button)
action_layout.addStretch(1)
form = self._three_column_form(
[
("状态", self.enabled_checkbox),
("服务商名", self.name_edit),
("类别", self.category_combo),
("api_type", self.api_type_combo),
("模型ID", self.model_id_edit),
("连接超时(秒)", self.connect_timeout_spin),
("网址", self.url_edit, True),
("密钥", self.api_key_edit, True),
]
)
direct_role_form = self._three_column_form(
[
("标题大模型", self.default_text_model_combo),
("图片大模型", self.default_image_model_combo),
]
)
ai_form = self._three_column_form(
[
("标题并发数", self.title_concurrency_spin),
("图片并发数", self.image_concurrency_spin),
("失败重试次数", self.retry_spin),
("分辨率", self.resolution_combo),
("返回超时", self.response_timeout_label),
("jpg质量", self.jpg_quality_spin),
]
)
port_range_layout = QHBoxLayout()
port_range_layout.setContentsMargins(0, 0, 0, 0)
port_range_layout.addWidget(self.debug_port_start_spin)
port_range_layout.addWidget(QLabel("到"))
port_range_layout.addWidget(self.debug_port_end_spin)
port_range_widget = QWidget()
port_range_widget.setLayout(port_range_layout)
path_form = self._three_column_form(
[
("Chrome路径", self.chrome_path_edit, True),
("账号数据根目录", self.user_data_root_edit),
("图片目录", self.image_dir_edit),
("DB路径", self.db_path_edit),
("默认调试端口", self.default_debug_port_spin),
("调试端口范围", port_range_widget),
("CDP就绪超时(秒)", self.cdp_ready_timeout_spin),
]
)
self.shopee_update_form_layout = self._three_column_form(
[
("每批最大更新条数", self.max_items_per_run_spin),
("", self.allow_real_submit_checkbox),
("", self.close_success_tab_checkbox),
("", self.allow_cover_update_checkbox),
("", self.parallel_accounts_group, 2),
]
)
self.model_picker_panel = QWidget()
self.model_picker_panel.setLayout(model_picker_layout)
self.model_detail_panel = QWidget()
model_detail_layout = QVBoxLayout(self.model_detail_panel)
model_detail_layout.setContentsMargins(0, 0, 0, 0)
model_detail_layout.setSpacing(8)
model_detail_layout.addLayout(form)
model_detail_layout.addLayout(action_layout)
model_detail_layout.addWidget(self.test_result_label)
self.direct_role_panel = QWidget()
self.direct_role_panel.setLayout(direct_role_form)
cmhub_form = self._three_column_form(
[
("网关 Base URL", self.cmhub_base_url_edit, True),
("API Key", self.cmhub_api_key_edit, True),
("连接超时(秒)", self.cmhub_connect_timeout_spin),
("生文别名", self.cmhub_title_alias_combo),
("生图别名", self.cmhub_image_alias_combo),
("", self.cmhub_check_balance_checkbox),
]
)
cmhub_action_layout = QHBoxLayout()
cmhub_action_layout.setContentsMargins(0, 0, 0, 0)
cmhub_action_layout.addWidget(self.cmhub_refresh_button)
cmhub_action_layout.addWidget(self.cmhub_test_button)
cmhub_action_layout.addStretch(1)
self.cmhub_panel = QWidget()
self.cmhub_panel.setObjectName("cmhubSettingsPanel")
cmhub_panel_layout = QVBoxLayout(self.cmhub_panel)
cmhub_panel_layout.setContentsMargins(0, 0, 0, 0)
cmhub_panel_layout.setSpacing(8)
cmhub_panel_layout.addLayout(cmhub_form)
cmhub_panel_layout.addWidget(self.cmhub_key_hint_label)
cmhub_panel_layout.addLayout(cmhub_action_layout)
cmhub_panel_layout.addWidget(self.cmhub_result_label)
panel = QWidget()
panel.setMaximumWidth(1800)
panel_layout = QVBoxLayout(panel)
self.settings_panel_layout = panel_layout
panel_layout.setContentsMargins(13, 18, 13, 18)
self.ai_model_section_title = self._section_title(
"cmhub 网关",
"settingsAiModelSectionTitle",
)
self.model_detail_section_title = self._section_title(
"模型详情",
"settingsModelDetailSectionTitle",
)
self.generation_section_title = self._section_title(
"角色与生成参数",
"settingsGenerationSectionTitle",
)
self.shopee_update_section_title = self._section_title(
"Shopee 更新安全 / 执行模式",
"settingsShopeeUpdateSectionTitle",
)
self.infrastructure_section_title = self._section_title(
"基础设施(路径与端口)",
"settingsInfrastructureSectionTitle",
)
panel_layout.addWidget(self.ai_model_section_title)
panel_layout.addWidget(self.model_picker_panel)
panel_layout.addSpacing(14)
panel_layout.addWidget(self.model_detail_section_title)
panel_layout.addWidget(self.model_detail_panel)
panel_layout.addWidget(self.cmhub_panel)
panel_layout.addSpacing(18)
panel_layout.addWidget(self.generation_section_title)
panel_layout.addWidget(self.direct_role_panel)
panel_layout.addLayout(ai_form)
panel_layout.addSpacing(18)
panel_layout.addWidget(self.shopee_update_section_title)
panel_layout.addLayout(self.shopee_update_form_layout)
panel_layout.addSpacing(18)
panel_layout.addWidget(self.infrastructure_section_title)
panel_layout.addLayout(path_form)
panel_layout.addWidget(self.save_config_button)
panel_layout.addStretch(1)
scroll = QScrollArea()
scroll.setWidgetResizable(True)
scroll_content = QWidget()
scroll_layout = QHBoxLayout(scroll_content)
scroll_layout.setContentsMargins(0, 0, 0, 0)
scroll_layout.addStretch(1)
scroll_layout.addWidget(panel)
scroll_layout.addStretch(1)
scroll.setWidget(scroll_content)
layout = QVBoxLayout(self)
layout.setContentsMargins(18, 18, 18, 18)
layout.addWidget(scroll, 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.backend_combo.currentIndexChanged.connect(self._on_backend_changed)
self.cmhub_refresh_button.clicked.connect(self.refresh_cmhub_models)
self.cmhub_test_button.clicked.connect(self.test_cmhub_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 _three_column_form(self, fields):
layout = QGridLayout()
layout.setHorizontalSpacing(18)
layout.setVerticalSpacing(8)
for column in (1, 3, 5):
layout.setColumnStretch(column, 1)
row = 0
column_pair = 0
for field in fields:
label = field[0]
widget = field[1]
span_pairs = self._form_field_span_pairs(field)
if span_pairs > 3 - column_pair:
row += 1
column_pair = 0
column = column_pair * 2
self._add_form_field(layout, row, column, label, widget, span_pairs)
column_pair += span_pairs
if column_pair >= 3:
row += 1
column_pair = 0
return layout
def _form_field_span_pairs(self, field):
if len(field) <= 2:
return 1
span = field[2]
if isinstance(span, bool):
return 3 if span else 1
return max(1, min(3, int(span or 1)))
def _add_form_field(self, layout, row, column, label, widget, span_pairs):
if label:
layout.addWidget(QLabel(label), row, column)
layout.addWidget(widget, row, column + 1, 1, span_pairs * 2 - 1)
else:
layout.addWidget(widget, row, column, 1, span_pairs * 2)
def _section_title(self, text, object_name):
label = QLabel(text)
label.setObjectName(object_name)
label.setStyleSheet("color: #24292f; font-weight: 600; padding-top: 4px;")
return label
def showEvent(self, event):
super().showEvent(event)
self._maybe_auto_refresh_cmhub_models()
def _set_status(self, message):
if self.status_callback is not None:
self.status_callback(message)
def _maybe_auto_refresh_cmhub_models(self):
if self._cmhub_auto_refresh_done:
return
if not self.cmhub_base_url_edit.text().strip() or not self.cmhub_api_key_edit.text():
return
self._cmhub_auto_refresh_done = True
self.refresh_cmhub_models()
def _default_cmhub_config_path(self, config_path):
if config_path and config_path != appconfig.CONFIG_PATH:
return os.path.join(
os.path.dirname(os.path.abspath(config_path)),
"config",
"cmhub.json",
)
return appconfig.CMHUB_CONFIG_PATH
def _on_backend_changed(self, index=None):
self.backend_combo.setVisible(False)
self.model_picker_panel.setVisible(False)
self.model_detail_section_title.setVisible(False)
self.model_detail_panel.setVisible(False)
self.direct_role_panel.setVisible(False)
self.cmhub_panel.setVisible(True)
self._set_cmhub_running(self.cmhub_thread is not None)
self._update_button_state()
def refresh_models(self, selected=None):
try:
self.models = appconfig.list_ai_models(
path=self.ai_models_path,
reveal_api_key=True,
)
except Exception as exc:
self.models = []
self.current_model_name = None
self._show_error(exc)
current = selected or self.current_model_name
self.model_combo.blockSignals(True)
self.model_combo.clear()
for model in self.models:
label = f"{model['name']} · {self._category_label(model['category'])}"
if not model.get("enabled", True):
label += " · 已停用"
self.model_combo.addItem(label, model["name"])
index = self.model_combo.findData(current)
self.model_combo.setCurrentIndex(index if index >= 0 else (0 if self.models else -1))
self.model_combo.blockSignals(False)
self.load_selected_model()
if hasattr(self, "default_text_model_combo"):
self._populate_role_model_combos()
def load_selected_model(self, index=None):
name = self.model_combo.currentData()
model = self._model_by_name(name)
self.current_model_name = model["name"] if model else None
self._populate_form(model)
self._update_button_state()
def add_model(self, checked=False):
name = self._unique_model_name("新文本模型")
model = {
"name": name,
"category": "text",
"enabled": True,
"url": "",
"model": "",
"api_key": "",
"api_type": "chat",
"connect_timeout_seconds": 30,
"timeout_seconds": 0,
"extra_body": {},
}
try:
appconfig.add_ai_model(model, path=self.ai_models_path)
except Exception as exc:
self._show_error(exc)
return
self.refresh_models(selected=name)
self._set_status(f"AI 模型已新增:{name}")
def save_model(self, checked=False):
model = self._form_values()
if model is None:
return
current = self._current_model()
if self._should_warn_plaintext_api_key(model, current):
self._show_plaintext_api_key_warning()
try:
if self.current_model_name is None:
appconfig.add_ai_model(model, path=self.ai_models_path)
else:
appconfig.update_ai_model(
self.current_model_name,
path=self.ai_models_path,
**model,
)
except Exception as exc:
self._show_error(exc)
return
self.refresh_models(selected=model["name"])
self._set_status(f"AI 模型已保存:{model['name']}")
def delete_model(self, checked=False):
model = self._current_model()
if model is None:
return
if not self._can_delete_model(model):
self._set_status("每个类别至少保留一个模型,当前模型不能删除")
return
answer = QMessageBox.question(
self,
"删除 AI 模型",
f"确认删除模型「{model['name']}」?",
QMessageBox.Yes | QMessageBox.No,
QMessageBox.No,
)
if answer != QMessageBox.Yes:
return
try:
appconfig.delete_ai_model(model["name"], path=self.ai_models_path)
except Exception as exc:
self._show_error(exc)
return
self.refresh_models()
self._set_status(f"AI 模型已删除:{model['name']}")
def test_connection(self, checked=False):
if self.test_thread is not None:
self._set_status("模型连接测试正在进行...")
return
model = self._current_model()
if model is None:
return
if self.name_edit.text().strip() != model["name"]:
self._set_status("请先保存模型名称变更后再测试连接")
return
worker = AIModelTestWorker(
model["name"],
ai_models_path=self.ai_models_path,
db_path=_database_path(config=self.config),
diagnostic_log_dir=diagnostics.DEFAULT_LOG_DIR,
)
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
cmhub_key = self.cmhub_api_key_edit.text()
if self._should_warn_plaintext_cmhub_api_key(cmhub_key):
self._show_plaintext_cmhub_api_key_warning()
try:
appconfig.save_cmhub_config(
{"api_key": cmhub_key},
path=self.cmhub_config_path,
)
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("设置已保存")
QMessageBox.information(self, "保存设置", "设置已保存")
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
ai_cfg = appconfig.ai_config(self.config)
backend = "cmhub"
text_model = self.default_text_model_combo.currentData() or ai_cfg.get("default_text_model")
image_model = self.default_image_model_combo.currentData() or ai_cfg.get("default_image_model")
cmhub_cfg = self._cmhub_settings_values(backend)
if cmhub_cfg is None:
return None
ai_cfg.update(
{
"backend": backend,
"default_text_model": text_model,
"default_image_model": image_model,
"cmhub": cmhub_cfg,
"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", "cmhub_config_path"}
}
settings.update(
{
"chrome_path": self.chrome_path_edit.text().strip(),
"user_data_root": self.user_data_root_edit.text().strip(),
"image_dir": self.image_dir_edit.text().strip(),
"db_path": self.db_path_edit.text().strip(),
"default_debug_port": default_port,
"debug_port_range": [start_port, end_port],
"cdp_ready_timeout": self.cdp_ready_timeout_spin.value(),
"ai": ai_cfg,
"shopee_update": {
"test_item_id": str(self._compat_test_item_id or ""),
"allow_real_submit": self.allow_real_submit_checkbox.isChecked(),
"allow_cover_update": self.allow_cover_update_checkbox.isChecked(),
"max_items_per_run": self.max_items_per_run_spin.value(),
"close_success_tab": self.close_success_tab_checkbox.isChecked(),
"dry_run": False,
"parallel_accounts": self.parallel_accounts_checkbox.isChecked(),
"max_parallel_accounts": self.max_parallel_accounts_spin.value(),
},
}
)
return settings
def _cmhub_settings_values(self, backend):
current = appconfig.cmhub_config(self.config)
values = {
"base_url": self.cmhub_base_url_edit.text().strip(),
"title_alias": self.cmhub_title_alias_combo.currentData() or "",
"image_alias": self.cmhub_image_alias_combo.currentData() or "",
"connect_timeout": self.cmhub_connect_timeout_spin.value(),
"check_balance_before_batch": self.cmhub_check_balance_checkbox.isChecked(),
}
merged = dict(current)
merged.update(values)
return merged
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
if self.cmhub_config_path != appconfig.CMHUB_CONFIG_PATH:
internal["cmhub_config_path"] = self.cmhub_config_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.backend_combo, "cmhub")
cmhub_cfg = appconfig.cmhub_config(self.config)
self.cmhub_base_url_edit.setText(cmhub_cfg.get("base_url", ""))
self._loaded_cmhub_api_key = appconfig.get_cmhub_api_key(path=self.cmhub_config_path)
self.cmhub_api_key_edit.setText(self._loaded_cmhub_api_key)
self.cmhub_connect_timeout_spin.setValue(
max(1, int(cmhub_cfg.get("connect_timeout", 10) or 10))
)
self.cmhub_check_balance_checkbox.setChecked(
bool(cmhub_cfg.get("check_balance_before_batch", False))
)
self._populate_cmhub_alias_combos(
self.cmhub_models,
title_selected=cmhub_cfg.get("title_alias", ""),
image_selected=cmhub_cfg.get("image_alias", ""),
)
self._set_combo_by_data(
self.default_text_model_combo,
ai_cfg.get("default_text_model", ""),
)
self._set_combo_by_data(
self.default_image_model_combo,
ai_cfg.get("default_image_model", ""),
)
self.title_concurrency_spin.setValue(
int(ai_cfg.get("title_concurrency", 4) or 4)
)
self.image_concurrency_spin.setValue(
int(ai_cfg.get("image_concurrency", 4) or 4)
)
self.retry_spin.setValue(int(ai_cfg.get("retry", 2) or 0))
self._set_combo_by_data(
self.resolution_combo,
str(ai_cfg.get("resolution", "1k")),
)
self.jpg_quality_spin.setValue(int(ai_cfg.get("jpg_quality", 90) or 90))
self.chrome_path_edit.setText(appconfig.chrome_path(self.config))
self.user_data_root_edit.setText(appconfig.user_data_root(self.config))
self.image_dir_edit.setText(appconfig.image_dir(self.config))
self.db_path_edit.setText(appconfig.db_path(self.config))
self.default_debug_port_spin.setValue(
int(appconfig.default_debug_port(self.config))
)
start_port, end_port = appconfig.debug_port_range(self.config)
self.debug_port_start_spin.setValue(int(start_port))
self.debug_port_end_spin.setValue(int(end_port))
self.cdp_ready_timeout_spin.setValue(
int(appconfig.cdp_ready_timeout(self.config))
)
update_cfg = self._shopee_update_config()
self._compat_test_item_id = str(update_cfg.get("test_item_id", ""))
self.allow_real_submit_checkbox.setChecked(
bool(update_cfg.get("allow_real_submit", False))
)
self.allow_cover_update_checkbox.setChecked(
bool(update_cfg.get("allow_cover_update", False))
)
self.max_items_per_run_spin.setValue(
max(1, int(update_cfg.get("max_items_per_run", 1) or 1))
)
self.close_success_tab_checkbox.setChecked(
bool(update_cfg.get("close_success_tab", False))
)
self.parallel_accounts_checkbox.setChecked(
bool(update_cfg.get("parallel_accounts", False))
)
self.max_parallel_accounts_spin.setValue(
max(1, int(update_cfg.get("max_parallel_accounts", 2) or 2))
)
self._update_response_timeout_label()
self._on_backend_changed()
def _shopee_update_config(self):
defaults = appconfig.default_config().get("shopee_update", {})
loaded = self.config.get("shopee_update", {})
if not isinstance(loaded, dict):
loaded = {}
merged = dict(defaults)
merged.update(loaded)
return merged
def _populate_role_model_combos(self):
ai_cfg = appconfig.ai_config(self.config)
self._populate_role_combo(
self.default_text_model_combo,
"text",
ai_cfg.get("default_text_model"),
)
self._populate_role_combo(
self.default_image_model_combo,
"image",
ai_cfg.get("default_image_model"),
)
def _populate_role_combo(self, combo, category, selected):
combo.blockSignals(True)
combo.clear()
for model in self.models:
if model.get("category") == category and model.get("enabled", True):
combo.addItem(model.get("name", ""), model.get("name", ""))
if combo.count() == 0:
combo.addItem("无可用模型", None)
index = combo.findData(selected)
combo.setCurrentIndex(index if index >= 0 else 0)
combo.blockSignals(False)
def _update_response_timeout_label(self, index=None):
ai_cfg = appconfig.ai_config(self.config)
resolution = self.resolution_combo.currentData() or ai_cfg.get("resolution", "1k")
timeouts = ai_cfg.get("resolution_timeouts", {})
timeout = timeouts.get(str(resolution))
if timeout is None:
self.response_timeout_label.setText("未配置")
return
self.response_timeout_label.setText(f"{int(timeout)} 秒")
def _form_values(self):
current = self._current_model() or {}
name = self.name_edit.text().strip()
if not name:
self._show_error("AI 模型服务商名不能为空")
return None
extra_body = current.get("extra_body", {})
if not isinstance(extra_body, dict):
extra_body = {}
return {
"name": name,
"category": self.category_combo.currentData() or "text",
"enabled": self.enabled_checkbox.isChecked(),
"url": self.url_edit.text().strip(),
"model": self.model_id_edit.text().strip(),
"api_key": self.api_key_edit.text(),
"api_type": self.api_type_combo.currentData() or "auto",
"connect_timeout_seconds": self.connect_timeout_spin.value(),
"timeout_seconds": int(current.get("timeout_seconds", 0) or 0),
"extra_body": dict(extra_body),
}
def _populate_form(self, model):
widgets = [
self.enabled_checkbox,
self.name_edit,
self.category_combo,
self.api_type_combo,
self.model_id_edit,
self.url_edit,
self.api_key_edit,
self.connect_timeout_spin,
]
for widget in widgets:
widget.blockSignals(True)
if model is None:
self.enabled_checkbox.setChecked(False)
self.name_edit.clear()
self.category_combo.setCurrentIndex(0)
self.api_type_combo.setCurrentIndex(0)
self.model_id_edit.clear()
self.url_edit.clear()
self.api_key_edit.clear()
self.connect_timeout_spin.setValue(30)
else:
self.enabled_checkbox.setChecked(bool(model.get("enabled", True)))
self.name_edit.setText(model.get("name", ""))
self._set_combo_by_data(self.category_combo, model.get("category", "text"))
self._set_combo_by_data(self.api_type_combo, model.get("api_type", "auto"))
self.model_id_edit.setText(model.get("model", ""))
self.url_edit.setText(model.get("url", ""))
self.api_key_edit.setText(model.get("api_key", ""))
self.connect_timeout_spin.setValue(
int(model.get("connect_timeout_seconds", 30) or 30)
)
for widget in widgets:
widget.blockSignals(False)
def _set_combo_by_data(self, combo, value):
index = combo.findData(value)
combo.setCurrentIndex(index if index >= 0 else 0)
def _update_button_state(self):
has_model = self._current_model() is not None
testing = self.test_thread is not None
for widget in (
self.enabled_checkbox,
self.name_edit,
self.category_combo,
self.api_type_combo,
self.model_id_edit,
self.url_edit,
self.api_key_edit,
self.connect_timeout_spin,
self.save_model_button,
):
widget.setEnabled(has_model and not testing)
self.add_model_button.setEnabled(not testing)
self.delete_model_button.setEnabled(
has_model and not testing and self._can_delete_model(self._current_model())
)
self.test_connection_button.setEnabled(has_model and not testing)
def _set_test_running(self, running):
self._update_button_state()
self.test_connection_button.setEnabled(
not running and self._current_model() is not None
)
def _forget_test_thread(self, thread):
if self.test_thread is thread:
self.test_thread = None
self.test_worker = None
self._set_test_running(False)
def _on_test_finished(self, payload):
if payload.get("ok"):
status = payload.get("status")
suffix = f"(HTTP {status})" if status else ""
message = f"测试连接成功:{payload.get('name')}{suffix}"
else:
error = payload.get("error") or "连接失败"
status = payload.get("status")
status_text = f"HTTP {status}," if status else ""
message = f"测试连接失败:{status_text}{error}"
self.test_result_label.setText(message)
self._set_status(message)
def _on_test_failed(self, _task_id, error):
message = f"测试连接失败:{error}"
self.test_result_label.setText(message)
self._set_status(message)
def refresh_cmhub_models(self, checked=False):
self._start_cmhub_worker(include_balance=False)
def test_cmhub_connection(self, checked=False):
self._start_cmhub_worker(include_balance=True)
def _start_cmhub_worker(self, include_balance):
if self.cmhub_thread is not None:
self._set_status("cmhub 检测正在进行...")
return
base_url = self.cmhub_base_url_edit.text().strip()
api_key = self.cmhub_api_key_edit.text()
missing = []
if not base_url:
missing.append("Base URL")
if not api_key:
missing.append("API Key")
if missing:
self._show_error("cmhub 配置不完整:缺少 " + "、".join(missing))
return
worker = CMHubSettingsWorker(
base_url,
api_key,
connect_timeout=self.cmhub_connect_timeout_spin.value(),
include_balance=include_balance,
db_path=_database_path(config=self.config),
diagnostic_log_dir=diagnostics.DEFAULT_LOG_DIR,
)
worker.finished.connect(self._on_cmhub_finished)
worker.failed.connect(self._on_cmhub_failed)
thread = run_worker(worker, thread_name="CMHubSettingsWorker", start=False)
thread.finished.connect(lambda: self._forget_cmhub_thread(thread))
self.cmhub_worker = worker
self.cmhub_thread = thread
self._set_cmhub_running(True)
message = "正在测试 cmhub 连接并查询余额..." if include_balance else "正在刷新 cmhub 别名..."
self.cmhub_result_label.setText(message)
self._set_status(message)
thread.start()
def _set_cmhub_running(self, running):
enabled = not running
for widget in (
self.cmhub_base_url_edit,
self.cmhub_api_key_edit,
self.cmhub_title_alias_combo,
self.cmhub_image_alias_combo,
self.cmhub_connect_timeout_spin,
self.cmhub_check_balance_checkbox,
self.cmhub_refresh_button,
self.cmhub_test_button,
):
widget.setEnabled(enabled)
def _forget_cmhub_thread(self, thread):
if self.cmhub_thread is thread:
self.cmhub_thread = None
self.cmhub_worker = None
self._set_cmhub_running(False)
def _on_cmhub_finished(self, payload):
if not payload.get("ok", True):
self._on_cmhub_failed(-1, payload.get("error") or "连接失败")
return
models = [model for model in payload.get("models", []) if isinstance(model, dict)]
self.cmhub_models = models
current_cfg = appconfig.cmhub_config(self.config)
title_selected = self.cmhub_title_alias_combo.currentData() or current_cfg.get("title_alias", "")
image_selected = self.cmhub_image_alias_combo.currentData() or current_cfg.get("image_alias", "")
self._populate_cmhub_alias_combos(
models,
title_selected=title_selected,
image_selected=image_selected,
)
title_count = self._cmhub_alias_count("title")
image_count = self._cmhub_alias_count("image")
balance = payload.get("points_balance")
balance_text = f";余额 {balance}" if balance is not None else ""
message = f"cmhub 连接成功:生文别名 {title_count} 个,生图别名 {image_count} 个{balance_text}"
self.cmhub_result_label.setText(message)
self._set_status(message)
def _on_cmhub_failed(self, _task_id, error):
message = f"cmhub 连接失败:{error}"
self.cmhub_result_label.setText(message)
self._set_status(message)
def _populate_cmhub_alias_combos(self, models, title_selected="", image_selected=""):
self._populate_cmhub_alias_combo(
self.cmhub_title_alias_combo,
models,
"title",
title_selected,
)
self._populate_cmhub_alias_combo(
self.cmhub_image_alias_combo,
models,
"image",
image_selected,
)
def _populate_cmhub_alias_combo(self, combo, models, operation, selected):
combo.blockSignals(True)
combo.clear()
selected = str(selected or "").strip()
added = set()
for model in self._cmhub_priced_models(models, operation):
alias = str(model.get("alias") or "").strip()
if not alias or alias in added:
continue
combo.addItem(self._cmhub_alias_label(model), alias)
added.add(alias)
if selected and selected not in added:
combo.addItem(f"{selected}(已保存)", selected)
if combo.count() == 0:
combo.addItem("无可用别名", None)
index = combo.findData(selected)
combo.setCurrentIndex(index if index >= 0 else 0)
combo.blockSignals(False)
def _cmhub_priced_models(self, models, operation):
items = []
for model in models or []:
if not isinstance(model, dict):
continue
alias = str(model.get("alias") or "").strip()
op = str(model.get("operation_type") or "").lower()
pricing_status = str(model.get("pricing_status") or "").lower()
if alias and op == operation and pricing_status != "unpriced":
items.append(model)
return items
def _cmhub_alias_label(self, model):
alias = str(model.get("alias") or "").strip()
price_text = self._cmhub_price_text(model.get("prices"))
parts = [alias]
if price_text:
parts.append(price_text)
if model.get("requires_image"):
parts.append("需参考图")
return " · ".join(parts)
def _cmhub_price_text(self, prices):
if not isinstance(prices, list):
return ""
parts = []
for item in prices[:3]:
if not isinstance(item, dict):
continue
cost = item.get("points_cost")
if cost is None:
cost = item.get("cost")
if cost is None:
continue
resolution = item.get("resolution") or item.get("name") or ""
parts.append(f"{resolution}:{cost}点" if resolution else f"{cost}点")
return "/".join(parts)
def _cmhub_alias_count(self, operation):
combo = self.cmhub_title_alias_combo if operation == "title" else self.cmhub_image_alias_combo
return sum(1 for index in range(combo.count()) if combo.itemData(index))
def _show_error(self, error):
message = str(error)
QMessageBox.warning(self, "设置", message)
self._set_status(message)
def _should_warn_plaintext_api_key(self, model, current):
new_key = str((model or {}).get("api_key") or "")
current_key = str((current or {}).get("api_key") or "")
return bool(new_key) and new_key != current_key
def _should_warn_plaintext_cmhub_api_key(self, key):
new_key = str(key or "")
return bool(new_key) and new_key != str(self._loaded_cmhub_api_key or "")
def _show_plaintext_api_key_warning(self):
QMessageBox.warning(
self,
PLAINTEXT_SECRET_TITLE,
PLAINTEXT_API_KEY_WARNING,
)
def _show_plaintext_cmhub_api_key_warning(self):
QMessageBox.warning(
self,
PLAINTEXT_SECRET_TITLE,
PLAINTEXT_CMHUB_API_KEY_WARNING,
)
def _current_model(self):
return self._model_by_name(self.current_model_name)
def _model_by_name(self, name):
for model in self.models:
if model.get("name") == name:
return model
return None
def _unique_model_name(self, base):
names = {model.get("name") for model in self.models}
if base not in names:
return base
counter = 2
while f"{base} {counter}" in names:
counter += 1
return f"{base} {counter}"
def _can_delete_model(self, model):
if model is None:
return False
category = model.get("category")
return sum(1 for item in self.models if item.get("category") == category) > 1
def _category_label(self, category):
return {"text": "文本", "image": "图像"}.get(category, category)