chengma/cmshoppe
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(ai): mark cover reset history in generate table

Browse Source
This commit is contained in:
chengma
2026-07-11 12:06:17 +08:00
parent b71fa60b9e
commit 5c01377173
5 changed files with 280 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_db.py
+163 -1
View File
@@ -34,7 +34,88 @@ class DbTests(TempDirMixin, unittest.TestCase):
row["name"]
for row in conn.execute("PRAGMA table_info(tasks)").fetchall()
}
self.assertTrue({"image_task_id", "image_task_key"}.issubset(task_columns))
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()
@@ -87,6 +168,8 @@ class DbTests(TempDirMixin, unittest.TestCase):
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)
@@ -174,6 +257,8 @@ class DbTests(TempDirMixin, unittest.TestCase):
)["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,
@@ -183,6 +268,8 @@ class DbTests(TempDirMixin, unittest.TestCase):
)["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)
@@ -313,6 +400,8 @@ class DbTests(TempDirMixin, unittest.TestCase):
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)
@@ -368,6 +457,8 @@ class DbTests(TempDirMixin, unittest.TestCase):
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,
@@ -385,6 +476,8 @@ class DbTests(TempDirMixin, unittest.TestCase):
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)
@@ -443,6 +536,7 @@ class DbTests(TempDirMixin, unittest.TestCase):
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)
@@ -478,6 +572,8 @@ class DbTests(TempDirMixin, unittest.TestCase):
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)
@@ -563,6 +659,72 @@ class DbTests(TempDirMixin, unittest.TestCase):
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")
tests/test_gui.py
+52
View File
@@ -384,6 +384,58 @@ class GuiTests(TempDirMixin, unittest.TestCase):
apply_model.data(apply_model.index(0, 5), gui.Qt.ToolTipRole),
)
def test_generate_table_marks_cover_reset_history_on_item_id_column(self):
account = SimpleNamespace(alias="papa", account_name="papa 店铺")
reset_task = SimpleNamespace(
alias="papa",
account_name="papa 店铺",
item_id="26887160467",
old_title="旧标题",
new_title="新标题",
new_cover_path="new.jpg",
stage="generated",
status="success",
last_error="",
collect_attempts=1,
generate_attempts=1,
apply_attempts=0,
committed=0,
cover_reset_count=2,
cover_reset_at="2026-07-11T10:00:00",
)
normal_task = SimpleNamespace(
alias="papa",
account_name="papa 店铺",
item_id="28431952912",
old_title="旧标题",
new_title="新标题",
new_cover_path="new.jpg",
stage="generated",
status="success",
last_error="",
collect_attempts=1,
generate_attempts=1,
apply_attempts=0,
committed=0,
cover_reset_count=0,
cover_reset_at=None,
)
model = gui.GenerateTaskTableModel()
model.set_tasks([reset_task, normal_task], [account])
self.assert_foreground(model, 0, 1, gui.COLOR_WARNING)
self.assertIsNone(model.data(model.index(1, 1), gui.Qt.ForegroundRole))
self.assertIn(
"该商品封面已重置 2 次",
model.data(model.index(0, 1), gui.Qt.ToolTipRole),
)
self.assertIn(
"双击可查看封面画廊",
model.data(model.index(0, 1), gui.Qt.ToolTipRole),
)
self.assert_foreground(model, 0, 4, gui.COLOR_SUCCESS)
self.assert_foreground(model, 0, 5, gui.COLOR_SUCCESS)
def test_main_window_has_five_tabs_in_workflow_order(self):
with self.make_temp_dir() as temp_dir:
window = MainWindow(config=self.make_config(temp_dir))