"""AI 穿搭页签（docs/11 §10）。 以 Excel 为数据源，逐行调 AI 图像 API 生成「人物穿着该衣服」的效果图， 写回 Excel D/E/F。界面分三栏：左设置 / 中（最近结果 + 处理明细）/ 右运行日志。 后台用 QThread + Worker(QObject) 包住 core.outfit_batch.OutfitBatchRunner， 通过 Qt signal 回主线程刷新 UI（子线程不直接碰控件）。 """ import logging from PySide6.QtCore import QObject, QSize, Qt, QThread, QUrl, Signal from PySide6.QtGui import QDesktopServices, QIcon, QPixmap from PySide6.QtWidgets import ( QAbstractItemView, QCheckBox, QComboBox, QInputDialog, QDoubleSpinBox, QFileDialog, QGridLayout, QGroupBox, QHBoxLayout, QHeaderView, QLabel, QLineEdit, QListWidget, QListWidgetItem, QMessageBox, QPlainTextEdit, QProgressBar, QPushButton, QScrollArea, QSizePolicy, QSpinBox, QSplitter, QTableWidget, QTableWidgetItem, QVBoxLayout, QWidget, ) from services.config_service import ( DEFAULT_OUTFIT_PROMPT, load_ai_models, load_outfit_prompts, save_outfit_prompts, ) logger = logging.getLogger(__name__) _RESOLUTIONS = ["512", "1K", "2K", "4K"] _QUALITIES = ["小文件", "均衡", "高清"] _COLS = ["行", "标题", "货号", "衣服图", "状态", "结果 / 原因"] # --------------------------------------------------------------------------- # Background worker # --------------------------------------------------------------------------- class _OutfitWorker(QObject): """Runs the batch on a QThread; reports back via queued signals.""" tasks_loaded = Signal(object) # List[OutfitTask] log = Signal(str) progress = Signal(int, int, object) # completed, total, OutfitResult finished = Signal(object) # OutfitBatchSummary failed = Signal(str) # fatal pre-run error (e.g. Excel locked) def __init__(self, excel_path, output_dir, model_config, prompt, options, resolution, quality, retry_failed): super().__init__() self._excel_path = excel_path self._output_dir = output_dir self._model_config = model_config self._prompt = prompt self._options = options self._resolution = resolution self._quality = quality self._retry_failed = retry_failed self._runner = None def stop(self): if self._runner is not None: self._runner.stop() def run(self): # Imported lazily so the UI thread never pulls in Pillow/requests at import. from core.ai_outfit import generate_outfit_image from core.outfit_batch import OutfitBatchRunner, OutfitBatchSummary from services.excel_service import ( ensure_excel_writable, load_outfit_tasks, write_outfit_result, ) try: ensure_excel_writable(self._excel_path) tasks = load_outfit_tasks(self._excel_path, retry_failed=self._retry_failed) except Exception as exc: # noqa: BLE001 - report to UI self.failed.emit(str(exc)) return self.tasks_loaded.emit(tasks) if not tasks: self.finished.emit(OutfitBatchSummary(total=0)) return def gen(task): return generate_outfit_image( task, self._prompt, self._output_dir, self._model_config, quality=self._quality, resolution=self._resolution, ) def on_progress(completed, total, result): try: write_outfit_result(self._excel_path, result) except Exception as exc: # noqa: BLE001 - keep going self.log.emit("⚠ 写回 Excel 失败（第 {} 行）：{}".format( result.task.row_index, exc)) self.progress.emit(completed, total, result) self._runner = OutfitBatchRunner( tasks, gen, options=self._options, progress_callback=on_progress, log_callback=self.log.emit, ) try: summary = self._runner.run() except Exception as exc: # noqa: BLE001 - report to UI self.failed.emit(str(exc)) return self.finished.emit(summary) # --------------------------------------------------------------------------- # Main panel # --------------------------------------------------------------------------- class AiOutfitPanel(QWidget): """The "2 AI 穿搭" page.""" # Emitted when an outfit setting changes; MainWindow merges + persists it. config_changed = Signal(dict) def __init__(self, parent=None): super().__init__(parent) self._models = [] # list of model dicts from ai_models.json self._thread = None self._worker = None self._row_to_table = {} # excel row_index -> table row self._failures = [] # list of OutfitResult (failed) self._last_resolution = "" # for "value actually changed" check (§10.2) self._last_model = "" self._prompts = [] # list of {name, text} (§7.2) self._current_prompt_name = "" self._saved_text = "" # stored text of the selected template (dirty check) self._build_ui() # -- construction --------------------------------------------------- def _build_ui(self): outer = QVBoxLayout(self) outer.setContentsMargins(0, 0, 0, 0) splitter = QSplitter(Qt.Horizontal) splitter.setHandleWidth(1) splitter.addWidget(self._build_left()) splitter.addWidget(self._build_center()) splitter.addWidget(self._build_right()) splitter.setStretchFactor(0, 0) splitter.setStretchFactor(1, 1) splitter.setStretchFactor(2, 0) splitter.setSizes([360, 540, 400]) outer.addWidget(splitter) def _build_left(self): scroll = QScrollArea() scroll.setWidgetResizable(True) # AsNeeded (not AlwaysOff): if a child ever exceeds the column it scrolls # instead of being clipped under the center panel (docs/11 §10.1). scroll.setHorizontalScrollBarPolicy(Qt.ScrollBarAsNeeded) scroll.setMinimumWidth(320) inner = QWidget() col = QVBoxLayout(inner) col.setContentsMargins(12, 12, 12, 12) col.setSpacing(10) # 数据源 / 输出：行内一行（标签 + 路径 + 浏览），省纵向空间给话术/预览 self._excel_edit = QLineEdit() self._excel_edit.setPlaceholderText("选择商品表 .xlsx") col.addLayout(self._inline_path_row("Excel", self._excel_edit, self._browse_excel)) self._output_edit = QLineEdit() self._output_edit.setPlaceholderText("默认：程序旁的「合并后的图片」") col.addLayout(self._inline_path_row("输出", self._output_edit, self._browse_output)) # 穿搭生成话术（多套模板；加大，随窗口高度拉伸） prm = QGroupBox() pv = QVBoxLayout(prm) # 模板选择行：下拉 + 新建/另存为/重命名/删除 prompt_header = QHBoxLayout() prompt_header.addWidget(QLabel("穿搭生成话术")) self._prompt_combo = QComboBox() self._compact_combo(self._prompt_combo) self._prompt_combo.activated.connect(self._on_prompt_template_activated) prompt_header.addWidget(self._prompt_combo, stretch=1) pv.addLayout(prompt_header) trow = QHBoxLayout() for text, slot in (("新建", self._prompt_new), ("另存为", self._prompt_save_as), ("重命名", self._prompt_rename), ("删除", self._prompt_delete)): b = QPushButton(text) b.clicked.connect(slot) trow.addWidget(b) pv.addLayout(trow) self._prompt_edit = QPlainTextEdit() self._prompt_edit.setMinimumHeight(150) self._prompt_edit.textChanged.connect(self._refresh_preview) pv.addWidget(self._prompt_edit) prow = QHBoxLayout() insert_title_btn = QPushButton("插入标题") insert_title_btn.clicked.connect( lambda: self._prompt_edit.insertPlainText("{title}")) prow.addWidget(insert_title_btn) save_btn = QPushButton("保存话术") save_btn.clicked.connect(self._save_prompt) prow.addWidget(save_btn) pv.addLayout(prow) col.addWidget(prm, stretch=2) # 最终生成要求预览（加大；内嵌、实时；替换样本行占位符 + 附加输出要求） prev = QGroupBox() pvw = QVBoxLayout(prev) preview_header = QHBoxLayout() preview_header.addWidget(QLabel("最终生成要求预览")) self._sample_combo = QComboBox() self._compact_combo(self._sample_combo) self._sample_combo.currentIndexChanged.connect(self._refresh_preview) preview_header.addWidget(self._sample_combo, stretch=1) pvw.addLayout(preview_header) self._preview_warn = QLabel("") self._preview_warn.setStyleSheet("color:#b87a00;") self._preview_warn.setVisible(False) pvw.addWidget(self._preview_warn) self._preview_view = QPlainTextEdit() self._preview_view.setReadOnly(True) self._preview_view.setMinimumHeight(200) pvw.addWidget(self._preview_view) col.addWidget(prev, stretch=3) scroll.setWidget(inner) return scroll def _create_settings_group(self): """生成设置 group (lives in the right run column); two params per row.""" gen = QGroupBox() gv = QVBoxLayout(gen) header = QHBoxLayout() header.addWidget(QLabel("生成设置")) header.addStretch(1) self._retry_failed_chk = QCheckBox("重试上次失败的行") header.addWidget(self._retry_failed_chk) gv.addLayout(header) self._concurrency = QSpinBox() self._concurrency.setRange(1, 16) self._interval = QDoubleSpinBox() self._interval.setRange(0.0, 60.0) self._interval.setSuffix(" 秒") self._cooldown = QDoubleSpinBox() self._cooldown.setRange(0.0, 60.0) self._cooldown.setSuffix(" 秒") self._retry_count = QSpinBox() self._retry_count.setRange(0, 10) self._resolution = QComboBox() self._resolution.addItems(_RESOLUTIONS) self._resolution.currentIndexChanged.connect(self._refresh_preview) # activated = user click only; programmatic sets won't pop (docs/11 §10.2) self._resolution.activated.connect(self._on_resolution_activated) self._quality = QComboBox() self._quality.addItems(_QUALITIES) pairs = [ ("并发数", self._concurrency), ("新请求间隔", self._interval), ("单任务冷却", self._cooldown), ("失败重试", self._retry_count), ("分辨率", self._resolution), ("JPG 质量", self._quality), ] grid = QGridLayout() grid.setHorizontalSpacing(10) grid.setVerticalSpacing(8) for c in range(3): grid.setColumnStretch(c, 1) for i, (label, widget) in enumerate(pairs): grid.addWidget(self._field(label, widget), i // 3, i % 3) gv.addLayout(grid) # AI 模型 下拉（移到生成设置下方） self._model_combo = QComboBox() self._compact_combo(self._model_combo) self._model_combo.activated.connect(self._on_model_activated) model_row = QHBoxLayout() model_row.addWidget(QLabel("AI 模型")) model_row.addWidget(self._model_combo, stretch=1) gv.addLayout(model_row) return gen def _field(self, label_text, widget): """A compact label-above-control cell for the settings grid.""" cell = QWidget() v = QVBoxLayout(cell) v.setContentsMargins(0, 0, 0, 0) v.setSpacing(2) v.addWidget(QLabel(label_text)) v.addWidget(widget) return cell def _compact_combo(self, combo): """Keep a combo from dictating column width: elide long items instead of expanding its minimum size hint (docs/11 §10.1).""" combo.setSizeAdjustPolicy(QComboBox.AdjustToMinimumContentsLengthWithIcon) combo.setMinimumContentsLength(6) combo.setSizePolicy(QSizePolicy.Ignored, QSizePolicy.Preferred) def _inline_path_row(self, label_text, line_edit, on_browse): """Label + path field + 浏览 button on one row.""" row = QHBoxLayout() lbl = QLabel(label_text) lbl.setFixedWidth(40) row.addWidget(lbl) row.addWidget(line_edit, stretch=1) btn = QPushButton("浏览…") btn.clicked.connect(on_browse) row.addWidget(btn) return row def _build_center(self): wrap = QWidget() col = QVBoxLayout(wrap) col.setContentsMargins(12, 12, 12, 12) col.setSpacing(10) col.addWidget(QLabel("最近结果（单击查看大图）")) self._results = QListWidget() self._results.setViewMode(QListWidget.IconMode) self._results.setFlow(QListWidget.LeftToRight) self._results.setWrapping(False) self._results.setMovement(QListWidget.Static) self._results.setIconSize(QSize(96, 120)) self._results.setFixedHeight(168) self._results.itemClicked.connect(self._open_result) col.addWidget(self._results) col.addWidget(QLabel("处理明细（按 Excel 行顺序，每完成一行即写回并保存）")) self._table = QTableWidget(0, len(_COLS)) self._table.setHorizontalHeaderLabels(_COLS) self._table.setEditTriggers(QAbstractItemView.NoEditTriggers) self._table.setSelectionBehavior(QAbstractItemView.SelectRows) self._table.verticalHeader().setVisible(False) header = self._table.horizontalHeader() header.setSectionResizeMode(1, QHeaderView.Stretch) header.setSectionResizeMode(5, QHeaderView.Stretch) col.addWidget(self._table, stretch=1) return wrap def _build_right(self): wrap = QWidget() wrap.setMinimumWidth(360) col = QVBoxLayout(wrap) col.setContentsMargins(12, 12, 12, 12) col.setSpacing(10) col.addWidget(self._create_settings_group()) col.addWidget(QLabel("本次进度")) self._progress = QProgressBar() self._progress.setValue(0) col.addWidget(self._progress) self._stats = QLabel("完成 0 · 失败 0 · 待处理 0") col.addWidget(self._stats) run_row = QHBoxLayout() self._start_btn = QPushButton("开始生成") self._start_btn.setObjectName("primaryBtn") self._start_btn.clicked.connect(self._start) run_row.addWidget(self._start_btn) self._stop_btn = QPushButton("停止生成") self._stop_btn.setEnabled(False) self._stop_btn.clicked.connect(self._stop) run_row.addWidget(self._stop_btn) col.addLayout(run_row) row = QHBoxLayout() self._export_fail_btn = QPushButton("导出失败清单") self._export_fail_btn.setEnabled(False) self._export_fail_btn.clicked.connect(self._export_failures) row.addWidget(self._export_fail_btn) open_btn = QPushButton("打开输出目录") open_btn.clicked.connect(self._open_output_dir) row.addWidget(open_btn) col.addLayout(row) col.addWidget(QLabel("实时日志")) self._log = QPlainTextEdit() self._log.setReadOnly(True) col.addWidget(self._log, stretch=1) return wrap # -- config wiring -------------------------------------------------- def apply_config(self, config): """Populate widgets from the merged app config + load models/prompt.""" self._excel_edit.setText(config.get("outfit_excel", "")) self._output_edit.setText(config.get("outfit_output_dir", "")) self._concurrency.setValue(int(config.get("outfit_concurrency", 1) or 1)) self._interval.setValue(float(config.get("outfit_request_interval", 2.0) or 0.0)) self._cooldown.setValue(float(config.get("outfit_task_cooldown", 1.0) or 0.0)) self._retry_count.setValue(int(config.get("outfit_retry_count", 2) or 0)) self._retry_failed_chk.setChecked(bool(config.get("outfit_retry_failed", False))) self._set_combo(self._resolution, config.get("outfit_resolution", "1K")) self._set_combo(self._quality, config.get("outfit_quality", "均衡")) # 话术模板：载入多套 + 选中上次（§7.2） self._prompts = load_outfit_prompts() names = [p["name"] for p in self._prompts] name = config.get("outfit_prompt_name", "") if name not in names: name = names[0] self._rebuild_prompt_combo() self._apply_prompt(name) self._models = load_ai_models() self._model_combo.clear() if not self._models: self._model_combo.addItem("（未配置模型，请在 ai_models.json 添加）") self._model_combo.setEnabled(False) else: self._model_combo.setEnabled(True) for m in self._models: self._model_combo.addItem(m.get("name") or m.get("model") or "(未命名)") self._set_combo(self._model_combo, config.get("outfit_model", "")) # Snapshot current dropdown values so a later user re-select of the same # item doesn't trigger the info popup (§10.2). self._last_resolution = self._resolution.currentText() self._last_model = self._model_combo.currentText() if self._models else "" # Fill the preview's sample-row dropdown from the remembered Excel. self._reload_sample_rows() def _set_combo(self, combo, value): idx = combo.findText(str(value)) if idx >= 0: combo.setCurrentIndex(idx) def _emit_config(self): self.config_changed.emit({ "outfit_excel": self._excel_edit.text(), "outfit_output_dir": self._output_edit.text(), "outfit_model": self._model_combo.currentText() if self._models else "", "outfit_concurrency": self._concurrency.value(), "outfit_request_interval": self._interval.value(), "outfit_task_cooldown": self._cooldown.value(), "outfit_retry_count": self._retry_count.value(), "outfit_resolution": self._resolution.currentText(), "outfit_quality": self._quality.currentText(), "outfit_retry_failed": self._retry_failed_chk.isChecked(), "outfit_prompt_name": self._current_prompt_name, }) # -- left actions --------------------------------------------------- def _browse_excel(self): path, _ = QFileDialog.getOpenFileName( self, "选择 Excel 文件", self._excel_edit.text(), "Excel 文件 (*.xlsx)") if path: self._excel_edit.setText(path) self._emit_config() self._reload_sample_rows() def _browse_output(self): path = QFileDialog.getExistingDirectory( self, "选择输出目录", self._output_edit.text()) if path: self._output_edit.setText(path) self._emit_config() def _save_prompt(self): self._store_current_text() self.statusBar_message("话术已保存") # -- 话术模板（§7.2）------------------------------------------------ def _rebuild_prompt_combo(self): """Refill the template dropdown from self._prompts (no signal).""" self._prompt_combo.blockSignals(True) self._prompt_combo.clear() self._prompt_combo.addItems([p["name"] for p in self._prompts]) self._prompt_combo.blockSignals(False) def _select_prompt_in_combo(self, name): idx = self._prompt_combo.findText(name) if idx >= 0: self._prompt_combo.blockSignals(True) self._prompt_combo.setCurrentIndex(idx) self._prompt_combo.blockSignals(False) def _prompt_text(self, name): return next((p["text"] for p in self._prompts if p["name"] == name), "") def _apply_prompt(self, name): """Load template *name* into the editor (no persistence).""" self._current_prompt_name = name self._saved_text = self._prompt_text(name) self._select_prompt_in_combo(name) self._prompt_edit.setPlainText(self._saved_text) # fires _refresh_preview def _store_current_text(self): """Save the editor text into the current template + persist to disk.""" text = self._prompt_edit.toPlainText() for p in self._prompts: if p["name"] == self._current_prompt_name: p["text"] = text break save_outfit_prompts(self._prompts) self._saved_text = text def _is_dirty(self): return self._prompt_edit.toPlainText() != self._saved_text def _maybe_save_dirty(self): """Handle unsaved edits before switching away. Return False = cancel.""" if not self._is_dirty(): return True ans = QMessageBox.question( self, "未保存", "当前话术「{}」有未保存的修改，是否保存？".format( self._current_prompt_name), QMessageBox.Save | QMessageBox.Discard | QMessageBox.Cancel, QMessageBox.Save) if ans == QMessageBox.Cancel: return False if ans == QMessageBox.Save: self._store_current_text() return True def _name_exists(self, name): return any(p["name"] == name for p in self._prompts) def _ask_name(self, title, default=""): """Prompt for a unique non-empty template name; None if cancelled/invalid.""" name, ok = QInputDialog.getText(self, title, "模板名称：", text=default) if not ok: return None name = name.strip() if not name: QMessageBox.warning(self, "名称无效", "模板名称不能为空。") return None if self._name_exists(name): QMessageBox.warning(self, "名称重复", "已存在同名模板：{}".format(name)) return None return name def _on_prompt_template_activated(self, index): name = self._prompt_combo.itemText(index) if name == self._current_prompt_name: return if not self._maybe_save_dirty(): self._select_prompt_in_combo(self._current_prompt_name) # cancel: revert return self._apply_prompt(name) self._emit_config() def _prompt_new(self): if not self._maybe_save_dirty(): return name = self._ask_name("新建话术") if name is None: return self._prompts.append({"name": name, "text": DEFAULT_OUTFIT_PROMPT}) save_outfit_prompts(self._prompts) self._rebuild_prompt_combo() self._apply_prompt(name) self._emit_config() def _prompt_save_as(self): name = self._ask_name("另存为", default=self._current_prompt_name) if name is None: return self._prompts.append({"name": name, "text": self._prompt_edit.toPlainText()}) save_outfit_prompts(self._prompts) self._rebuild_prompt_combo() self._apply_prompt(name) self._emit_config() def _prompt_rename(self): new = self._ask_name("重命名", default=self._current_prompt_name) if new is None: return for p in self._prompts: if p["name"] == self._current_prompt_name: p["name"] = new break self._current_prompt_name = new save_outfit_prompts(self._prompts) self._rebuild_prompt_combo() self._select_prompt_in_combo(new) self._emit_config() def _prompt_delete(self): if len(self._prompts) <= 1: QMessageBox.information(self, "无法删除", "至少保留一套话术。") return ans = QMessageBox.question( self, "删除话术", "确定删除话术「{}」？".format(self._current_prompt_name), QMessageBox.Yes | QMessageBox.No, QMessageBox.No) if ans != QMessageBox.Yes: return idx = next((i for i, p in enumerate(self._prompts) if p["name"] == self._current_prompt_name), 0) self._prompts.pop(idx) save_outfit_prompts(self._prompts) self._rebuild_prompt_combo() self._apply_prompt(self._prompts[min(idx, len(self._prompts) - 1)]["name"]) self._emit_config() # -- inline prompt preview ------------------------------------------ def _fill_sample_combo(self, tasks): """Populate the sample-row dropdown (None data = no sample).""" self._sample_combo.blockSignals(True) self._sample_combo.clear() if tasks: for t in tasks: self._sample_combo.addItem( "第 {} 行 · {} · {}".format(t.row_index, t.product_id, t.title), t) else: self._sample_combo.addItem("（选 Excel 后显示替换效果）", None) self._sample_combo.blockSignals(False) self._refresh_preview() def _reload_sample_rows(self): """Best-effort: read the chosen Excel to fill the sample dropdown. Uses read_all_rows (status-independent) so preview still works after the whole sheet is 完成 (docs/11 §10.3); generation still uses load_outfit_tasks. """ excel = self._excel_edit.text().strip() rows = [] if excel: try: from services.excel_service import read_all_rows rows = read_all_rows(excel) except Exception as exc: # noqa: BLE001 - silent for preview logger.info("Sample rows unavailable: %s", exc) self._fill_sample_combo(rows) def _refresh_preview(self): if not hasattr(self, "_preview_view"): return from core.ai_outfit import build_output_requirements, render_prompt template = self._prompt_edit.toPlainText() self._preview_warn.setVisible("{title}" not in template) if "{title}" not in template: self._preview_warn.setText("⚠ 话术缺少 {title} 占位符") # Mirror what actually gets sent: append the output-requirements block # for the currently selected resolution (docs/11 §7.1). resolution = self._resolution.currentText() if hasattr(self, "_resolution") else None task = self._sample_combo.currentData() if hasattr(self, "_sample_combo") else None if task is None: self._preview_view.setPlainText(template + build_output_requirements(resolution)) else: self._preview_view.setPlainText(render_prompt(template, task, resolution)) # -- switch info popups (user-only; §10.2) -------------------------- def _on_resolution_activated(self, _index): res = self._resolution.currentText() if res == self._last_resolution: return self._last_resolution = res from services.ai_image_service import resolution_timeout QMessageBox.information( self, "分辨率已切换", "已切换分辨率到 {}。\n\n" "· 单任务超时约 {} 秒，分辨率越高越慢。\n" "· 仅在下次「开始生成」生效，不影响正在进行的批次。".format( res, resolution_timeout(res))) def _on_model_activated(self, _index): name = self._model_combo.currentText() if name == self._last_model: return self._last_model = name api_type = "auto" idx = self._model_combo.currentIndex() if self._models and 0 <= idx < len(self._models): api_type = self._models[idx].get("api_type", "auto") or "auto" QMessageBox.information( self, "AI 模型已切换", "已切换模型到 {}。\n\n" "· 调用方式：{}。\n" "· 不同模型的计费与效果可能不同。\n" "· 仅在下次「开始生成」生效。".format(name, api_type)) # -- run control ---------------------------------------------------- def _start(self): if self._thread is not None: return excel = self._excel_edit.text().strip() if not excel: QMessageBox.information(self, "提示", "请先选择 Excel 文件。") return model_config = self._selected_model_config() if model_config is None: return prompt = self._prompt_edit.toPlainText() if "{title}" not in prompt: answer = QMessageBox.question( self, "缺少占位符", "话术中没有 {title} 占位符，生成时不会带入商品标题。仍要继续吗？", QMessageBox.Yes | QMessageBox.No, QMessageBox.No) if answer != QMessageBox.Yes: return self._store_current_text() # persist editor into the selected template self._emit_config() output_dir = self._output_edit.text().strip() if not output_dir: from services.file_service import get_output_dir output_dir = str(get_output_dir()) from core.outfit_batch import OutfitBatchOptions options = OutfitBatchOptions( concurrency=self._concurrency.value(), request_interval=self._interval.value(), task_cooldown=self._cooldown.value(), retry_count=self._retry_count.value(), ) # reset run state self._table.setRowCount(0) self._results.clear() self._row_to_table = {} self._failures = [] self._progress.setValue(0) self._log.clear() self._worker = _OutfitWorker( excel, output_dir, model_config, prompt, options, self._resolution.currentText(), self._quality.currentText(), self._retry_failed_chk.isChecked(), ) self._thread = QThread(self) self._worker.moveToThread(self._thread) self._thread.started.connect(self._worker.run) self._worker.tasks_loaded.connect(self._on_tasks_loaded) self._worker.log.connect(self._append_log) self._worker.progress.connect(self._on_progress) self._worker.finished.connect(self._on_finished) self._worker.failed.connect(self._on_failed) self._worker.finished.connect(self._thread.quit) self._worker.failed.connect(self._thread.quit) self._thread.finished.connect(self._cleanup_thread) self._thread.start() self._set_running(True) def _stop(self): if self._worker is not None: self._worker.stop() self._append_log("已请求停止：不再提交新任务，进行中的任务会收尾。") self._stop_btn.setEnabled(False) def _selected_model_config(self): if not self._models: QMessageBox.warning( self, "未配置模型", "尚未配置 AI 模型。请在 ~/.cmbot/config/ai_models.json 添加后重试。") return None data = self._models[self._model_combo.currentIndex()] from services.ai_image_service import AiModelConfig, api_config_errors errors = api_config_errors(data) if errors: QMessageBox.warning(self, "模型配置有误", "；".join(errors)) return None return AiModelConfig.from_dict(data) def _set_running(self, running): self._start_btn.setEnabled(not running) self._stop_btn.setEnabled(running) self._excel_edit.setEnabled(not running) self._model_combo.setEnabled(not running and bool(self._models)) # -- worker callbacks (UI thread) ----------------------------------- def _on_tasks_loaded(self, tasks): self._table.setRowCount(len(tasks)) for row, task in enumerate(tasks): self._row_to_table[task.row_index] = row self._set_cell(row, 0, str(task.row_index)) self._set_cell(row, 1, task.title) self._set_cell(row, 2, task.product_id) self._set_cell(row, 3, _basename(task.garment_path)) self._set_cell(row, 4, "待处理") self._set_cell(row, 5, "—") self._progress.setMaximum(max(1, len(tasks))) self._update_stats(0, 0, len(tasks)) self._fill_sample_combo(tasks) self._append_log("已加载 {} 行待处理任务".format(len(tasks))) def _on_progress(self, completed, total, result): row = self._row_to_table.get(result.task.row_index) if row is not None: if result.success: self._set_cell(row, 4, "完成") self._set_cell(row, 5, result.output_path) else: self._set_cell(row, 4, "失败") self._set_cell(row, 5, result.error) self._progress.setValue(completed) if result.success: self._add_result_thumb(result) else: self._failures.append(result) self._export_fail_btn.setEnabled(True) failed = len(self._failures) self._update_stats(completed - failed, failed, total - completed) def _on_finished(self, summary): self._set_running(False) msg = "完成 {}，失败 {}{}".format( summary.success_count, summary.failure_count, "（已停止）" if getattr(summary, "stopped", False) else "") self._append_log("批量结束：" + msg) QMessageBox.information(self, "AI 穿搭", "本次生成结束。\n" + msg) def _on_failed(self, message): self._set_running(False) self._append_log("无法开始：" + message) QMessageBox.warning(self, "无法开始", message) def _cleanup_thread(self): self._thread = None self._worker = None # -- helpers -------------------------------------------------------- def _add_result_thumb(self, result): pix = QPixmap(result.output_path) item = QListWidgetItem(result.task.product_id) if not pix.isNull(): item.setIcon(QIcon(pix)) item.setData(Qt.UserRole, result.output_path) self._results.insertItem(0, item) def _open_result(self, item): path = item.data(Qt.UserRole) if path: QDesktopServices.openUrl(QUrl.fromLocalFile(path)) def _open_output_dir(self): path = self._output_edit.text().strip() if not path: from services.file_service import get_output_dir path = str(get_output_dir()) QDesktopServices.openUrl(QUrl.fromLocalFile(path)) def _export_failures(self): if not self._failures: return path, _ = QFileDialog.getSaveFileName( self, "导出失败清单", "失败清单.csv", "CSV 文件 (*.csv)") if not path: return try: with open(path, "w", encoding="utf-8-sig", newline="") as f: f.write("行,标题,货号,原因\n") for r in self._failures: f.write("{},{},{},{}\n".format( r.task.row_index, _csv(r.task.title), _csv(r.task.product_id), _csv(r.error))) self.statusBar_message("失败清单已导出") except OSError as exc: QMessageBox.warning(self, "导出失败", str(exc)) def _append_log(self, message): self._log.appendPlainText(message) def _update_stats(self, done, failed, pending): self._stats.setText("完成 {} · 失败 {} · 待处理 {}".format( max(0, done), max(0, failed), max(0, pending))) def _set_cell(self, row, col, text): self._table.setItem(row, col, QTableWidgetItem(str(text))) def statusBar_message(self, text): win = self.window() if hasattr(win, "statusBar"): win.statusBar().showMessage(text, 4000) def _basename(path): import os return os.path.basename(str(path)) def _csv(value): text = str(value).replace('"', '""') if any(c in text for c in (",", "\n", '"')): return '"' + text + '"' return text