feat(product-suite): add grouped generation history

This commit is contained in:
chengma
2026-07-16 23:47:47 +08:00
parent f3defdeb95
commit cd150ef6c7
8 changed files with 879 additions and 35 deletions
+36
View File
@@ -981,6 +981,42 @@ class ImageStudioTests(TempDirMixin, unittest.TestCase):
[failed_round, round_one, None],
[round_.generation_round_key for round_ in rounds],
)
self.assertEqual(
[failed_round],
[
round_.generation_round_key
for round_ in image_studio.list_generation_rounds(
project.id,
limit=1,
offset=0,
path=db_path,
)
],
)
self.assertEqual(
[round_one],
[
round_.generation_round_key
for round_ in image_studio.list_generation_rounds(
project.id,
limit=1,
offset=1,
path=db_path,
)
],
)
self.assertEqual(
[None],
[
round_.generation_round_key
for round_ in image_studio.list_generation_rounds(
project.id,
limit=1,
offset=2,
path=db_path,
)
],
)
summary = rounds[1]
self.assertTrue(summary.is_current)
self.assertEqual(3, summary.job_count)
+304 -11
View File
@@ -25,7 +25,10 @@ from app.gui.tabs.product_suite import (
ORIGINAL_CHECK_STATE_ROLE,
ProductOriginalDelegate,
ProductOriginalList,
ProductSuiteHistoryDialog,
ProductSuitePreviewDialog,
ProductSuiteTab,
SuiteHistoryImageCard,
SuiteResultCard,
)
from app.gui.product_suite_prompt_dialog import ProductSuitePromptDialog
@@ -85,6 +88,44 @@ class ProductSuiteGuiTests(TempDirMixin, unittest.TestCase):
)
return project, assets
def _create_history_job(
self,
project,
source,
db_path,
*,
round_key=None,
slot_index=None,
status="succeeded",
local_path=None,
job_type="白底图",
):
asset = None
if local_path is not None:
asset = image_studio.add_asset(
project.id,
"generated_main",
local_path=local_path,
parent_asset_id=source.id,
path=db_path,
)
job = image_studio.create_job(
project.id,
source_asset_id=source.id,
job_type=job_type,
prompt="历史图片",
generation_round_key=round_key,
generation_slot_index=slot_index,
path=db_path,
)
return image_studio.update_job_status(
job.id,
status,
error="测试失败" if status == "failed" else None,
output_asset_id=asset.id if asset is not None else None,
path=db_path,
)
def test_tab_builds_suite_controls_without_old_detail_workspace(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
@@ -1349,13 +1390,6 @@ class ProductSuiteGuiTests(TempDirMixin, unittest.TestCase):
)
self.assertEqual("图片重试成功", messages[-1][0])
state.show_history = True
history_ids = {job.id for job in tab._jobs_for_state(state)}
self.assertEqual(
{success_job.id, failed_job.id, retry_job.id},
history_ids,
)
self.assert_removed(temp_dir)
def test_generation_round_restores_after_project_rebind_and_retry_keeps_slot(self):
@@ -1572,7 +1606,7 @@ class ProductSuiteGuiTests(TempDirMixin, unittest.TestCase):
self.assert_removed(temp_dir)
def test_retry_tracks_only_new_job_and_preserves_history_view(self):
def test_retry_tracks_only_new_job_and_keeps_current_results(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)
@@ -1670,7 +1704,6 @@ class ProductSuiteGuiTests(TempDirMixin, unittest.TestCase):
)
)
state.show_history = True
tab._load_state(state)
with mock.patch.object(
tab,
@@ -1690,14 +1723,274 @@ class ProductSuiteGuiTests(TempDirMixin, unittest.TestCase):
retry_job_id=failed_job.id,
)
)
self.assertTrue(state.show_history)
self.assertTrue(tab.history_button.isChecked())
self.assertFalse(tab.history_button.isCheckable())
self.assertEqual("历史生成", tab.history_button.text())
state.worker = None
state.thread = None
state.generation_run_token = ""
self.assert_removed(temp_dir)
def test_history_dialog_groups_current_round_retries_and_paginates(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]
db_path = config["db_path"]
other_project = image_studio.create_or_get_project(
account_alias="其他店",
account_slug="other-shop",
item_id="51100639511",
path=db_path,
)
other_source = image_studio.add_asset(
other_project.id,
image_studio.ASSET_KIND_ORIGINAL,
path=db_path,
)
other_job = self._create_history_job(
other_project,
other_source,
db_path,
round_key="other-round",
slot_index=0,
)
legacy_job = self._create_history_job(project, source, db_path)
for index in range(20):
self._create_history_job(
project,
source,
db_path,
round_key="old-round-%02d" % index,
slot_index=0,
job_type="场景图",
)
current_key = "current-round"
self._create_history_job(
project,
source,
db_path,
round_key=current_key,
slot_index=0,
status="failed",
)
missing_path = os.path.join(temp_dir, "missing-history-image.png")
retry_job = self._create_history_job(
project,
source,
db_path,
round_key=current_key,
slot_index=0,
local_path=missing_path,
)
usable_path = os.path.join(temp_dir, "usable-history-image.png")
self._write_image(usable_path)
usable_job = self._create_history_job(
project,
source,
db_path,
round_key=current_key,
slot_index=1,
local_path=usable_path,
job_type="卖点图",
)
image_studio.set_current_generation_round(
project.id,
current_key,
path=db_path,
)
dialog = ProductSuiteHistoryDialog(project.id, db_path=db_path)
dialog.show()
self.app.processEvents()
self.assertIn("店铺:主店", dialog.context_label.text())
self.assertIn("商品ID:51100639510", dialog.context_label.text())
self.assertEqual(20, dialog._round_count)
self.assertEqual(1, dialog._available_image_count)
self.assertTrue(dialog.load_more_button.isVisible())
self.assertTrue(
any(label.text() == "当前" for label in dialog.findChildren(QLabel))
)
cards = dialog.findChildren(SuiteHistoryImageCard)
cards_by_job = {card.job.id: card for card in cards}
self.assertIn(retry_job.id, cards_by_job)
self.assertIn(usable_job.id, cards_by_job)
self.assertNotIn(other_job.id, cards_by_job)
self.assertNotIn(legacy_job.id, cards_by_job)
self.assertIn("本槽位已重试1次", cards_by_job[retry_job.id].toolTip())
self.assertIn("本地图片文件不可用", cards_by_job[retry_job.id].toolTip())
dialog.load_more()
self.app.processEvents()
self.assertEqual(22, dialog._round_count)
self.assertFalse(dialog.load_more_button.isVisible())
self.assertTrue(
any(
label.text() == "旧版历史记录"
for label in dialog.findChildren(QLabel)
)
)
cards_by_job = {
card.job.id: card
for card in dialog.findChildren(SuiteHistoryImageCard)
}
self.assertIn(legacy_job.id, cards_by_job)
self.assert_removed(temp_dir)
def test_history_dialog_actions_are_read_only(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)
image_path = os.path.join(temp_dir, "history-image.png")
self._write_image(image_path)
job = self._create_history_job(
project,
sources[0],
config["db_path"],
round_key="current-round",
slot_index=0,
local_path=image_path,
)
image_studio.set_current_generation_round(
project.id,
"current-round",
path=config["db_path"],
)
asset = image_studio.get_asset(job.output_asset_id, path=config["db_path"])
dialog = ProductSuiteHistoryDialog(project.id, db_path=config["db_path"])
with mock.patch.object(ProductSuitePreviewDialog, "exec", return_value=0) as preview:
dialog._preview_job(job, asset)
preview.assert_called_once_with()
class MenuAction:
def __init__(self, text):
self._text = text
def text(self):
return self._text
class MenuStub:
selected_text = ""
observed_actions = []
def __init__(self, parent=None):
self._actions = []
def addAction(self, text):
action = MenuAction(text)
self._actions.append(action)
return action
def exec(self, _position):
type(self).observed_actions.extend(
action.text() for action in self._actions
)
return next(
(
action
for action in self._actions
if action.text() == type(self).selected_text
),
None,
)
with mock.patch("app.gui.tabs.product_suite.QMenu", MenuStub):
dialog._show_job_menu(job, asset, None)
self.assertEqual(
["预览", "复制路径", "打开所在文件夹"],
MenuStub.observed_actions,
)
MenuStub.selected_text = "复制路径"
with mock.patch("app.gui.tabs.product_suite.QMenu", MenuStub):
dialog._show_job_menu(job, asset, None)
self.assertEqual(image_path, QApplication.clipboard().text())
MenuStub.selected_text = "打开所在文件夹"
with mock.patch("app.gui.tabs.product_suite.QMenu", MenuStub), mock.patch(
"app.gui.tabs.product_suite.file_manager.open_in_file_manager"
) as open_folder:
dialog._show_job_menu(job, asset, None)
open_folder.assert_called_once_with(os.path.dirname(image_path))
self.assert_removed(temp_dir)
def test_history_button_reuses_project_dialog_and_closes_with_task(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)
image_path = os.path.join(temp_dir, "current-image.png")
self._write_image(image_path)
job = self._create_history_job(
project,
sources[0],
config["db_path"],
round_key="current-round",
slot_index=0,
local_path=image_path,
)
image_studio.set_current_generation_round(
project.id,
"current-round",
path=config["db_path"],
)
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 = project.item_id
state.project_id = project.id
state.project_binding_state = project.binding_state
state.current_generation_round_key = "current-round"
state.current_job_ids = [job.id]
tab._load_state(state)
tab.show()
self.app.processEvents()
self.assertFalse(tab.history_button.isCheckable())
self.assertEqual([job.id], [entry.id for entry in tab._jobs_for_state(state)])
tab.open_history_dialog()
self.app.processEvents()
dialog = tab._history_dialog
self.assertIsInstance(dialog, ProductSuiteHistoryDialog)
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.assert_removed(temp_dir)
def test_history_button_explains_empty_project_without_opening_dialog(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
project, _ = 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 = 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.assertEqual("暂无历史生成记录", messages[-1][0])
self.assertIsNone(tab._history_dialog)
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)