feat(product-suite): support temporary item drafts

tests/test_image_studio.py

@@ -57,6 +57,8 @@ class ImageStudioTests(TempDirMixin, unittest.TestCase):
                         "account_alias",
                         "account_slug",
                         "item_id",
+                        "storage_key",
+                        "binding_state",
                         "target_main_count",
                         "target_detail_count",
                         "suite_settings_json",
@@ -114,6 +116,8 @@ class ImageStudioTests(TempDirMixin, unittest.TestCase):
 
             self.assertEqual("{}", project.suite_settings_json)
             self.assertEqual({}, image_studio.project_suite_settings(project))
+            self.assertEqual("51100639510", project.storage_key)
+            self.assertEqual(image_studio.PROJECT_BINDING_BOUND, project.binding_state)
 
         self.assert_removed(temp_dir)
 
@@ -152,6 +156,8 @@ class ImageStudioTests(TempDirMixin, unittest.TestCase):
             self.assertEqual("alias_a_slug", project.account_slug)
             self.assertEqual("店铺A", project.account_name)
             self.assertEqual("初始提示词", project.draft_prompt)
+            self.assertEqual("51100639510", project.storage_key)
+            self.assertEqual(image_studio.PROJECT_BINDING_BOUND, project.binding_state)
 
             updated = image_studio.update_project_prompt(project.id, "二次提示词", path=db_path)
             self.assertEqual("二次提示词", updated.draft_prompt)
@@ -213,6 +219,68 @@ class ImageStudioTests(TempDirMixin, unittest.TestCase):
 
         self.assert_removed(temp_dir)
 
+    def test_draft_projects_bind_in_place_recover_and_conflict_safely(self):
+        with self.make_temp_dir() as temp_dir:
+            db_path = os.path.join(temp_dir, "cmshopee.db")
+            db.init_db(db_path)
+            account = SimpleNamespace(
+                alias="alias-a",
+                account_name="主店",
+                slug="alias_a",
+            )
+            draft = image_studio.create_draft_project(
+                account,
+                draft_prompt="临时草稿提示词",
+                path=db_path,
+            )
+
+            self.assertTrue(draft.item_id.startswith(image_studio.TEMPORARY_ITEM_PREFIX))
+            self.assertEqual(draft.item_id, draft.storage_key)
+            self.assertEqual(image_studio.PROJECT_BINDING_DRAFT, draft.binding_state)
+            self.assertFalse(image_studio.project_has_content(draft.id, path=db_path))
+            self.assertEqual([], image_studio.list_recoverable_draft_projects(path=db_path))
+            before_dirs = image_studio.project_image_dirs(os.path.join(temp_dir, "images"), draft)
+
+            original = image_studio.add_asset(
+                draft.id,
+                image_studio.ASSET_KIND_ORIGINAL,
+                local_path=os.path.join(temp_dir, "draft.png"),
+                path=db_path,
+            )
+            job = image_studio.create_job(
+                draft.id,
+                source_asset_id=original.id,
+                path=db_path,
+            )
+            image_studio.replace_selections(draft.id, "main", [original.id], path=db_path)
+            self.assertTrue(image_studio.project_has_content(draft.id, path=db_path))
+            self.assertEqual([draft.id], [project.id for project in image_studio.list_recoverable_draft_projects(path=db_path)])
+
+            conflict = image_studio.create_or_get_project(
+                account,
+                item_id="51100639510",
+                path=db_path,
+            )
+            image_studio.soft_delete_project(conflict.id, "历史项目", path=db_path)
+            with self.assertRaises(image_studio.ImageStudioProjectConflictError):
+                image_studio.bind_draft_project(draft.id, "51100639510", path=db_path)
+
+            bound = image_studio.bind_draft_project(draft.id, "51100639511", path=db_path)
+            self.assertEqual(draft.id, bound.id)
+            self.assertEqual("51100639511", bound.item_id)
+            self.assertEqual(image_studio.PROJECT_BINDING_BOUND, bound.binding_state)
+            self.assertEqual(draft.storage_key, bound.storage_key)
+            self.assertEqual(before_dirs, image_studio.project_image_dirs(os.path.join(temp_dir, "images"), bound))
+            self.assertEqual(job.id, image_studio.list_jobs(bound.id, path=db_path)[0].id)
+            self.assertEqual(original.id, image_studio.list_selections(bound.id, "main", path=db_path)[0].asset_id)
+            self.assertEqual(bound, image_studio.bind_draft_project(bound.id, "51100639511", path=db_path))
+
+            empty = image_studio.create_draft_project(account, path=db_path)
+            discarded = image_studio.discard_empty_draft_project(empty.id, path=db_path)
+            self.assertIsNotNone(discarded.deleted_at)
+
+        self.assert_removed(temp_dir)
+
     def test_asset_crud_parent_status_and_sorting(self):
         with self.make_temp_dir() as temp_dir:
             db_path = os.path.join(temp_dir, "cmshopee.db")

tests/test_image_studio_images.py

@@ -294,6 +294,48 @@ class ImageStudioImageTests(TempDirMixin, unittest.TestCase):
 
         self.assert_removed(temp_dir)
 
+    def test_draft_import_uses_stable_storage_key_after_formal_binding(self):
+        with self.make_temp_dir() as temp_dir:
+            db_path = os.path.join(temp_dir, "cmshopee.db")
+            db.init_db(db_path)
+            draft = image_studio.create_draft_project(
+                account_alias="alias",
+                account_slug="alias_slug",
+                path=db_path,
+            )
+            config = {
+                "data_dir": temp_dir,
+                "db_path": db_path,
+                "image_dir": os.path.join(temp_dir, "images"),
+            }
+
+            first = image_studio_images.import_original_bytes(
+                draft.id,
+                self._png_bytes(color=(10, 20, 30, 255)),
+                filename_hint="first.png",
+                path=db_path,
+                config=config,
+            )
+            first_path = first.local_path
+            bound = image_studio.bind_draft_project(draft.id, "51100639510", path=db_path)
+            second = image_studio_images.import_original_bytes(
+                bound.id,
+                self._png_bytes(color=(40, 50, 60, 255)),
+                filename_hint="second.png",
+                path=db_path,
+                config=config,
+            )
+
+            self.assertEqual(image_studio.PROJECT_BINDING_BOUND, bound.binding_state)
+            self.assertEqual(draft.storage_key, bound.storage_key)
+            self.assertEqual(os.path.dirname(first_path), os.path.dirname(second.local_path))
+            self.assertTrue(os.path.isfile(first_path))
+            self.assertTrue(os.path.isfile(second.local_path))
+            self.assertIn(draft.storage_key, first_path)
+            self.assertNotIn("51100639510", first_path)
+
+        self.assert_removed(temp_dir)
+
     def test_import_original_ignores_missing_history_when_enforcing_limit(self):
         with self.make_temp_dir() as temp_dir:
             db_path = os.path.join(temp_dir, "cmshopee.db")

tests/test_product_suite_gui.py

@@ -651,6 +651,148 @@ class ProductSuiteGuiTests(TempDirMixin, unittest.TestCase):
 
         self.assert_removed(temp_dir)
 
+    def test_temporary_draft_allows_local_work_but_blocks_shopee_pull_and_recovers(self):
+        with self.make_temp_dir() as temp_dir:
+            config = self._config(temp_dir)
+            accounts.create_account("主店", "alias-a", debug_port=9222, config=config)
+            tab = ProductSuiteTab(config=config, db_path=config["db_path"])
+            self.addCleanup(tab.close)
+            state = tab._displayed_state
+
+            self.assertTrue(tab.add_images_button.isEnabled())
+            self.assertFalse(tab.item_id_hint_label.isHidden())
+            self.assertEqual("请输入正确的商品ID", tab.item_id_hint_label.text())
+            with mock.patch(
+                "app.gui.tabs.product_suite.QFileDialog.getOpenFileNames",
+                return_value=([], ""),
+            ):
+                tab.choose_images()
+            self.assertEqual([], image_studio.list_projects(path=config["db_path"]))
+
+            source_path = os.path.join(temp_dir, "draft-source.png")
+            self._write_image(source_path)
+            with mock.patch.object(tab, "_start_thread", return_value=object()):
+                tab._start_import(file_paths=[source_path])
+            worker = state.import_worker
+            draft = image_studio.get_project(state.project_id, path=config["db_path"])
+            self.assertTrue(image_studio.is_draft_project(draft))
+            self.assertEqual("", state.item_id)
+            self.assertIn("临时草稿", tab.task_tabs.tabText(tab.task_tabs.currentIndex()))
+            self.assertFalse(tab.item_id_hint_label.isHidden())
+            self.assertTrue(tab.pull_button.isEnabled())
+            self.assertEqual("需要先绑定正式商品ID", tab.pull_button.toolTip())
+            tab._on_import_finished(state, worker.execute())
+            self.assertEqual(1, len(image_studio.list_assets(draft.id, path=config["db_path"])))
+
+            captured = {}
+
+            class _Signal:
+                def connect(self, callback):
+                    self.callback = callback
+
+            class _AiWriteWorker:
+                def __init__(self, instruction, context, **kwargs):
+                    captured["instruction"] = instruction
+                    captured["context"] = context
+                    self.finished = _Signal()
+                    self.cancelled = _Signal()
+                    self.failed = _Signal()
+
+                def cancel(self):
+                    pass
+
+            with mock.patch(
+                "app.gui.tabs.product_suite.ProductSuiteAiWriteWorker",
+                _AiWriteWorker,
+            ), mock.patch.object(tab, "_start_thread", return_value=object()):
+                tab.start_ai_write()
+            self.assertIn("未绑定商品", captured["context"])
+            self.assertNotIn("draft_", captured["context"])
+            state.ai_worker = None
+            state.ai_thread = None
+            state.ai_started_at = None
+            tab._apply_running_state(state)
+
+            pull_message = mock.Mock()
+            with mock.patch.object(tab, "_message", pull_message):
+                tab.pull_main_images()
+            self.assertIsNone(state.pull_worker)
+            self.assertEqual("无法拉取蝦皮主图", pull_message.call_args.args[0])
+            self.assertIn("当前为临时项目", pull_message.call_args.args[1])
+
+            tab.item_id_edit.setText("51100639510")
+            with mock.patch.object(tab, "_confirm", return_value=True):
+                tab._on_item_finished()
+            bound = image_studio.get_project(draft.id, path=config["db_path"])
+            self.assertEqual("51100639510", bound.item_id)
+            self.assertEqual(image_studio.PROJECT_BINDING_BOUND, bound.binding_state)
+            self.assertTrue(tab.item_id_hint_label.isHidden())
+            self.assertIn("套图任务", tab.task_tabs.tabText(tab.task_tabs.currentIndex()))
+
+            draft_state = tab.add_task(inherit=False)
+            draft = tab._create_draft_project(draft_state)
+            image_studio.add_asset(
+                draft.id,
+                image_studio.ASSET_KIND_ORIGINAL,
+                local_path=source_path,
+                path=config["db_path"],
+            )
+            draft_state.ai_worker = mock.Mock()
+            with mock.patch.object(tab, "_draft_close_action", return_value="cancel"):
+                tab.close_task(tab.task_tabs.currentIndex())
+            draft_state.ai_worker.cancel.assert_not_called()
+            self.assertIn(draft_state.key, tab._states)
+            draft_state.ai_worker = None
+            with mock.patch.object(tab, "_draft_close_action", return_value="keep"):
+                tab.close_task(tab.task_tabs.currentIndex())
+            self.assertIsNotNone(image_studio.get_project(draft.id, path=config["db_path"]))
+
+            tab.close()
+            restored = ProductSuiteTab(config=config, db_path=config["db_path"])
+            self.addCleanup(restored.close)
+            restored_state = next(
+                state for state in restored._states.values() if state.project_id == draft.id
+            )
+            self.assertEqual("", restored_state.item_id)
+            self.assertTrue(restored._is_draft_state(restored_state))
+            restored_index = next(
+                index
+                for index in range(restored.task_tabs.count())
+                if restored.task_tabs.tabData(index) == restored_state.key
+            )
+            self.assertIn("临时草稿", restored.task_tabs.tabText(restored_index))
+
+        self.assert_removed(temp_dir)
+
+    def test_first_failed_import_discards_new_empty_temporary_draft(self):
+        with self.make_temp_dir() as temp_dir:
+            config = self._config(temp_dir)
+            accounts.create_account("主店", "alias-a", debug_port=9222, config=config)
+            tab = ProductSuiteTab(config=config, db_path=config["db_path"])
+            self.addCleanup(tab.close)
+            bad_path = os.path.join(temp_dir, "not-an-image.png")
+            with open(bad_path, "wb") as handle:
+                handle.write(b"not an image")
+
+            state = tab._displayed_state
+            with mock.patch.object(tab, "_start_thread", return_value=object()):
+                tab._start_import(file_paths=[bad_path])
+            worker = state.import_worker
+            draft_id = state.project_id
+            with mock.patch.object(tab, "_message"):
+                tab._on_import_finished(state, worker.execute())
+
+            discarded = image_studio.get_project(
+                draft_id,
+                path=config["db_path"],
+                include_deleted=True,
+            )
+            self.assertIsNotNone(discarded.deleted_at)
+            self.assertIsNone(state.project_id)
+            self.assertEqual([], image_studio.list_recoverable_draft_projects(path=config["db_path"]))
+
+        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)