cmshoppe/tests/test_db.py

import os
import sys
import unittest
from unittest import mock

sys.path.insert(0, os.path.dirname(__file__))

from _helpers import TempDirMixin

from app import db


class DbTests(TempDirMixin, unittest.TestCase):
    def test_init_db_is_idempotent_and_sets_pragmas(self):
        with self.make_temp_dir() as temp_dir:
            db_path = os.path.join(temp_dir, "cmshopee.db")

            db.init_db(db_path)
            db.init_db(db_path)

            conn = db.connect(db_path)
            try:
                self.assertEqual(1, conn.execute("PRAGMA foreign_keys").fetchone()[0])
                self.assertEqual("wal", conn.execute("PRAGMA journal_mode").fetchone()[0])
                self.assertEqual(5000, conn.execute("PRAGMA busy_timeout").fetchone()[0])
                tables = {
                    row["name"]
                    for row in conn.execute(
                        "SELECT name FROM sqlite_master WHERE type = 'table'"
                    ).fetchall()
                }
                self.assertTrue({"batches", "accounts", "tasks"}.issubset(tables))
                task_columns = {
                    row["name"]
                    for row in conn.execute("PRAGMA table_info(tasks)").fetchall()
                }
                self.assertTrue(
                    {
                        "image_task_id",
                        "image_task_key",
                        "cover_reset_count",
                        "cover_reset_at",
                    }.issubset(task_columns)
                )
            finally:
                conn.close()

        self.assert_removed(temp_dir)

    def test_init_db_migrates_cover_reset_columns_for_legacy_tasks(self):
        with self.make_temp_dir() as temp_dir:
            db_path = os.path.join(temp_dir, "legacy.sqlite")
            conn = db.connect(db_path)
            try:
                with conn:
                    conn.executescript(
                        """
                        CREATE TABLE batches (
                          id TEXT PRIMARY KEY,
                          source_files_json TEXT NOT NULL,
                          status TEXT NOT NULL DEFAULT 'active',
                          note TEXT,
                          created_at TEXT NOT NULL,
                          updated_at TEXT NOT NULL
                        );
                        CREATE TABLE tasks (
                          id INTEGER PRIMARY KEY,
                          batch_id TEXT NOT NULL REFERENCES batches(id),
                          source_file TEXT NOT NULL,
                          source_file_abs TEXT NOT NULL,
                          source_sheet TEXT NOT NULL,
                          source_row INTEGER NOT NULL,
                          row_key TEXT NOT NULL UNIQUE,
                          account_name TEXT,
                          alias TEXT NOT NULL,
                          item_id TEXT NOT NULL,
                          old_title TEXT,
                          old_cover_path TEXT,
                          new_title TEXT,
                          new_cover_path TEXT,
                          image_task_id TEXT,
                          image_task_key TEXT,
                          committed INTEGER NOT NULL DEFAULT 0,
                          stage TEXT NOT NULL DEFAULT 'imported',
                          status TEXT NOT NULL DEFAULT 'pending',
                          last_error TEXT,
                          collect_attempts INTEGER NOT NULL DEFAULT 0,
                          generate_attempts INTEGER NOT NULL DEFAULT 0,
                          apply_attempts INTEGER NOT NULL DEFAULT 0,
                          imported_at TEXT NOT NULL,
                          collected_at TEXT,
                          generated_at TEXT,
                          applied_at TEXT,
                          updated_at TEXT NOT NULL,
                          UNIQUE(batch_id, source_file_abs, source_sheet, source_row)
                        );
                        INSERT INTO batches
                          (id, source_files_json, status, note, created_at, updated_at)
                        VALUES ('batch', '[]', 'active', NULL, '2026-07-11T00:00:00', '2026-07-11T00:00:00');
                        INSERT INTO tasks
                          (batch_id, source_file, source_file_abs, source_sheet, source_row,
                           row_key, account_name, alias, item_id, imported_at, updated_at)
                        VALUES
                          ('batch', 'input.xlsx', 'input.xlsx', 'Sheet1', 2,
                           'row-key', 'shop', 'alias', '51100639510',
                           '2026-07-11T00:00:00', '2026-07-11T00:00:00');
                        """
                    )
                db.init_db(conn=conn)
                columns = {
                    row["name"]
                    for row in conn.execute("PRAGMA table_info(tasks)").fetchall()
                }
                self.assertIn("cover_reset_count", columns)
                self.assertIn("cover_reset_at", columns)
                task = db.get_task(1, conn=conn)
                self.assertEqual(0, task.cover_reset_count)
                self.assertIsNone(task.cover_reset_at)
            finally:
                conn.close()

        self.assert_removed(temp_dir)

    def test_account_batch_task_lifecycle(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"], note="导入", path=db_path)
            batch = db.get_batch(batch_id, path=db_path)
            self.assertEqual("导入", batch.note)
            self.assertEqual([os.path.abspath("input.xlsx")], batch.source_files)

            account = db.add_account(
                "shop",
                "alias",
                "shopee.tw",
                9222,
                note="备注",
                path=db_path,
            )
            self.assertEqual("alias", account.alias)
            self.assertEqual("alias_cdb6fdbe", account.slug)

            db.update_account("alias", path=db_path, debug_port=9333)
            self.assertEqual(9333, db.get_account_by_alias("alias", path=db_path).debug_port)
            self.assertEqual(1, len(db.list_accounts(path=db_path)))

            count = db.insert_tasks(
                batch_id,
                [
                    {
                        "source_file": "input.xlsx",
                        "source_file_abs": os.path.abspath("input.xlsx"),
                        "source_sheet": "Sheet1",
                        "source_row": 2,
                        "account_name": "shop",
                        "alias": "alias",
                        "item_id": "51100639510",
                    }
                ],
                path=db_path,
            )
            self.assertEqual(1, count)

            task = db.list_tasks(batch_id=batch_id, alias="alias", path=db_path)[0]
            self.assertEqual("imported", task.stage)
            self.assertEqual("pending", task.status)
            self.assertIsNone(task.image_task_id)
            self.assertIsNone(task.image_task_key)
            self.assertEqual(0, task.cover_reset_count)
            self.assertIsNone(task.cover_reset_at)

            db.mark_running(task.id, "collect", path=db_path)
            self.assertEqual("running", db.list_tasks(path=db_path)[0].status)

            db.mark_failed(task.id, "collect", "采集失败", path=db_path)
            failed = db.list_tasks(path=db_path)[0]
            self.assertEqual("imported", failed.stage)
            self.assertEqual("failed", failed.status)
            self.assertEqual(1, failed.collect_attempts)
            self.assertEqual("采集失败", failed.last_error)

            db.set_collected(task.id, "旧标题", "old.jpg", path=db_path)
            collected = db.list_tasks(path=db_path)[0]
            self.assertEqual("collected", collected.stage)
            self.assertEqual("success", collected.status)
            self.assertEqual("旧标题", collected.old_title)

            db.set_generated(task.id, "新标题", "new.jpg", path=db_path)
            generated = db.list_tasks(path=db_path)[0]
            self.assertEqual("generated", generated.stage)
            self.assertEqual("新标题", generated.new_title)

            db.mark_failed(task.id, "generate", "标题需调整", path=db_path)
            db.update_generated_title(task.id, "人工标题", path=db_path)
            edited = db.list_tasks(path=db_path)[0]
            self.assertEqual("generated", edited.stage)
            self.assertEqual("pending", edited.status)
            self.assertEqual("人工标题", edited.new_title)
            self.assertIsNone(edited.last_error)

            db.set_applied(task.id, False, "按钮禁用", path=db_path)
            apply_failed = db.list_tasks(path=db_path)[0]
            self.assertEqual("generated", apply_failed.stage)
            self.assertEqual("failed", apply_failed.status)
            self.assertEqual(1, apply_failed.apply_attempts)

            db.set_applied(task.id, True, path=db_path)
            applied = db.list_tasks(path=db_path)[0]
            self.assertEqual("applied", applied.stage)
            self.assertEqual("success", applied.status)
            self.assertEqual(1, applied.committed)
            self.assertEqual(2, applied.apply_attempts)
            with self.assertRaises(db.DbError):
                db.update_generated_title(task.id, "线上后改标题", path=db_path)

        self.assert_removed(temp_dir)

    def test_set_generated_cover_preserves_title_and_records_ai_success(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": row,
                        "account_name": "shop",
                        "alias": "alias",
                        "item_id": str(51100639510 + row),
                    }
                    for row in (2, 3)
                ],
                path=db_path,
            )
            tasks = db.list_tasks(batch_id=batch_id, path=db_path)
            for task in tasks:
                db.set_collected(task.id, "旧标题", "old.jpg", path=db_path)
            db.set_generated(tasks[1].id, "已有新标题", None, path=db_path)
            before_with_title = db.get_task(tasks[1].id, path=db_path)

            db.set_generated_cover(tasks[0].id, "cover-a.jpg", path=db_path)
            db.set_generated_cover(tasks[1].id, "cover-b.jpg", path=db_path)

            without_title = db.get_task(tasks[0].id, path=db_path)
            with_title = db.get_task(tasks[1].id, path=db_path)
            self.assertIsNone(without_title.new_title)
            self.assertEqual("cover-a.jpg", without_title.new_cover_path)
            self.assertEqual("generated", without_title.stage)
            self.assertEqual("success", without_title.status)
            self.assertEqual(1, without_title.generate_attempts)
            self.assertIsNotNone(without_title.generated_at)
            self.assertEqual("已有新标题", with_title.new_title)
            self.assertEqual("cover-b.jpg", with_title.new_cover_path)
            self.assertEqual(before_with_title.generate_attempts + 1, with_title.generate_attempts)

        self.assert_removed(temp_dir)

    def test_image_task_helpers_and_reset_lifecycle(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]
            db.set_collected(task.id, "旧标题", "old.jpg", path=db_path)
            db.set_generated(task.id, "新标题", "new.jpg", path=db_path)

            key = db.ensure_image_task_key(task.id, path=db_path)
            self.assertTrue(key.startswith(f"cmshopee-task-{task.id}-"))
            self.assertEqual(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)
            submitted = db.get_task(task.id, path=db_path)
            self.assertEqual("cmhub-task-1", submitted.image_task_id)
            self.assertEqual(key, submitted.image_task_key)

            title_only = db.reset_generated(
                task.id,
                reset_title=True,
                reset_cover=False,
                path=db_path,
            )["after"]
            self.assertEqual("cmhub-task-1", title_only.image_task_id)
            self.assertEqual(key, title_only.image_task_key)
            self.assertEqual(0, title_only.cover_reset_count)
            self.assertIsNone(title_only.cover_reset_at)

            cover_reset = db.reset_generated(
                task.id,
                reset_title=False,
                reset_cover=True,
                path=db_path,
            )["after"]
            self.assertIsNone(cover_reset.image_task_id)
            self.assertIsNone(cover_reset.image_task_key)
            self.assertEqual(1, cover_reset.cover_reset_count)
            self.assertIsNotNone(cover_reset.cover_reset_at)

            new_key = db.ensure_image_task_key(task.id, path=db_path)
            self.assertNotEqual(key, new_key)
            db.clear_image_task(task.id, path=db_path)
            cleared = db.get_task(task.id, path=db_path)
            self.assertIsNone(cleared.image_task_id)
            self.assertIsNone(cleared.image_task_key)

        self.assert_removed(temp_dir)

    def test_mark_failed_prepends_chinese_step_without_duplicate_prefix(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]

            db.mark_failed(task.id, "collect", "图片超过大小上限", path=db_path, step="read_cover")
            failed = db.get_task(task.id, path=db_path)
            self.assertEqual("读封面失败：图片超过大小上限", failed.last_error)

            db.mark_failed(task.id, "collect", "读封面失败：图片超过大小上限", path=db_path, step="read_cover")
            failed = db.get_task(task.id, path=db_path)
            self.assertEqual("读封面失败：图片超过大小上限", failed.last_error)

            db.mark_failed(task.id, "collect", "未知错误", path=db_path, step="custom_step")
            failed = db.get_task(task.id, path=db_path)
            self.assertEqual("custom_step失败：未知错误", failed.last_error)

            db.mark_failed(task.id, "collect", "保持原样", path=db_path)
            failed = db.get_task(task.id, path=db_path)
            self.assertEqual("保持原样", failed.last_error)

        self.assert_removed(temp_dir)

    def test_set_applied_prepends_failed_step_for_update_errors(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]
            db.set_generated(task.id, "新标题", "new.jpg", path=db_path)

            db.set_applied(task.id, False, "按钮禁用", path=db_path, step="click_update")

            failed = db.get_task(task.id, path=db_path)
            self.assertEqual("点击更新失败：按钮禁用", failed.last_error)
            self.assertEqual("failed", failed.status)

        self.assert_removed(temp_dir)


    def test_reset_generated_and_apply_status_keep_local_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]
            new_cover = os.path.join(temp_dir, "new.jpg")
            with open(new_cover, "wb") as fh:
                fh.write(b"jpeg")

            db.set_collected(task.id, "旧标题", "old.jpg", path=db_path)
            db.set_generated(task.id, "新标题", new_cover, path=db_path)
            db.set_applied(task.id, True, path=db_path)

            reset_apply = db.reset_apply_status(task.id, path=db_path)
            self.assertEqual("applied", reset_apply["before"].stage)
            after_apply = reset_apply["after"]
            self.assertEqual("generated", after_apply.stage)
            self.assertEqual("pending", after_apply.status)
            self.assertEqual("新标题", after_apply.new_title)
            self.assertEqual(new_cover, after_apply.new_cover_path)
            self.assertEqual(1, after_apply.committed)

            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.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)
            self.assertIsNone(after_generated.new_cover_path)
            self.assertIsNone(after_generated.last_error)
            self.assertEqual(1, after_generated.committed)
            self.assertEqual(1, after_generated.cover_reset_count)
            self.assertIsNotNone(after_generated.cover_reset_at)

        self.assert_removed(temp_dir)

    def test_reset_generated_can_reset_title_or_cover_components(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",
                    },
                    {
                        "source_file_abs": os.path.join(temp_dir, "input.xlsx"),
                        "source_sheet": "Sheet1",
                        "source_row": 3,
                        "account_name": "shop",
                        "alias": "alias",
                        "item_id": "51100639511",
                    },
                ],
                path=db_path,
            )
            first, second = db.list_tasks(batch_id=batch_id, path=db_path)
            first_cover = os.path.join(temp_dir, "first_new.jpg")
            second_cover = os.path.join(temp_dir, "second_new.jpg")
            for cover in [first_cover, second_cover]:
                with open(cover, "wb") as fh:
                    fh.write(b"jpeg")
            db.set_collected(first.id, "旧标题A", "old-a.jpg", path=db_path)
            db.set_generated(first.id, "手动标题A", first_cover, path=db_path)
            db.mark_failed(first.id, "generate", "图片不满意", path=db_path)
            db.set_collected(second.id, "旧标题B", "old-b.jpg", path=db_path)
            db.set_generated(second.id, "手动标题B", second_cover, path=db_path)

            title_reset = db.reset_generated(
                first.id,
                reset_title=True,
                reset_cover=False,
                path=db_path,
            )
            after_title = title_reset["after"]
            self.assertEqual("generated", after_title.stage)
            self.assertEqual("success", after_title.status)
            self.assertIsNone(after_title.new_title)
            self.assertEqual(first_cover, after_title.new_cover_path)
            self.assertIsNone(after_title.last_error)
            self.assertTrue(os.path.exists(first_cover))
            self.assertEqual(0, after_title.cover_reset_count)
            self.assertIsNone(after_title.cover_reset_at)

            cover_reset = db.reset_generated(
                second.id,
                reset_title=False,
                reset_cover=True,
                delete_file=True,
                path=db_path,
            )
            after_cover = cover_reset["after"]
            self.assertEqual("generated", after_cover.stage)
            self.assertEqual("success", after_cover.status)
            self.assertEqual("手动标题B", after_cover.new_title)
            self.assertIsNone(after_cover.new_cover_path)
            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"]))
            self.assertEqual(1, after_cover.cover_reset_count)
            self.assertIsNotNone(after_cover.cover_reset_at)

            with self.assertRaises(db.DbError):
                db.reset_generated(first.id, reset_title=False, reset_cover=False, path=db_path)
            with self.assertRaises(db.DbError):
                db.reset_generated(first.id, reset_title=True, reset_cover=False, delete_file=True, path=db_path)

        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.assertEqual(1, after.cover_reset_count)

        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.assertEqual(0, after.cover_reset_count)
            self.assertIsNone(after.cover_reset_at)
            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_cover_reset_history_persists_after_selecting_or_generating_cover(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",
                    },
                    {
                        "source_file_abs": os.path.join(temp_dir, "input.xlsx"),
                        "source_sheet": "Sheet1",
                        "source_row": 3,
                        "account_name": "shop",
                        "alias": "alias",
                        "item_id": "51100639511",
                    },
                ],
                path=db_path,
            )
            first, second = db.list_tasks(batch_id=batch_id, path=db_path)
            first_cover = os.path.join(temp_dir, "first.jpg")
            selected_cover = os.path.join(temp_dir, "selected.jpg")
            regenerated_cover = os.path.join(temp_dir, "regenerated.jpg")
            for cover in (first_cover, selected_cover, regenerated_cover):
                with open(cover, "wb") as fh:
                    fh.write(b"jpeg")
            db.set_generated(first.id, "新标题", first_cover, path=db_path)

            reset = db.reset_generated(
                first.id,
                reset_title=False,
                reset_cover=True,
                path=db_path,
            )["after"]
            self.assertEqual(1, reset.cover_reset_count)
            self.assertIsNotNone(reset.cover_reset_at)

            db.update_generated_cover(first.id, selected_cover, path=db_path)
            selected = db.get_task(first.id, path=db_path)
            self.assertEqual(1, selected.cover_reset_count)
            self.assertIsNotNone(selected.cover_reset_at)

            db.set_generated(first.id, selected.new_title, regenerated_cover, path=db_path)
            regenerated = db.get_task(first.id, path=db_path)
            self.assertEqual(1, regenerated.cover_reset_count)
            self.assertIsNotNone(regenerated.cover_reset_at)

            empty_reset = db.reset_generated(
                second.id,
                reset_title=False,
                reset_cover=True,
                path=db_path,
            )["after"]
            self.assertEqual(0, empty_reset.cover_reset_count)
            self.assertIsNone(empty_reset.cover_reset_at)

        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")
            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",
                    },
                    {
                        "source_file_abs": os.path.join(temp_dir, "input.xlsx"),
                        "source_sheet": "Sheet1",
                        "source_row": 3,
                        "account_name": "shop",
                        "alias": "alias",
                        "item_id": "51100639511",
                    },
                ],
                path=db_path,
            )
            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)

            result = db.delete_batch(batch_id, reason="导错文件", path=db_path)

            self.assertEqual(batch_id, result["batch_id"])
            self.assertEqual(2, result["task_count"])
            self.assertEqual(1, result["committed_count"])
            self.assertEqual([old_cover, new_cover], result["image_paths"])
            self.assertEqual([], db.list_batches(path=db_path))
            self.assertEqual([], db.list_tasks(path=db_path))
            self.assertIsNone(db.get_batch(batch_id, path=db_path))
            self.assertIsNone(db.get_task(tasks[0].id, path=db_path))
            deleted_task = db.get_task(tasks[0].id, path=db_path, include_deleted=True)
            self.assertEqual(tasks[0].id, deleted_task.id)
            deleted_batch = db.get_batch(batch_id, path=db_path, include_deleted=True)
            self.assertIsNotNone(deleted_batch.deleted_at)
            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)

    def test_duplicate_task_and_invalid_update_raise_clear_errors(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)
            row = {
                "source_file": "input.xlsx",
                "source_file_abs": os.path.abspath("input.xlsx"),
                "source_sheet": "Sheet1",
                "source_row": 2,
                "alias": "alias",
                "item_id": "51100639510",
            }

            db.insert_tasks(batch_id, [row], path=db_path)

            with self.assertRaises(db.DbError):
                db.insert_tasks(batch_id, [row], path=db_path)

            with self.assertRaises(db.DbError):
                db.update_account("alias", path=db_path, unknown_field=True)

        self.assert_removed(temp_dir)

    def test_run_logs_and_events_are_persisted(self):
        with self.make_temp_dir() as temp_dir:
            db_path = os.path.join(temp_dir, "cmshopee.db")
            db.init_db(db_path)

            run_id = db.create_run_log(
                "apply",
                dry_run=True,
                total=2,
                options={"api_key": "secret", "mode": "preview"},
                path=db_path,
            )
            db.add_run_log_event(
                run_id,
                "dry-run 预览任务",
                task_id=7,
                alias="alias",
                item_id="51100639510",
                path=db_path,
            )
            db.finish_run_log(
                run_id,
                status="done",
                done=2,
                success_count=1,
                skipped_count=1,
                failed_count=0,
                summary_json={"password": "secret", "done": 2},
                path=db_path,
            )

            run = db.list_run_logs(path=db_path)[0]
            self.assertEqual(run_id, run.id)
            self.assertEqual("apply", run.run_type)
            self.assertEqual(1, run.dry_run)
            self.assertEqual("done", run.status)
            self.assertEqual(2, run.done)
            self.assertEqual("***", run.options["api_key"])
            self.assertEqual("***", run.summary["password"])

            events = db.list_run_log_events(run_id, path=db_path)
            self.assertEqual(1, len(events))
            self.assertEqual("alias", events[0].alias)
            self.assertEqual("51100639510", events[0].item_id)
            self.assertIn("dry-run", events[0].message)

        self.assert_removed(temp_dir)


if __name__ == "__main__":
    unittest.main()