feat(generate): confirm product status scope

app/product_status.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import hashlib
+import json
 import re
 
 
@@ -101,7 +103,37 @@ def partition_tasks(tasks) -> dict:
     return grouped
 
 
+def build_generation_plan(tasks, generate_mode, scope=SCOPE_NORMAL_ONLY) -> dict:
+    """Build a frozen, status-aware generation plan without mutating tasks."""
+
+    base_candidates = _deduplicate_tasks(tasks)
+    status_counts = {status: 0 for status in VALID_PRODUCT_STATUSES}
+    for task in base_candidates:
+        status_counts[_task_status(task)] += 1
+
+    normalized_scope = normalize_scope(scope)
+    if normalized_scope == SCOPE_ALL:
+        execution_tasks = list(base_candidates)
+    else:
+        execution_tasks = [
+            task for task in base_candidates if is_normal(_task_status(task))
+        ]
+
+    return {
+        "base_candidates": base_candidates,
+        "status_counts": status_counts,
+        "execution_tasks": execution_tasks,
+        "scope": normalized_scope,
+        "scope_excluded": len(base_candidates) - len(execution_tasks),
+        "fingerprint": _generation_fingerprint(base_candidates, generate_mode),
+    }
+
+
 def normalize_collect_scope(value) -> str:
+    return normalize_scope(value)
+
+
+def normalize_scope(value) -> str:
     value = str(value or "").strip().lower()
     return SCOPE_ALL if value == SCOPE_ALL else SCOPE_NORMAL_ONLY
 
@@ -114,6 +146,51 @@ def collect_skip_reason(status) -> str:
     return f"{status_label(status)}，按本轮范围略过"
 
 
+def _deduplicate_tasks(tasks) -> list:
+    seen = set()
+    unique = []
+    for task in list(tasks or []):
+        task_id = _task_value(task, "id")
+        key = ("id", str(task_id)) if task_id is not None else ("object", id(task))
+        if key in seen:
+            continue
+        seen.add(key)
+        unique.append(task)
+    return unique
+
+
+def _generation_fingerprint(tasks, generate_mode) -> str:
+    snapshots = [
+        {
+            "task_id": _task_value(task, "id"),
+            "updated_at": _task_value(task, "updated_at"),
+            "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 ""),
+        }
+        for task in tasks
+    ]
+    encoded = json.dumps(
+        snapshots,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+        default=str,
+    ).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+def _task_status(task) -> str:
+    return normalize_status(_task_value(task, "product_status"))
+
+
+def _task_value(task, name, default=None):
+    if isinstance(task, dict):
+        return task.get(name, default)
+    return getattr(task, name, default)
+
+
 def _alert_note(title: str, description: str) -> str:
     parts = []
     if title: