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

feat(collect): support soft deletion of one task

Browse Source
This commit is contained in:
chengma
2026-07-20 10:27:21 +08:00
parent a92f1c86f4
commit 9852b1c9c1
8 changed files with 333 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_db.py
+76 -1
View File
@@ -119,6 +119,8 @@ class DbTests(TempDirMixin, unittest.TestCase):
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)
@@ -146,6 +148,7 @@ class DbTests(TempDirMixin, unittest.TestCase):
"boundary",
"after",
"deleted",
"task-deleted",
),
start=1,
):
@@ -170,13 +173,14 @@ class DbTests(TempDirMixin, unittest.TestCase):
try:
with conn:
conn.execute(
"UPDATE batches SET created_at = ? WHERE id IN (?, ?, ?, ?)",
"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(
@@ -195,6 +199,10 @@ class DbTests(TempDirMixin, unittest.TestCase):
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,),
@@ -207,6 +215,10 @@ class DbTests(TempDirMixin, unittest.TestCase):
"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()
@@ -221,6 +233,11 @@ class DbTests(TempDirMixin, unittest.TestCase):
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)
@@ -231,6 +248,7 @@ class DbTests(TempDirMixin, unittest.TestCase):
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)
@@ -1005,6 +1023,63 @@ class DbTests(TempDirMixin, unittest.TestCase):
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")
tests/test_gui.py
+109
View File
@@ -8827,6 +8827,115 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assert_removed(temp_dir)
def test_collect_tab_soft_deletes_selected_task_from_context_menu(self):
with self.make_temp_dir() as temp_dir:
cfg = self.make_config(temp_dir)
accounts.create_account("主店", "alias-a", debug_port=9222, config=cfg)
batch_id = db.create_batch(["input.xlsx"], path=cfg["db_path"])
db.insert_tasks(
batch_id,
[
{
"source_file_abs": os.path.join(temp_dir, "input.xlsx"),
"source_sheet": "商品",
"source_row": 2,
"account_name": "Excel主店",
"alias": "alias-a",
"item_id": "51100639510",
},
{
"source_file_abs": os.path.join(temp_dir, "input.xlsx"),
"source_sheet": "商品",
"source_row": 3,
"account_name": "Excel主店",
"alias": "alias-a",
"item_id": "51100639511",
},
],
path=cfg["db_path"],
)
first, second = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
db.set_collected(first.id, "旧标题", "old.jpg", path=cfg["db_path"])
db.set_generated(first.id, "新标题", "new.jpg", path=cfg["db_path"])
db.set_applied(first.id, True, path=cfg["db_path"])
db.set_collected(second.id, "旧标题2", "old2.jpg", path=cfg["db_path"])
db.set_generated(second.id, "新标题2", "new2.jpg", path=cfg["db_path"])
statuses = []
refresh_calls = []
generate_tab = GenerateTab(config=cfg)
apply_tab = ApplyTab(config=cfg)
apply_tab.status_filter.setCurrentIndex(apply_tab.status_filter.findData("all"))
def refresh_workflow():
refresh_calls.append(True)
generate_tab.refresh_tasks()
apply_tab.refresh_tasks()
collect_tab = CollectTab(
config=cfg,
status_callback=statuses.append,
refresh_workflow_callback=refresh_workflow,
)
self.addCleanup(collect_tab.close)
self.addCleanup(generate_tab.close)
self.addCleanup(apply_tab.close)
collect_tab.resize(1000, 600)
collect_tab.show()
QApplication.processEvents()
first_index = collect_tab.model.index(0, 0)
first_rect = collect_tab.table.visualRect(first_index)
self.assertTrue(first_rect.isValid())
class FakeMenu:
def __init__(self, parent=None):
self.actions = []
def addAction(self, text):
action = SimpleNamespace(
text=text,
enabled=True,
triggered=DummySignal(),
)
action.setEnabled = lambda enabled: setattr(action, "enabled", bool(enabled))
self.actions.append(action)
return action
def exec(self, position):
return None
with mock.patch("app.gui.tabs.collect.QMenu", FakeMenu):
collect_tab._show_task_context_menu(first_rect.center())
self.assertEqual(0, collect_tab.table.currentIndex().row())
with mock.patch(
"app.gui.QMessageBox.question",
return_value=gui.QMessageBox.No,
) as question:
collect_tab.delete_selected_task()
self.assertIsNotNone(db.get_task(first.id, path=cfg["db_path"]))
self.assertIn("已取消删除本条记录", statuses[-1])
self.assertIn("不会回滚蝦皮线上商品", question.call_args[0][2])
with mock.patch(
"app.gui.QMessageBox.question",
return_value=gui.QMessageBox.Yes,
), mock.patch("app.gui.QMessageBox.information") as information:
collect_tab.delete_selected_task()
self.assertIsNone(db.get_task(first.id, path=cfg["db_path"]))
self.assertEqual([second.id], [task.id for task in db.list_tasks(batch_id=batch_id, path=cfg["db_path"])])
self.assertTrue(all(task.id != first.id for task in generate_tab.model.tasks))
self.assertTrue(all(task.id != first.id for task in apply_tab.model.tasks))
self.assertEqual([True], refresh_calls)
self.assertIn("已删除本地记录：商品 51100639510", statuses[-1])
self.assertIn("已删除本地记录：商品 51100639510", information.call_args[0][2])
collect_tab.collect_thread = object()
collect_tab.delete_selected_task()
self.assertIn("采集或回写正在进行", statuses[-1])
self.assert_removed(temp_dir)
def test_collect_tab_can_filter_unmatched_tasks_from_summary_bar(self):
with self.make_temp_dir() as temp_dir:
cfg = self.make_config(temp_dir)