From d64e3c74046036121f41351e604db797dde20b98 Mon Sep 17 00:00:00 2001
From: chengma <chengma2027@gmail.com>
Date: Sat, 18 Jul 2026 17:10:12 +0800
Subject: [PATCH] feat(apply): block abnormal product statuses

---
 app/gui/tabs/apply.py        | 117 +++++++++++++---
 app/gui/workers.py           |  24 +++-
 app/product_status.py        |  60 ++++++++-
 docs/04-architecture.md      |   3 +-
 docs/api.md                  |  10 +-
 docs/routes.md               |   8 +-
 docs/tasks/T-662d.md         |   8 +-
 tests/test_gui.py            | 255 +++++++++++++++++++++++++++++++++--
 tests/test_product_status.py |  65 +++++++++
 9 files changed, 499 insertions(+), 51 deletions(-)

diff --git a/app/gui/tabs/apply.py b/app/gui/tabs/apply.py
index cc46409..ab20145 100644
--- a/app/gui/tabs/apply.py
+++ b/app/gui/tabs/apply.py
@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from ... import product_status
 from ..models import ApplyTaskTableModel
 from ..widgets import *
 from ..workers import ApplyWorker as _RealApplyWorker, WriteBackWorker as _RealWriteBackWorker
@@ -244,21 +245,23 @@ class ApplyTab(QWidget):
         if self.apply_thread is not None:
             self._set_status("更新正在进行...")
             return
-        tasks = [
-            task for task in self.model.tasks
-            if self._is_update_candidate_task(task)
-        ]
-        if not tasks:
+        base_candidates = self._update_candidates()
+        if not base_candidates:
             self._set_status("当前筛选结果没有可更新任务")
             return
         update_cfg = self._shopee_update_config()
         update_mode = self._current_update_mode()
-        content_plan = self._partition_update_content_tasks(tasks, update_mode)
-        executable_tasks = content_plan["executable"]
+        update_plan = product_status.build_apply_plan(base_candidates, update_mode)
+        executable_tasks = update_plan["executable"]
         if not executable_tasks:
-            content_error = self._update_content_error(content_plan, update_mode)
-            QMessageBox.warning(self, "更新内容未生成", content_error)
-            self._set_status(content_error.replace("\n", " "))
+            preflight_error = self._update_preflight_error(update_plan, update_mode)
+            title = (
+                "商品状态不允许更新"
+                if update_plan["status_scope_excluded"]
+                else "更新内容未生成"
+            )
+            QMessageBox.warning(self, title, preflight_error)
+            self._set_status(preflight_error.replace("\n", " "))
             return
         dry_run = bool(dry_run)
         answer = QMessageBox.question(
@@ -267,7 +270,7 @@ class ApplyTab(QWidget):
             self._confirmation_message(
                 executable_tasks,
                 dry_run=dry_run,
-                content_plan=content_plan,
+                content_plan=update_plan,
             ),
             QMessageBox.Yes | QMessageBox.No,
             QMessageBox.No,
@@ -275,6 +278,18 @@ class ApplyTab(QWidget):
         if answer != QMessageBox.Yes:
             self._set_status("已取消检查本轮更新" if dry_run else "已取消开始更新")
             return
+        current_plan = product_status.build_apply_plan(
+            self._update_candidates(),
+            update_mode,
+        )
+        if current_plan["fingerprint"] != update_plan["fingerprint"]:
+            self._set_status("当前任务数据已变化，请重新开始更新")
+            return
+        executable_tasks = current_plan["executable"]
+        if not executable_tasks:
+            preflight_error = self._update_preflight_error(current_plan, update_mode)
+            self._set_status(preflight_error.replace("\n", " "))
+            return
         batch_size = max(1, int(update_cfg.get("max_items_per_run", 1) or 1))
         worker = ApplyWorker(
             executable_tasks,
@@ -288,6 +303,10 @@ class ApplyTab(QWidget):
             ),
             batch_size=batch_size,
             diagnostic_log_dir=diagnostics.DEFAULT_LOG_DIR,
+            product_status_counts=current_plan["status_counts"],
+            status_scope_excluded=current_plan["status_scope_excluded"],
+            content_scope_excluded=current_plan["content_scope_excluded"],
+            apply_plan_fingerprint=current_plan["fingerprint"],
         )
         worker.progress.connect(self._on_apply_progress)
         worker.row_updated.connect(self._on_apply_row_updated)
@@ -304,9 +323,9 @@ class ApplyTab(QWidget):
             executable_tasks,
             dry_run=dry_run,
             batch_size=batch_size,
-            content_plan=content_plan,
+            content_plan=current_plan,
         )
-        skip_status = self._content_skip_status(content_plan)
+        skip_status = self._content_skip_status(current_plan)
         if dry_run:
             self._set_status(f"开始检查本轮更新：{len(executable_tasks)} 条{skip_status}")
         else:
@@ -407,6 +426,13 @@ class ApplyTab(QWidget):
             and getattr(task, "status", None) in {"success", "pending", "failed"}
         )
 
+    def _update_candidates(self):
+        return [
+            task
+            for task in self.model.tasks
+            if self._is_update_candidate_task(task)
+        ]
+
     def _populate_batch_filter(self, batches, selected_batch):
         had_previous_items = self.batch_filter.count() > 0
         self.batch_filter.blockSignals(True)
@@ -617,6 +643,22 @@ class ApplyTab(QWidget):
             "missing_cover": missing_cover,
         }
 
+    def _update_preflight_error(self, update_plan, update_mode):
+        if update_plan["executable"]:
+            return None
+        lines = []
+        status_text = self._status_skip_text(update_plan)
+        if status_text:
+            lines.append("当前筛选结果没有状态正常且可更新的商品。")
+            lines.append(status_text)
+            lines.append("请先回到①导入采集重新确认商品状态。")
+        content_error = self._update_content_error(update_plan, update_mode)
+        if content_error:
+            if lines:
+                lines.append("")
+            lines.append(content_error)
+        return "\n".join(lines) or "当前筛选结果没有可更新任务。"
+
     def _update_content_error(self, tasks_or_plan, update_mode):
         if isinstance(tasks_or_plan, dict):
             content_plan = tasks_or_plan
@@ -643,12 +685,13 @@ class ApplyTab(QWidget):
     def _content_skip_text(self, content_plan):
         if not content_plan:
             return ""
+        status_text = self._status_skip_text(content_plan)
         missing_title = content_plan.get("missing_title", [])
         missing_cover = content_plan.get("missing_cover", [])
         skipped_count = len({id(task) for task in missing_title + missing_cover})
-        if not skipped_count:
-            return ""
-        lines = [f"本轮将跳过 {skipped_count} 条缺少所选更新内容的记录："]
+        lines = [status_text] if status_text else []
+        if skipped_count:
+            lines.append(f"本轮将跳过 {skipped_count} 条缺少所选更新内容的记录：")
         if missing_title:
             lines.append(
                 f"缺少新标题：{len(missing_title)} 条（示例商品ID：{self._sample_item_ids(missing_title)}）"
@@ -662,16 +705,43 @@ class ApplyTab(QWidget):
     def _content_skip_status(self, content_plan):
         if not content_plan:
             return ""
+        status_skipped = int(content_plan.get("status_scope_excluded", 0) or 0)
         missing_title = content_plan.get("missing_title", [])
         missing_cover = content_plan.get("missing_cover", [])
         skipped_count = len({id(task) for task in missing_title + missing_cover})
-        if not skipped_count:
+        if not skipped_count and not status_skipped:
             return ""
-        return (
-            f"，跳过 {skipped_count} 条（缺标题 {len(missing_title)} / "
-            f"缺封面 {len(missing_cover)}）"
+        parts = []
+        if status_skipped:
+            parts.append(f"状态异常 {status_skipped}")
+        if skipped_count:
+            parts.append(
+                f"缺标题 {len(missing_title)} / 缺封面 {len(missing_cover)}"
+            )
+        return "，跳过 {total} 条（{details}）".format(
+            total=status_skipped + skipped_count,
+            details="；".join(parts),
         )
 
+    def _status_skip_text(self, update_plan):
+        if not update_plan:
+            return ""
+        rows = [
+            ("未上架", update_plan.get("unlisted", [])),
+            ("审核中", update_plan.get("reviewing", [])),
+            ("状态未知", update_plan.get("unknown", [])),
+        ]
+        skipped = sum(len(tasks) for _label, tasks in rows)
+        if not skipped:
+            return ""
+        lines = [f"本轮将排除 {skipped} 条商品状态异常记录："]
+        for label, tasks in rows:
+            if tasks:
+                lines.append(
+                    f"{label}：{len(tasks)} 条（示例商品ID：{self._sample_item_ids(tasks)}）"
+                )
+        return "\n".join(lines)
+
     def _sample_item_ids(self, tasks):
         values = [
             str(getattr(task, "item_id", "") or "").strip()
@@ -784,8 +854,13 @@ class ApplyTab(QWidget):
         )
         skip_status = self._content_skip_status(content_plan)
         if skip_status:
+            excluded_label = (
+                "预检排除记录"
+                if content_plan and content_plan.get("status_scope_excluded")
+                else "缺失记录"
+            )
             self._append_run_log(
-                f"本轮预检：可执行 {len(tasks)} 条{skip_status}；缺失记录未进入本轮执行。"
+                f"本轮预检：可执行 {len(tasks)} 条{skip_status}；{excluded_label}未进入本轮执行。"
             )
 
     def _load_latest_run_log(self):
diff --git a/app/gui/workers.py b/app/gui/workers.py
index 71182a4..d1176d6 100644
--- a/app/gui/workers.py
+++ b/app/gui/workers.py
@@ -1251,6 +1251,10 @@ class ApplyWorker(BaseWorker):
         max_parallel_accounts=1,
         batch_size=None,
         diagnostic_log_dir=None,
+        product_status_counts=None,
+        status_scope_excluded=0,
+        content_scope_excluded=0,
+        apply_plan_fingerprint=None,
     ):
         super().__init__()
         self.tasks = list(tasks)
@@ -1271,6 +1275,13 @@ class ApplyWorker(BaseWorker):
         self._batch_count = 0
         self._progress_lock = threading.Lock()
         self.diagnostic_log_dir = diagnostic_log_dir
+        self.product_status_counts = {
+            status: int((product_status_counts or {}).get(status, 0) or 0)
+            for status in product_status.VALID_PRODUCT_STATUSES
+        }
+        self.status_scope_excluded = max(0, int(status_scope_excluded or 0))
+        self.content_scope_excluded = max(0, int(content_scope_excluded or 0))
+        self.apply_plan_fingerprint = str(apply_plan_fingerprint or "") or None
         self._run_id = None
 
     def execute(self):
@@ -1295,7 +1306,7 @@ class ApplyWorker(BaseWorker):
         }
         self._run_id = self._create_run_log(eligible, batch_ids)
         self._log_run_event(
-            "step=start result=start detail=运行开始：{mode}，更新内容{update_mode}，任务{total}，每批最多{batch_size}，批次{batch_count}，{parallel}".format(
+            "step=start result=start detail=运行开始：{mode}，更新内容{update_mode}，任务{total}，每批最多{batch_size}，批次{batch_count}，{parallel}；商品状态异常排除{status_excluded}条；内容缺失排除{content_excluded}条".format(
                 mode="检查本轮更新" if self.dry_run else "真实更新",
                 update_mode=_update_mode_label(self.update_mode),
                 total=total,
@@ -1306,6 +1317,8 @@ class ApplyWorker(BaseWorker):
                     if self.max_parallel_accounts > 1
                     else "串行"
                 ),
+                status_excluded=self.status_scope_excluded,
+                content_excluded=self.content_scope_excluded,
             )
         )
 
@@ -1363,6 +1376,7 @@ class ApplyWorker(BaseWorker):
         return (
             getattr(task, "stage", None) == "generated"
             and getattr(task, "status", None) in {"success", "pending", "failed"}
+            and product_status.is_normal(getattr(task, "product_status", None))
             and (
                 (
                     appconfig.update_mode_includes_title(self.update_mode)
@@ -1766,6 +1780,10 @@ class ApplyWorker(BaseWorker):
             "batch_count": self._batch_count,
             "update_mode": self.update_mode,
             "run_id": self._run_id,
+            "product_status_counts": dict(self.product_status_counts),
+            "status_scope_excluded": self.status_scope_excluded,
+            "content_scope_excluded": self.content_scope_excluded,
+            "apply_plan_fingerprint": self.apply_plan_fingerprint,
         }
         if blocked:
             summary["blocked"] = True
@@ -1787,6 +1805,10 @@ class ApplyWorker(BaseWorker):
                     "batch_size": self._current_batch_size,
                     "batch_count": self._batch_count,
                     "update_mode": self.update_mode,
+                    "product_status_counts": dict(self.product_status_counts),
+                    "status_scope_excluded": self.status_scope_excluded,
+                    "content_scope_excluded": self.content_scope_excluded,
+                    "apply_plan_fingerprint": self.apply_plan_fingerprint,
                 },
                 path=self.db_path,
             )
diff --git a/app/product_status.py b/app/product_status.py
index e90f0d0..eefe8da 100644
--- a/app/product_status.py
+++ b/app/product_status.py
@@ -125,7 +125,61 @@ def build_generation_plan(tasks, generate_mode, scope=SCOPE_NORMAL_ONLY) -> dict
         "execution_tasks": execution_tasks,
         "scope": normalized_scope,
         "scope_excluded": len(base_candidates) - len(execution_tasks),
-        "fingerprint": _generation_fingerprint(base_candidates, generate_mode),
+        "fingerprint": _task_plan_fingerprint(base_candidates, generate_mode),
+    }
+
+
+def build_apply_plan(tasks, update_mode) -> dict:
+    """Return a status-first, content-aware plan for Shopee update actions."""
+
+    base_candidates = _deduplicate_tasks(tasks)
+    status_counts = {status: 0 for status in VALID_PRODUCT_STATUSES}
+    status_excluded = {
+        STATUS_UNLISTED: [],
+        STATUS_REVIEWING: [],
+        STATUS_UNKNOWN: [],
+    }
+    normal_candidates = []
+    for task in base_candidates:
+        status = _task_status(task)
+        status_counts[status] += 1
+        if status == STATUS_NORMAL:
+            normal_candidates.append(task)
+        else:
+            status_excluded[status].append(task)
+
+    mode = str(update_mode or "").strip().lower()
+    requires_title = mode in {"title", "title_cover"}
+    requires_cover = mode in {"cover", "title_cover"}
+    missing_title = [
+        task
+        for task in normal_candidates
+        if requires_title and not str(_task_value(task, "new_title") or "").strip()
+    ]
+    missing_cover = [
+        task
+        for task in normal_candidates
+        if requires_cover and not str(_task_value(task, "new_cover_path") or "").strip()
+    ]
+    content_excluded_ids = {id(task) for task in missing_title + missing_cover}
+    executable_tasks = [
+        task for task in normal_candidates if id(task) not in content_excluded_ids
+    ]
+    status_excluded_count = sum(len(rows) for rows in status_excluded.values())
+
+    return {
+        "base_candidates": base_candidates,
+        "status_counts": status_counts,
+        "executable": executable_tasks,
+        "unlisted": status_excluded[STATUS_UNLISTED],
+        "reviewing": status_excluded[STATUS_REVIEWING],
+        "unknown": status_excluded[STATUS_UNKNOWN],
+        "missing_title": missing_title,
+        "missing_cover": missing_cover,
+        "status_scope_excluded": status_excluded_count,
+        "content_scope_excluded": len(content_excluded_ids),
+        "scope_excluded": status_excluded_count + len(content_excluded_ids),
+        "fingerprint": _task_plan_fingerprint(base_candidates, update_mode),
     }
 
 
@@ -159,7 +213,7 @@ def _deduplicate_tasks(tasks) -> list:
     return unique
 
 
-def _generation_fingerprint(tasks, generate_mode) -> str:
+def _task_plan_fingerprint(tasks, mode) -> str:
     snapshots = [
         {
             "task_id": _task_value(task, "id"),
@@ -167,7 +221,7 @@ def _generation_fingerprint(tasks, generate_mode) -> str:
             "product_status": _task_status(task),
             "new_title": _task_value(task, "new_title"),
             "new_cover_path": _task_value(task, "new_cover_path"),
-            "generate_mode": str(generate_mode or ""),
+            "mode": str(mode or ""),
         }
         for task in tasks
     ]
diff --git a/docs/04-architecture.md b/docs/04-architecture.md
index 5b5b89d..1a39a59 100644
--- a/docs/04-architecture.md
+++ b/docs/04-architecture.md
@@ -476,7 +476,8 @@ data/images/<batch_id>/<slug>/<task_id>_<item_id>_new.<ext>    # AI 生成的新
 ### 6.3 应用更新（③ Tab）
 
 - ③ 顶部筛选确定本次作用范围；点击「开始更新」后弹窗展示筛选条件、任务数量和“将提交线上”的风险提示。
-- ③ 左下角提供「更新内容」下拉：`只更新标题` / `只更新封面` / `更新标题和封面`。点击「开始更新」或「检查本轮更新」时，先按当前模式分出具备 `new_title` / `new_cover_path` 的可执行任务与缺失记录；有可执行任务时缺失记录只在确认框和本轮日志中列为预检跳过，不创建其 `ApplyWorker` 执行项、不打开其 Chrome、不写其失败状态。仅当全部记录缺少所选内容时，才弹「更新内容未生成」并整体阻断。
+- ③ 左下角提供「更新内容」下拉：`只更新标题` / `只更新封面` / `更新标题和封面`。点击「开始更新」或「检查本轮更新」时，`product_status.build_apply_plan()` 先把候选按商品状态冻结：仅 `normal` 可继续检查 `new_title` / `new_cover_path`，`unlisted`、`reviewing`、`unknown` 与历史 NULL 均从 Worker 输入剔除；再对正常商品分出可执行项和内容缺失项。状态异常与内容缺失仅在确认框、状态栏和本轮日志列为预检排除，不打开其 Chrome、不写其失败状态、Excel 或运行内 worker 统计。无可执行项时，优先展示真实商品状态原因；只有没有状态异常时才沿用「更新内容未生成」文案。
+- 计划指纹覆盖 `task_id/updated_at/product_status/new_title/new_cover_path/更新模式`；用户确认后重新计算，任何变化均废弃旧计划并要求重新开始。`ApplyWorker` 仍在执行层再次验证 `product_status=normal`，防止直接构造 Worker 绕过 GUI 预检。①、②本轮的“所有状态”范围选择不向③传递更新授权。
 - ③ 提供「检查本轮更新」按钮：只读取当前筛选结果和写运行日志，不打开 Shopee、不提交、不改任务状态；检查汇总展示总数、店铺分布、每批最大条数、预计批次数、更新内容和略过原因。
 - 弹确认前先读取 `data/config.json` 的 `shopee_update` 执行参数。普通正式更新不再检查 `test_item_id` 或旧真实提交开关，当前筛选结果可以包含多个真实商品 ID。`max_items_per_run` 作为每批最大任务数，当前筛选总数超过该值时自动分批，不再按总数阻断。
 - 用户在③确认弹窗点「是/确认」才开始批量更新；点「否/取消」不执行、不改库。
diff --git a/docs/api.md b/docs/api.md
index bfb6c66..621cbb6 100644
--- a/docs/api.md
+++ b/docs/api.md
@@ -455,13 +455,13 @@ main() -> int                              # 创建 QApplication + MainWindow
 class MainWindow(QMainWindow)              # QTabWidget: ①、②、③、账号管理、设置、商品套图；支持注入 db_path/config/config_path/ai_models_path 便于测试
 class CollectTab(QWidget)                  # ① 导入采集：导入 Excel + 汇总栏 + QTableView 任务列表 + 未匹配略过标记
 class GenerateTab(QWidget)                 # ② AI生成：提示词管理 + 筛选任务 + 状态范围确认 + 开始/停止生成 + 新旧封面预览 + AI生成运行日志
-class ApplyTab(QWidget)                    # ③ 更新蝦皮：筛选已生成任务 + 缺失内容预检剔除 + 检查本轮更新 + 确认后分批真实更新 + 运行日志
+class ApplyTab(QWidget)                    # ③ 更新蝦皮：筛选已生成任务 + 商品状态/内容预检剔除 + 检查本轮更新 + 确认后分批真实更新 + 运行日志
 class SettingsTab(QWidget)                 # 设置：cmhub 网关配置 + 响应式三列布局 + 角色/生成参数/路径端口 + 蝦皮更新安全 + 未保存状态追踪
 class ProductSuiteTab(QWidget)             # 商品套图：多任务、原图、结构配置、AI帮写、cmhub生成、历史结果
 class ImageStudioTab(QWidget)              # 旧AI工场兼容实现；主窗口不再创建
 class CollectWorker(BaseWorker)            # ① 后台采集：范围 normal_only/all + 账号预检 -> editor.collect -> 状态独立落库，采集或略过
 class GenerateWorker(BaseWorker)           # ② 后台生成：确认后的 normal_only/all 精确任务 -> ai.generate_batch -> 写库 + 进度
-class ApplyWorker(BaseWorker)              # ③ 后台更新：账号就绪预检 -> 检查或按批调用 editor.apply_task(...) -> db.set_applied/mark_skipped
+class ApplyWorker(BaseWorker)              # ③ 后台更新：再次拒绝非正常状态 -> 账号就绪预检 -> 检查或按批调用 editor.apply_task(...) -> db.set_applied/mark_skipped
 class WriteBackWorker(BaseWorker)          # ①/③ 后台回写：旧字段或更新结果写回原 Excel
 class AIModelTestWorker(BaseWorker)        # 设置 后台测试 AI 模型连接：appconfig.test_ai_model
 class ImageStudioPullImagesWorker(BaseWorker)      # 商品套图 后台只读拉蝦皮原主图 URL；安全边界协作停止
@@ -546,14 +546,14 @@ T-523 后 GUI 已从旧 `app/gui.py` 拆为 `app/gui/` 包：`__init__.py` 负
 ③ 更新蝦皮当前要点（T-303b/T-401/T-402/T-403）：
 
 - `ApplyTab` 顶部筛选栏包含：批次、店铺、商品ID、状态、刷新。批次来自 `db.list_batches()`；店铺来自当前更新候选任务别名并优先显示匹配账号名；商品ID输入框按包含匹配 `item_id`，清空表示全部。
-- ③ 只列出已生成或可查看的更新候选任务：`stage=generated/applied`，或已有新字段且 `status=failed/skipped` 的任务；真正开始更新前再按③「更新内容」模式分出可执行任务与缺少 `new_title` / `new_cover_path` 的预检跳过记录。
+- ③ 只列出已生成或可查看的更新候选任务：`stage=generated/applied`，或已有新字段且 `status=failed/skipped` 的任务；真正开始更新前先按 `product_status` 分出可进入线上更新的 `normal` 与必须排除的 `unlisted/reviewing/unknown`（历史 NULL 归入未知），再按③「更新内容」模式分出可执行任务与缺少 `new_title` / `new_cover_path` 的预检跳过记录。
 - 状态筛选支持：已生成（默认，`stage=generated` 且 `status=success/pending`）、失败、已更新、略过、全部状态。
 - 任务列表使用 `QTableView + ApplyTaskTableModel`，列为：店铺、商品ID、新标题、新封面、阶段、结果。
 - 「开始更新」只读取当前筛选结果；无任务时只提示，不弹确认、不改库；该按钮是③的主操作，视觉上强于检查、停止和回写。
 - 「检查本轮更新」只读取当前筛选结果并创建检查运行日志，不打开 Shopee、不调用 `editor.apply_task()`、不写任务状态、不回写 Excel；检查内容包含任务总数、店铺分布、每批最大条数、预计批次数、更新内容、会更新字段和略过原因。
-- 点击「开始更新」先按③「更新内容」模式预检当前筛选任务：只更新标题要求 `new_title`，只更新封面要求 `new_cover_path`，更新标题和封面要求两者都有。部分缺失时将这些记录从本轮 `ApplyWorker` 任务中剔除，在确认框、状态栏摘要和本轮日志列出缺标题/缺封面条数与示例商品ID；不打开其 Chrome、不写其失败状态或 Excel。只有全部记录缺失时才弹「更新内容未生成」并阻断。再读取 `shopee_update` 执行参数：普通正式更新不再检查 `test_item_id` 或旧真实提交开关，当前筛选结果可包含多个真实商品 ID；`max_items_per_run` 作为每批最大更新条数，当前可执行任务超过该值时自动分批。通过后才弹窗展示批次/店铺/商品ID/状态/更新内容/可执行任务数、预检跳过项、每批最大条数、预计批次数、提交线上风险和当前执行设置。
+- 点击「开始更新」和「检查本轮更新」都先使用同一 `build_apply_plan()` 预检当前筛选任务：仅商品状态 `normal` 继续按③「更新内容」模式检查，只更新标题要求 `new_title`，只更新封面要求 `new_cover_path`，更新标题和封面要求两者都有。未上架、审核中、状态未知和历史 NULL 均不能传入正式或检查模式的 `ApplyWorker`；状态异常与内容缺失在确认框、状态栏摘要和本轮日志分类列出数量与示例商品ID，不打开其 Chrome、不写其失败状态或 Excel，也不计入 worker 成功/略过/失败统计。所有记录被排除时，若存在状态异常则弹「商品状态不允许更新」，否则弹「更新内容未生成」。确认后按 `task_id/updated_at/product_status/new_title/new_cover_path/更新模式` 重新验证计划指纹，变化即中止。再读取 `shopee_update` 执行参数：普通正式更新不再检查 `test_item_id` 或旧真实提交开关，当前筛选结果可包含多个真实商品 ID；`max_items_per_run` 作为每批最大更新条数，当前可执行任务超过该值时自动分批。通过后才弹窗展示批次/店铺/商品ID/状态/更新内容/可执行任务数、预检跳过项、每批最大条数、预计批次数、提交线上风险和当前执行设置。
 - 用户点否/取消时不执行、不改库；用户点是后才创建 `ApplyWorker` 做真实提交。
-- `ApplyWorker` 只处理经过 `ApplyTab` 预检后的任务：原候选范围为 `stage=generated`、状态为 `success/pending/failed`，缺少当前 `update_mode` 所需内容的记录不传入 worker；只更新标题时不替换封面，只更新封面时不改标题。预检跳过记录保持原 stage/status/Excel，不被计入本轮 worker 的成功、略过或失败统计。已更新和略过记录仅查看，不会再次提交，除非用户先用 T-404a 的「重置更新状态」把选中记录退回可更新。
+- `ApplyWorker` 只处理经过 `ApplyTab` 预检后的任务，并在执行层再次拒绝商品状态不是 `normal` 的记录：原候选范围为 `stage=generated`、状态为 `success/pending/failed`，缺少当前 `update_mode` 所需内容的记录不传入 worker；只更新标题时不替换封面，只更新封面时不改标题。预检跳过记录保持原 stage/status/Excel，不被计入本轮 worker 的成功、略过或失败统计。已更新和略过记录仅查看，不会再次提交，除非用户先用 T-404a 的「重置更新状态」把选中记录退回可更新。
 - 检查本轮更新：不做账号登录预检，不调用 `editor.apply_task()`，不写任务状态，不回写 Excel；只把每条“将更新/将略过”写入运行日志并弹汇总。
 - 真实更新前先做账号就绪预检：无账号、当前筛选结果匹配账号 Chrome 未启动、CDP 端口不可访问、未登录，或本轮涉及账号调试端口冲突时，返回 `blocked=True`，GUI 弹窗汇总并跳转/引导去账号管理；预检不通过时不调用 `editor.apply_task()`、不写失败状态、不自动启动 Chrome。
 - 预检通过后默认串行；若 `max_parallel_accounts>1`，按账号分组并行执行，不同账号可同时跑，同一账号内仍串行。每条执行 `db.mark_running(..., "apply")` → `editor.apply_task(account, task)` → `db.set_applied()`；成功推进 `stage=applied/status=success/committed=1`，失败保持原 stage、`status=failed/committed=0/last_error`，单条失败继续下一条。
diff --git a/docs/routes.md b/docs/routes.md
index d0f3849..ac23010 100644
--- a/docs/routes.md
+++ b/docs/routes.md
@@ -129,7 +129,7 @@
 - 顶部**按批次 / 店铺 / 商品ID / 状态筛选**（与 ①②一致）；商品ID输入框按包含匹配 `item_id`，清空表示全部；「开始更新」作用于**当前筛选结果**，是一道范围控制。
   - 店铺筛选：建议**逐店铺更新**（每店铺需先启动其 Chrome 并登录）。
   - 状态筛选：`已生成` 只跑未更新的；`失败` 用于**失败重试**；`已更新成功/略过` 仅查看。
-- 「更新内容」下拉支持只更新标题、只更新封面、更新标题和封面；点击「检查本轮更新」或「开始更新」前先按当前模式检查 `new_title` / `new_cover_path`。有合格记录时，缺失记录从本轮 worker 任务中剔除，确认框与日志列出缺标题/缺封面条数和示例商品ID；仅全部记录缺失时才中文弹窗阻断，不打开 Chrome、不改任务状态。
+- 「更新内容」下拉支持只更新标题、只更新封面、更新标题和封面；点击「检查本轮更新」或「开始更新」前统一用状态优先的预检计划：仅 `product_status=normal` 可进入标题/封面完整性检查，`unlisted`、`reviewing`、`unknown` 及历史 NULL 一律排除，不传给 `ApplyWorker`。有可执行记录时，状态异常和缺失内容记录都在确认框、状态栏和本轮日志列出分类、数量与示例商品ID；全部被排除时按真实原因中文弹窗阻断，不打开 Chrome、不改任务状态。用户确认后重新计算计划指纹，任务状态、更新时间、新标题或新封面变化则中止并要求重新开始。
 - 「检查本轮更新」只读取当前筛选结果并写运行日志，不打开 Shopee、不提交、不改任务状态；弹窗/日志展示总任务数、店铺分布、当前更新内容、会更新标题/封面、略过原因、每批最大条数和预计批次数。
 - 完成缺失内容预检并剔除不合格记录后，点击「开始更新」读取设置中的 `shopee_update` 执行设置：普通正式更新不再以测试商品 ID 或旧真实提交开关限制当前筛选结果，允许当前筛选结果包含多个真实商品 ID；`max_items_per_run` 作为**每批最大更新条数**，当前可执行任务超过该值时不阻断，而是自动分批执行。
 - 弹窗展示本次筛选条件、更新内容、任务总数、每批最大条数、预计批次数、执行设置和“将提交线上”的风险提示；用户点「是/确认」才开始，点「否/取消」不执行。
@@ -178,7 +178,7 @@
 - 路径与端口（T-501b/T-506/T-539/T-580 已接入）：组件组改为 3 个组件一组；普通设置页只显示 Chrome 路径、默认端口、端口起止、Chrome 就绪超时。T-538 后账号数据根目录、图片目录、DB 路径固定解析到 `data/` 下，普通 UI 不再提供输入框，避免用户误改后数据分裂；`config.json` 中 `user_data_root` / `image_dir` / `db_path` 字段继续作为内部兼容字段保留，手工配置值仍会被读取和保存。
 - 蝦皮更新执行（T-580 已接入）：组件组改为 3 个组件一组；普通设置页只保留「每批最大更新条数」和「同时更新蝦皮账号(1..5)」两个执行参数。`1` 表示逐个账号串行，`2..5` 表示按账号分组并行；“dry-run”不作为用户可见开关，改到③成为「检查本轮更新」按钮；测试商品 ID 和旧封面开关仅作为历史/调试兼容字段读取，普通设置页无入口，保存后不再写回。
   - ③「更新内容」默认只更新标题，每批最大更新条数默认 1，同时更新蝦皮账号默认 1。
-- ③ 点击「开始更新」会先按「更新内容」剔除缺失内容记录并在确认框说明；仅无可执行记录时中止。
+- ③ 点击「开始更新」和「检查本轮更新」会先排除非正常商品状态，再按「更新内容」剔除缺失内容记录并在确认框说明；仅无可执行记录时中止。①的“采集所有状态”和②的“生成所有状态”不构成③线上更新授权。
 
 ## 商品套图
 
@@ -242,13 +242,13 @@
 | `MainWindow(QMainWindow)` | 根窗口 | 持有 `QTabWidget`、状态栏、全局消息 |
 | `CollectTab(QWidget)` | ① | 导入、任务表、采集、回写 |
 | `GenerateTab(QWidget)` | ② | 左提示词管理 + 右筛选/任务列表；双击看新旧封面；先确认商品状态生成范围，再按本轮「生成内容」下拉接入 `GenerateWorker` |
-| `ApplyTab(QWidget)` | ③ | 已生成任务筛选 +「更新内容」下拉 + 缺失内容预检剔除 +「检查本轮更新」+ 分批开始更新确认 + 检查/真实更新运行日志 + 结果回写与结束汇总 |
+| `ApplyTab(QWidget)` | ③ | 已生成任务筛选 +「更新内容」下拉 + 商品状态优先/内容完整性预检剔除 +「检查本轮更新」+ 分批开始更新确认 + 检查/真实更新运行日志 + 结果回写与结束汇总 |
 | `AccountsTab(QWidget)` | 账号管理 | 账号增删改、启动登录、检测登录、生成快捷方式；首次启动复用初始卖家中心页，启动中防重复触发，登录检测把 Shopee accounts 登录页判为未登录 |
 | `SettingsTab(QWidget)` | 设置 | cmhub 网关配置 + 响应式三列设置表单 + 生成参数 + Chrome/端口配置 + 蝦皮更新执行；数据路径字段隐藏但保留配置兼容 |
 | `ProductSuiteTab(QWidget)` | 商品套图 | 商品套图多任务、账号+商品ID上下文、原图导入/排序、套图分类、AI帮写、cmhub 异步生成、结果历史与删除撤销 |
 | `TaskTableModel(QAbstractTableModel)` | ①②③ | 任务表格数据模型，供 `QTableView` 使用 |
 | `BaseWorker(QObject)` | 后台 | 定义 `progress/log/row_updated/failed/finished/cancelled` signals |
-| `ApplyWorker(BaseWorker)` | ③ | 账号就绪预检、检查本轮更新、按每批最大条数分批、按账号并行或串行调用 `editor.apply_task(...)`、逐条 `set_applied()`，失败继续，写运行日志 |
+| `ApplyWorker(BaseWorker)` | ③ | 账号就绪预检、检查本轮更新、按每批最大条数分批、按账号并行或串行调用 `editor.apply_task(...)`、逐条 `set_applied()`，失败继续，写运行日志；执行层再次拒绝非正常商品状态 |
 | `AIModelTestWorker(BaseWorker)` | 设置 | 后台调用 `appconfig.test_ai_model()` 测试模型连接 |
 | `WriteBackWorker(BaseWorker)` | ①③ | ①回写旧字段；③回写新标题/新封面/更新状态 |
 | `ImageStudioPullImagesWorker / ImageStudioDownloadOriginalWorker / ProductSuiteImportImagesWorker / ProductSuiteGenerateWorker / ProductSuiteAiWriteWorker / CMHubModelCatalogWorker` | 商品套图 | 后台执行只读拉主图、远程原图下载、本地图片校验复制、cmhub 套图生成、AI帮写和只读模型目录；拉图和本轮下载支持安全边界协作停止，worker 不直接操作 QWidget |
diff --git a/docs/tasks/T-662d.md b/docs/tasks/T-662d.md
index 2550bbb..0b5fc64 100644
--- a/docs/tasks/T-662d.md
+++ b/docs/tasks/T-662d.md
@@ -1,7 +1,7 @@
 ---
 id: T-662d
 title: 更新蝦皮商品状态后台拦截与T-659预检整合
-status: TODO
+status: DONE
 phase: 7
 deps: [T-662a, T-659]
 created: 2026-07-18
@@ -55,4 +55,8 @@ git diff --check
 
 ## 执行记录
 
-- 待实现。
+- 新增 `product_status.build_apply_plan()`：先按商品状态去重分组，仅 `normal` 进入标题/封面完整性检查；未上架、审核中、未知与历史 NULL 均不传入执行列表。
+- `ApplyTab` 的「检查本轮更新」和「开始更新」共用该计划器；确认框、状态栏与本轮日志展示状态异常和内容缺失的分类排除，确认后重新核对 `task_id/updated_at/product_status/new_title/new_cover_path/更新模式` 指纹。
+- `ApplyWorker` 记录预检元数据，并在执行层再次拒绝非正常商品状态，避免直接构造 Worker 绕过 GUI 预检。
+- 同步更新 `docs/routes.md`、`docs/04-architecture.md`、`docs/api.md`；补充状态优先、混合排除、确认期变更和计划指纹测试。
+- 验证通过：`py -3.10 -m unittest tests.test_product_status tests.test_gui`（207 项）、`py -3.10 -m unittest discover -s tests`（616 项）、`py -3.10 -m ruff check app tests main.py`、`py -3.10 -m compileall app main.py`、`git diff --check`。
diff --git a/tests/test_gui.py b/tests/test_gui.py
index f01f7d1..e8b1457 100644
--- a/tests/test_gui.py
+++ b/tests/test_gui.py
@@ -6682,7 +6682,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
                 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"])
+            db.set_collected(
+                task.id,
+                "旧标题",
+                "old.jpg",
+                product_status_value="normal",
+                path=cfg["db_path"],
+            )
             db.set_generated(task.id, "新标题", "new.jpg", path=cfg["db_path"])
             statuses = []
             tab = ApplyTab(config=cfg, status_callback=statuses.append)
@@ -6730,7 +6736,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
                 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"])
+            db.set_collected(
+                task.id,
+                "旧标题",
+                "old.jpg",
+                product_status_value="normal",
+                path=cfg["db_path"],
+            )
             db.set_generated(task.id, "新标题", "new.jpg", path=cfg["db_path"])
             statuses = []
             tab = ApplyTab(config=cfg, status_callback=statuses.append)
@@ -6813,7 +6825,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
                 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"])
+            db.set_collected(
+                task.id,
+                "旧标题",
+                "old.jpg",
+                product_status_value="normal",
+                path=cfg["db_path"],
+            )
             db.set_generated(task.id, "新标题", "new.jpg", path=cfg["db_path"])
             statuses = []
             tab = ApplyTab(config=cfg, status_callback=statuses.append)
@@ -6877,7 +6895,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
                 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"])
+            db.set_collected(
+                task.id,
+                "旧标题",
+                "old.jpg",
+                product_status_value="normal",
+                path=cfg["db_path"],
+            )
             db.set_generated(task.id, "新标题", "new.jpg", path=cfg["db_path"])
             statuses = []
             tab = ApplyTab(config=cfg, status_callback=statuses.append)
@@ -6933,7 +6957,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
                 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"])
+            db.set_collected(
+                task.id,
+                "旧标题",
+                "old.jpg",
+                product_status_value="normal",
+                path=cfg["db_path"],
+            )
             db.set_generated(task.id, "新标题", None, path=cfg["db_path"])
             tab = ApplyTab(config=cfg)
             self.addCleanup(tab.close)
@@ -6983,6 +7013,149 @@ class GuiTests(TempDirMixin, unittest.TestCase):
 
         self.assert_removed(temp_dir)
 
+    def test_apply_tab_blocks_all_abnormal_statuses_before_content_or_worker(self):
+        with self.make_temp_dir() as temp_dir:
+            cfg = self.make_config(temp_dir)
+            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": index + 2,
+                        "alias": "alias-a",
+                        "item_id": str(51100639510 + index),
+                    }
+                    for index in range(3)
+                ],
+                path=cfg["db_path"],
+            )
+            statuses = ["unlisted", "reviewing", None]
+            for task, status in zip(
+                db.list_tasks(batch_id=batch_id, path=cfg["db_path"]),
+                statuses,
+            ):
+                db.set_collected(
+                    task.id,
+                    "旧标题",
+                    "old.jpg",
+                    product_status_value=status,
+                    path=cfg["db_path"],
+                )
+                db.set_generated(task.id, "新标题", "new.jpg", path=cfg["db_path"])
+            states = []
+            tab = ApplyTab(config=cfg, status_callback=states.append)
+            self.addCleanup(tab.close)
+
+            with mock.patch("app.gui.QMessageBox.warning") as warning, mock.patch(
+                "app.gui.QMessageBox.question"
+            ) as question, mock.patch("app.gui.run_worker") as run_worker:
+                tab.preview_update()
+
+            self.assertEqual("商品状态不允许更新", warning.call_args[0][1])
+            message = warning.call_args[0][2]
+            self.assertIn("未上架：1 条", message)
+            self.assertIn("审核中：1 条", message)
+            self.assertIn("状态未知：1 条", message)
+            self.assertNotIn("缺少新标题", message)
+            question.assert_not_called()
+            run_worker.assert_not_called()
+            self.assertIn("状态正常且可更新", states[-1])
+            unchanged = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
+            self.assertTrue(all(task.stage == "generated" for task in unchanged))
+            self.assertTrue(all(task.status == "success" for task in unchanged))
+
+        self.assert_removed(temp_dir)
+
+    def test_apply_tab_confirmation_combines_status_and_content_exclusions(self):
+        with self.make_temp_dir() as temp_dir:
+            cfg = self.make_config(temp_dir)
+            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": index + 2,
+                        "alias": "alias-a",
+                        "item_id": str(51100639510 + index),
+                    }
+                    for index in range(3)
+                ],
+                path=cfg["db_path"],
+            )
+            tasks = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
+            for task, status in zip(tasks, ["normal", "reviewing", "normal"]):
+                db.set_collected(
+                    task.id,
+                    "旧标题",
+                    "old.jpg",
+                    product_status_value=status,
+                    path=cfg["db_path"],
+                )
+            db.set_generated(tasks[0].id, "新标题", "new.jpg", path=cfg["db_path"])
+            db.set_generated(tasks[1].id, "新标题", "new.jpg", path=cfg["db_path"])
+            db.set_generated(tasks[2].id, None, "new.jpg", path=cfg["db_path"])
+            tab = ApplyTab(config=cfg)
+            self.addCleanup(tab.close)
+            thread = FakeThread()
+
+            with mock.patch("app.gui.QMessageBox.question", return_value=gui.QMessageBox.Yes) as question, mock.patch(
+                "app.gui.run_worker",
+                return_value=thread,
+            ):
+                tab.preview_update()
+
+            message = question.call_args[0][2]
+            self.assertIn("商品状态异常记录", message)
+            self.assertIn("审核中：1 条", message)
+            self.assertIn("缺少新标题：1 条", message)
+            self.assertEqual([tasks[0].id], [task.id for task in tab.apply_worker.tasks])
+            self.assertTrue(tab.apply_worker.dry_run)
+            self.assertTrue(thread.started)
+
+        self.assert_removed(temp_dir)
+
+    def test_apply_tab_discards_plan_when_candidates_change_during_confirmation(self):
+        with self.make_temp_dir() as temp_dir:
+            states = []
+            tab = ApplyTab(
+                config=self.make_config(temp_dir),
+                status_callback=states.append,
+            )
+            self.addCleanup(tab.close)
+            before = SimpleNamespace(
+                id=1,
+                item_id="51100639510",
+                updated_at="2026-07-18T10:00:00",
+                product_status="normal",
+                new_title="新标题",
+                new_cover_path="new.jpg",
+            )
+            after = SimpleNamespace(
+                id=1,
+                item_id="51100639510",
+                updated_at="2026-07-18T10:01:00",
+                product_status="normal",
+                new_title="新标题",
+                new_cover_path="new.jpg",
+            )
+
+            with mock.patch.object(tab, "_update_candidates", side_effect=[[before], [after]]), mock.patch(
+                "app.gui.QMessageBox.question",
+                return_value=gui.QMessageBox.Yes,
+            ), mock.patch("app.gui.run_worker") as run_worker:
+                tab.preview_update()
+
+            run_worker.assert_not_called()
+            self.assertIn("当前任务数据已变化", states[-1])
+
+        self.assert_removed(temp_dir)
+
     def test_apply_tab_skips_missing_content_but_starts_eligible_update(self):
         with self.make_temp_dir() as temp_dir:
             cfg = self.make_config(temp_dir)
@@ -7020,7 +7193,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
             )
             tasks = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
             for task in tasks:
-                db.set_collected(task.id, "旧标题", "old.jpg", path=cfg["db_path"])
+                db.set_collected(
+                    task.id,
+                    "旧标题",
+                    "old.jpg",
+                    product_status_value="normal",
+                    path=cfg["db_path"],
+                )
             db.set_generated(tasks[0].id, "新标题", "new.jpg", path=cfg["db_path"])
             db.set_generated(tasks[1].id, "新标题", None, path=cfg["db_path"])
             db.set_generated(tasks[2].id, None, "new.jpg", path=cfg["db_path"])
@@ -7080,7 +7259,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
             )
             tasks = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
             for task in tasks:
-                db.set_collected(task.id, "旧标题", "old.jpg", path=cfg["db_path"])
+                db.set_collected(
+                    task.id,
+                    "旧标题",
+                    "old.jpg",
+                    product_status_value="normal",
+                    path=cfg["db_path"],
+                )
             db.set_generated(tasks[0].id, "新标题", "new.jpg", path=cfg["db_path"])
             db.set_generated(tasks[1].id, None, "new.jpg", path=cfg["db_path"])
             statuses = []
@@ -7176,7 +7361,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
                 path=cfg["db_path"],
             )
             for task in db.list_tasks(batch_id=batch_id, path=cfg["db_path"]):
-                db.set_collected(task.id, "旧标题", "old.jpg", path=cfg["db_path"])
+                db.set_collected(
+                    task.id,
+                    "旧标题",
+                    "old.jpg",
+                    product_status_value="normal",
+                    path=cfg["db_path"],
+                )
                 db.set_generated(task.id, "新标题", "new.jpg", path=cfg["db_path"])
             tasks = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
             applied_aliases = []
@@ -7280,7 +7471,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
                 path=cfg["db_path"],
             )
             for task in db.list_tasks(batch_id=batch_id, path=cfg["db_path"]):
-                db.set_collected(task.id, "旧标题", "old.jpg", path=cfg["db_path"])
+                db.set_collected(
+                    task.id,
+                    "旧标题",
+                    "old.jpg",
+                    product_status_value="normal",
+                    path=cfg["db_path"],
+                )
                 db.set_generated(task.id, "新标题", "new.jpg", path=cfg["db_path"])
             tasks = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
             foreground_flags = []
@@ -7335,7 +7532,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
                 path=cfg["db_path"],
             )
             for task in db.list_tasks(batch_id=batch_id, path=cfg["db_path"]):
-                db.set_collected(task.id, "旧标题", "old.jpg", path=cfg["db_path"])
+                db.set_collected(
+                    task.id,
+                    "旧标题",
+                    "old.jpg",
+                    product_status_value="normal",
+                    path=cfg["db_path"],
+                )
                 db.set_generated(task.id, "新标题", "new.jpg", path=cfg["db_path"])
             tasks = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
             logs = []
@@ -7402,7 +7605,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
                 path=cfg["db_path"],
             )
             for task in db.list_tasks(batch_id=batch_id, path=cfg["db_path"]):
-                db.set_collected(task.id, "旧标题", "old.jpg", path=cfg["db_path"])
+                db.set_collected(
+                    task.id,
+                    "旧标题",
+                    "old.jpg",
+                    product_status_value="normal",
+                    path=cfg["db_path"],
+                )
                 db.set_generated(task.id, "新标题", "new.jpg", path=cfg["db_path"])
             tasks = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
             started = {"alias-a": threading.Event(), "alias-b": threading.Event()}
@@ -7469,7 +7678,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
                 path=cfg["db_path"],
             )
             for task in db.list_tasks(batch_id=batch_id, path=cfg["db_path"]):
-                db.set_collected(task.id, "旧标题", "old.jpg", path=cfg["db_path"])
+                db.set_collected(
+                    task.id,
+                    "旧标题",
+                    "old.jpg",
+                    product_status_value="normal",
+                    path=cfg["db_path"],
+                )
                 db.set_generated(task.id, "新标题", "new.jpg", path=cfg["db_path"])
             tasks = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
 
@@ -7604,7 +7819,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
                 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"])
+            db.set_collected(
+                task.id,
+                "旧标题",
+                "old.jpg",
+                product_status_value="normal",
+                path=cfg["db_path"],
+            )
             db.set_generated(task.id, "新标题", "new.jpg", path=cfg["db_path"])
             tasks = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
 
@@ -9595,7 +9816,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
                 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"])
+            db.set_collected(
+                task.id,
+                "旧标题",
+                "old.jpg",
+                product_status_value="normal",
+                path=cfg["db_path"],
+            )
             db.set_generated(task.id, "新标题", "new.jpg", path=cfg["db_path"])
             task = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])[0]
             diagnostic_log_dir = os.path.join(temp_dir, "logs")
diff --git a/tests/test_product_status.py b/tests/test_product_status.py
index b4c2bd0..009321d 100644
--- a/tests/test_product_status.py
+++ b/tests/test_product_status.py
@@ -132,3 +132,68 @@ class ProductStatusTests(unittest.TestCase):
         )
         self.assertEqual([1, 2, 3], [task.id if hasattr(task, "id") else task["id"] for task in all_statuses["execution_tasks"]])
         self.assertNotEqual(normal_only["fingerprint"], changed["fingerprint"])
+
+    def test_apply_plan_excludes_abnormal_before_content_and_deduplicates_skips(self):
+        tasks = [
+            SimpleNamespace(
+                id=1,
+                updated_at="2026-07-18T10:00:00",
+                product_status="normal",
+                new_title="新标题",
+                new_cover_path="new.jpg",
+            ),
+            SimpleNamespace(
+                id=2,
+                updated_at="2026-07-18T10:00:01",
+                product_status="unlisted",
+                new_title=None,
+                new_cover_path=None,
+            ),
+            SimpleNamespace(
+                id=3,
+                updated_at="2026-07-18T10:00:02",
+                product_status="reviewing",
+                new_title=None,
+                new_cover_path="new.jpg",
+            ),
+            SimpleNamespace(
+                id=4,
+                updated_at="2026-07-18T10:00:03",
+                product_status=None,
+                new_title="新标题",
+                new_cover_path=None,
+            ),
+            SimpleNamespace(
+                id=5,
+                updated_at="2026-07-18T10:00:04",
+                product_status="normal",
+                new_title=None,
+                new_cover_path=None,
+            ),
+        ]
+
+        plan = product_status.build_apply_plan(tasks + [tasks[0]], "title_cover")
+        changed = product_status.build_apply_plan(
+            tasks[:4]
+            + [
+                SimpleNamespace(
+                    id=5,
+                    updated_at="2026-07-18T10:01:00",
+                    product_status="normal",
+                    new_title=None,
+                    new_cover_path=None,
+                )
+            ],
+            "title_cover",
+        )
+
+        self.assertEqual([1], [task.id for task in plan["executable"]])
+        self.assertEqual([2], [task.id for task in plan["unlisted"]])
+        self.assertEqual([3], [task.id for task in plan["reviewing"]])
+        self.assertEqual([4], [task.id for task in plan["unknown"]])
+        self.assertEqual([5], [task.id for task in plan["missing_title"]])
+        self.assertEqual([5], [task.id for task in plan["missing_cover"]])
+        self.assertEqual(3, plan["status_scope_excluded"])
+        self.assertEqual(1, plan["content_scope_excluded"])
+        self.assertEqual(4, plan["scope_excluded"])
+        self.assertNotEqual(plan["fingerprint"], changed["fingerprint"])