import os import sqlite3 import sys import unittest from unittest import mock sys.path.insert(0, os.path.dirname(__file__)) from _helpers import TempDirMixin from app import db, product_status 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", "product_status", "product_status_note", "product_status_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) self.assertIn("product_status", columns) self.assertIn("product_status_note", columns) self.assertIn("product_status_at", columns) self.assertIn("deleted_at", columns) self.assertIn("deleted_reason", columns) task = db.get_task(1, conn=conn) self.assertEqual(0, task.cover_reset_count) self.assertIsNone(task.cover_reset_at) self.assertEqual("normal", task.product_status) self.assertEqual( db.LEGACY_PRODUCT_STATUS_DEFAULT_NOTE, task.product_status_note, ) self.assertIsNone(task.product_status_at) finally: conn.close() self.assert_removed(temp_dir) def test_init_db_defaults_only_pre_feature_active_missing_status_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_ids = {} for index, name in enumerate( ( "legacy-null", "legacy-blank", "legacy-unknown", "boundary", "after", "deleted", "task-deleted", ), start=1, ): batch_id = db.create_batch([f"{name}.xlsx"], path=db_path) batch_ids[name] = batch_id db.insert_tasks( batch_id, [ { "source_file_abs": os.path.join(temp_dir, f"{name}.xlsx"), "source_sheet": "Sheet1", "source_row": 2, "account_name": "店铺", "alias": "alias", "item_id": str(51100639700 + index), } ], path=db_path, ) conn = db.connect(db_path) try: with conn: conn.execute( "UPDATE batches SET created_at = ? WHERE id IN (?, ?, ?, ?, ?)", ( "2026-07-18T16:34:36", batch_ids["legacy-null"], batch_ids["legacy-blank"], batch_ids["legacy-unknown"], batch_ids["deleted"], batch_ids["task-deleted"], ), ) conn.execute( "UPDATE batches SET created_at = ? WHERE id = ?", (db.PRODUCT_STATUS_FEATURE_INTRODUCED_AT, batch_ids["boundary"]), ) conn.execute( "UPDATE batches SET created_at = ? WHERE id = ?", ("2026-07-18T16:34:38", batch_ids["after"]), ) blank_task = db.list_tasks( batch_id=batch_ids["legacy-blank"], conn=conn, )[0] unknown_task = db.list_tasks( batch_id=batch_ids["legacy-unknown"], conn=conn, )[0] task_deleted = db.list_tasks( batch_id=batch_ids["task-deleted"], conn=conn, )[0] conn.execute( "UPDATE tasks SET product_status = ' ' WHERE id = ?", (blank_task.id,), ) conn.execute( "UPDATE tasks SET product_status = 'unknown', product_status_note = '已有未知状态' WHERE id = ?", (unknown_task.id,), ) conn.execute( "UPDATE batches SET deleted_at = ? WHERE id = ?", ("2026-07-19T00:00:00", batch_ids["deleted"]), ) conn.execute( "UPDATE tasks SET deleted_at = ? WHERE id = ?", ("2026-07-19T00:00:00", task_deleted.id), ) finally: conn.close() db.init_db(db_path) legacy_null = db.list_tasks(batch_id=batch_ids["legacy-null"], path=db_path)[0] legacy_blank = db.list_tasks(batch_id=batch_ids["legacy-blank"], path=db_path)[0] legacy_unknown = db.list_tasks(batch_id=batch_ids["legacy-unknown"], path=db_path)[0] boundary = db.list_tasks(batch_id=batch_ids["boundary"], path=db_path)[0] after = db.list_tasks(batch_id=batch_ids["after"], path=db_path)[0] deleted = db.list_tasks( batch_id=batch_ids["deleted"], path=db_path, include_deleted=True, )[0] task_deleted = db.list_tasks( batch_id=batch_ids["task-deleted"], path=db_path, include_deleted=True, )[0] for task in (legacy_null, legacy_blank): self.assertEqual("normal", task.product_status) self.assertEqual(db.LEGACY_PRODUCT_STATUS_DEFAULT_NOTE, task.product_status_note) self.assertIsNone(task.product_status_at) self.assertEqual("unknown", legacy_unknown.product_status) self.assertEqual("已有未知状态", legacy_unknown.product_status_note) self.assertIsNone(boundary.product_status) self.assertIsNone(after.product_status) self.assertIsNone(deleted.product_status) self.assertIsNone(task_deleted.product_status) first_updated_at = legacy_null.updated_at db.init_db(db_path) self.assertEqual( first_updated_at, db.get_task(legacy_null.id, path=db_path).updated_at, ) self.assert_removed(temp_dir) def test_init_db_defaults_processed_missing_status_and_creates_one_backup(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(["processed.xlsx"], path=db_path) deleted_batch_id = db.create_batch(["deleted.xlsx"], path=db_path) item_ids = { "stage": "51100639801", "timestamp": "51100639802", "content": "51100639803", "unprocessed": "51100639804", "unknown": "51100639805", "deleted_task": "51100639806", "deleted_batch": "51100639807", } rows = [] for source_row, item_id in enumerate(item_ids.values(), start=2): rows.append( { "source_file_abs": os.path.join(temp_dir, f"{item_id}.xlsx"), "source_sheet": "Sheet1", "source_row": source_row, "account_name": "店铺", "alias": "alias", "item_id": item_id, } ) db.insert_tasks(batch_id, rows[:-1], path=db_path) db.insert_tasks(deleted_batch_id, rows[-1:], path=db_path) conn = db.connect(db_path) try: tasks = { task.item_id: task for task in db.list_tasks(batch_id=batch_id, conn=conn) } deleted_batch_task = db.list_tasks( batch_id=deleted_batch_id, conn=conn, )[0] with conn: conn.execute( "UPDATE batches SET created_at = ? WHERE id IN (?, ?)", ("2026-07-27T12:00:00", batch_id, deleted_batch_id), ) conn.execute( "UPDATE tasks SET stage = 'generated', new_cover_path = ? WHERE id = ?", ( os.path.join(temp_dir, "stage.jpg"), tasks[item_ids["stage"]].id, ), ) conn.execute( "UPDATE tasks SET generated_at = ?, new_cover_path = ? WHERE id = ?", ( "2026-07-27T12:01:00", os.path.join(temp_dir, "timestamp.jpg"), tasks[item_ids["timestamp"]].id, ), ) conn.execute( "UPDATE tasks SET new_title = ? WHERE id = ?", ("历史生成标题", tasks[item_ids["content"]].id), ) conn.execute( """ UPDATE tasks SET product_status = 'unknown', product_status_note = '已有未知状态', new_title = '不应覆盖' WHERE id = ? """, (tasks[item_ids["unknown"]].id,), ) conn.execute( "UPDATE tasks SET stage = 'generated', deleted_at = ? WHERE id = ?", ("2026-07-27T12:02:00", tasks[item_ids["deleted_task"]].id), ) conn.execute( "UPDATE tasks SET stage = 'generated' WHERE id = ?", (deleted_batch_task.id,), ) conn.execute( "UPDATE batches SET deleted_at = ? WHERE id = ?", ("2026-07-27T12:03:00", deleted_batch_id), ) finally: conn.close() db.init_db(db_path) migrated = { task.item_id: task for task in db.list_tasks( batch_id=batch_id, path=db_path, include_deleted=True, ) } for key in ("stage", "timestamp", "content"): task = migrated[item_ids[key]] self.assertEqual("normal", task.product_status) self.assertEqual( db.LEGACY_PRODUCT_STATUS_DEFAULT_NOTE, task.product_status_note, ) self.assertIsNone(task.product_status_at) self.assertIsNone(migrated[item_ids["unprocessed"]].product_status) self.assertEqual("unknown", migrated[item_ids["unknown"]].product_status) self.assertEqual( "已有未知状态", migrated[item_ids["unknown"]].product_status_note, ) self.assertIsNone(migrated[item_ids["deleted_task"]].product_status) deleted_batch_task = db.list_tasks( batch_id=deleted_batch_id, path=db_path, include_deleted=True, )[0] self.assertIsNone(deleted_batch_task.product_status) backup_dir = os.path.join(temp_dir, "backups") backups = [ os.path.join(backup_dir, name) for name in os.listdir(backup_dir) if f"before-{db.LEGACY_STATUS_BACKUP_TASK_ID}" in name ] self.assertEqual(1, len(backups)) backup_conn = sqlite3.connect(backups[0]) try: row = backup_conn.execute( "SELECT product_status FROM tasks WHERE item_id = ?", (item_ids["stage"],), ).fetchone() self.assertIsNone(row[0]) finally: backup_conn.close() content_task = migrated[item_ids["content"]] plan = product_status.build_apply_plan([content_task], "title") self.assertEqual([content_task.id], [task.id for task in plan["executable"]]) first_updated_at = content_task.updated_at db.init_db(db_path) self.assertEqual( first_updated_at, db.get_task(content_task.id, path=db_path).updated_at, ) self.assertEqual( 1, len( [ name for name in os.listdir(backup_dir) if f"before-{db.LEGACY_STATUS_BACKUP_TASK_ID}" in name ] ), ) self.assert_removed(temp_dir) def test_init_db_repairs_only_legacy_promotion_status_misclassification(self): with self.make_temp_dir() as temp_dir: db_path = os.path.join(temp_dir, "cmshopee.db") db.init_db(db_path) batch_ids = {} for index, name in enumerate( ( "legacy-promotion", "legacy-unknown", "after-promotion", "legacy-unlisted", "deleted-promotion", ), start=1, ): batch_id = db.create_batch([f"{name}.xlsx"], path=db_path) batch_ids[name] = batch_id db.insert_tasks( batch_id, [ { "source_file_abs": os.path.join(temp_dir, f"{name}.xlsx"), "source_sheet": "Sheet1", "source_row": 2, "account_name": "店铺", "alias": "alias", "item_id": str(51100639800 + index), } ], path=db_path, ) promotion_note = ( "标题：由於正在進行促銷，以下一些欄位無法進行編輯，" "數值將顯示為灰色。" ) conn = db.connect(db_path) try: with conn: legacy_ids = [ batch_ids["legacy-promotion"], batch_ids["legacy-unknown"], batch_ids["legacy-unlisted"], batch_ids["deleted-promotion"], ] placeholders = ", ".join("?" for _batch_id in legacy_ids) conn.execute( f"UPDATE batches SET created_at = ? WHERE id IN ({placeholders})", ("2026-07-18T16:34:36", *legacy_ids), ) task_by_batch = { batch_id: db.list_tasks(batch_id=batch_id, conn=conn)[0] for batch_id in batch_ids.values() } conn.execute( """ UPDATE tasks SET product_status = 'unknown', product_status_note = ?, product_status_at = ?, new_cover_path = ?, stage = 'generated', status = 'success' WHERE id = ? """, ( promotion_note, "2026-07-20T09:18:10", "new.jpg", task_by_batch[batch_ids["legacy-promotion"]].id, ), ) conn.execute( "UPDATE tasks SET product_status = 'unknown', product_status_note = ? WHERE id = ?", ( "其他未知状态提示", task_by_batch[batch_ids["legacy-unknown"]].id, ), ) conn.execute( "UPDATE tasks SET product_status = 'unknown', product_status_note = ? WHERE id = ?", ( promotion_note, task_by_batch[batch_ids["after-promotion"]].id, ), ) conn.execute( "UPDATE tasks SET product_status = 'unlisted', product_status_note = ? WHERE id = ?", ( promotion_note, task_by_batch[batch_ids["legacy-unlisted"]].id, ), ) conn.execute( "UPDATE tasks SET product_status = 'unknown', product_status_note = ? WHERE id = ?", ( promotion_note, task_by_batch[batch_ids["deleted-promotion"]].id, ), ) conn.execute( "UPDATE batches SET deleted_at = ? WHERE id = ?", ("2026-07-19T00:00:00", batch_ids["deleted-promotion"]), ) finally: conn.close() db.init_db(db_path) legacy_promotion = db.list_tasks( batch_id=batch_ids["legacy-promotion"], path=db_path )[0] legacy_unknown = db.list_tasks( batch_id=batch_ids["legacy-unknown"], path=db_path )[0] after_promotion = db.list_tasks( batch_id=batch_ids["after-promotion"], path=db_path )[0] legacy_unlisted = db.list_tasks( batch_id=batch_ids["legacy-unlisted"], path=db_path )[0] deleted_promotion = db.list_tasks( batch_id=batch_ids["deleted-promotion"], include_deleted=True, path=db_path, )[0] self.assertEqual("normal", legacy_promotion.product_status) self.assertEqual(db.LEGACY_PROMOTION_STATUS_REPAIR_NOTE, legacy_promotion.product_status_note) self.assertEqual("2026-07-20T09:18:10", legacy_promotion.product_status_at) cover_plan = product_status.build_apply_plan([legacy_promotion], "cover") self.assertEqual([legacy_promotion.id], [task.id for task in cover_plan["executable"]]) self.assertEqual("unknown", legacy_unknown.product_status) self.assertEqual("unknown", after_promotion.product_status) self.assertEqual("unlisted", legacy_unlisted.product_status) self.assertEqual("unknown", deleted_promotion.product_status) first_updated_at = legacy_promotion.updated_at db.init_db(db_path) self.assertEqual( first_updated_at, db.get_task(legacy_promotion.id, path=db_path).updated_at, ) 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_product_status_snapshot_preserves_collection_lifecycle_and_content(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", db_path) db.set_product_status( task.id, "reviewing", "审核说明", "2026-07-18T09:00:00", path=db_path, ) status_only = db.get_task(task.id, path=db_path) self.assertEqual("collected", status_only.stage) self.assertEqual("success", status_only.status) self.assertEqual("旧标题", status_only.old_title) self.assertEqual("old.jpg", status_only.old_cover_path) self.assertEqual("reviewing", status_only.product_status) self.assertEqual("审核说明", status_only.product_status_note) self.assertEqual("2026-07-18T09:00:00", status_only.product_status_at) db.set_collected( task.id, "重新采集标题", "new-old.jpg", product_status_value="not-a-status", product_status_note="x" * 2200, product_status_at="2026-07-18T10:00:00", path=db_path, ) refreshed = db.get_task(task.id, path=db_path) self.assertEqual("unknown", refreshed.product_status) self.assertEqual("重新采集标题", refreshed.old_title) self.assertEqual("new-old.jpg", refreshed.old_cover_path) self.assertEqual("2026-07-18T10:00:00", refreshed.product_status_at) self.assertEqual(2000, len(refreshed.product_status_note)) 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_delete_task_soft_hides_only_one_inactive_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", }, { "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) db.set_collected(first.id, "旧标题", "old.jpg", path=db_path) db.set_generated(first.id, "新标题", "new.jpg", path=db_path) db.set_applied(first.id, True, path=db_path) result = db.delete_task(first.id, reason="导入了错误商品", path=db_path) self.assertEqual(first.id, result["task_id"]) self.assertEqual(batch_id, result["batch_id"]) self.assertEqual("51100639510", result["item_id"]) self.assertEqual(1, result["committed"]) self.assertEqual([second.id], [task.id for task in db.list_tasks(batch_id=batch_id, path=db_path)]) self.assertIsNone(db.get_task(first.id, path=db_path)) deleted = db.get_task(first.id, path=db_path, include_deleted=True) self.assertIsNotNone(deleted.deleted_at) self.assertEqual("导入了错误商品", deleted.deleted_reason) self.assertEqual("旧标题", deleted.old_title) self.assertEqual("新标题", deleted.new_title) self.assertEqual("new.jpg", deleted.new_cover_path) self.assertEqual(1, deleted.committed) self.assertEqual([batch_id], [batch.id for batch in db.list_batches(path=db_path)]) with self.assertRaisesRegex(db.DbError, "不存在或已删除"): db.delete_task(first.id, path=db_path) db.mark_running(second.id, "collect", path=db_path) with self.assertRaisesRegex(db.DbError, "正在处理"): db.delete_task(second.id, path=db_path) 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()