"""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, QDoubleSpinBox, QFileDialog, QGroupBox, QHBoxLayout, QHeaderView, QLabel, QLineEdit, QListWidget, QListWidgetItem, QMessageBox, QPlainTextEdit, QProgressBar, QPushButton, QScrollArea, QSpinBox, QSplitter, QTableWidget, QTableWidgetItem, QVBoxLayout, QWidget, ) from services.config_service import ( load_ai_models, load_outfit_prompt, save_outfit_prompt, ) 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._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([430, 760, 300]) outer.addWidget(splitter) def _build_left(self): scroll = QScrollArea() scroll.setWidgetResizable(True) scroll.setHorizontalScrollBarPolicy(Qt.ScrollBarAlwaysOff) scroll.setMinimumWidth(360) inner = QWidget() col = QVBoxLayout(inner) col.setContentsMargins(12, 12, 12, 12) col.setSpacing(12) # 数据源 src = QGroupBox("数据源（Excel）") sv = QVBoxLayout(src) self._excel_edit = QLineEdit() self._excel_edit.setPlaceholderText("选择商品表 .xlsx") sv.addLayout(self._path_row(self._excel_edit, self._browse_excel)) col.addWidget(src) # 输出 + 模型 out = QGroupBox("输出") ov = QVBoxLayout(out) self._output_edit = QLineEdit() self._output_edit.setPlaceholderText("默认：程序旁的「合并后的图片」") ov.addLayout(self._path_row(self._output_edit, self._browse_output)) ov.addWidget(QLabel("AI 模型")) self._model_combo = QComboBox() ov.addWidget(self._model_combo) col.addWidget(out) # 通用话术 prm = QGroupBox("通用话术") pv = QVBoxLayout(prm) self._prompt_edit = QPlainTextEdit() self._prompt_edit.setFixedHeight(120) 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) # 最终提示词预览（内嵌、实时；替换样本行的占位符） prev = QGroupBox("最终提示词预览") pvw = QVBoxLayout(prev) srow = QHBoxLayout() srow.addWidget(QLabel("样本行")) self._sample_combo = QComboBox() self._sample_combo.currentIndexChanged.connect(self._refresh_preview) srow.addWidget(self._sample_combo, stretch=1) pvw.addLayout(srow) 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(140) pvw.addWidget(self._preview_view) col.addWidget(prev) col.addStretch() scroll.setWidget(inner) return scroll def _create_settings_group(self): """生成设置 group (lives in the right run column).""" from PySide6.QtWidgets import QFormLayout gen = QGroupBox("生成设置") gv = QVBoxLayout(gen) self._retry_failed_chk = QCheckBox("重试上次失败的行") gv.addWidget(self._retry_failed_chk) form = QFormLayout() 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._quality = QComboBox() self._quality.addItems(_QUALITIES) form.addRow("并发数", self._concurrency) form.addRow("新请求间隔", self._interval) form.addRow("单任务冷却", self._cooldown) form.addRow("失败重试", self._retry_count) form.addRow("分辨率", self._resolution) form.addRow("JPG 质量", self._quality) gv.addLayout(form) return gen def _path_row(self, line_edit, on_browse): row = QHBoxLayout() 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(300) 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) self._start_btn = QPushButton("开始生成") self._start_btn.setObjectName("primaryBtn") self._start_btn.clicked.connect(self._start) col.addWidget(self._start_btn) self._stop_btn = QPushButton("停止生成") self._stop_btn.setEnabled(False) self._stop_btn.clicked.connect(self._stop) col.addWidget(self._stop_btn) 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", "均衡")) self._prompt_edit.setPlainText(load_outfit_prompt()) 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", "")) # 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(), }) # -- 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): save_outfit_prompt(self._prompt_edit.toPlainText()) self.statusBar_message("话术已保存") # -- 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.""" excel = self._excel_edit.text().strip() tasks = [] if excel: try: from services.excel_service import load_outfit_tasks tasks = load_outfit_tasks( excel, retry_failed=self._retry_failed_chk.isChecked()) except Exception as exc: # noqa: BLE001 - silent for preview logger.info("Sample rows unavailable: %s", exc) self._fill_sample_combo(tasks) def _refresh_preview(self): if not hasattr(self, "_preview_view"): return template = self._prompt_edit.toPlainText() self._preview_warn.setVisible("{title}" not in template) if "{title}" not in template: self._preview_warn.setText("⚠ 话术缺少 {title} 占位符") task = self._sample_combo.currentData() if hasattr(self, "_sample_combo") else None if task is None: self._preview_view.setPlainText(template) else: rendered = template.replace("{title}", task.title).replace( "{product_id}", task.product_id) self._preview_view.setPlainText(rendered) # -- 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 save_outfit_prompt(prompt) 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