fix(db): migrate processed tasks with missing product status

app/db.py

@@ -19,6 +19,7 @@ DEFAULT_BUSY_TIMEOUT_MS = 5000
 PRODUCT_STATUS_FEATURE_INTRODUCED_AT = "2026-07-18T16:34:37"
 LEGACY_PRODUCT_STATUS_DEFAULT_NOTE = "历史批次默认按架上商品处理（未实时检测）"
 LEGACY_PROMOTION_STATUS_REPAIR_NOTE = "历史商品状态修正：促销编辑提示不属于商品状态异常"
+LEGACY_STATUS_BACKUP_TASK_ID = "T704"
 VALID_BATCH_FIELDS = {
     "source_files_json",
     "status",
@@ -514,6 +515,10 @@ def init_db(path=None, conn=None) -> None:
             _ensure_image_studio_job_reference_asset_ids_column(database)
             _ensure_image_studio_asset_output_columns(database)
             _ensure_image_studio_job_run_session_column(database)
+        if _has_processed_tasks_missing_product_status(database):
+            _backup_before_processed_status_migration(database)
+            with database:
+                _migrate_processed_task_product_status_defaults(database)
 
 
 def _ensure_batch_delete_columns(database):
@@ -586,6 +591,101 @@ def _migrate_legacy_product_status_defaults(database):
     )
 
 
+def _processed_task_status_condition() -> str:
+    return """
+        (tasks.stage IN ('collected', 'generated', 'applied')
+         OR tasks.collected_at IS NOT NULL
+         OR tasks.generated_at IS NOT NULL
+         OR tasks.applied_at IS NOT NULL
+         OR NULLIF(TRIM(COALESCE(tasks.old_title, '')), '') IS NOT NULL
+         OR NULLIF(TRIM(COALESCE(tasks.old_cover_path, '')), '') IS NOT NULL
+         OR NULLIF(TRIM(COALESCE(tasks.new_title, '')), '') IS NOT NULL
+         OR NULLIF(TRIM(COALESCE(tasks.new_cover_path, '')), '') IS NOT NULL)
+    """
+
+
+def _has_processed_tasks_missing_product_status(database) -> bool:
+    row = database.execute(
+        f"""
+        SELECT 1
+        FROM tasks
+        INNER JOIN batches ON batches.id = tasks.batch_id
+        WHERE (tasks.product_status IS NULL OR TRIM(tasks.product_status) = '')
+          AND tasks.deleted_at IS NULL
+          AND batches.deleted_at IS NULL
+          AND {_processed_task_status_condition()}
+        LIMIT 1
+        """
+    ).fetchone()
+    return row is not None
+
+
+def _main_database_path(database) -> str:
+    for row in database.execute("PRAGMA database_list").fetchall():
+        if row["name"] == "main":
+            return str(row["file"] or "").strip()
+    return ""
+
+
+def _backup_before_processed_status_migration(database) -> Optional[str]:
+    """Create a consistent file backup before T-704 changes user task rows."""
+
+    database_path = _main_database_path(database)
+    if not database_path or database_path == ":memory:":
+        return None
+
+    backup_dir = os.path.join(os.path.dirname(os.path.abspath(database_path)), "backups")
+    stem, ext = os.path.splitext(os.path.basename(database_path))
+    ext = ext or ".db"
+    timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
+    backup_path = os.path.join(
+        backup_dir,
+        f"{stem}-before-{LEGACY_STATUS_BACKUP_TASK_ID}-{timestamp}-{uuid.uuid4().hex[:8]}{ext}",
+    )
+    try:
+        os.makedirs(backup_dir, exist_ok=True)
+        backup_database = sqlite3.connect(backup_path)
+        try:
+            database.backup(backup_database)
+        finally:
+            backup_database.close()
+    except (OSError, sqlite3.Error) as exc:
+        try:
+            if os.path.exists(backup_path):
+                os.remove(backup_path)
+        except OSError:
+            pass
+        raise DbError(f"兼容历史商品状态前备份数据库失败: {exc}") from exc
+    return backup_path
+
+
+def _migrate_processed_task_product_status_defaults(database):
+    """Default only already-processed missing snapshots, regardless of batch date."""
+
+    database.execute(
+        f"""
+        UPDATE tasks
+        SET product_status = 'normal',
+            product_status_note = ?,
+            product_status_at = NULL,
+            updated_at = ?
+        WHERE id IN (
+            SELECT tasks.id
+            FROM tasks
+            INNER JOIN batches ON batches.id = tasks.batch_id
+            WHERE (tasks.product_status IS NULL OR TRIM(tasks.product_status) = '')
+              AND tasks.deleted_at IS NULL
+              AND batches.deleted_at IS NULL
+              AND {_processed_task_status_condition()}
+        )
+        """,
+        (
+            LEGACY_PRODUCT_STATUS_DEFAULT_NOTE,
+            _now(),
+        ),
+    )
+
+
 def _repair_legacy_promotion_status_defaults(database):
     """Repair only legacy unknown snapshots caused by a known benign warning."""