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

feat(product-suite): add global generation history

Browse Source
This commit is contained in:
chengma
2026-07-17 09:59:55 +08:00
parent f1dc7f28dc
commit e232097ba5
13 changed files with 1409 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_image_studio.py
+154
View File
@@ -1028,6 +1028,160 @@ class ImageStudioTests(TempDirMixin, unittest.TestCase):
self.assert_removed(temp_dir)
def test_global_generation_rounds_filter_paginate_and_exclude_deleted_projects(self):
with self.make_temp_dir() as temp_dir:
db_path = os.path.join(temp_dir, "cmshopee.db")
db.init_db(db_path)
first_project = image_studio.create_or_get_project(
account_alias="main-shop",
account_name="主店",
account_slug="main-shop",
item_id="51100639510",
path=db_path,
)
second_project = image_studio.create_or_get_project(
account_alias="second-shop",
account_name="副店",
account_slug="second-shop",
item_id="51100639511",
path=db_path,
)
deleted_project = image_studio.create_or_get_project(
account_alias="deleted-shop",
account_slug="deleted-shop",
item_id="51100639512",
path=db_path,
)
first_source = image_studio.add_asset(
first_project.id,
image_studio.ASSET_KIND_ORIGINAL,
path=db_path,
)
second_source = image_studio.add_asset(
second_project.id,
image_studio.ASSET_KIND_ORIGINAL,
path=db_path,
)
deleted_source = image_studio.add_asset(
deleted_project.id,
image_studio.ASSET_KIND_ORIGINAL,
path=db_path,
)
first = image_studio.create_job(
first_project.id,
source_asset_id=first_source.id,
task_key="global-first-slot",
generation_round_key="first-round",
generation_slot_index=0,
path=db_path,
)
image_studio.update_job_status(first.id, "failed", path=db_path)
retry = image_studio.create_job(
first_project.id,
source_asset_id=first_source.id,
task_key="global-first-retry",
generation_round_key="first-round",
generation_slot_index=0,
path=db_path,
)
image_studio.update_job_status(retry.id, "succeeded", path=db_path)
second = image_studio.create_job(
second_project.id,
source_asset_id=second_source.id,
task_key="global-second-round",
generation_round_key="second-round",
generation_slot_index=0,
path=db_path,
)
image_studio.update_job_status(second.id, "succeeded", path=db_path)
deleted = image_studio.create_job(
deleted_project.id,
source_asset_id=deleted_source.id,
task_key="global-deleted-round",
generation_round_key="deleted-round",
generation_slot_index=0,
path=db_path,
)
image_studio.update_job_status(deleted.id, "succeeded", path=db_path)
legacy = image_studio.create_job(
first_project.id,
source_asset_id=first_source.id,
task_key="global-legacy-history",
path=db_path,
)
image_studio.update_job_status(legacy.id, "cancelled", path=db_path)
image_studio.set_current_generation_round(
first_project.id,
"first-round",
path=db_path,
)
image_studio.soft_delete_project(
deleted_project.id,
reason="测试软删除",
path=db_path,
)
rounds = image_studio.list_global_generation_rounds(path=db_path)
self.assertEqual(
{first_project.id, second_project.id},
{round_.project_id for round_ in rounds},
)
self.assertIsNone(rounds[0].generation_round_key)
self.assertTrue(rounds[0].is_legacy)
self.assertEqual(first_project.id, rounds[0].project_id)
self.assertEqual(second_project.id, rounds[1].project_id)
first_round = next(
round_
for round_ in rounds
if (
round_.project_id == first_project.id
and round_.generation_round_key == "first-round"
)
)
self.assertEqual("first-round", first_round.generation_round_key)
self.assertEqual("主店", first_round.account_name)
self.assertEqual(1, first_round.job_count)
self.assertEqual(1, first_round.slot_count)
self.assertEqual(1, first_round.retry_count)
self.assertEqual(1, first_round.succeeded_count)
self.assertEqual(0, first_round.failed_count)
self.assertTrue(first_round.is_current)
self.assertEqual(
[first_project.id, first_project.id],
[
round_.project_id
for round_ in image_studio.list_global_generation_rounds(
account_query="主店",
path=db_path,
)
],
)
self.assertEqual(
[second_project.id],
[
round_.project_id
for round_ in image_studio.list_global_generation_rounds(
item_query="51100639511",
path=db_path,
)
],
)
self.assertEqual(
[first_project.id, first_project.id],
[
round_.project_id
for round_ in image_studio.list_global_generation_rounds(
project_id=first_project.id,
path=db_path,
)
],
)
self.assertEqual(1, len(image_studio.list_global_generation_rounds(limit=1, path=db_path)))
self.assertEqual(1, len(image_studio.list_global_generation_rounds(limit=1, offset=1, path=db_path)))
self.assert_removed(temp_dir)
def test_selections_are_consecutive_unique_and_replaceable(self):
with self.make_temp_dir() as temp_dir:
db_path = os.path.join(temp_dir, "cmshopee.db")
tests/test_image_studio_export.py
+116
View File
@@ -53,6 +53,122 @@ class ImageStudioExportTests(TempDirMixin, unittest.TestCase):
image_studio.replace_selections(project.id, "detail", [detail.id], path=cfg["db_path"])
return cfg, project, main, detail
def test_export_generation_round_copies_successes_and_uses_new_directory_on_repeat(self):
with self.make_temp_dir() as temp_dir:
cfg, project, _main, _detail = self._project_with_assets(temp_dir)
source = image_studio.add_asset(
project.id,
image_studio.ASSET_KIND_ORIGINAL,
path=cfg["db_path"],
)
first_source = self._write_image(os.path.join(temp_dir, "round-first.png"))
first_asset = image_studio.add_asset(
project.id,
"generated_main",
local_path=first_source,
parent_asset_id=source.id,
path=cfg["db_path"],
)
first_job = image_studio.create_job(
project.id,
source_asset_id=source.id,
job_type="白底图",
task_key="round-export-first",
generation_round_key="round-export",
generation_slot_index=0,
path=cfg["db_path"],
)
image_studio.update_job_status(
first_job.id,
"succeeded",
output_asset_id=first_asset.id,
path=cfg["db_path"],
)
missing_asset = image_studio.add_asset(
project.id,
"generated_main",
local_path=os.path.join(temp_dir, "missing-round.png"),
parent_asset_id=source.id,
path=cfg["db_path"],
)
missing_job = image_studio.create_job(
project.id,
source_asset_id=source.id,
job_type="场景图",
task_key="round-export-missing",
generation_round_key="round-export",
generation_slot_index=1,
path=cfg["db_path"],
)
image_studio.update_job_status(
missing_job.id,
"succeeded",
output_asset_id=missing_asset.id,
path=cfg["db_path"],
)
second_source = self._write_image(
os.path.join(temp_dir, "round-second.png"),
color=(220, 80, 40, 255),
)
second_asset = image_studio.add_asset(
project.id,
"generated_main",
local_path=second_source,
parent_asset_id=source.id,
path=cfg["db_path"],
)
second_job = image_studio.create_job(
project.id,
source_asset_id=source.id,
job_type="卖点图",
task_key="round-export-second",
generation_round_key="round-export",
generation_slot_index=2,
path=cfg["db_path"],
)
image_studio.update_job_status(
second_job.id,
"succeeded",
output_asset_id=second_asset.id,
path=cfg["db_path"],
)
parent = os.path.join(temp_dir, "exports")
os.makedirs(parent)
result = image_studio_export.export_generation_round(
project.id,
"round-export",
parent,
path=cfg["db_path"],
)
self.assertEqual(2, len(result.files))
self.assertEqual(1, result.skipped_count)
self.assertFalse(result.cancelled)
self.assertTrue(os.path.isdir(result.target_dir))
self.assertEqual(
["01_白底图.png", "03_卖点图.png"],
sorted(os.path.basename(item.output_path) for item in result.files),
)
self.assertTrue(os.path.isfile(first_source))
self.assertTrue(os.path.isfile(second_source))
with open(first_source, "rb") as source_file, open(
result.files[0].output_path,
"rb",
) as exported_file:
self.assertEqual(source_file.read(), exported_file.read())
second_result = image_studio_export.export_generation_round(
project.id,
"round-export",
parent,
path=cfg["db_path"],
)
self.assertNotEqual(result.target_dir, second_result.target_dir)
self.assertTrue(second_result.target_dir.endswith("_2"))
self.assert_removed(temp_dir)
def test_export_partial_selection_outputs_ordered_jpegs(self):
with self.make_temp_dir() as temp_dir:
cfg, project, _main, _detail = self._project_with_assets(temp_dir)
tests/test_product_suite_gui.py
+116 -9
View File
@@ -25,9 +25,13 @@ from app.gui.tabs.product_suite import (
ORIGINAL_CHECK_STATE_ROLE,
ProductOriginalDelegate,
ProductOriginalList,
ProductSuiteGlobalHistoryDialog,
ProductSuiteHistoryDialog,
ProductSuitePreviewDialog,
ProductSuiteRoundPreviewDialog,
ProductSuiteTab,
SuiteGlobalHistoryRoundRow,
SuiteGlobalHistoryThumbnail,
SuiteHistoryImageCard,
SuiteResultCard,
)
@@ -1976,7 +1980,103 @@ class ProductSuiteGuiTests(TempDirMixin, unittest.TestCase):
self.assert_removed(temp_dir)
def test_history_button_reuses_project_dialog_and_closes_with_task(self):
def test_global_history_dialog_lists_filters_and_previews_round_images(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
project, sources = self._create_project_with_assets(temp_dir, config, 1)
source = sources[0]
other_project = image_studio.create_or_get_project(
account_alias="other-shop",
account_name="副店",
account_slug="other-shop",
item_id="51100639511",
path=config["db_path"],
)
other_source = image_studio.add_asset(
other_project.id,
image_studio.ASSET_KIND_ORIGINAL,
path=config["db_path"],
)
for index in range(7):
image_path = os.path.join(temp_dir, "global-%d.png" % index)
self._write_image(image_path)
self._create_history_job(
project,
source,
config["db_path"],
round_key="main-round",
slot_index=index,
local_path=image_path,
job_type="场景图",
)
other_path = os.path.join(temp_dir, "other-global.png")
self._write_image(other_path)
self._create_history_job(
other_project,
other_source,
config["db_path"],
round_key="other-round",
slot_index=0,
local_path=other_path,
job_type="卖点图",
)
image_studio.set_current_generation_round(
project.id,
"main-round",
path=config["db_path"],
)
dialog = ProductSuiteGlobalHistoryDialog(
current_project_id=project.id,
db_path=config["db_path"],
)
dialog.show()
self.app.processEvents()
self.assertEqual(2, dialog._round_count)
rows = dialog.findChildren(SuiteGlobalHistoryRoundRow)
main_row = next(
row for row in rows if row.round_info.project_id == project.id
)
self.assertEqual(5, len(main_row.findChildren(SuiteGlobalHistoryThumbnail)))
self.assertTrue(
any(
label.text() == "+2"
for label in main_row.findChildren(QLabel)
)
)
self.assertTrue(
any(label.text() == "当前" for label in main_row.findChildren(QLabel))
)
with mock.patch.object(
ProductSuiteRoundPreviewDialog,
"exec",
return_value=0,
) as preview:
dialog._preview_round(main_row, 1)
preview.assert_called_once_with()
dialog.account_filter_edit.setText("副店")
dialog.refresh_history()
self.app.processEvents()
self.assertEqual(1, dialog._round_count)
self.assertEqual(
[other_project.id],
[row.round_info.project_id for row in dialog._history_rows],
)
dialog.account_filter_edit.clear()
dialog.current_project_checkbox.setChecked(True)
self.app.processEvents()
self.assertEqual(1, dialog._round_count)
self.assertEqual(
[project.id],
[row.round_info.project_id for row in dialog._history_rows],
)
self.assert_removed(temp_dir)
def test_history_button_reuses_global_dialog_and_keeps_it_when_task_closes(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
project, sources = self._create_project_with_assets(temp_dir, config, 1)
@@ -2014,18 +2114,20 @@ class ProductSuiteGuiTests(TempDirMixin, unittest.TestCase):
tab.open_history_dialog()
self.app.processEvents()
dialog = tab._history_dialog
self.assertIsInstance(dialog, ProductSuiteHistoryDialog)
self.assertIsInstance(dialog, ProductSuiteGlobalHistoryDialog)
self.assertEqual(project.id, dialog.current_project_id)
self.assertTrue(dialog.isVisible())
tab.open_history_dialog()
self.assertIs(dialog, tab._history_dialog)
tab.close_task(0)
self.app.processEvents()
self.assertIsNone(tab._history_dialog)
self.assertIs(dialog, tab._history_dialog)
self.assertTrue(dialog.isVisible())
self.assert_removed(temp_dir)
def test_history_button_explains_empty_project_without_opening_dialog(self):
def test_history_button_opens_global_dialog_for_empty_current_project(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
project, _ = self._create_project_with_assets(temp_dir, config, 0)
@@ -2036,13 +2138,18 @@ class ProductSuiteGuiTests(TempDirMixin, unittest.TestCase):
state.item_id = project.item_id
state.project_id = project.id
state.project_binding_state = project.binding_state
messages = []
tab._message = lambda title, message, **kwargs: messages.append((title, message))
tab.open_history_dialog()
self.app.processEvents()
self.assertEqual("暂无历史生成记录", messages[-1][0])
self.assertIsNone(tab._history_dialog)
dialog = tab._history_dialog
self.assertIsInstance(dialog, ProductSuiteGlobalHistoryDialog)
self.assertTrue(dialog.isVisible())
self.assertTrue(
any(
label.text() == "暂无套图历史生成记录，完成套图生成后会自动出现在这里"
for label in dialog.findChildren(QLabel)
)
)
self.assert_removed(temp_dir)
tests/test_workers.py
+38
View File
@@ -22,6 +22,7 @@ from app.gui.workers import (
ImageStudioPullImagesWorker,
ProductSuiteAiWriteWorker,
ProductSuiteGenerateWorker,
ProductSuiteHistoryExportWorker,
)
@@ -229,6 +230,43 @@ class WorkerTests(unittest.TestCase):
gen_title.assert_not_called()
self.assertEqual(expected, result)
def test_product_suite_history_export_worker_uses_round_export_service(self):
worker = ProductSuiteHistoryExportWorker(
7,
"round-key",
"D:/exports",
db_path="suite.db",
)
expected = SimpleNamespace(
target_dir="D:/exports/main-shop_51100639510_20260717",
files=(object(), object()),
skipped_count=1,
cancelled=False,
)
with mock.patch(
"app.gui.workers.image_studio_export.export_generation_round",
return_value=expected,
) as export_round:
result = worker.execute()
export_round.assert_called_once_with(
7,
"round-key",
"D:/exports",
path="suite.db",
should_stop=worker.should_cancel,
)
self.assertEqual(
{
"target_dir": expected.target_dir,
"file_count": 2,
"skipped_count": 1,
"cancelled": False,
},
result,
)
def test_product_suite_worker_writes_generation_round_and_stable_slots(self):
with tempfile.TemporaryDirectory() as temp_dir:
db_path = os.path.join(temp_dir, "cmshopee.db")