feat: 完成T-303 AI批量生成

- 新增 generate_batch，先并发生成标题再并发生成封面，成功逐条 set_generated 落库

- Tab② 接入开始生成、停止、进度展示和双击新旧封面预览

- 新增 GenerateWorker，通过 worker signal 回传进度与行刷新

- 补充批量生成成功、失败、停止取消和 GUI worker 单元测试

- 同步任务看板、API、路由、当前状态与 progress 文档

app/ai.py

@@ -1,6 +1,7 @@
 """AI generation helpers backed by configurable HTTP model endpoints."""
 
 import base64
+from concurrent.futures import CancelledError, ThreadPoolExecutor, as_completed
 import copy
 import json
 import mimetypes
@@ -10,7 +11,9 @@ import urllib.error
 import urllib.request
 import uuid
 
-from . import appconfig
+from . import appconfig, db
+from . import prompts as prompt_module
+from .config import make_slug
 
 
 class AIError(RuntimeError):
@@ -105,6 +108,144 @@ def gen_cover(
     return _save_jpeg(image_bytes, out_path, resolution, quality)
 
 
+def generate_batch(tasks, prompts, ai_cfg=None, on_progress=None, should_stop=None):
+    """Generate titles first, then covers, and persist each successful task."""
+
+    runtime = dict(ai_cfg or {})
+    config = _runtime_config(runtime)
+    generation_cfg = appconfig.ai_config(config)
+    generation_cfg.update(
+        {
+            key: value
+            for key, value in runtime.items()
+            if key in {
+                "title_concurrency",
+                "image_concurrency",
+                "retry",
+                "jpg_quality",
+                "resolution",
+            }
+        }
+    )
+    db_path = runtime.get("db_path")
+    models_path = runtime.get("models_path", appconfig.AI_MODELS_PATH)
+    image_root = runtime.get("image_dir") or appconfig.image_dir(config)
+    account_by_alias = runtime.get("account_by_alias") or {}
+    on_task_update = runtime.get("on_task_update")
+    title_prompt = _prompt_value(prompts, "title")
+    cover_prompt = _prompt_value(prompts, "cover")
+    should_stop = should_stop or (lambda: False)
+    eligible = [
+        task for task in list(tasks)
+        if getattr(task, "stage", None) == "collected"
+    ]
+    summary = {
+        "ok": True,
+        "total": len(eligible),
+        "title_done": 0,
+        "cover_done": 0,
+        "failed": 0,
+        "cancelled": False,
+    }
+    _emit_generation_progress(on_progress, summary)
+    title_results = {}
+
+    with ThreadPoolExecutor(
+        max_workers=max(1, int(generation_cfg.get("title_concurrency", 1)))
+    ) as executor:
+        futures = {}
+        for task in eligible:
+            if should_stop():
+                summary["cancelled"] = True
+                break
+            futures[
+                executor.submit(
+                    gen_title,
+                    title_prompt,
+                    getattr(task, "old_title", "") or "",
+                    retry=generation_cfg.get("retry"),
+                    config=config,
+                    models_path=models_path,
+                )
+            ] = task
+        for future in as_completed(futures):
+            task = futures[future]
+            if should_stop():
+                summary["cancelled"] = True
+                _cancel_pending(futures)
+            try:
+                title_results[task.id] = future.result()
+                summary["title_done"] += 1
+            except CancelledError:
+                summary["cancelled"] = True
+            except Exception as exc:
+                summary["failed"] += 1
+                summary["ok"] = False
+                _mark_generate_failed(task, exc, db_path, on_task_update)
+            _emit_generation_progress(on_progress, summary)
+
+    cover_tasks = [
+        task for task in eligible
+        if task.id in title_results
+    ]
+    with ThreadPoolExecutor(
+        max_workers=max(1, int(generation_cfg.get("image_concurrency", 1)))
+    ) as executor:
+        futures = {}
+        for task in cover_tasks:
+            if should_stop():
+                summary["cancelled"] = True
+                break
+            new_title = title_results[task.id]
+            rendered_cover_prompt = prompt_module.render_prompt(
+                cover_prompt,
+                _prompt_context(task, new_title, account_by_alias),
+            )
+            futures[
+                executor.submit(
+                    gen_cover,
+                    rendered_cover_prompt,
+                    getattr(task, "old_cover_path", "") or "",
+                    _new_cover_path(task, account_by_alias, image_root),
+                    resolution=generation_cfg.get("resolution"),
+                    jpg_quality=generation_cfg.get("jpg_quality"),
+                    retry=generation_cfg.get("retry"),
+                    config=config,
+                    models_path=models_path,
+                )
+            ] = (task, new_title)
+        for future in as_completed(futures):
+            task, new_title = futures[future]
+            if should_stop():
+                summary["cancelled"] = True
+                _cancel_pending(futures)
+            try:
+                new_cover_path = future.result()
+                db.set_generated(task.id, new_title, new_cover_path, path=db_path)
+                summary["cover_done"] += 1
+                if on_task_update is not None:
+                    on_task_update(
+                        task.id,
+                        {
+                            "stage": "generated",
+                            "status": "success",
+                            "new_title": new_title,
+                            "new_cover_path": new_cover_path,
+                        },
+                    )
+            except CancelledError:
+                summary["cancelled"] = True
+            except Exception as exc:
+                summary["failed"] += 1
+                summary["ok"] = False
+                _mark_generate_failed(task, exc, db_path, on_task_update)
+            _emit_generation_progress(on_progress, summary)
+
+    if summary["cancelled"]:
+        summary["ok"] = False
+    return summary
+
+
 def _role_model(category, name, models_path):
     if not name:
         raise AIError("未配置默认 %s 模型" % category)
@@ -123,6 +264,100 @@ def _role_model(category, name, models_path):
     return model
 
 
+def _runtime_config(runtime):
+    if runtime.get("config") is not None:
+        return runtime["config"]
+    config = appconfig.load_config()
+    ai_updates = {
+        key: value
+        for key, value in runtime.items()
+        if key in {
+            "default_text_model",
+            "default_image_model",
+            "title_concurrency",
+            "image_concurrency",
+            "retry",
+            "jpg_quality",
+            "resolution",
+            "resolution_timeouts",
+        }
+    }
+    if ai_updates:
+        config = copy.deepcopy(config)
+        config.setdefault("ai", {}).update(ai_updates)
+    return config
+
+
+def _prompt_value(prompt_values, name):
+    if isinstance(prompt_values, dict):
+        return str(
+            prompt_values.get(name)
+            or prompt_values.get(f"{name}_prompt")
+            or ""
+        )
+    return ""
+
+
+def _prompt_context(task, new_title, account_by_alias):
+    return {
+        "old_title": getattr(task, "old_title", ""),
+        "new_title": new_title,
+        "item_id": getattr(task, "item_id", ""),
+        "account_name": _account_name(task, account_by_alias),
+        "alias": getattr(task, "alias", ""),
+    }
+
+
+def _account_name(task, account_by_alias):
+    alias = str(getattr(task, "alias", "") or "").strip()
+    account = account_by_alias.get(alias)
+    if account is not None:
+        return getattr(account, "account_name", "") or alias
+    return getattr(task, "account_name", "") or alias
+
+
+def _new_cover_path(task, account_by_alias, image_root):
+    alias = str(getattr(task, "alias", "") or "").strip()
+    account = account_by_alias.get(alias)
+    slug = getattr(account, "slug", None) if account is not None else None
+    if not slug:
+        slug = make_slug(alias or getattr(task, "account_name", "") or "unknown")
+    return os.path.abspath(
+        os.path.join(
+            image_root,
+            slug,
+            "%s_new.jpg" % getattr(task, "item_id", ""),
+        )
+    )
+
+
+def _mark_generate_failed(task, exc, db_path, on_task_update):
+    error = str(exc) or exc.__class__.__name__
+    db.mark_failed(task.id, "generate", error, path=db_path)
+    if on_task_update is not None:
+        on_task_update(task.id, {"status": "failed", "last_error": error})
+
+
+def _cancel_pending(futures):
+    for future in futures:
+        if not future.done():
+            future.cancel()
+
+
+def _emit_generation_progress(on_progress, summary):
+    if on_progress is None:
+        return
+    payload = dict(summary)
+    try:
+        on_progress(payload)
+    except TypeError:
+        on_progress(
+            payload.get("title_done", 0),
+            payload.get("cover_done", 0),
+            payload.get("failed", 0),
+        )
+
+
 def _attempt_count(ai_cfg, retry):
     retry_count = ai_cfg.get("retry", 2) if retry is None else retry
     return max(1, int(retry_count) + 1)

app/gui.py

@@ -7,6 +7,7 @@ import sys
 
 try:
     from PySide6.QtCore import QAbstractTableModel, QModelIndex, Qt
+    from PySide6.QtGui import QPixmap
     from PySide6.QtWidgets import (
         QAbstractItemView,
         QApplication,
@@ -78,7 +79,7 @@ QTabBar::tab:hover:!selected {
 
 
 if QT_IMPORT_ERROR is None:
-    from . import accounts, appconfig, chrome, db, editor, excel, prompts
+    from . import accounts, ai, appconfig, chrome, db, editor, excel, prompts
     from . import config as account_config
 
 
@@ -316,6 +317,8 @@ if QT_IMPORT_ERROR is None:
             self.title_prompt_path = title_prompt_path or prompts.TITLE_PROMPT_PATH
             self.cover_prompts_dir = cover_prompts_dir or prompts.COVER_PROMPTS_DIR
             self.current_cover_template = None
+            self.generate_worker = None
+            self.generate_thread = None
 
             self.title_prompt_edit = QPlainTextEdit()
             self.title_prompt_edit.setObjectName("titlePromptEdit")
@@ -336,6 +339,10 @@ if QT_IMPORT_ERROR is None:
             self.delete_cover_template_button = QPushButton("删除")
             self.insert_title_button = QPushButton("插入标题")
             self.preview_prompt_button = QPushButton("预览")
+            self.generate_button = QPushButton("开始生成")
+            self.stop_generate_button = QPushButton("停止")
+            self.stop_generate_button.setEnabled(False)
+            self.progress_label = QLabel("进度：标题0/0 · 封面0/0 · 失败0")
 
             left_panel = QWidget()
             left_layout = QVBoxLayout(left_panel)
@@ -401,9 +408,16 @@ if QT_IMPORT_ERROR is None:
             self.splitter.setStretchFactor(1, 3)
             self.splitter.setSizes([280, 860])
 
+            bottom_layout = QHBoxLayout()
+            bottom_layout.addWidget(self.progress_label)
+            bottom_layout.addStretch(1)
+            bottom_layout.addWidget(self.generate_button)
+            bottom_layout.addWidget(self.stop_generate_button)
+
             layout = QVBoxLayout(self)
             layout.setContentsMargins(18, 18, 18, 18)
             layout.addWidget(self.splitter, 1)
+            layout.addLayout(bottom_layout)
 
             self.batch_filter.currentIndexChanged.connect(self.refresh_tasks)
             self.shop_filter.currentIndexChanged.connect(self.refresh_tasks)
@@ -418,6 +432,9 @@ if QT_IMPORT_ERROR is None:
             self.delete_cover_template_button.clicked.connect(self.delete_cover_template)
             self.insert_title_button.clicked.connect(self.insert_title_placeholder)
             self.preview_prompt_button.clicked.connect(self.preview_cover_prompt)
+            self.generate_button.clicked.connect(self.start_generate)
+            self.stop_generate_button.clicked.connect(self.stop_generate)
+            self.task_table.doubleClicked.connect(self.show_task_images)
 
             self.refresh_cover_templates()
             self.refresh_tasks()
@@ -566,6 +583,147 @@ if QT_IMPORT_ERROR is None:
             QMessageBox.information(self, "封面提示词预览", rendered)
             self._set_status("封面提示词预览已生成")
 
+        def start_generate(self, checked=False):
+            if self.generate_thread is not None:
+                self._set_status("AI 生成正在进行...")
+                return
+            tasks = [
+                task for task in self.model.tasks
+                if getattr(task, "stage", None) == "collected"
+            ]
+            if not tasks:
+                self._set_status("当前筛选结果没有可生成任务")
+                return
+            prompt_values = {
+                "title": self.title_prompt_edit.toPlainText(),
+                "cover": self.cover_prompt_edit.toPlainText(),
+            }
+            worker = GenerateWorker(
+                tasks,
+                prompt_values,
+                db_path=self.db_path,
+                config=self.config,
+            )
+            worker.progress.connect(self._on_generate_progress)
+            worker.row_updated.connect(self._on_generate_row_updated)
+            worker.log.connect(self._set_status)
+            worker.failed.connect(self._on_generate_failed)
+            worker.finished.connect(self._on_generate_finished)
+            worker.cancelled.connect(self._on_generate_cancelled)
+            thread = run_worker(worker, thread_name="GenerateWorker", start=False)
+            thread.finished.connect(lambda: self._forget_generate_thread(thread))
+            self.generate_worker = worker
+            self.generate_thread = thread
+            self._set_generate_running(True)
+            self._update_generate_progress(
+                {"total": len(tasks), "title_done": 0, "cover_done": 0, "failed": 0}
+            )
+            self._set_status(f"开始 AI 生成：{len(tasks)} 条")
+            thread.start()
+
+        def stop_generate(self, checked=False):
+            if self.generate_worker is not None:
+                self.generate_worker.cancel()
+                self._set_status("正在停止 AI 生成...")
+
+        def show_task_images(self, index):
+            task = self.model.task_at(index.row()) if index.isValid() else self._selected_task()
+            if task is None:
+                self._set_status("没有可预览的任务")
+                return
+            dialog = QDialog(self)
+            dialog.setWindowTitle(f"封面对照：{task.item_id}")
+            layout = QVBoxLayout(dialog)
+            images_layout = QHBoxLayout()
+            images_layout.addWidget(self._image_panel("旧封面", task.old_cover_path))
+            images_layout.addWidget(self._image_panel("新封面", task.new_cover_path))
+            layout.addLayout(images_layout)
+            buttons = QDialogButtonBox(QDialogButtonBox.Close)
+            buttons.rejected.connect(dialog.reject)
+            layout.addWidget(buttons)
+            dialog.resize(720, 420)
+            dialog.exec()
+
+        def _image_panel(self, title, path):
+            panel = QWidget()
+            layout = QVBoxLayout(panel)
+            layout.addWidget(QLabel(title))
+            image_label = QLabel()
+            image_label.setAlignment(Qt.AlignCenter)
+            image_label.setMinimumSize(260, 260)
+            image_label.setWordWrap(True)
+            if path and os.path.exists(str(path)):
+                pixmap = QPixmap(str(path))
+                if not pixmap.isNull():
+                    image_label.setPixmap(
+                        pixmap.scaled(
+                            260,
+                            260,
+                            Qt.KeepAspectRatio,
+                            Qt.SmoothTransformation,
+                        )
+                    )
+                else:
+                    image_label.setText(f"图片无法读取\n{path}")
+            else:
+                image_label.setText(f"无图片\n{path or ''}".strip())
+            layout.addWidget(image_label, 1)
+            return panel
+
+        def _set_generate_running(self, running):
+            self.generate_button.setEnabled(not running)
+            self.stop_generate_button.setEnabled(running)
+            self.refresh_button.setEnabled(not running)
+            self.batch_filter.setEnabled(not running)
+            self.shop_filter.setEnabled(not running)
+            self.status_filter.setEnabled(not running)
+            self.save_title_button.setEnabled(not running)
+            self.save_cover_template_button.setEnabled(not running)
+            self.save_cover_template_as_button.setEnabled(not running)
+            self.rename_cover_template_button.setEnabled(not running)
+            self.delete_cover_template_button.setEnabled(not running)
+
+        def _forget_generate_thread(self, thread):
+            if self.generate_thread is thread:
+                self.generate_thread = None
+                self.generate_worker = None
+
+        def _on_generate_progress(self, payload):
+            self._update_generate_progress(payload)
+            self._set_status("生成进度：" + self._generate_progress_text(payload))
+
+        def _on_generate_row_updated(self, task_id, fields):
+            self.refresh_tasks()
+
+        def _on_generate_failed(self, task_id, error):
+            self._set_status(f"AI 生成失败：{error}")
+
+        def _on_generate_finished(self, payload):
+            self._set_generate_running(False)
+            self.refresh_tasks()
+            self._update_generate_progress(payload)
+            if payload.get("error"):
+                self._set_status(f"AI 生成失败：{payload.get('error')}")
+                return
+            self._set_status("AI 生成完成：" + self._generate_progress_text(payload))
+
+        def _on_generate_cancelled(self, payload):
+            self._set_generate_running(False)
+            self.refresh_tasks()
+            self._update_generate_progress(payload)
+            self._set_status("AI 生成已停止：" + self._generate_progress_text(payload))
+
+        def _update_generate_progress(self, payload):
+            self.progress_label.setText("进度：" + self._generate_progress_text(payload))
+
+        def _generate_progress_text(self, payload):
+            return "标题{title}/{total} · 封面{cover}/{total} · 失败{failed}".format(
+                title=payload.get("title_done", 0),
+                cover=payload.get("cover_done", 0),
+                total=payload.get("total", 0),
+                failed=payload.get("failed", 0),
+            )
+
         def _selected_task(self):
             index = self.task_table.currentIndex()
             if index.isValid():
@@ -1186,6 +1344,41 @@ if QT_IMPORT_ERROR is None:
     from .workers import BaseWorker, run_worker
 
 
+    class GenerateWorker(BaseWorker):
+        """Generate titles and covers for collected tasks."""
+
+        def __init__(self, tasks, prompt_values, db_path=None, config=None):
+            super().__init__()
+            self.tasks = list(tasks)
+            self.prompt_values = dict(prompt_values or {})
+            self.db_path = db_path
+            self.config = config
+
+        def execute(self):
+            account_rows = accounts.list_accounts(path=self.db_path, config=self.config)
+            account_by_alias = {
+                str(account.alias).strip(): account
+                for account in account_rows
+                if str(account.alias).strip()
+            }
+            return ai.generate_batch(
+                self.tasks,
+                self.prompt_values,
+                ai_cfg={
+                    "config": self.config,
+                    "db_path": self.db_path,
+                    "image_dir": appconfig.image_dir(self.config),
+                    "account_by_alias": account_by_alias,
+                    "on_task_update": self._emit_row_update,
+                },
+                on_progress=self.progress.emit,
+                should_stop=self.should_cancel,
+            )
+
+        def _emit_row_update(self, task_id, fields):
+            self.row_updated.emit(int(task_id), dict(fields or {}))
+
+
     class CollectWorker(BaseWorker):
         """Collect old title and cover for imported tasks."""
 

docs/06-tasks.md

@@ -62,7 +62,7 @@
 | T-301 | 确定 AI 服务商/模型并接入 `app/ai.py`（`gen_title`/`gen_cover`，带重试/分辨率/jpg质量） | T-005 | 从 `config/ai_models.json` 读取模型与本地明文 Key；`gen_cover` 支持 resolution+jpg_quality；失败按 retry 重试；错误明确；日志脱敏 | DONE |
 | T-302 | Tab② 左右布局：左提示词（标题/封面），右按批次/店铺/状态筛选 + 任务列表 | T-301, T-203 | 左 ~1/4 提示词多行；右筛选+列表（店铺/商品id/旧标题/新标题/状态） | DONE |
 | T-302p | `app/prompts.py` + Tab② 提示词管理 | T-302 | 标题保存/启动回显 title_prompt.txt；封面多模板（下拉+新建/保存/另存为/重命名/删除，存 prompts/cover/）；插入 `{新标题}`；预览变量替换；render_prompt 接入生成 | DONE |
-| T-303 | Tab② 开始生成（单按钮）+ 停止 + 进度：**先并发标题再并发图片** | T-302, T-104b | `generate_batch` 先 title_concurrency 并发标题、再 image_concurrency 并发图片；worker/signal 回传进度；每条 set_generated 立即写库；停止可取消未开始项；进度 标题/封面/失败 计数；双击弹窗看新旧封面 | TODO |
+| T-303 | Tab② 开始生成（单按钮）+ 停止 + 进度：**先并发标题再并发图片** | T-302, T-104b | `generate_batch` 先 title_concurrency 并发标题、再 image_concurrency 并发图片；worker/signal 回传进度；每条 set_generated 立即写库；停止可取消未开始项；进度 标题/封面/失败 计数；双击弹窗看新旧封面 | DONE |
 
 ## Phase 4 · 更新 shopee（③）
 

docs/api.md

@@ -230,11 +230,11 @@ gen_cover(cover_prompt, old_cover_path, out_path, resolution=None, jpg_quality=N
 # 图像生成(image-to-image)：读取 default_image_model；chat 多模态 JSON 或 images_edits multipart；
 # 支持返回 url / data URL / b64_json，按 resolution resize 并以 jpg_quality 保存 JPEG，返回路径
 
-generate_batch(tasks, prompts, ai_cfg, on_progress, should_stop) -> None
+generate_batch(tasks, prompts, ai_cfg=None, on_progress=None, should_stop=None) -> dict
 # 编排：先以 title_concurrency 线程池并发跑 gen_title，再以 image_concurrency 并发跑 gen_cover
 # 每条完成即 db.set_generated（实时落库）；should_stop() 为真则取消未开始项
-# on_progress(标题完成数, 封面完成数, 失败数) 回调刷新进度
-# T-303 实现
+# on_progress({"total","title_done","cover_done","failed","cancelled","ok"}) 回调刷新进度
+# 返回同结构 summary；失败任务 mark_failed(..., "generate", error)，不阻塞其余
 ```
 
 要点：
@@ -283,8 +283,9 @@ render_prompt(template_text, task) -> str
 main() -> int                              # 创建 QApplication + MainWindow
 class MainWindow(QMainWindow)              # QTabWidget: ①②③④⑤；支持注入 db_path/config 便于测试
 class CollectTab(QWidget)                  # ① 导入采集：导入 Excel + 汇总栏 + QTableView 任务列表 + 未匹配略过标记
-class GenerateTab(QWidget)                 # ② AI生成：提示词管理 + 批次/店铺/状态筛选 + 任务列表
+class GenerateTab(QWidget)                 # ② AI生成：提示词管理 + 筛选任务 + 开始/停止生成 + 新旧封面预览
 class CollectWorker(BaseWorker)            # ① 后台采集：账号就绪预检 -> editor.collect -> db.set_collected/mark_skipped/mark_failed
+class GenerateWorker(BaseWorker)           # ② 后台生成：ai.generate_batch -> db.set_generated/mark_failed + 进度
 class WriteBackWorker(BaseWorker)          # ① 后台回写：excel.write_back(batch_id) 写旧标题/旧封面到原 Excel
 class TaskTableModel(QAbstractTableModel)  # 任务表格模型：账号/别名/商品ID/阶段；未匹配别名显示“略过”
 class GenerateTaskTableModel(QAbstractTableModel) # ② 任务表格模型：店铺/商品ID/旧标题/新标题/状态
@@ -294,7 +295,7 @@ TAB_TITLES: list[str]                      # 固定 Tab 顺序
 TAB_STYLE: str                              # 顶层 Tab 栏防误点样式：最小宽度/padding/间距/当前态
 ```
 
-`MainWindow` 已实现五 Tab、① 导入采集任务列表、② AI生成布局与任务列表、④ 账号管理；③⑤ 仍为空壳，业务控件按后续 Tab 任务逐步填充。缺 PySide6 时 `main()` 返回 1 并输出明确提示。
+`MainWindow` 已实现五 Tab、① 导入采集任务列表、② AI生成布局/提示词/开始生成/停止/封面对照预览、④ 账号管理；③⑤ 仍为空壳，业务控件按后续 Tab 任务逐步填充。缺 PySide6 时 `main()` 返回 1 并输出明确提示。
 
 主 Tab 栏必须在 `MainWindow` 初始化时应用 `TAB_STYLE`：5 个 Tab 不使用 Qt 默认紧凑宽度，需保证点击区域稳定、间距清晰、当前 Tab 高亮明显。该样式属于全局导航基础，不归后续业务 Tab 任务重复实现。
 
@@ -317,7 +318,7 @@ TAB_STYLE: str                              # 顶层 Tab 栏防误点样式：
 - 「回写旧数据到 Excel」通过 `WriteBackWorker` 后台调用 `excel.write_back()`，把已采集旧标题/旧封面路径按原 Excel 行定位写回；该按钮主要作为自动回写失败后的手动重试入口。原文件被占用时弹窗提示关闭后重试，SQLite 采集结果不回滚。
 - 采集前由 `CollectWorker` 做账号就绪预检：无账号、当前批次匹配账号未启动 CDP 端口或未登录时，返回 `blocked=True`，GUI 弹窗汇总并跳转/引导去④账号管理；不无提示批量启动所有账号 Chrome。匹配账号未登录属于预检阻断，不是逐条 skipped。
 
-② AI生成当前要点（T-302/T-302p）：
+② AI生成当前要点（T-302/T-302p/T-303）：
 
 - 左右 `QSplitter`：左侧约 1/4 为标题提示词、封面提示词两个多行输入；右侧为筛选栏 + 任务列表。
 - 标题提示词启动时从 `title_prompt.txt` 回显；点击「保存标题提示词」写回该文件。
@@ -325,7 +326,10 @@ TAB_STYLE: str                              # 顶层 Tab 栏防误点样式：
 - 「插入标题」在封面提示词光标处插入 `{新标题}`；「预览」使用当前选中任务（无选择则用第一条）调用 `prompts.render_prompt()` 并弹窗展示。
 - 筛选栏包含：批次、店铺、状态、刷新。批次来自 `db.list_batches()`；店铺来自当前任务别名并优先显示匹配账号名；状态支持全部/待生成/已生成/失败/略过/已更新。
 - 任务列表使用 `QTableView + GenerateTaskTableModel`，列为：店铺、商品ID、旧标题、新标题、状态。`stage=collected` 显示“待生成”，`stage=generated` 显示“已生成”，`status=failed/skipped/running` 优先显示对应状态。
-- T-302/T-302p 不调用 `app.ai`、不写 SQLite；开始生成/停止/进度与 `set_generated()` 留给 T-303。
+- 双击任务行弹窗展示旧封面与新封面路径对应图片；图片不存在时显示空态/路径提示，只做查看，不做审核。
+- 底部「开始生成」只处理当前筛选结果里 `stage=collected` 的任务；通过 `GenerateWorker` 调 `ai.generate_batch()`，先并发标题、再并发封面。
+- 「停止」调用 worker 的协作式 `cancel()`；未开始的 Future 取消，不记失败；已完成的 `db.set_generated()` 结果保留，可再次开始生成剩余任务。
+- 进度标签显示：`标题x/n · 封面x/n · 失败z`；每条封面生成成功后立即 `db.set_generated()`，失败 `db.mark_failed(..., "generate", error)`，GUI 刷新任务表。
 
 ## workers 模块（`app/workers.py`，已建，PySide6）
 

docs/current-state.md

@@ -6,10 +6,10 @@
 ## 当前快照
 
 - 日期：2026-06-27
-- 阶段：V0 单账号 CDP 流程已验证；V1 已完成 T-000 正式代码包结构、T-001 `app/editor.py` 模块化、T-002 `app/appconfig.py` 应用配置、T-003 SQLite 持久化地基、T-004 本地数据忽略规则、T-005 AI 模型清单后端、T-006 单元测试基座、T-101 账号 user-data-dir 工具、T-102 Chrome 启动器、T-103 登录保活与检测、T-104 PySide6 主窗口骨架、T-104b PySide6 worker 基座、T-105 Tab④ 账号管理、T-106 账号快捷方式、T-201 Excel 导入入库、T-202 Tab① 任务列表与导入按钮、T-202b Tab① 导入汇总栏、T-203 采集旧标题旧封面、T-204 回写旧字段到原 Excel、T-204b 采集完成自动回写、T-205 首次未配账号/Chrome 未启动/未登录引导保护、T-301 AI 生成接口、T-302 Tab② AI 生成布局与任务列表、T-302p 提示词管理。
+- 阶段：V0 单账号 CDP 流程已验证；V1 已完成 T-000 正式代码包结构、T-001 `app/editor.py` 模块化、T-002 `app/appconfig.py` 应用配置、T-003 SQLite 持久化地基、T-004 本地数据忽略规则、T-005 AI 模型清单后端、T-006 单元测试基座、T-101 账号 user-data-dir 工具、T-102 Chrome 启动器、T-103 登录保活与检测、T-104 PySide6 主窗口骨架、T-104b PySide6 worker 基座、T-105 Tab④ 账号管理、T-106 账号快捷方式、T-201 Excel 导入入库、T-202 Tab① 任务列表与导入按钮、T-202b Tab① 导入汇总栏、T-203 采集旧标题旧封面、T-204 回写旧字段到原 Excel、T-204b 采集完成自动回写、T-205 首次未配账号/Chrome 未启动/未登录引导保护、T-301 AI 生成接口、T-302 Tab② AI 生成布局与任务列表、T-302p 提示词管理、T-303 Tab② 开始生成/停止/进度。
 - 技术栈：Python 3.10+，自研 CDP（websocket-client + requests），SQLite（sqlite3）+ `config.json` + openpyxl + AI（`config/ai_models.json` 通用 HTTP，chat JSON / images_edits），GUI PySide6 5 Tab（已定）。
-- 生产代码：已建立 `app/` 包 + 根入口 `main.py`；`app/cdp.py` 为已验证 CDP 底座；`app/editor.py` 已封装登录状态检测、标题/封面/采集/更新按钮能力；`app/appconfig.py` 已实现 `config.json` 默认值/读写/更新、AI 参数与端口读取，以及 `config/ai_models.json` 模型清单 CRUD/过滤/打码/测试连接；`app/ai.py` 已实现 `gen_title()`/`gen_cover()`，按默认文本/图片模型发起通用 HTTP 调用，支持重试、错误脱敏、图片 URL/base64 解析、resolution resize 与 jpg_quality 保存；`app/prompts.py` 已实现标题提示词读写、封面模板 CRUD 与变量替换；`app/db.py` 已实现 SQLite schema、连接 PRAGMA、批次/账号/任务与阶段写库函数；`app/excel.py` 已实现多 Excel 输入列解析、整文件列校验、脏行统计跳过、导入批次与任务入库、别名匹配统计、旧标题/旧封面路径回写原 Excel 与另存副本；`app/config.py` 已实现账号 slug 与 user-data-dir 创建；`app/accounts.py` 已实现账号 CRUD 服务、端口默认分配、启动登录、检测登录、生成快捷方式；`app/chrome.py` 已实现 Chrome 参数拼装、启动、CDP 端口探测、PowerShell `.lnk` 快捷方式生成；`app/gui.py` 已实现 PySide6 `MainWindow`、五 Tab、顶部 Tab 栏防误点样式、① 导入采集的 Excel 导入按钮/导入汇总栏/QTableView 任务列表/未匹配筛选与略过标记/采集旧标题旧封面 worker/采集前账号就绪预检与④引导/采集完成自动回写/旧数据回写重试按钮与 worker、② AI生成左右布局/标题与封面提示词管理/批次店铺状态筛选/任务列表/变量预览、④ 账号管理表格/弹窗/按钮/快捷方式与状态栏；`app/workers.py` 已实现 `BaseWorker`、通用 signals、取消标记和 `QThread` 启动包装。
-- 测试：`tests/` 已建立；T-006 后纯逻辑改动必须运行 `python -m unittest discover -s tests`，当前覆盖 appconfig/db/config/accounts/chrome 启动与快捷方式/editor 登录检测/excel 导入/ai 标题与封面 HTTP 解析/prompts 读写与渲染/gui ① 导入采集/gui ② AI生成布局筛选与提示词管理/gui ④ 账号管理/worker signal 与线程包装；CDP/Shopee 改动仍需测试商品手动验证。
+- 生产代码：已建立 `app/` 包 + 根入口 `main.py`；`app/cdp.py` 为已验证 CDP 底座；`app/editor.py` 已封装登录状态检测、标题/封面/采集/更新按钮能力；`app/appconfig.py` 已实现 `config.json` 默认值/读写/更新、AI 参数与端口读取，以及 `config/ai_models.json` 模型清单 CRUD/过滤/打码/测试连接；`app/ai.py` 已实现 `gen_title()`/`gen_cover()`/`generate_batch()`，按默认文本/图片模型发起通用 HTTP 调用，支持重试、错误脱敏、图片 URL/base64 解析、resolution resize、jpg_quality 保存、先并发标题再并发封面、逐条 `set_generated`、失败 `mark_failed` 与停止取消未开始项；`app/prompts.py` 已实现标题提示词读写、封面模板 CRUD 与变量替换；`app/db.py` 已实现 SQLite schema、连接 PRAGMA、批次/账号/任务与阶段写库函数；`app/excel.py` 已实现多 Excel 输入列解析、整文件列校验、脏行统计跳过、导入批次与任务入库、别名匹配统计、旧标题/旧封面路径回写原 Excel 与另存副本；`app/config.py` 已实现账号 slug 与 user-data-dir 创建；`app/accounts.py` 已实现账号 CRUD 服务、端口默认分配、启动登录、检测登录、生成快捷方式；`app/chrome.py` 已实现 Chrome 参数拼装、启动、CDP 端口探测、PowerShell `.lnk` 快捷方式生成；`app/gui.py` 已实现 PySide6 `MainWindow`、五 Tab、顶部 Tab 栏防误点样式、① 导入采集的 Excel 导入按钮/导入汇总栏/QTableView 任务列表/未匹配筛选与略过标记/采集旧标题旧封面 worker/采集前账号就绪预检与④引导/采集完成自动回写/旧数据回写重试按钮与 worker、② AI生成左右布局/标题与封面提示词管理/批次店铺状态筛选/任务列表/变量预览/开始生成/停止/进度/双击新旧封面预览与 `GenerateWorker`、④ 账号管理表格/弹窗/按钮/快捷方式与状态栏；`app/workers.py` 已实现 `BaseWorker`、通用 signals、取消标记和 `QThread` 启动包装。
+- 测试：`tests/` 已建立；T-006 后纯逻辑改动必须运行 `python -m unittest discover -s tests`，当前覆盖 appconfig/db/config/accounts/chrome 启动与快捷方式/editor 登录检测/excel 导入/ai 标题与封面 HTTP 解析/`generate_batch` 正常、失败与停止/prompts 读写与渲染/gui ① 导入采集/gui ② AI生成布局筛选、提示词管理、生成 worker 与双击预览/gui ④ 账号管理/worker signal 与线程包装；CDP/Shopee 改动仍需测试商品手动验证。
 - 数据：`config.json`、`config/ai_models.json`、`cmshopee.db`、`chrome_user_data_dir/`、`images/` 已由 `.gitignore` 排除；运营填写后的 Excel 业务文件默认忽略，标准空模板 `shopee待处理任务模板.xlsx` 可提交；`app/appconfig.py` 首次读取缺失的 `config.json` 时会在本地写默认配置，`app/db.py` 调用 `init_db()` 时会在本地创建 SQLite DB。
 
 ## 既定设计要点（文档已定）
@@ -32,17 +32,17 @@
 | `prototypes/` | 已有 | 已验证原型/探查脚本（demo/set_title/set_cover/get_title/cookies/inspect_images/grab/1.py），保留作人工回归与探查参考；见 `prototypes/README.md` |
 | `chrome-remote-debug-lan.md` | 已有 | WSL→Windows CDP 转发排查记录 |
 | `app/__init__.py` / `app/__main__.py` / `main.py` | 已有 | 正式包与启动入口；`python main.py` / `python -m app` 可运行占位入口 |
-| `app/gui.py` | 已有 | T-104/T-105/T-106/T-202/T-202b/T-203/T-204/T-204b/T-205/T-302/T-302p 产出：PySide6 `QMainWindow` + 五 Tab；顶部 Tab 栏防误点样式；① 导入采集导入按钮、导入汇总栏、`QTableView` 任务列表、未匹配筛选与略过标记、采集旧标题旧封面 worker、采集前账号就绪预检与④引导、采集完成自动回写与手动重试；② AI生成左右布局、提示词管理、筛选栏和任务列表；④ 账号管理表格、账号弹窗、启动登录、检测登录、快捷方式 |
+| `app/gui.py` | 已有 | T-104/T-105/T-106/T-202/T-202b/T-203/T-204/T-204b/T-205/T-302/T-302p/T-303 产出：PySide6 `QMainWindow` + 五 Tab；顶部 Tab 栏防误点样式；① 导入采集导入按钮、导入汇总栏、`QTableView` 任务列表、未匹配筛选与略过标记、采集旧标题旧封面 worker、采集前账号就绪预检与④引导、采集完成自动回写与手动重试；② AI生成左右布局、提示词管理、筛选栏、任务列表、开始生成/停止/进度、双击新旧封面预览与 `GenerateWorker`；④ 账号管理表格、账号弹窗、启动登录、检测登录、快捷方式 |
 | `app/workers.py` | 已有 | T-104b 产出：`BaseWorker` + 通用 signals + 取消标记 + `run_worker()` QThread 包装 |
 | `app/accounts.py` | 已有 | T-105/T-106 产出：账号 CRUD 服务、目录创建、端口分配、启动登录、检测登录、快捷方式 |
 | `app/editor.py` | 已有 | T-001/T-103 产出：登录状态检测、打开商品页、读/写标题、读/下载封面、上传拖封面、更新按钮、apply_task |
 | `app/appconfig.py` | 已有 | T-002 产出：`config.json` 默认值、读写、更新、路径/端口/AI 参数读取；拒绝敏感字段写入 |
-| `app/ai.py` | 已有 | T-301 产出：`gen_title()`/`gen_cover()`；读取默认模型；通用 HTTP 调用；失败重试；错误脱敏；封面按 resolution/jpg_quality 保存 |
+| `app/ai.py` | 已有 | T-301/T-303 产出：`gen_title()`/`gen_cover()`/`generate_batch()`；读取默认模型；通用 HTTP 调用；失败重试；错误脱敏；封面按 resolution/jpg_quality 保存；批量生成先标题后封面、进度回调、逐条落库、失败标记、停止取消未开始项 |
 | `app/prompts.py` | 已有 | T-302p 产出：标题提示词读写、封面模板列表/读取/保存/重命名/删除、变量替换 |
 | `app/db.py` | 已有 | T-003 产出：batches/accounts/tasks schema；WAL/busy_timeout/foreign_keys；账号/批次/任务与 set_* 阶段写库 |
 | `app/config.py` | 已有 | T-101 产出：别名→稳定 slug；创建并返回绝对 user-data-dir |
 | `app/chrome.py` | 已有 | T-102/T-106 产出：Chrome 启动参数、`subprocess.Popen` 启动、`/json/version` 端口探测、PowerShell `.lnk` 快捷方式 |
-| `tests/` | 已有 | T-006/T-201/T-202/T-202b/T-203/T-204/T-204b/T-205/T-301/T-302/T-302p 产出：stdlib unittest 基座；覆盖 appconfig/db/config/accounts/chrome/editor/excel/ai/prompts/gui/workers |
+| `tests/` | 已有 | T-006/T-201/T-202/T-202b/T-203/T-204/T-204b/T-205/T-301/T-302/T-302p/T-303 产出：stdlib unittest 基座；覆盖 appconfig/db/config/accounts/chrome/editor/excel/ai/prompts/gui/workers |
 | `app/excel.py` | 已有 | T-201/T-204 产出：多文件 Excel 输入列解析、必需列整文件拒绝、脏行逐行跳过、批次/任务入库、匹配统计；按源文件/工作表/行号回写旧标题与旧封面路径；支持原文件被占用时另存副本 |
 | `shopee待处理任务模板.xlsx` | 已有，已提交 | 标准空 Excel 模板；单工作表 `待处理任务`，表头 `账号名 | 别名 | 商品id | 旧标题 | 旧封面图片路径 | 新标题 | 新封面图片路径 | 更新状态`；运营复制后填写，填写副本不提交 |
 | `config.json` / `config/ai_models.json` / `cmshopee.db` / `chrome_user_data_dir/` / `images/` | 本地存在或按需生成，已忽略 | 含配置、密钥、业务、登录态、图片，不提交版本库 |
@@ -60,14 +60,14 @@
 
 任务状态以 [`06-tasks.md`](06-tasks.md) 为准，历史记录见 [`../progress.md`](../progress.md)。
 
-- 已完成：T-000（正式代码包结构）、T-001（`app/editor.py` 模块化）、T-002（`app/appconfig.py` + `config.json`）、T-003（SQLite 建表）、T-004（本地数据 gitignore）、T-005（AI 模型清单后端）、T-006（单元测试基座）、T-101（账号 slug/user-data-dir）、T-102（Chrome 启动器）、T-103（登录保活与检测）、T-104（PySide6 五 Tab 主窗口骨架）、T-104b（PySide6 worker 基座）、T-105（Tab④ 账号管理）、T-106（账号快捷方式）、T-201（Excel 导入：解析多文件输入列入库）、T-202（Tab① 任务列表 + 导入按钮 + 别名匹配标记）、T-202b（Tab① 导入汇总栏）、T-203（采集旧标题+旧封面）、T-204（回写旧字段到原 Excel）、T-204b（采集完成自动回写旧字段）、T-205（首次未配账号 / Chrome 未启动 / 未登录引导保护）、T-301（AI 生成接口）、T-302（Tab② 左右布局与任务列表）、T-302p（提示词管理）。
+- 已完成：T-000（正式代码包结构）、T-001（`app/editor.py` 模块化）、T-002（`app/appconfig.py` + `config.json`）、T-003（SQLite 建表）、T-004（本地数据 gitignore）、T-005（AI 模型清单后端）、T-006（单元测试基座）、T-101（账号 slug/user-data-dir）、T-102（Chrome 启动器）、T-103（登录保活与检测）、T-104（PySide6 五 Tab 主窗口骨架）、T-104b（PySide6 worker 基座）、T-105（Tab④ 账号管理）、T-106（账号快捷方式）、T-201（Excel 导入：解析多文件输入列入库）、T-202（Tab① 任务列表 + 导入按钮 + 别名匹配标记）、T-202b（Tab① 导入汇总栏）、T-203（采集旧标题+旧封面）、T-204（回写旧字段到原 Excel）、T-204b（采集完成自动回写旧字段）、T-205（首次未配账号 / Chrome 未启动 / 未登录引导保护）、T-301（AI 生成接口）、T-302（Tab② 左右布局与任务列表）、T-302p（提示词管理）、T-303（Tab② 开始生成 + 停止 + 进度）。
 - 正在进行：无。
-- 下一个可领取任务：**T-303（Tab② 开始生成 + 停止 + 进度）**。
+- 下一个可领取任务：**T-401（Tab③ 列出已生成任务 + 按批次/店铺/状态筛选 + 开始更新确认弹窗）**。
 
 ## 当前已知限制
 
 - ① 采集依赖对应账号 Chrome 已用专属 user-data-dir 和 CDP 端口启动并登录；T-205 已在采集前拦截未配置账号、Chrome 未启动、未登录，并引导去④账号管理，但不会无提示批量启动所有账号 Chrome。
-- T-301 已完成通用 HTTP AI 接口和 mock 单测；真实 AI 生成还需要在 `config/ai_models.json` 填入可用 url/model/api_key 后做一次成本可控的实测。
+- T-301/T-303 已完成通用 HTTP AI 接口、批量生成编排和 GUI 接入 mock 单测；真实 AI 生成还需要在 `config/ai_models.json` 填入可用 url/model/api_key 后做一次成本可控的小样本实测。
 
 ## 当前可运行内容
 

docs/routes.md

@@ -148,7 +148,7 @@
 | --- | --- | --- |
 | `MainWindow(QMainWindow)` | 根窗口 | 持有 `QTabWidget`、状态栏、全局消息 |
 | `CollectTab(QWidget)` | ① | 导入、任务表、采集、回写 |
-| `GenerateTab(QWidget)` | ② | 左提示词管理 + 右筛选/任务列表；双击看图与开始生成由后续 T-303 接入 |
+| `GenerateTab(QWidget)` | ② | 左提示词管理 + 右筛选/任务列表；双击看新旧封面；开始生成/停止/进度已接入 `GenerateWorker` |
 | `ApplyTab(QWidget)` | ③ | 已生成任务、开始更新确认、换标题+封面+提交、回写 |
 | `AccountsTab(QWidget)` | ④ | 账号增删改、启动登录、检测登录、生成快捷方式 |
 | `SettingsTab(QWidget)` | ⑤ | AI/目录/Chrome 配置 |

progress.md

@@ -499,3 +499,13 @@
 - 测试：新增 `tests/test_prompts.py`；`tests/test_gui.py` 覆盖标题回显/保存、封面模板另存/重命名/删除、插入标题与变量预览。
 - 文档：`docs/06-tasks.md` 将 T-302p 标为 DONE；同步 `docs/api.md`、`docs/routes.md`、`docs/current-state.md`，下一个可领取任务更新为 T-303。
 - 验证：`python -m unittest discover -s tests -p "test_prompts.py"` 通过（3 tests）；`python -m unittest discover -s tests -p "test_gui.py"` 通过（21 tests）；`python -m compileall app main.py tests` 通过；`python -m unittest discover -s tests` 通过（69 tests）；`py -3 -m compileall app main.py tests` 通过；`py -3 -m unittest discover -s tests` 通过（40 tests，skipped=4，py -3 环境缺 openpyxl/PySide6/Pillow，相关测试按设计跳过）。
+
+## 【2026-06-27】T-303 Tab② 开始生成 + 停止 + 进度
+
+- 状态：DONE
+- 变更：`app/ai.py` 新增 `generate_batch()` 批量编排，按 `title_concurrency` 先并发生成标题，再按 `image_concurrency` 并发生成封面；封面成功后立即 `db.set_generated()`，标题/封面失败 `db.mark_failed(..., "generate", error)`，失败不阻塞其余任务；`should_stop()` 为真时取消未开始的 Future，被取消任务不记失败。
+- GUI：`GenerateTab` 底部接入「开始生成」「停止」和进度标签；开始生成只处理当前筛选结果里 `stage=collected` 的任务；新增 `GenerateWorker` 通过 signal 回传进度、行刷新、完成/取消状态；任务表双击弹窗查看旧封面与新封面对照。
+- 测试：`tests/test_ai.py` 覆盖 `generate_batch` 正常落库、单条失败不阻塞、停止前取消且任务保持 collected；`tests/test_gui.py` 覆盖生成控件、双击预览弹窗、`GenerateWorker` 调用 `ai.generate_batch()` 并发出 progress/row_updated。
+- 文档：`docs/06-tasks.md` 将 T-303 标为 DONE；同步 `docs/api.md`、`docs/routes.md`、`docs/current-state.md`，下一个可领取任务更新为 T-401。
+- 验证：`python -m unittest discover -s tests -p "test_ai.py"` 通过（6 tests）；`python -m unittest discover -s tests -p "test_gui.py"` 通过（22 tests）；`python -m compileall app main.py tests` 通过；`python -m unittest discover -s tests` 通过（73 tests）。
+- 注意：本轮仍为 mock AI 单测，未调用真实 AI 服务；真实生成需在 `config/ai_models.json` 配好可用 text/image 模型后做小样本实测。

tests/test_ai.py

@@ -4,13 +4,14 @@ import json
 import os
 import sys
 import unittest
+from types import SimpleNamespace
 from unittest import mock
 
 sys.path.insert(0, os.path.dirname(__file__))
 
 from _helpers import TempDirMixin
 
-from app import ai, appconfig
+from app import ai, appconfig, db
 
 
 class _Response:
@@ -64,6 +65,35 @@ class AITests(TempDirMixin, unittest.TestCase):
         cfg["ai"]["jpg_quality"] = 80
         return cfg
 
+    def _collected_tasks(self, temp_dir, cfg, titles=None):
+        titles = titles or ["旧标题A", "旧标题B"]
+        db.init_db(cfg["db_path"])
+        batch_id = db.create_batch([os.path.join(temp_dir, "input.xlsx")], path=cfg["db_path"])
+        db.insert_tasks(
+            batch_id,
+            [
+                {
+                    "source_file_abs": os.path.join(temp_dir, "input.xlsx"),
+                    "source_sheet": "商品",
+                    "source_row": index + 2,
+                    "account_name": "Excel主店",
+                    "alias": "alias-a",
+                    "item_id": "5110063951%s" % index,
+                }
+                for index in range(len(titles))
+            ],
+            path=cfg["db_path"],
+        )
+        tasks = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
+        for task, title in zip(tasks, titles):
+            db.set_collected(
+                task.id,
+                title,
+                os.path.join(temp_dir, "%s_old.jpg" % task.item_id),
+                path=cfg["db_path"],
+            )
+        return batch_id, db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
+
     def test_gen_title_uses_configured_model_and_retries(self):
         with self.make_temp_dir() as temp_dir:
             models_path = os.path.join(temp_dir, "ai_models.json")
@@ -159,6 +189,129 @@ class AITests(TempDirMixin, unittest.TestCase):
 
         self.assert_removed(temp_dir)
 
+    def test_generate_batch_persists_titles_and_covers_per_task(self):
+        with self.make_temp_dir() as temp_dir:
+            cfg = self._config()
+            cfg["db_path"] = os.path.join(temp_dir, "cmshopee.db")
+            cfg["image_dir"] = os.path.join(temp_dir, "images")
+            cfg["ai"]["title_concurrency"] = 2
+            cfg["ai"]["image_concurrency"] = 2
+            batch_id, tasks = self._collected_tasks(temp_dir, cfg)
+            cover_prompts = []
+            progress = []
+
+            def fake_title(title_prompt, old_title, **kwargs):
+                self.assertEqual("标题提示", title_prompt)
+                return "新" + old_title
+
+            def fake_cover(cover_prompt, old_cover_path, out_path, **kwargs):
+                cover_prompts.append(cover_prompt)
+                os.makedirs(os.path.dirname(out_path), exist_ok=True)
+                with open(out_path, "wb") as fh:
+                    fh.write(b"jpeg")
+                return out_path
+
+            with mock.patch("app.ai.gen_title", side_effect=fake_title), \
+                    mock.patch("app.ai.gen_cover", side_effect=fake_cover):
+                summary = ai.generate_batch(
+                    tasks,
+                    {
+                        "title": "标题提示",
+                        "cover": "封面 {新标题} {店铺} {商品id}",
+                    },
+                    ai_cfg={
+                        "config": cfg,
+                        "db_path": cfg["db_path"],
+                        "account_by_alias": {
+                            "alias-a": SimpleNamespace(account_name="主店", slug="main")
+                        },
+                    },
+                    on_progress=progress.append,
+                )
+
+            self.assertTrue(summary["ok"])
+            self.assertEqual(2, summary["title_done"])
+            self.assertEqual(2, summary["cover_done"])
+            self.assertEqual(0, summary["failed"])
+            self.assertTrue(progress)
+            updated = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
+            self.assertTrue(all(task.stage == "generated" for task in updated))
+            self.assertEqual({"新旧标题A", "新旧标题B"}, {task.new_title for task in updated})
+            self.assertTrue(all(task.new_cover_path.endswith("_new.jpg") for task in updated))
+            self.assertTrue(all(os.path.exists(task.new_cover_path) for task in updated))
+            self.assertIn("主店", "\n".join(cover_prompts))
+            self.assertIn("新旧标题A", "\n".join(cover_prompts))
+
+        self.assert_removed(temp_dir)
+
+    def test_generate_batch_marks_failed_task_without_blocking_others(self):
+        with self.make_temp_dir() as temp_dir:
+            cfg = self._config()
+            cfg["db_path"] = os.path.join(temp_dir, "cmshopee.db")
+            cfg["image_dir"] = os.path.join(temp_dir, "images")
+            batch_id, tasks = self._collected_tasks(temp_dir, cfg, ["好标题", "坏标题"])
+
+            def fake_title(title_prompt, old_title, **kwargs):
+                if old_title == "坏标题":
+                    raise ai.AIError("标题生成失败")
+                return "新" + old_title
+
+            def fake_cover(cover_prompt, old_cover_path, out_path, **kwargs):
+                os.makedirs(os.path.dirname(out_path), exist_ok=True)
+                with open(out_path, "wb") as fh:
+                    fh.write(b"jpeg")
+                return out_path
+
+            with mock.patch("app.ai.gen_title", side_effect=fake_title), \
+                    mock.patch("app.ai.gen_cover", side_effect=fake_cover):
+                summary = ai.generate_batch(
+                    tasks,
+                    {"title": "标题提示", "cover": "封面 {新标题}"},
+                    ai_cfg={"config": cfg, "db_path": cfg["db_path"]},
+                )
+
+            self.assertFalse(summary["ok"])
+            self.assertEqual(1, summary["title_done"])
+            self.assertEqual(1, summary["cover_done"])
+            self.assertEqual(1, summary["failed"])
+            updated = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
+            by_title = {task.old_title: task for task in updated}
+            self.assertEqual("generated", by_title["好标题"].stage)
+            self.assertEqual("success", by_title["好标题"].status)
+            self.assertEqual("collected", by_title["坏标题"].stage)
+            self.assertEqual("failed", by_title["坏标题"].status)
+            self.assertIn("标题生成失败", by_title["坏标题"].last_error)
+
+        self.assert_removed(temp_dir)
+
+    def test_generate_batch_stop_before_scheduling_keeps_tasks_collected(self):
+        with self.make_temp_dir() as temp_dir:
+            cfg = self._config()
+            cfg["db_path"] = os.path.join(temp_dir, "cmshopee.db")
+            cfg["image_dir"] = os.path.join(temp_dir, "images")
+            batch_id, tasks = self._collected_tasks(temp_dir, cfg)
+
+            with mock.patch("app.ai.gen_title") as gen_title, \
+                    mock.patch("app.ai.gen_cover") as gen_cover:
+                summary = ai.generate_batch(
+                    tasks,
+                    {"title": "标题提示", "cover": "封面"},
+                    ai_cfg={"config": cfg, "db_path": cfg["db_path"]},
+                    should_stop=lambda: True,
+                )
+
+            self.assertFalse(summary["ok"])
+            self.assertTrue(summary["cancelled"])
+            self.assertEqual(0, summary["title_done"])
+            self.assertEqual(0, summary["cover_done"])
+            self.assertEqual(0, summary["failed"])
+            gen_title.assert_not_called()
+            gen_cover.assert_not_called()
+            updated = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
+            self.assertTrue(all(task.stage == "collected" for task in updated))
+
+        self.assert_removed(temp_dir)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_gui.py

@@ -22,6 +22,7 @@ from app.gui import (
     AccountsTab,
     CollectWorker,
     CollectTab,
+    GenerateWorker,
     GenerateTab,
     MainWindow,
     TAB_STYLE,
@@ -91,6 +92,10 @@ class GuiTests(TempDirMixin, unittest.TestCase):
             self.assertEqual("标题提示词", tab.title_prompt_edit.placeholderText())
             self.assertEqual("封面提示词", tab.cover_prompt_edit.placeholderText())
             self.assertEqual("保存标题提示词", tab.save_title_button.text())
+            self.assertEqual("开始生成", tab.generate_button.text())
+            self.assertEqual("停止", tab.stop_generate_button.text())
+            self.assertFalse(tab.stop_generate_button.isEnabled())
+            self.assertEqual("进度：标题0/0 · 封面0/0 · 失败0", tab.progress_label.text())
             self.assertEqual("默认", tab.cover_template_combo.currentText())
             self.assertEqual(["店铺", "商品ID", "旧标题", "新标题", "状态"], tab.model.HEADERS)
             self.assertEqual("任务 0/0 条", tab.summary_label.text())
@@ -161,6 +166,10 @@ class GuiTests(TempDirMixin, unittest.TestCase):
                 tab.preview_cover_prompt()
             self.assertIn("预览 旧标题 新标题 51100639510 主店", info.call_args[0][2])
 
+            with mock.patch("app.gui.QDialog.exec", return_value=0) as exec_dialog:
+                tab.show_task_images(tab.model.index(0, 0))
+            exec_dialog.assert_called_once()
+
             tab.cover_prompt_edit.moveCursor(QTextCursor.End)
             tab.insert_title_placeholder()
             self.assertTrue(tab.cover_prompt_edit.toPlainText().endswith("{新标题}"))
@@ -171,6 +180,59 @@ class GuiTests(TempDirMixin, unittest.TestCase):
 
         self.assert_removed(temp_dir)
 
+    def test_generate_worker_calls_generate_batch_and_emits_signals(self):
+        with self.make_temp_dir() as temp_dir:
+            cfg = self.make_config(temp_dir)
+            account = accounts.create_account("主店", "alias-a", debug_port=9222, config=cfg)
+            batch_id = db.create_batch(["input.xlsx"], path=cfg["db_path"])
+            db.insert_tasks(
+                batch_id,
+                [
+                    {
+                        "source_file_abs": os.path.join(temp_dir, "input.xlsx"),
+                        "source_sheet": "商品",
+                        "source_row": 2,
+                        "account_name": "Excel主店",
+                        "alias": "alias-a",
+                        "item_id": "51100639510",
+                    }
+                ],
+                path=cfg["db_path"],
+            )
+            task = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])[0]
+            db.set_collected(task.id, "旧标题", "old.jpg", path=cfg["db_path"])
+            tasks = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
+            progress = []
+            rows = []
+
+            def fake_generate_batch(tasks_arg, prompt_values, ai_cfg=None, on_progress=None, should_stop=None):
+                self.assertEqual(tasks, tasks_arg)
+                self.assertEqual({"title": "标题提示", "cover": "封面提示"}, prompt_values)
+                self.assertEqual(account, ai_cfg["account_by_alias"]["alias-a"])
+                self.assertEqual(cfg["db_path"], ai_cfg["db_path"])
+                self.assertFalse(should_stop())
+                on_progress({"total": 1, "title_done": 1, "cover_done": 0, "failed": 0})
+                ai_cfg["on_task_update"](tasks[0].id, {"stage": "generated"})
+                return {"ok": True, "total": 1, "title_done": 1, "cover_done": 1, "failed": 0}
+
+            worker = GenerateWorker(
+                tasks,
+                {"title": "标题提示", "cover": "封面提示"},
+                db_path=cfg["db_path"],
+                config=cfg,
+            )
+            worker.progress.connect(progress.append)
+            worker.row_updated.connect(lambda task_id, fields: rows.append((task_id, fields)))
+
+            with mock.patch("app.gui.ai.generate_batch", side_effect=fake_generate_batch):
+                summary = worker.execute()
+
+            self.assertEqual(1, summary["cover_done"])
+            self.assertEqual([{"total": 1, "title_done": 1, "cover_done": 0, "failed": 0}], progress)
+            self.assertEqual([(tasks[0].id, {"stage": "generated"})], rows)
+
+        self.assert_removed(temp_dir)
+
     def test_generate_tab_lists_tasks_and_filters_by_shop_status_and_batch(self):
         with self.make_temp_dir() as temp_dir:
             cfg = self.make_config(temp_dir)