T-566 封面历史归档数据层

tests/test_db.py

@@ -1,6 +1,7 @@
 import os
 import sys
 import unittest
+from unittest import mock
 
 sys.path.insert(0, os.path.dirname(__file__))
 
@@ -302,8 +303,10 @@ class DbTests(TempDirMixin, unittest.TestCase):
             reset_generated = db.reset_generated(task.id, path=db_path)
             after_generated = reset_generated["after"]
             self.assertEqual(new_cover, reset_generated["new_cover_path"])
+            self.assertIsNotNone(reset_generated["archived_file"])
             self.assertIsNone(reset_generated["deleted_file"])
-            self.assertTrue(os.path.exists(new_cover))
+            self.assertFalse(os.path.exists(new_cover))
+            self.assertTrue(os.path.exists(reset_generated["archived_file"]))
             self.assertEqual("generated", after_generated.stage)
             self.assertEqual("success", after_generated.status)
             self.assertIsNone(after_generated.new_title)
@@ -378,8 +381,10 @@ class DbTests(TempDirMixin, unittest.TestCase):
             self.assertEqual("success", after_cover.status)
             self.assertEqual("手动标题B", after_cover.new_title)
             self.assertIsNone(after_cover.new_cover_path)
-            self.assertEqual(second_cover, cover_reset["deleted_file"])
+            self.assertIsNone(cover_reset["deleted_file"])
+            self.assertIsNotNone(cover_reset["archived_file"])
             self.assertFalse(os.path.exists(second_cover))
+            self.assertTrue(os.path.exists(cover_reset["archived_file"]))
 
             with self.assertRaises(db.DbError):
                 db.reset_generated(first.id, reset_title=False, reset_cover=False, path=db_path)
@@ -389,6 +394,175 @@ class DbTests(TempDirMixin, unittest.TestCase):
         self.assert_removed(temp_dir)
 
 
+    def test_reset_generated_archives_cover_with_collision_name(self):
+        with self.make_temp_dir() as temp_dir:
+            db_path = os.path.join(temp_dir, "cmshopee.db")
+            db.init_db(db_path)
+            batch_id = db.create_batch(["input.xlsx"], path=db_path)
+            db.insert_tasks(
+                batch_id,
+                [
+                    {
+                        "source_file_abs": os.path.join(temp_dir, "input.xlsx"),
+                        "source_sheet": "Sheet1",
+                        "source_row": 2,
+                        "account_name": "shop",
+                        "alias": "alias",
+                        "item_id": "51100639510",
+                    }
+                ],
+                path=db_path,
+            )
+            task = db.list_tasks(batch_id=batch_id, path=db_path)[0]
+            cover = os.path.join(temp_dir, "42_51100639510_new.jpg")
+            collision = os.path.join(temp_dir, "42_51100639510_new_20260709094800.jpg")
+            expected_archive = os.path.join(temp_dir, "42_51100639510_new_20260709094800_2.jpg")
+            with open(cover, "wb") as fh:
+                fh.write(b"current")
+            with open(collision, "wb") as fh:
+                fh.write(b"existing")
+            db.set_collected(task.id, "旧标题", "old.jpg", path=db_path)
+            db.set_generated(task.id, "新标题", cover, path=db_path)
+
+            with mock.patch("app.db._cover_archive_timestamp", return_value="20260709094800"):
+                result = db.reset_generated(
+                    task.id,
+                    reset_title=False,
+                    reset_cover=True,
+                    path=db_path,
+                )
+
+            self.assertEqual(expected_archive, result["archived_file"])
+            self.assertFalse(os.path.exists(cover))
+            self.assertTrue(os.path.exists(collision))
+            self.assertTrue(os.path.exists(expected_archive))
+            with open(collision, "rb") as fh:
+                self.assertEqual(b"existing", fh.read())
+            with open(expected_archive, "rb") as fh:
+                self.assertEqual(b"current", fh.read())
+            after = db.get_task(task.id, path=db_path)
+            self.assertEqual("新标题", after.new_title)
+            self.assertIsNone(after.new_cover_path)
+
+        self.assert_removed(temp_dir)
+
+    def test_reset_generated_permission_error_keeps_db_pointer(self):
+        with self.make_temp_dir() as temp_dir:
+            db_path = os.path.join(temp_dir, "cmshopee.db")
+            db.init_db(db_path)
+            batch_id = db.create_batch(["input.xlsx"], path=db_path)
+            db.insert_tasks(
+                batch_id,
+                [
+                    {
+                        "source_file_abs": os.path.join(temp_dir, "input.xlsx"),
+                        "source_sheet": "Sheet1",
+                        "source_row": 2,
+                        "account_name": "shop",
+                        "alias": "alias",
+                        "item_id": "51100639510",
+                    }
+                ],
+                path=db_path,
+            )
+            task = db.list_tasks(batch_id=batch_id, path=db_path)[0]
+            cover = os.path.join(temp_dir, "new.jpg")
+            with open(cover, "wb") as fh:
+                fh.write(b"jpeg")
+            db.set_generated(task.id, "新标题", cover, path=db_path)
+
+            with mock.patch("app.db.os.rename", side_effect=PermissionError):
+                with self.assertRaisesRegex(db.DbError, "请先关闭正在查看的封面图片再重置"):
+                    db.reset_generated(task.id, reset_title=False, reset_cover=True, path=db_path)
+
+            after = db.get_task(task.id, path=db_path)
+            self.assertEqual(cover, after.new_cover_path)
+            self.assertEqual("新标题", after.new_title)
+            self.assertTrue(os.path.exists(cover))
+
+        self.assert_removed(temp_dir)
+
+    def test_update_generated_cover_switches_pointer_and_keeps_apply_history(self):
+        with self.make_temp_dir() as temp_dir:
+            db_path = os.path.join(temp_dir, "cmshopee.db")
+            db.init_db(db_path)
+            batch_id = db.create_batch(["input.xlsx"], path=db_path)
+            db.insert_tasks(
+                batch_id,
+                [
+                    {
+                        "source_file_abs": os.path.join(temp_dir, "input.xlsx"),
+                        "source_sheet": "Sheet1",
+                        "source_row": 2,
+                        "account_name": "shop",
+                        "alias": "alias",
+                        "item_id": "51100639510",
+                    }
+                ],
+                path=db_path,
+            )
+            task = db.list_tasks(batch_id=batch_id, path=db_path)[0]
+            current_cover = os.path.join(temp_dir, "current.jpg")
+            selected_cover = os.path.join(temp_dir, "selected.jpg")
+            for cover in (current_cover, selected_cover):
+                with open(cover, "wb") as fh:
+                    fh.write(b"jpeg")
+            db.set_collected(task.id, "旧标题", "old.jpg", path=db_path)
+            db.set_generated(task.id, "新标题", current_cover, path=db_path)
+            key = db.ensure_image_task_key(task.id, path=db_path)
+            db.set_image_task_submitted(task.id, "cmhub-task-1", key, path=db_path)
+            db.set_applied(task.id, True, path=db_path)
+            before = db.get_task(task.id, path=db_path)
+
+            db.update_generated_cover(task.id, selected_cover, path=db_path)
+
+            after = db.get_task(task.id, path=db_path)
+            self.assertEqual(os.path.abspath(selected_cover), after.new_cover_path)
+            self.assertEqual("generated", after.stage)
+            self.assertEqual("pending", after.status)
+            self.assertIsNone(after.last_error)
+            self.assertEqual(before.generate_attempts, after.generate_attempts)
+            self.assertEqual(before.apply_attempts, after.apply_attempts)
+            self.assertEqual(before.committed, after.committed)
+            self.assertEqual(before.applied_at, after.applied_at)
+            self.assertEqual(before.image_task_id, after.image_task_id)
+            self.assertEqual(before.image_task_key, after.image_task_key)
+
+            with self.assertRaisesRegex(db.DbError, "新封面图片不存在"):
+                db.update_generated_cover(task.id, os.path.join(temp_dir, "missing.jpg"), path=db_path)
+
+        self.assert_removed(temp_dir)
+
+    def test_update_generated_cover_rejects_running_task(self):
+        with self.make_temp_dir() as temp_dir:
+            db_path = os.path.join(temp_dir, "cmshopee.db")
+            db.init_db(db_path)
+            batch_id = db.create_batch(["input.xlsx"], path=db_path)
+            db.insert_tasks(
+                batch_id,
+                [
+                    {
+                        "source_file_abs": os.path.join(temp_dir, "input.xlsx"),
+                        "source_sheet": "Sheet1",
+                        "source_row": 2,
+                        "account_name": "shop",
+                        "alias": "alias",
+                        "item_id": "51100639510",
+                    }
+                ],
+                path=db_path,
+            )
+            task = db.list_tasks(batch_id=batch_id, path=db_path)[0]
+            cover = os.path.join(temp_dir, "selected.jpg")
+            with open(cover, "wb") as fh:
+                fh.write(b"jpeg")
+            db.mark_running(task.id, "generate", path=db_path)
+
+            with self.assertRaisesRegex(db.DbError, "任务正在运行"):
+                db.update_generated_cover(task.id, cover, path=db_path)
+
+        self.assert_removed(temp_dir)
+
     def test_delete_batch_soft_hides_batch_and_tasks(self):
         with self.make_temp_dir() as temp_dir:
             db_path = os.path.join(temp_dir, "cmshopee.db")
@@ -419,6 +593,11 @@ class DbTests(TempDirMixin, unittest.TestCase):
             tasks = db.list_tasks(batch_id=batch_id, path=db_path)
             old_cover = os.path.join(temp_dir, "old.jpg")
             new_cover = os.path.join(temp_dir, "new.jpg")
+            batch_image_dir = os.path.join(temp_dir, "images", batch_id, "alias")
+            os.makedirs(batch_image_dir, exist_ok=True)
+            generated_cover = os.path.join(batch_image_dir, "1_51100639510_new.jpg")
+            with open(generated_cover, "wb") as fh:
+                fh.write(b"jpeg")
             db.set_collected(tasks[0].id, "旧标题", old_cover, path=db_path)
             db.set_generated(tasks[0].id, "新标题", new_cover, path=db_path)
             db.set_applied(tasks[0].id, True, path=db_path)
@@ -440,6 +619,8 @@ class DbTests(TempDirMixin, unittest.TestCase):
             self.assertEqual("导错文件", deleted_batch.deleted_reason)
             deleted_tasks = db.list_tasks(batch_id=batch_id, path=db_path, include_deleted=True)
             self.assertEqual(2, len(deleted_tasks))
+            self.assertTrue(os.path.exists(batch_image_dir))
+            self.assertTrue(os.path.exists(generated_cover))
 
         self.assert_removed(temp_dir)