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

feat(product-suite): add original image batch management

Browse Source
This commit is contained in:
chengma
2026-07-14 15:35:59 +08:00
parent 21d2dd5253
commit e86b9189ab
8 changed files with 759 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_image_studio.py
+133
View File
@@ -335,6 +335,139 @@ class ImageStudioTests(TempDirMixin, unittest.TestCase):
self.assert_removed(temp_dir)
def test_remove_original_assets_is_atomic_and_reorders_remaining_assets(self):
with self.make_temp_dir() as temp_dir:
db_path = os.path.join(temp_dir, "cmshopee.db")
db.init_db(db_path)
project = image_studio.create_or_get_project(
account_alias="alias",
account_slug="alias_slug",
item_id="51100639510",
path=db_path,
)
local_paths = []
assets = []
for index in range(1, 5):
local_path = os.path.join(temp_dir, "original-%d.png" % index)
with open(local_path, "wb") as fh:
fh.write(b"image-%d" % index)
local_paths.append(local_path)
assets.append(
image_studio.add_asset(
project.id,
image_studio.ASSET_KIND_ORIGINAL,
local_path=local_path,
source_order=index * 10,
path=db_path,
)
)
removed = image_studio.remove_original_assets_if_unused(
project.id,
[assets[0].id, assets[2].id, assets[0].id],
path=db_path,
)
self.assertEqual([assets[0].id, assets[2].id], [asset.id for asset in removed])
remaining = image_studio.list_assets(
project.id,
kind=image_studio.ASSET_KIND_ORIGINAL,
path=db_path,
)
self.assertEqual([assets[1].id, assets[3].id], [asset.id for asset in remaining])
self.assertEqual([1, 2], [asset.source_order for asset in remaining])
self.assertTrue(all(os.path.isfile(path) for path in local_paths))
self.assert_removed(temp_dir)
def test_remove_original_assets_rejects_invalid_or_referenced_batch_without_partial_delete(self):
with self.make_temp_dir() as temp_dir:
db_path = os.path.join(temp_dir, "cmshopee.db")
db.init_db(db_path)
project = image_studio.create_or_get_project(
account_alias="alias",
account_slug="alias_slug",
item_id="51100639510",
path=db_path,
)
other_project = image_studio.create_or_get_project(
account_alias="alias",
account_slug="alias_slug",
item_id="51100639511",
path=db_path,
)
free_asset = image_studio.add_asset(
project.id,
image_studio.ASSET_KIND_ORIGINAL,
source_order=1,
path=db_path,
)
referenced_asset = image_studio.add_asset(
project.id,
image_studio.ASSET_KIND_ORIGINAL,
source_order=2,
path=db_path,
)
selected_asset = image_studio.add_asset(
project.id,
image_studio.ASSET_KIND_ORIGINAL,
source_order=3,
path=db_path,
)
generated_asset = image_studio.add_asset(
project.id,
"generated",
path=db_path,
)
foreign_asset = image_studio.add_asset(
other_project.id,
image_studio.ASSET_KIND_ORIGINAL,
path=db_path,
)
image_studio.create_job(
project.id,
source_asset_id=referenced_asset.id,
path=db_path,
)
image_studio.replace_selections(
project.id,
"main",
[selected_asset.id],
path=db_path,
)
with self.assertRaisesRegex(db.DbError, "引用"):
image_studio.remove_original_assets_if_unused(
project.id,
[free_asset.id, referenced_asset.id],
path=db_path,
)
self.assertIsNotNone(image_studio.get_asset(free_asset.id, path=db_path))
self.assertIsNotNone(image_studio.get_asset(referenced_asset.id, path=db_path))
with self.assertRaisesRegex(db.DbError, "引用"):
image_studio.remove_original_assets_if_unused(
project.id,
[free_asset.id, selected_asset.id],
path=db_path,
)
self.assertIsNotNone(image_studio.get_asset(free_asset.id, path=db_path))
self.assertIsNotNone(image_studio.get_asset(selected_asset.id, path=db_path))
for invalid_id in (generated_asset.id, foreign_asset.id, 999999):
with self.assertRaisesRegex(db.DbError, "不属于当前项目"):
image_studio.remove_original_assets_if_unused(
project.id,
[free_asset.id, invalid_id],
path=db_path,
)
self.assertIsNotNone(image_studio.get_asset(free_asset.id, path=db_path))
with self.assertRaisesRegex(db.DbError, "请选择"):
image_studio.remove_original_assets_if_unused(project.id, [], path=db_path)
self.assert_removed(temp_dir)
def test_sync_original_asset_urls_is_idempotent_and_marks_missing(self):
with self.make_temp_dir() as temp_dir:
db_path = os.path.join(temp_dir, "cmshopee.db")
tests/test_product_suite_gui.py
+237 -3
View File
@@ -13,10 +13,18 @@ from app import gui
if gui.QT_IMPORT_ERROR is not None:
raise unittest.SkipTest("PySide6 未安装")
from PySide6.QtGui import QImage
from PySide6.QtWidgets import QApplication, QLabel, QPushButton
from PySide6.QtCore import Qt
from PySide6.QtGui import QIcon, QImage, QPixmap
from PySide6.QtTest import QTest
from PySide6.QtWidgets import QApplication, QLabel, QListWidgetItem, QPushButton
from app.gui.tabs.product_suite import ProductSuiteTab, SuiteResultCard
from app.gui.tabs.product_suite import (
ORIGINAL_CHECK_STATE_ROLE,
ProductOriginalDelegate,
ProductOriginalList,
ProductSuiteTab,
SuiteResultCard,
)
class ProductSuiteGuiTests(TempDirMixin, unittest.TestCase):
@@ -46,6 +54,33 @@ class ProductSuiteGuiTests(TempDirMixin, unittest.TestCase):
image.fill(0xFF336699)
self.assertTrue(image.save(path))
def _create_project_with_assets(self, temp_dir, config, count, item_id="51100639510"):
account = accounts.create_account(
"主店",
"alias-a",
debug_port=9222,
config=config,
)
project = image_studio.create_or_get_project(
account,
item_id=item_id,
path=config["db_path"],
)
assets = []
for index in range(count):
source_path = os.path.join(temp_dir, "%s-%02d.png" % (item_id, index + 1))
self._write_image(source_path)
assets.append(
image_studio.add_asset(
project.id,
image_studio.ASSET_KIND_ORIGINAL,
local_path=source_path,
source_order=index + 1,
path=config["db_path"],
)
)
return project, assets
def test_tab_builds_suite_controls_without_old_detail_workspace(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
@@ -276,6 +311,205 @@ class ProductSuiteGuiTests(TempDirMixin, unittest.TestCase):
self.assert_removed(temp_dir)
def test_original_list_expands_without_internal_scrollbars(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
project, assets = self._create_project_with_assets(temp_dir, config, 0)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
state.account_alias = "alias-a"
state.item_id = "51100639510"
state.project_id = project.id
tab.resize(1180, 760)
tab.show()
tab._load_state(state)
self.app.processEvents()
self.assertEqual(Qt.ScrollBarAlwaysOff, tab.original_list.horizontalScrollBarPolicy())
self.assertEqual(Qt.ScrollBarAlwaysOff, tab.original_list.verticalScrollBarPolicy())
self.assertEqual(6, tab.original_list.count())
self.assertEqual(2, tab.original_list.content_row_count())
empty_height = tab.original_list.height()
for target_count in (6, 7, 16):
for index in range(len(assets), target_count):
source_path = os.path.join(temp_dir, "added-%02d.png" % (index + 1))
self._write_image(source_path)
assets.append(
image_studio.add_asset(
project.id,
image_studio.ASSET_KIND_ORIGINAL,
local_path=source_path,
source_order=index + 1,
path=config["db_path"],
)
)
tab._refresh_originals(state)
self.app.processEvents()
self.assertEqual(target_count, len(tab.original_list.asset_ids()))
expected_rows = (max(6, target_count) + 2) // 3
self.assertEqual(expected_rows, tab.original_list.content_row_count())
if target_count == 6:
self.assertEqual(empty_height, tab.original_list.height())
wide_height = tab.original_list.height()
tab.original_list.setFixedWidth(250)
self.app.processEvents()
self.assertEqual(2, tab.original_list.content_column_count())
self.assertEqual(8, tab.original_list.content_row_count())
self.assertGreater(tab.original_list.height(), wide_height)
self.assert_removed(temp_dir)
def test_original_checks_preserve_on_refresh_and_clear_on_task_switch(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
project, assets = self._create_project_with_assets(temp_dir, config, 2)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
state.account_alias = "alias-a"
state.item_id = "51100639510"
state.project_id = project.id
tab._load_state(state)
self.assertEqual(6, tab.original_list.count())
self.assertTrue(tab.select_all_originals_button.isEnabled())
self.assertTrue(tab.invert_originals_button.isEnabled())
for row in range(2):
self.assertIsNotNone(tab.original_list.item(row).data(ORIGINAL_CHECK_STATE_ROLE))
for row in range(2, 6):
self.assertIsNone(tab.original_list.item(row).data(ORIGINAL_CHECK_STATE_ROLE))
tab.select_all_originals_button.click()
self.assertEqual([asset.id for asset in assets], tab.original_list.checked_asset_ids())
self.assertEqual("已选 2 张", tab.original_selected_label.text())
tab.invert_originals_button.click()
self.assertEqual([], tab.original_list.checked_asset_ids())
tab.original_list.set_checked_asset_ids([assets[1].id])
tab._refresh_originals(state)
self.assertEqual([assets[1].id], tab.original_list.checked_asset_ids())
tab.reorder_originals([assets[1].id, assets[0].id])
self.assertEqual([assets[1].id], tab.original_list.checked_asset_ids())
tab.add_task(inherit=True)
self.assertEqual([], tab.original_list.checked_asset_ids())
self.assertFalse(tab.select_all_originals_button.isEnabled())
tab.task_tabs.setCurrentIndex(0)
self.assertEqual([], tab.original_list.checked_asset_ids())
self.assertEqual("已选 0 张", tab.original_selected_label.text())
self.assert_removed(temp_dir)
def test_original_checkbox_click_and_keyboard_delete_keep_actions_separate(self):
original_list = ProductOriginalList()
self.addCleanup(original_list.close)
original_list.setFixedSize(240, 210)
for asset_id in (11, 12):
item = QListWidgetItem("原图%d" % asset_id)
item.setData(Qt.UserRole, asset_id)
item.setData(ORIGINAL_CHECK_STATE_ROLE, Qt.Unchecked)
pixmap = QPixmap(82, 64)
pixmap.fill(0xFF336699)
item.setIcon(QIcon(pixmap))
original_list.addItem(item)
original_list.show()
self.app.processEvents()
clicked = []
double_clicked = []
batch_deleted = []
single_deleted = []
original_list.itemClicked.connect(lambda item: clicked.append(item.data(Qt.UserRole)))
original_list.itemDoubleClicked.connect(
lambda item: double_clicked.append(item.data(Qt.UserRole))
)
original_list.deleteAssetsRequested.connect(lambda ids: batch_deleted.append(ids))
original_list.deleteRequested.connect(lambda asset_id: single_deleted.append(asset_id))
first_rect = original_list.visualItemRect(original_list.item(0))
check_point = ProductOriginalDelegate.checkbox_hit_rect(first_rect).center()
QTest.mouseClick(original_list.viewport(), Qt.LeftButton, pos=check_point)
self.assertEqual([11], original_list.checked_asset_ids())
self.assertEqual([], clicked)
QTest.mouseDClick(original_list.viewport(), Qt.LeftButton, pos=check_point)
self.assertEqual([], double_clicked)
QTest.mouseClick(original_list.viewport(), Qt.LeftButton, pos=first_rect.center())
self.assertEqual([11], clicked)
original_list.set_checked_asset_ids([11, 12])
self.assertEqual(
[("删除选中的2张图片…", [11, 12])],
original_list.context_delete_options(11),
)
original_list.set_checked_asset_ids([12])
self.assertEqual(
[("删除这张图片…", [11]), ("删除选中图片…", [12])],
original_list.context_delete_options(11),
)
original_list.setFocus()
original_list.set_checked_asset_ids([11, 12])
QTest.keyClick(original_list, Qt.Key_Delete)
self.assertEqual([[11, 12]], batch_deleted)
original_list.clear_checks()
original_list.setCurrentRow(1)
QTest.keyClick(original_list, Qt.Key_Backspace)
self.assertEqual([12], single_deleted)
def test_batch_delete_confirms_main_image_and_blocks_running_or_downloading(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
project, assets = self._create_project_with_assets(temp_dir, config, 3)
statuses = []
tab = ProductSuiteTab(
config=config,
db_path=config["db_path"],
status_callback=lambda message, level=None: statuses.append((message, level)),
)
self.addCleanup(tab.close)
state = tab._displayed_state
state.account_alias = "alias-a"
state.item_id = "51100639510"
state.project_id = project.id
tab._load_state(state)
confirmations = []
tab._confirm = lambda title, message, **kwargs: confirmations.append(
(title, message, kwargs)
) or True
tab.original_list.set_checked_asset_ids([assets[0].id, assets[1].id])
tab.delete_originals(tab.original_list.checked_asset_ids())
self.assertEqual([assets[2].id], tab.original_list.asset_ids())
self.assertEqual([], tab.original_list.checked_asset_ids())
self.assertIn("选中的2张", confirmations[0][1])
self.assertIn("下一张图片将成为主图", confirmations[0][1])
self.assertIn("不会删除蝦皮线上图片", confirmations[0][1])
self.assertEqual(("已移除2张商品原图", "success"), statuses[-1])
messages = []
tab._message = lambda title, message, **kwargs: messages.append((title, message))
state.download_queue = [assets[2].id]
tab.delete_originals([assets[2].id])
self.assertIn("仍在后台下载", messages[-1][1])
self.assertIsNotNone(image_studio.get_asset(assets[2].id, path=config["db_path"]))
class RunningWorker:
def cancel(self):
return None
state.download_queue = []
state.worker = RunningWorker()
tab.delete_originals([assets[2].id])
self.assertEqual(("生成中不能删除当前任务的商品原图", "warning"), statuses[-1])
self.assertIsNotNone(image_studio.get_asset(assets[2].id, path=config["db_path"]))
state.worker = None
self.assert_removed(temp_dir)
if __name__ == "__main__":
unittest.main()