fix: clarify AI generation failure retry states

2026-07-06 11:58:51 +08:00
parent 19eaa794cf
commit 7ac0dcd00e
9 changed files with 195 additions and 9 deletions
@@ -757,6 +757,66 @@ class AITests(TempDirMixin, unittest.TestCase):

        self.assert_removed(temp_dir)

+    def test_generate_batch_retries_failed_generation_record_after_existing_result(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"]["generate_cover"] = False
+            batch_id, tasks = self._collected_tasks(temp_dir, cfg, ["旧标题"])
+            db.set_generated(tasks[0].id, "旧AI标题", "old-new.jpg", path=cfg["db_path"])
+            db.mark_failed(tasks[0].id, "generate", "上次生成失败", path=cfg["db_path"])
+            retry_task = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])[0]
+
+            with mock.patch("app.ai.gen_title", return_value="重新生成标题") as gen_title, \
+                    mock.patch("app.ai.gen_cover") as gen_cover:
+                summary = ai.generate_batch(
+                    [retry_task],
+                    {"title": "标题提示", "cover": "封面"},
+                    ai_cfg={"config": cfg, "db_path": cfg["db_path"]},
+                )
+
+            self.assertTrue(summary["ok"])
+            self.assertEqual(1, summary["total"])
+            self.assertEqual(1, summary["generated_done"])
+            gen_title.assert_called_once()
+            gen_cover.assert_not_called()
+            updated = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])[0]
+            self.assertEqual("generated", updated.stage)
+            self.assertEqual("success", updated.status)
+            self.assertEqual("重新生成标题", updated.new_title)
+            self.assertIsNone(updated.new_cover_path)
+
+        self.assert_removed(temp_dir)
+
+    def test_generate_batch_does_not_retry_apply_failed_records(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, ["旧标题"])
+            db.set_generated(tasks[0].id, "新标题", "new.jpg", path=cfg["db_path"])
+            db.mark_failed(tasks[0].id, "apply", "更新失败", path=cfg["db_path"])
+            update_failed_task = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])[0]
+
+            with mock.patch("app.ai.gen_title") as gen_title, \
+                    mock.patch("app.ai.gen_cover") as gen_cover:
+                summary = ai.generate_batch(
+                    [update_failed_task],
+                    {"title": "标题提示", "cover": "封面"},
+                    ai_cfg={"config": cfg, "db_path": cfg["db_path"]},
+                )
+
+            self.assertTrue(summary["ok"])
+            self.assertEqual(0, summary["total"])
+            gen_title.assert_not_called()
+            gen_cover.assert_not_called()
+            unchanged = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])[0]
+            self.assertEqual("generated", unchanged.stage)
+            self.assertEqual("failed", unchanged.status)
+            self.assertEqual(1, unchanged.apply_attempts)
+
+        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()