"""Tab 5: settings UI.""" from __future__ import annotations from contextlib import contextmanager import os from ... import ai as ai_module from ... import chrome from ... import cmhub_models from ..widgets import * from ..workers import AIModelTestWorker as _RealAIModelTestWorker from ..workers import CMHubSettingsWorker as _RealCMHubSettingsWorker PLAINTEXT_CMHUB_API_KEY_WARNING = ( "cmhub API Key 会以本地明文保存到 data/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 data/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_config_path(self.config) ) self.cmhub_config_path = ( self.config.get("cmhub_config_path") or appconfig.cmhub_config_file_path(self.config) ) 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._dirty = False self._suspend_dirty = 0 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://host（不要带 /api 或 /api/v1）") 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_vision_alias_combo = QComboBox() self.cmhub_vision_alias_combo.setObjectName("cmhubVisionAliasCombo") 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(appconfig.CMHUB_CONNECT_TIMEOUT_DEFAULT) 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_base_url_hint_label = QLabel("Base URL 只填网关根，如 https://host；不要带 /api 或 /api/v1。") self.cmhub_base_url_hint_label.setObjectName("cmhubBaseUrlHintLabel") self.cmhub_base_url_hint_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( appconfig.AI_CONCURRENCY_MIN, appconfig.AI_CONCURRENCY_MAX, ) self.image_concurrency_spin = QSpinBox() self.image_concurrency_spin.setObjectName("imageConcurrencySpin") self.image_concurrency_spin.setRange( appconfig.AI_CONCURRENCY_MIN, appconfig.AI_CONCURRENCY_MAX, ) self.retry_spin = QSpinBox() self.retry_spin.setObjectName("retrySpin") self.retry_spin.setRange(appconfig.AI_RETRY_MIN, appconfig.AI_RETRY_MAX) 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.chrome_path_edit = QLineEdit() self.chrome_path_edit.setObjectName("chromePathEdit") self.chrome_path_browse_button = QPushButton("选择...") self.chrome_path_browse_button.setObjectName("chromePathBrowseButton") self.chrome_path_detect_button = QPushButton("自动检测") self.chrome_path_detect_button.setObjectName("chromePathAutoDetectButton") self.chrome_path_widget = QWidget() self.chrome_path_widget.setObjectName("chromePathWidget") chrome_path_layout = QHBoxLayout(self.chrome_path_widget) chrome_path_layout.setContentsMargins(0, 0, 0, 0) chrome_path_layout.setSpacing(8) chrome_path_layout.addWidget(self.chrome_path_edit, 1) chrome_path_layout.addWidget(self.chrome_path_browse_button) chrome_path_layout.addWidget(self.chrome_path_detect_button) self.user_data_root_edit = QLineEdit() self.user_data_root_edit.setObjectName("userDataRootEdit") self.user_data_root_edit.setEnabled(False) self.user_data_root_edit.setVisible(False) self.image_dir_edit = QLineEdit() self.image_dir_edit.setObjectName("imageDirEdit") self.image_dir_edit.setEnabled(False) self.image_dir_edit.setVisible(False) self.db_path_edit = QLineEdit() self.db_path_edit.setObjectName("dbPathEdit") self.db_path_edit.setEnabled(False) self.db_path_edit.setVisible(False) 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.unsaved_changes_label = QLabel("● 未保存更改") self.unsaved_changes_label.setObjectName("settingsUnsavedChangesLabel") self.unsaved_changes_label.setStyleSheet("color: #bc4c00; font-weight: 600;") self.unsaved_changes_label.setVisible(False) 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.max_parallel_accounts_spin = QSpinBox() self.max_parallel_accounts_spin.setObjectName("maxParallelAccountsSpin") self.max_parallel_accounts_spin.setRange( appconfig.SHOPEE_PARALLEL_ACCOUNTS_MIN, appconfig.SHOPEE_PARALLEL_ACCOUNTS_MAX, ) self.max_parallel_accounts_spin.setToolTip("1=逐个更新，不并行；大于1时按账号并行更新，最多同时5个账号。") 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), ] ) 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_widget, True), ("默认调试端口", self.default_debug_port_spin), ("调试端口范围", port_range_widget), ("Chrome 就绪超时(秒)", self.cdp_ready_timeout_spin), ] ) self.infrastructure_form_layout = path_form self.shopee_update_form_layout = self._three_column_form( [ ("每批最大更新条数", self.max_items_per_run_spin), ("同时更新蝦皮账号", self.max_parallel_accounts_spin), ] ) 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_vision_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_base_url_hint_label) 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( "蝦皮更新执行", "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) save_settings_layout = QHBoxLayout() save_settings_layout.setContentsMargins(0, 0, 0, 0) save_settings_layout.addWidget(self.save_config_button) save_settings_layout.addWidget(self.unsaved_changes_label) save_settings_layout.addStretch(1) panel_layout.addLayout(save_settings_layout) 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.chrome_path_browse_button.clicked.connect(self.browse_chrome_path) self.chrome_path_detect_button.clicked.connect(self.detect_chrome_path) self.save_config_button.clicked.connect(self.save_app_settings) self._connect_dirty_signals() with self._dirty_tracking_suspended(): self.refresh_models() self._populate_app_settings() self._set_dirty(False) 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, level=None): _emit_status(self.status_callback, message, level=level) @contextmanager def _dirty_tracking_suspended(self): self._suspend_dirty += 1 try: yield finally: self._suspend_dirty -= 1 def _connect_dirty_signals(self): line_edits = ( self.cmhub_base_url_edit, self.cmhub_api_key_edit, self.name_edit, self.model_id_edit, self.url_edit, self.api_key_edit, self.chrome_path_edit, ) combos = ( self.backend_combo, self.cmhub_title_alias_combo, self.cmhub_image_alias_combo, self.cmhub_vision_alias_combo, self.category_combo, self.api_type_combo, self.default_text_model_combo, self.default_image_model_combo, self.resolution_combo, ) spin_boxes = ( self.cmhub_connect_timeout_spin, self.connect_timeout_spin, self.title_concurrency_spin, self.image_concurrency_spin, self.retry_spin, self.default_debug_port_spin, self.debug_port_start_spin, self.debug_port_end_spin, self.cdp_ready_timeout_spin, self.max_items_per_run_spin, self.max_parallel_accounts_spin, ) checkboxes = ( self.cmhub_check_balance_checkbox, self.enabled_checkbox, ) for widget in line_edits: widget.textChanged.connect(self._mark_dirty) for widget in combos: widget.currentIndexChanged.connect(self._mark_dirty) for widget in spin_boxes: widget.valueChanged.connect(self._mark_dirty) for widget in checkboxes: widget.toggled.connect(self._mark_dirty) def _mark_dirty(self, *args): if self._suspend_dirty > 0: return self._set_dirty(True) def _set_dirty(self, dirty): self._dirty = bool(dirty) self.unsaved_changes_label.setVisible(self._dirty) def is_dirty(self): return self._dirty def browse_chrome_path(self, checked=False): current = self.chrome_path_edit.text().strip() initial_dir = "" if current: parent_dir = os.path.dirname(os.path.expanduser(current)) if parent_dir and os.path.isdir(parent_dir): initial_dir = parent_dir path, _selected_filter = QFileDialog.getOpenFileName( self, "选择 Chrome 程序", initial_dir, "Chrome 程序 (chrome.exe);;可执行文件 (*.exe);;所有文件 (*)", ) if not path: return self.chrome_path_edit.setText(path) def detect_chrome_path(self, checked=False): path = chrome.detect_chrome_path() if not path: self._set_status( "未找到 Chrome，请点『选择...』手动指定，或先安装 Chrome", level="warning", ) return self.chrome_path_edit.setText(path) self._set_status(f"已自动定位 Chrome：{path}", level="success") def discard_unsaved_changes(self): try: saved = appconfig.load_config(self.config_path) except Exception as exc: self._show_error(exc) return False self._replace_config(saved) self._populate_app_settings() self._set_dirty(False) self._set_status("已放弃未保存更改") return True 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 _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 False 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 False self._replace_config(saved) self._populate_app_settings() self._set_dirty(False) self._set_status("设置已保存") QMessageBox.information(self, "保存设置", "设置已保存") return True 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(), "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", "data_dir"} } try: current_update_cfg = appconfig.shopee_update_config(appconfig.load_config(self.config_path)) except Exception: current_update_cfg = self._shopee_update_config() update_mode = appconfig.normalize_update_mode( current_update_cfg.get("update_mode"), allow_cover_update=current_update_cfg.get("allow_cover_update", False), ) settings.update( { "chrome_path": self.chrome_path_edit.text().strip(), "user_data_root": self._preserved_config_text( "user_data_root", "chrome_user_data_dir", ), "image_dir": self._preserved_config_text("image_dir", "images"), "db_path": self._preserved_config_text("db_path", "cmshopee.db"), "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 ""), "update_mode": update_mode, "max_items_per_run": self.max_items_per_run_spin.value(), "dry_run": False, "max_parallel_accounts": self.max_parallel_accounts_spin.value(), }, } ) return settings def _preserved_config_text(self, key, fallback): value = self.config.get(key) if value is None: value = fallback return str(value or "") def _cmhub_settings_values(self, backend): current = appconfig.cmhub_config(self.config) values = { "base_url": appconfig.normalize_cmhub_base_url(self.cmhub_base_url_edit.text()), "title_alias": self.cmhub_title_alias_combo.currentData() or "", "image_alias": self.cmhub_image_alias_combo.currentData() or "", "vision_alias": self.cmhub_vision_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 if self.config.get("data_dir"): internal["data_dir"] = self.config.get("data_dir") self.config.clear() self.config.update(saved) self.config.update(internal) def _populate_app_settings(self): if self._suspend_dirty <= 0: with self._dirty_tracking_suspended(): self._populate_app_settings() return 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(appconfig.normalize_cmhub_base_url(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", appconfig.CMHUB_CONNECT_TIMEOUT_DEFAULT, ) or appconfig.CMHUB_CONNECT_TIMEOUT_DEFAULT ), ) ) 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", ""), vision_selected=cmhub_cfg.get("vision_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.chrome_path_edit.setText(appconfig.chrome_path(self.config)) self.user_data_root_edit.setText( str(self.config.get("user_data_root", "chrome_user_data_dir") or "") ) self.image_dir_edit.setText(str(self.config.get("image_dir", "images") or "")) self.db_path_edit.setText(str(self.config.get("db_path", "cmshopee.db") or "")) 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.max_items_per_run_spin.setValue( max(1, int(update_cfg.get("max_items_per_run", 1) or 1)) ) self.max_parallel_accounts_spin.setValue( max( appconfig.SHOPEE_PARALLEL_ACCOUNTS_MIN, min( appconfig.SHOPEE_PARALLEL_ACCOUNTS_MAX, int(update_cfg.get("max_parallel_accounts", 1) or 1), ), ) ) self._update_response_timeout_label() self._on_backend_changed() def _shopee_update_config(self): return appconfig.shopee_update_config(self.config) 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) if appconfig.ai_backend(self.config) == "cmhub": self.response_timeout_label.setText( "标题 %s 秒 / 图片 %s 秒" % ( ai_module.CMHUB_TITLE_READ_TIMEOUT_SECONDS, ai_module.CMHUB_IMAGE_READ_TIMEOUT_SECONDS, ) ) return 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 = appconfig.normalize_cmhub_base_url(self.cmhub_base_url_edit.text()) if base_url != self.cmhub_base_url_edit.text().strip(): self.cmhub_base_url_edit.setText(base_url) 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_vision_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", "") vision_selected = self.cmhub_vision_alias_combo.currentData() or current_cfg.get("vision_alias", "") self._populate_cmhub_alias_combos( models, title_selected=title_selected, image_selected=image_selected, vision_selected=vision_selected, ) title_count = self._cmhub_alias_count("title") image_count = self._cmhub_alias_count("image") vision_aliases = { str(model.get("alias") or "").strip() for model in self._cmhub_priced_models(models, "vision") } vision_count = len(vision_aliases) balance = payload.get("points_balance") balance_text = f"；余额 {balance}" if balance is not None else "" subject = self._cmhub_success_subject(payload) vision_text = f"图片理解别名 {vision_count} 个" if vision_selected and vision_selected not in vision_aliases: vision_text += "（当前已保存值暂不可用）" message = f"{subject}：生文别名 {title_count} 个，生图别名 {image_count} 个，{vision_text}{balance_text}；别名已拉取，记得点『保存设置』持久化" self.cmhub_result_label.setText(message) self._set_status(message) def _cmhub_success_subject(self, payload): account_name = self._cmhub_account_display_name(payload) if account_name: return f"cmhub 账号「{account_name}」连接成功" return "cmhub 连接成功" def _cmhub_account_display_name(self, payload): if not isinstance(payload, dict): return "" sources = [] balance = payload.get("balance") if isinstance(balance, dict): account = balance.get("account") if isinstance(account, dict): sources.append(account) user = balance.get("user") if isinstance(user, dict): sources.append(user) elif user is not None: sources.append({"user": user}) sources.append(balance) account = payload.get("account") if isinstance(account, dict): sources.append(account) user = payload.get("user") if isinstance(user, dict): sources.append(user) elif user is not None: sources.append({"user": user}) sources.append(payload) for source in sources: name = self._cmhub_display_name_from_source(source) if name: return name return "" def _cmhub_display_name_from_source(self, source): if not isinstance(source, dict): return "" for field in ("display_name", "name", "account_name", "username", "user", "email", "id"): value = source.get(field) if value is None: continue if isinstance(value, dict): nested_name = self._cmhub_display_name_from_source(value) if nested_name: return nested_name continue text = str(value).strip() if not text: continue if field == "email" or "@" in text: return appconfig.mask_email(text) return text return "" 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="", vision_selected="", ): if self._suspend_dirty <= 0: with self._dirty_tracking_suspended(): self._populate_cmhub_alias_combos( models, title_selected, image_selected, vision_selected, ) return 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, ) self._populate_cmhub_alias_combo( self.cmhub_vision_alias_combo, models, "vision", vision_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(cmhub_models.alias_label(model), alias) combo.setItemData(combo.count() - 1, cmhub_models.alias_tooltip(model), Qt.ToolTipRole) 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() requires_image = bool(model.get("requires_image")) if ( alias and op == operation and pricing_status != "unpriced" and ( operation != "vision" or (pricing_status == "priced" and requires_image) ) ): items.append(model) return items def _cmhub_alias_count(self, operation): combos = { "title": self.cmhub_title_alias_combo, "image": self.cmhub_image_alias_combo, "vision": self.cmhub_vision_alias_combo, } combo = combos.get(operation) if combo is None: return 0 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)