"""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, QDialog, 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, load_title_prompt, save_outfit_prompts, save_title_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): # request_interval/image_log pace and narrate directory rows that # fan out into many images (docs/11 §9.1); single-file rows ignore them. return generate_outfit_image( task, self._prompt, self._output_dir, self._model_config, quality=self._quality, resolution=self._resolution, request_interval=self._options.request_interval, image_concurrency=self._options.concurrency, image_log=self.log.emit, ) 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) class _TitleWorker(QObject): """Generates titles row-by-row on a QThread (docs/11 §17); queued signals. Sequential by row: each row's garment image + the title prompt -> one title, written back to that row's A column immediately. Failures are logged and skipped. """ tasks_loaded = Signal(object) # List[OutfitTask] log = Signal(str) progress = Signal(int, int, object) # completed, total, TitleResult finished = Signal(int, int) # success_count, fail_count failed = Signal(str) # fatal pre-run error (e.g. Excel locked) def __init__(self, excel_path, model_config, prompt, request_interval): super().__init__() self._excel_path = excel_path self._model_config = model_config self._prompt = prompt self._interval = float(request_interval or 0.0) self._stop = False def stop(self): self._stop = True def run(self): import time from core.ai_title import generate_title from core.models import TitleResult from services.ai_text_service import AiTextClient from services.excel_service import ( ensure_excel_writable, read_all_rows, write_title_result, ) try: ensure_excel_writable(self._excel_path) rows = read_all_rows(self._excel_path) client = AiTextClient(self._model_config) # one client for the run except Exception as exc: # noqa: BLE001 - report to UI self.failed.emit(str(exc)) return self.tasks_loaded.emit(rows) total = len(rows) if total == 0: self.finished.emit(0, 0) return success = fail = 0 for index, task in enumerate(rows, start=1): if self._stop: break if index > 1 and self._interval > 0: time.sleep(self._interval) result = generate_title(task, self._prompt, self._model_config, api_client=client) if result.success: try: write_title_result(self._excel_path, task.row_index, result.generated_title) except Exception as exc: # noqa: BLE001 - keep going result = TitleResult(task=task, success=False, error="写回失败：{}".format(exc), attempts=1) if result.success: success += 1 self.log.emit("第 {} 行标题：{}".format( task.row_index, result.generated_title)) else: fail += 1 self.log.emit("第 {} 行标题失败：{}".format( task.row_index, result.error)) self.progress.emit(index, total, result) self.finished.emit(success, fail) class _OutfitPreviewDialog(QDialog): """Non-modal 最终提示词预览（docs/11 §7.3）。 数据行下拉 + 只读结果；内容 = render_prompt(话术, 选中行, 当前分辨率)。 话术/分辨率/数据行变化时由面板调 refresh() 刷新。下拉用 read_all_rows （状态无关）填，整表完成后仍可选。 """ def __init__(self, panel): super().__init__(panel) self._panel = panel self.setWindowTitle("最终提示词预览") self.setModal(False) self.setAttribute(Qt.WA_DeleteOnClose) self.resize(440, 380) v = QVBoxLayout(self) row = QHBoxLayout() row.addWidget(QLabel("数据行")) self._combo = QComboBox() panel._compact_combo(self._combo) self._combo.currentIndexChanged.connect(self.refresh) row.addWidget(self._combo, stretch=1) v.addLayout(row) self._warn = QLabel("") self._warn.setStyleSheet("color:#b87a00;") self._warn.setVisible(False) v.addWidget(self._warn) self._view = QPlainTextEdit() self._view.setReadOnly(True) v.addWidget(self._view, stretch=1) self.reload_rows() def reload_rows(self): """(Re)fill the data-row dropdown from the panel's Excel (status-independent).""" excel = self._panel._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("Preview rows unavailable: %s", exc) self._combo.blockSignals(True) self._combo.clear() if rows: for t in rows: pid = t.product_id if (t.product_id and t.product_id.strip()) else "(无货号)" self._combo.addItem( "第 {} 行 · {} · {}".format(t.row_index, pid, t.title), t) else: self._combo.addItem("（选 Excel 后显示替换效果）", None) self._combo.blockSignals(False) self.refresh() def refresh(self): from core.ai_outfit import build_output_requirements, render_prompt template = self._panel._prompt_edit.toPlainText() resolution = self._panel._resolution.currentText() missing = "{title}" not in template self._warn.setVisible(missing) if missing: self._warn.setText("⚠ 话术缺少 {title} 占位符，标题不会带入") task = self._combo.currentData() if task is None: self._view.setPlainText(template + build_output_requirements(resolution)) else: self._view.setPlainText(render_prompt(template, task, resolution)) # --------------------------------------------------------------------------- # 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._title_thread = None # 标题生成线程（§17） self._title_worker = None self._title_fail = 0 # running failure count for title stats self._title_model_name = "" # configured 标题模型 name (§17.3 配置定名) self._preview_dialog = None # 最终提示词预览弹窗（§7.3，按需） 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): self.setObjectName("aiOutfitPanel") 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) self._apply_styles() def _apply_styles(self): """Button palette mirroring the 添加印花 widgets (docs/11 §10.5). Scoped under #aiOutfitPanel so QMessageBox/QInputDialog buttons keep the native look. Blue = the single primary action (开始生成); red = stop/delete; everything else is the neutral secondary used across the print panels. """ self.setStyleSheet(""" #aiOutfitPanel QPushButton { font-size: 12px; padding: 4px 10px; border: 1px solid #d6d6d6; border-radius: 3px; background: #f0f0f0; color: #202020; } #aiOutfitPanel QPushButton:hover { background: #e0e0e0; } #aiOutfitPanel QPushButton:pressed { background: #d0d0d0; } #aiOutfitPanel QPushButton:disabled { color: #aaaaaa; background: #f7f7f7; } #aiOutfitPanel QPushButton#aiStartBtn { font-weight: bold; color: #ffffff; border: none; background: #0078d4; } #aiOutfitPanel QPushButton#aiStartBtn:hover { background: #106ebe; } #aiOutfitPanel QPushButton#aiStartBtn:pressed { background: #005a9e; } #aiOutfitPanel QPushButton#aiStartBtn:disabled { color: #aaaaaa; background: #f7f7f7; } #aiOutfitPanel QPushButton#aiStopBtn, #aiOutfitPanel QPushButton#aiPromptDeleteBtn { color: #c42b1c; background: #f0f0f0; border: 1px solid #d6d6d6; } #aiOutfitPanel QPushButton#aiStopBtn:hover, #aiOutfitPanel QPushButton#aiPromptDeleteBtn:hover { background: #fde7e9; border-color: #c42b1c; } #aiOutfitPanel QPushButton#aiStopBtn:pressed, #aiOutfitPanel QPushButton#aiPromptDeleteBtn:pressed { background: #f9d4d7; } #aiOutfitPanel QPushButton#aiStopBtn:disabled, #aiOutfitPanel QPushButton#aiPromptDeleteBtn:disabled { color: #aaaaaa; background: #f7f7f7; border-color: #d6d6d6; } """) 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)) # 标题生成（§17）：看衣服图 + 提示词 → AI 文字标题 → 写回 Excel A 列 col.addWidget(self._build_title_group(), stretch=2) # 穿搭生成话术（多套模板；加大，随窗口高度拉伸） 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() # 「保存」插在「重命名」「删除」之间（§7.2，原编辑框下方的「保存话术」上移改名） for text, slot in (("新建", self._prompt_new), ("另存为", self._prompt_save_as), ("重命名", self._prompt_rename), ("保存", self._save_prompt), ("删除", self._prompt_delete)): b = QPushButton(text) if text == "删除": b.setObjectName("aiPromptDeleteBtn") 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_dialog) 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) preview_btn = QPushButton("预览最终提示词") preview_btn.clicked.connect(self._open_preview) prow.addWidget(preview_btn) pv.addLayout(prow) col.addWidget(prm, stretch=3) scroll.setWidget(inner) return scroll def _build_title_group(self): """标题生成组（§17）：提示词 + 保存 + 生成标题。 无标题模型下拉——模型由 app_config.title_model「配置定名」（§17.3/§17.6）。 """ box = QGroupBox() v = QVBoxLayout(box) v.addWidget(QLabel("标题生成提示词")) self._title_prompt_edit = QPlainTextEdit() self._title_prompt_edit.setMinimumHeight(110) v.addWidget(self._title_prompt_edit) btn_row = QHBoxLayout() save_title_btn = QPushButton("保存") save_title_btn.clicked.connect(self._save_title_prompt) btn_row.addWidget(save_title_btn) self._title_btn = QPushButton("生成标题") self._title_btn.clicked.connect(self._start_title) btn_row.addWidget(self._title_btn) v.addLayout(btn_row) return box 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) # Keep the preview dialog's output-requirements block in sync (§7.3). self._resolution.currentIndexChanged.connect(self._refresh_preview_dialog) # 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("aiStartBtn") self._start_btn.clicked.connect(self._start) run_row.addWidget(self._start_btn) self._stop_btn = QPushButton("停止生成") self._stop_btn.setObjectName("aiStopBtn") 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", "")) outfit_output_dir = str(config.get("outfit_output_dir", "") or "").strip() if not outfit_output_dir: from services.file_service import get_outfit_output_dir outfit_output_dir = str(get_outfit_output_dir()) self._output_edit.setText(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) # 标题生成提示词（单份，§17.3） self._title_prompt_edit.setPlainText(load_title_prompt()) # 标题模型「配置定名」：记下名字，运行时按名查 ai_models.json（§17.3，无下拉） self._title_model_name = str(config.get("title_model", "") or "") self._models = load_ai_models() self._fill_model_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 "" def _fill_model_combo(self, combo, selected_name): """Fill a model dropdown from self._models (shared by 图片/标题 model, §17.3).""" combo.clear() if not self._models: combo.addItem("（未配置模型，请在 ai_models.json 添加）") combo.setEnabled(False) return combo.setEnabled(True) for m in self._models: combo.addItem(m.get("name") or m.get("model") or "(未命名)") self._set_combo(combo, selected_name) 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, # title_model 由 app_config 配置定名（§17.3），UI 不写、不覆盖。 }) # -- 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() if self._preview_dialog is not None: self._preview_dialog.reload_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.3）-------------------------------------- def _open_preview(self): """Open (or raise) the non-modal 最终提示词预览 dialog.""" if self._preview_dialog is not None: self._preview_dialog.raise_() self._preview_dialog.activateWindow() return dialog = _OutfitPreviewDialog(self) dialog.finished.connect(self._on_preview_closed) self._preview_dialog = dialog dialog.show() def _on_preview_closed(self, _result): self._preview_dialog = None def _refresh_preview_dialog(self): """Refresh the preview if open (话术/分辨率/数据行变化时；否则 no-op).""" if self._preview_dialog is not None: self._preview_dialog.refresh() # -- 话术模板（§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) 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() # -- 标题生成（§17）------------------------------------------------- def _save_title_prompt(self, silent=False): save_title_prompt(self._title_prompt_edit.toPlainText()) if not silent: self.statusBar_message("标题提示词已保存") def _find_title_model(self): """Resolve the configured 标题模型 → (AiModelConfig|None, error|None). Pure. Looks up app_config.title_model (a name) in ai_models.json by 'name' (matching how models are labelled), so 换模型 is a config edit (§17.3). """ if not self._models: return None, ("尚未配置 AI 模型。请在 ~/.cmbot/config/ai_models.json " "添加可生成文字的模型（chat/gemini）后重试。") name = (self._title_model_name or "").strip() data = next((m for m in self._models if (m.get("name") or m.get("model")) == name), None) if data is None: return None, ("未找到标题模型「{}」。请在 ai_models.json 添加可生成文字的" "模型（chat/gemini），并把 app_config 的 title_model 设为它的 " "name。".format(name or "（未配置）")) from services.ai_image_service import AiModelConfig, api_config_errors errors = api_config_errors(data) if errors: return None, "标题模型「{}」配置有误：{}".format(name, "；".join(errors)) return AiModelConfig.from_dict(data), None def _resolve_title_model_config(self): config, error = self._find_title_model() if error: QMessageBox.warning(self, "标题模型不可用", error) return None return config def _start_title(self): if self._thread is not None or self._title_thread is not None: return excel = self._excel_edit.text().strip() if not excel: QMessageBox.information(self, "提示", "请先选择 Excel 文件。") return model_config = self._resolve_title_model_config() if model_config is None: return prompt = self._title_prompt_edit.toPlainText().strip() if not prompt: QMessageBox.information(self, "提示", "请先填写标题生成提示词。") return self._save_title_prompt(silent=True) self._emit_config() # reset run state (shared with image generation) self._table.setRowCount(0) self._row_to_table = {} self._title_fail = 0 self._progress.setValue(0) self._log.clear() self._title_worker = _TitleWorker( excel, model_config, prompt, self._interval.value()) self._title_thread = QThread(self) self._title_worker.moveToThread(self._title_thread) self._title_thread.started.connect(self._title_worker.run) self._title_worker.tasks_loaded.connect(self._on_title_tasks_loaded) self._title_worker.log.connect(self._append_log) self._title_worker.progress.connect(self._on_title_progress) self._title_worker.finished.connect(self._on_title_finished) self._title_worker.failed.connect(self._on_title_failed) self._title_worker.finished.connect(self._title_thread.quit) self._title_worker.failed.connect(self._title_thread.quit) self._title_thread.finished.connect(self._cleanup_title_thread) self._title_thread.start() self._set_title_running(True) def _on_title_tasks_loaded(self, tasks): self._populate_table(tasks) self._update_stats(0, 0, len(tasks)) self._append_log("标题生成：已加载 {} 行".format(len(tasks))) def _on_title_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, 1, result.generated_title) # refresh 标题 列 self._set_cell(row, 4, "完成") self._set_cell(row, 5, "标题已生成") else: self._set_cell(row, 4, "失败") self._set_cell(row, 5, result.error) if not result.success: self._title_fail += 1 self._progress.setValue(completed) self._update_stats(completed - self._title_fail, self._title_fail, total - completed) def _on_title_finished(self, success, fail): self._set_title_running(False) msg = "标题生成结束：成功 {}，失败 {}".format(success, fail) self._append_log(msg) self._reload_after_titles() if success or fail: QMessageBox.information( self, "标题生成", msg + "。\n标题已写回 Excel，可点「开始生成」生成穿搭图。") else: QMessageBox.information(self, "标题生成", "该表没有可处理的行。") def _on_title_failed(self, message): self._set_title_running(False) self._append_log("标题生成无法开始：" + message) QMessageBox.warning(self, "无法开始", message) def _cleanup_title_thread(self): self._title_thread = None self._title_worker = None def _reload_after_titles(self): """Re-read Excel so the table + later image generation use new titles.""" excel = self._excel_edit.text().strip() if not excel: return try: from services.excel_service import read_all_rows rows = read_all_rows(excel) except Exception as exc: # noqa: BLE001 - best effort logger.info("Reload after titles failed: %s", exc) return self._populate_table(rows) self._update_stats(0, 0, len(rows)) def _set_title_running(self, running): self._title_btn.setEnabled(not running) self._title_btn.setText("生成中…" if running else "生成标题") self._start_btn.setEnabled(not running) # mutually exclusive with 开始生成 self._excel_edit.setEnabled(not running) # -- 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 or self._title_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_outfit_output_dir output_dir = str(get_outfit_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)) self._title_btn.setEnabled(not running) # mutually exclusive with 生成标题 # -- worker callbacks (UI thread) ----------------------------------- def _populate_table(self, tasks): """Fill the detail table from a task list (shared by image/title runs).""" self._table.setRowCount(len(tasks)) self._row_to_table = {} 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))) def _on_tasks_loaded(self, tasks): self._populate_table(tasks) self._update_stats(0, 0, len(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, "完成") if result.output_paths: # 目录行：子目录 + 张数 self._set_cell(row, 5, "{}（{} 张）".format( result.output_path, len(result.output_paths))) else: 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) if getattr(summary, "total", 0) == 0: self._append_log("没有待处理的行。") self._show_no_pending_message() return 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 _show_no_pending_message(self): """Explain why nothing ran: rows already 完成/失败, and how to redo (§10.4).""" done = failed = 0 excel = self._excel_edit.text().strip() if excel: try: from services.excel_service import ( STATUS_DONE, STATUS_FAILED, read_all_rows) for r in read_all_rows(excel): if r.status == STATUS_DONE: done += 1 elif r.status == STATUS_FAILED: failed += 1 except Exception: # noqa: BLE001 - best effort pass if done == 0 and failed == 0: text = "该表没有可处理的行（标题/货号/衣服图为空的行会被跳过）。" else: parts = [] if done: parts.append("已完成 {} 行（已跳过）".format(done)) if failed: parts.append("失败 {} 行（勾选「重试上次失败的行」可重做）".format(failed)) text = ("该表没有待处理的行：" + "；".join(parts) + "。\n要重做已完成的行，请清空对应行的状态(E)列后重试。") QMessageBox.information(self, "没有待处理的行", text) 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): # Directory rows produce several images (output_paths); single-file rows # one (output_path). Add a thumbnail for each (docs/11 §9.1). for path in (result.output_paths or [result.output_path]): if not path: continue pix = QPixmap(path) item = QListWidgetItem(result.task.product_id) if not pix.isNull(): item.setIcon(QIcon(pix)) item.setData(Qt.UserRole, 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_outfit_output_dir path = str(get_outfit_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 # normpath so a directory path "d:/images/a/" shows its leaf "a" (docs/11 §4.1). return os.path.basename(os.path.normpath(str(path))) def _csv(value): text = str(value).replace('"', '""') if any(c in text for c in (",", "\n", '"')): return '"' + text + '"' return text