diff --git a/src/app/main_window.py b/src/app/main_window.py index cb8c8c7..b7cfb5c 100644 --- a/src/app/main_window.py +++ b/src/app/main_window.py @@ -14,6 +14,7 @@ from PySide6.QtWidgets import ( QPushButton, QScrollArea, QSplitter, + QStackedWidget, QTabBar, QVBoxLayout, QWidget, @@ -23,6 +24,7 @@ from version import APP_NAME, APP_VERSION from core.models import BatchMode, TransformState from services.config_service import load_config, save_config from services.update_service import check_for_update +from app.widgets.ai_outfit_panel import AiOutfitPanel from app.widgets.export_panel import ExportPanel from app.widgets.image_canvas import ImageCanvas from app.widgets.image_list_panel import ImageListPanel @@ -80,8 +82,12 @@ class MainWindow(QMainWindow): # shows the app name and version, so an in-content header would just # duplicate it. Start straight from the workflow tab bar. layout.addWidget(self._create_tab_bar()) - layout.addWidget(self._create_work_area(), stretch=1) - layout.addWidget(self._create_queue_panel()) + + # Swappable workflow pages: 0 = 添加印花 (canvas + queue), 1 = AI 穿搭. + self._stack = QStackedWidget() + self._stack.addWidget(self._create_print_page()) + self._stack.addWidget(self._create_ai_outfit_page()) + layout.addWidget(self._stack, stretch=1) self.setCentralWidget(root) self._apply_stylesheet() @@ -104,8 +110,7 @@ class MainWindow(QMainWindow): self._tab_bar.addTab("1 添加印花") self._tab_bar.addTab("2 AI 穿搭") self._tab_bar.addTab("3 导出上架") - # Only first tab is implemented in phase 1 - self._tab_bar.setTabEnabled(1, False) + # Tabs 1 (添加印花) and 2 (AI 穿搭) are implemented; 3 (导出上架) is not yet. self._tab_bar.setTabEnabled(2, False) self._tab_bar.currentChanged.connect(self._on_tab_changed) @@ -189,6 +194,22 @@ class MainWindow(QMainWindow): self._update_hint.setToolTip("点击进入设置更新") self._update_hint.setVisible(bool(available)) + def _create_print_page(self): + """Page 0: the print workflow — work area (canvas) above the queue.""" + page = QWidget() + col = QVBoxLayout(page) + col.setContentsMargins(0, 0, 0, 0) + col.setSpacing(0) + col.addWidget(self._create_work_area(), stretch=1) + col.addWidget(self._create_queue_panel()) + return page + + def _create_ai_outfit_page(self): + """Page 1: the AI outfit panel (docs/11).""" + self.ai_outfit_panel = AiOutfitPanel() + self.ai_outfit_panel.config_changed.connect(self._on_outfit_config_changed) + return self.ai_outfit_panel + def _create_work_area(self): """Horizontal splitter: left material panel | canvas | right params.""" splitter = QSplitter(Qt.Horizontal) @@ -300,9 +321,12 @@ class MainWindow(QMainWindow): # ------------------------------------------------------------------ def _on_tab_changed(self, index): - """Keep focus on tab 0; notify user for unimplemented tabs.""" - if index != 0: - self._tab_bar.setCurrentIndex(0) + """Switch the stacked page for implemented tabs; block the rest.""" + if index in (0, 1): + self._stack.setCurrentIndex(index) + else: + # 导出上架 not implemented — snap back to the current page's tab. + self._tab_bar.setCurrentIndex(self._stack.currentIndex()) self.statusBar().showMessage("该功能暂未开放") def _toggle_queue(self): @@ -530,6 +554,9 @@ class MainWindow(QMainWindow): self.image_list_panel.set_garment_start_dir(self._config.get("last_garment_dir", "")) self.image_list_panel.set_print_start_dir(self._config.get("last_print_dir", "")) + # AI outfit panel: inject last paths/settings + load models/prompt + self.ai_outfit_panel.apply_config(self._config) + # Persist future changes (connected after restore to avoid echo writes) self.template_panel.template_changed.connect(self._on_template_persist) self.queue_panel.batch_mode_changed.connect(self._on_batch_mode_persist) @@ -552,6 +579,11 @@ class MainWindow(QMainWindow): self._config["last_print_dir"] = path save_config(self._config) + def _on_outfit_config_changed(self, changes): + """Persist AI-outfit settings (last Excel/output/model + batch options).""" + self._config.update(changes) + save_config(self._config) + # ------------------------------------------------------------------ # Style (ref: docs/07-ui-design.md §10) # ------------------------------------------------------------------ diff --git a/src/app/widgets/ai_outfit_panel.py b/src/app/widgets/ai_outfit_panel.py new file mode 100644 index 0000000..8c91de1 --- /dev/null +++ b/src/app/widgets/ai_outfit_panel.py @@ -0,0 +1,682 @@ +"""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, + QDialogButtonBox, + 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) + + +# --------------------------------------------------------------------------- +# Preview dialog (最终提示词预览) +# --------------------------------------------------------------------------- + +class _PromptPreviewDialog(QDialog): + def __init__(self, parent, template, tasks): + super().__init__(parent) + self.setWindowTitle("最终提示词预览") + self.resize(560, 420) + self._template = template + self._tasks = tasks + + col = QVBoxLayout(self) + row = QHBoxLayout() + row.addWidget(QLabel("样本行：")) + self._combo = QComboBox() + for t in tasks: + self._combo.addItem("第 {} 行 · {} · {}".format( + t.row_index, t.product_id, t.title), t) + self._combo.currentIndexChanged.connect(self._render) + row.addWidget(self._combo, stretch=1) + col.addLayout(row) + + self._text = QPlainTextEdit() + self._text.setReadOnly(True) + col.addWidget(self._text, stretch=1) + + self._warn = QLabel("") + self._warn.setStyleSheet("color:#b87a00;") + self._warn.setVisible(False) + col.addWidget(self._warn) + + buttons = QDialogButtonBox() + copy_btn = buttons.addButton("复制", QDialogButtonBox.ActionRole) + buttons.addButton("关闭", QDialogButtonBox.RejectRole) + copy_btn.clicked.connect(self._copy) + buttons.rejected.connect(self.reject) + col.addWidget(buttons) + + if "{title}" not in (template or ""): + self._warn.setText("⚠ 话术中未检测到 {title} 占位符") + self._warn.setVisible(True) + self._render() + + def _render(self): + from core.ai_outfit import render_prompt + task = self._combo.currentData() + if task is None: + self._text.setPlainText(self._template or "") + return + self._text.setPlainText(render_prompt(self._template or "", task)) + + def _copy(self): + from PySide6.QtWidgets import QApplication + QApplication.clipboard().setText(self._text.toPlainText()) + + +# --------------------------------------------------------------------------- +# 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(130) + pv.addWidget(self._prompt_edit) + prow = QHBoxLayout() + for text, ph in (("插入标题", "{title}"), ("插入货号", "{product_id}")): + b = QPushButton(text) + b.clicked.connect(lambda _=False, p=ph: self._prompt_edit.insertPlainText(p)) + prow.addWidget(b) + preview_btn = QPushButton("预览最终提示词") + preview_btn.clicked.connect(self._preview_prompt) + prow.addWidget(preview_btn) + pv.addLayout(prow) + save_btn = QPushButton("保存话术") + save_btn.clicked.connect(self._save_prompt) + pv.addWidget(save_btn) + col.addWidget(prm) + + # 生成设置 + gen = QGroupBox("生成设置") + gv = QVBoxLayout(gen) + self._retry_failed_chk = QCheckBox("重试上次失败的行") + gv.addWidget(self._retry_failed_chk) + + from PySide6.QtWidgets import QFormLayout + 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) + col.addWidget(gen) + + col.addStretch() + scroll.setWidget(inner) + return scroll + + 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(280) + col = QVBoxLayout(wrap) + col.setContentsMargins(12, 12, 12, 12) + col.setSpacing(10) + + 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", "")) + + 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() + + 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("话术已保存") + + def _preview_prompt(self): + tasks = self._peek_tasks() + if tasks is None: + return + dlg = _PromptPreviewDialog(self, self._prompt_edit.toPlainText(), tasks) + dlg.exec() + + def _peek_tasks(self): + """Load a few tasks just for preview; returns [] if none, None on error.""" + excel = self._excel_edit.text().strip() + if not excel: + QMessageBox.information(self, "提示", "请先选择 Excel 文件再预览。") + return None + try: + from services.excel_service import load_outfit_tasks + return load_outfit_tasks(excel, retry_failed=self._retry_failed_chk.isChecked()) + except Exception as exc: # noqa: BLE001 + QMessageBox.warning(self, "读取失败", "无法读取 Excel：{}".format(exc)) + return None + + # -- 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._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 diff --git a/src/services/config_service.py b/src/services/config_service.py index 85af2db..cdd7739 100644 --- a/src/services/config_service.py +++ b/src/services/config_service.py @@ -14,9 +14,29 @@ DEFAULT_CONFIG = { "update_source": "", # HTTP(S) manifest source (empty = no update check) "update_user": "", # HTTP Basic Auth user (empty = anonymous) "update_pass": "", # HTTP Basic Auth password (empty = anonymous) + # AI 穿搭（docs/11 §11）：上次路径与批量设置并入 app_config（密钥不在此，见 ai_models.json） + "outfit_excel": "", + "outfit_output_dir": "", + "outfit_model": "", # last-selected model name + "outfit_concurrency": 1, + "outfit_request_interval": 2.0, + "outfit_task_cooldown": 1.0, + "outfit_retry_count": 2, + "outfit_resolution": "1K", + "outfit_quality": "均衡", + "outfit_retry_failed": False, } _CONFIG_FILENAME = "app_config.json" +_AI_MODELS_FILENAME = "ai_models.json" +_OUTFIT_PROMPT_FILENAME = "outfit_prompt.txt" + +# Default outfit prompt (docs/11 §7). Persisted to outfit_prompt.txt on first save. +DEFAULT_OUTFIT_PROMPT = ( + "为商品「{title}」（货号 {product_id}）生成人物上身实穿效果图：真人模特正面" + "穿着这件衣服，完整保留款式、版型、颜色与印花图案，自然光、纯色棚拍背景，" + "电商主图风格，不加文字与促销标签。" +) def load_config(): @@ -65,3 +85,56 @@ def save_config(data): logger.info("Config saved to %s", config_file) except OSError as exc: logger.error("Failed to save config to %s: %s", config_file, exc) + + +def load_ai_models(): + """Load AI model configs from ai_models.json. + + Returns a list of dicts (each with a ``name`` plus AiModelConfig fields). + Missing/corrupt file → empty list (the UI then prompts the admin to add one). + Keys live only in ~/.cmbot and are never committed (docs/11 §11). + """ + from services.file_service import get_config_path + models_file = get_config_path(_AI_MODELS_FILENAME) + if not models_file.exists(): + logger.info("AI models file not found: %s", models_file) + return [] + try: + with open(str(models_file), encoding="utf-8-sig") as f: + data = json.load(f) + except (json.JSONDecodeError, ValueError, OSError) as exc: + logger.warning("AI models file unreadable (%s): %s", exc, models_file) + return [] + + models = data.get("models") if isinstance(data, dict) else data + if not isinstance(models, list): + logger.warning("AI models file has no model list: %s", models_file) + return [] + return [m for m in models if isinstance(m, dict)] + + +def load_outfit_prompt(): + """Return the saved outfit prompt template, or the built-in default.""" + from services.file_service import get_config_path + prompt_file = get_config_path(_OUTFIT_PROMPT_FILENAME) + if not prompt_file.exists(): + return DEFAULT_OUTFIT_PROMPT + try: + with open(str(prompt_file), encoding="utf-8-sig") as f: + return f.read() + except OSError as exc: + logger.warning("Outfit prompt unreadable (%s): %s", exc, prompt_file) + return DEFAULT_OUTFIT_PROMPT + + +def save_outfit_prompt(text): + """Persist the outfit prompt template (utf-8, no BOM). Does not raise.""" + from services.file_service import get_config_path + prompt_file = get_config_path(_OUTFIT_PROMPT_FILENAME) + try: + prompt_file.parent.mkdir(parents=True, exist_ok=True) + with open(str(prompt_file), "w", encoding="utf-8") as f: + f.write(text) + logger.info("Outfit prompt saved to %s", prompt_file) + except OSError as exc: + logger.error("Failed to save outfit prompt to %s: %s", prompt_file, exc) diff --git a/tasks.md b/tasks.md index 05539b1..19604d1 100644 --- a/tasks.md +++ b/tasks.md @@ -1063,16 +1063,18 @@ 前置阅读：`docs/11-ai-outfit.md`（§10）、`docs/07-ui-design.md`（§4.2）、`docs/ui-ai-outfit.png` -- [ ] 页签栏接 `QStackedWidget`（「1 添加印花」=现有工作区，「2 AI 穿搭」=新面板），启用当前禁用的页签 -- [ ] `app/widgets/ai_outfit_panel.py`：左设置（Excel/输出/模型/话术编辑+保存+插入占位符+预览弹窗/生成设置 3 列）、中（最近结果缩略图条 + 处理明细表）、右（进度环+统计+开始/停止+导出失败清单+实时日志） -- [ ] 接线后台线程、日志、进度、结束摘要、失败清单导出 +- [x] 页签栏接 `QStackedWidget`（页 0=添加印花工作区+队列，页 1=AI 穿搭面板）；启用页签 1（导出上架仍禁用） +- [x] `app/widgets/ai_outfit_panel.py`：左设置（Excel/输出/模型/话术编辑+保存+插入占位符+预览弹窗/生成设置）、中（最近结果缩略图 + 处理明细表）、右（进度+统计+开始/停止+导出失败清单+打开输出目录+实时日志） +- [x] 接线后台线程：`QThread` + `_OutfitWorker(QObject)` 包 `OutfitBatchRunner`，signal 回主线程刷日志/进度/明细/缩略图；每行写回 Excel；结束摘要弹窗；失败清单导出 CSV +- [ ] GUI 实测（真机）：选 Excel/模型跑通、停止生效、失败清单正确（离屏冒烟已过，待真机） ### 19.4 配置与提示词 — docs/11 §14 阶段 4 前置阅读：`docs/11-ai-outfit.md`（§7、§11） -- [ ] `~/.cmbot/config/ai_models.json`（密钥明文、**不入库**）、`outfit_prompt.txt`、并入 `app_config.json`（上次 Excel/输出路径、批量设置），统一经 `config_service` -- [ ] `requirements.txt` 增 `requests>=2.31,<3`、`openpyxl>=3.1,<4`、`urllib3<2`；`docs/03` 增列依赖 +- [x] `config_service` 增 `load_ai_models()` / `load_outfit_prompt()` / `save_outfit_prompt()`；`ai_models.json`（密钥明文、**不入库**、BOM 容错）、`outfit_prompt.txt`（无 BOM）、`app_config.json` 并入 `outfit_*`（上次 Excel/输出/模型 + 批量设置），UI 经 `config_changed` 信号回主窗口集中存；`tests/test_config_service.py` 8 用例 +- [x] `requirements.txt` 已锁 `requests 2.31`/`openpyxl 3.1.3`/`urllib3 1.26`；`docs/03` 增列依赖 +- [ ] 可选：应用内 AI 模型编辑界面（当前由管理员预置 `ai_models.json`，暂不做） ### 19.5 真机联调 — docs/11 §14 阶段 5 diff --git a/tests/test_config_service.py b/tests/test_config_service.py new file mode 100644 index 0000000..64e4f60 --- /dev/null +++ b/tests/test_config_service.py @@ -0,0 +1,73 @@ +"""Tests for AI-outfit config helpers in config_service (no GUI).""" +import json +import os +import shutil +import sys +import tempfile +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +import services.config_service as cs + + +class TestOutfitConfigHelpers(unittest.TestCase): + def setUp(self): + self.tmp = Path(tempfile.mkdtemp()) + self._env = os.environ.get("CMBOT_DATA_DIR") + os.environ["CMBOT_DATA_DIR"] = str(self.tmp) + self.config_dir = self.tmp / "config" + self.config_dir.mkdir(parents=True, exist_ok=True) + + def tearDown(self): + if self._env is None: + os.environ.pop("CMBOT_DATA_DIR", None) + else: + os.environ["CMBOT_DATA_DIR"] = self._env + shutil.rmtree(str(self.tmp), ignore_errors=True) + + # -- ai_models.json ------------------------------------------------- + + def test_load_ai_models_missing_returns_empty(self): + self.assertEqual(cs.load_ai_models(), []) + + def test_load_ai_models_object_with_models_list(self): + (self.config_dir / "ai_models.json").write_text( + json.dumps({"models": [{"name": "m1", "url": "https://x"}]}), + encoding="utf-8") + models = cs.load_ai_models() + self.assertEqual(len(models), 1) + self.assertEqual(models[0]["name"], "m1") + + def test_load_ai_models_bare_list(self): + (self.config_dir / "ai_models.json").write_text( + json.dumps([{"name": "a"}, {"name": "b"}]), encoding="utf-8") + self.assertEqual(len(cs.load_ai_models()), 2) + + def test_load_ai_models_corrupt_returns_empty(self): + (self.config_dir / "ai_models.json").write_text("{ not json", encoding="utf-8") + self.assertEqual(cs.load_ai_models(), []) + + def test_load_ai_models_tolerates_bom(self): + (self.config_dir / "ai_models.json").write_bytes( + "".encode("utf-8") + json.dumps([{"name": "z"}]).encode("utf-8")) + self.assertEqual(cs.load_ai_models()[0]["name"], "z") + + # -- outfit_prompt.txt ---------------------------------------------- + + def test_prompt_default_when_missing(self): + self.assertEqual(cs.load_outfit_prompt(), cs.DEFAULT_OUTFIT_PROMPT) + + def test_prompt_save_then_load_roundtrip(self): + cs.save_outfit_prompt("hello {title} {product_id}") + self.assertEqual(cs.load_outfit_prompt(), "hello {title} {product_id}") + + def test_prompt_saved_without_bom(self): + cs.save_outfit_prompt("abc") + raw = (self.config_dir / "outfit_prompt.txt").read_bytes() + self.assertFalse(raw.startswith(b"\xef\xbb\xbf")) + + +if __name__ == "__main__": + unittest.main()