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

feat(collect): choose product status scope

Browse Source
This commit is contained in:
chengma
2026-07-18 16:46:59 +08:00
parent 0cf23db403
commit 4658a1ca61
11 changed files with 420 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_editor_login.py
+27
View File
@@ -971,6 +971,33 @@ class EditorLoginTests(unittest.TestCase):
steps,
)
def test_collect_normal_only_skips_non_normal_before_reading_content(self):
cdp = FakeProductCDP(
"ws-status",
alerts=[{"title": "您的商品未上架", "description": "已下架"}],
)
with mock.patch("app.editor.open_product", return_value=cdp), mock.patch(
"app.editor.read_title"
) as read_title, mock.patch("app.editor.read_cover_src") as read_cover_src, mock.patch(
"app.editor.download_cover"
) as download_cover:
result = editor.collect(
{"debug_port": 9222},
{
"item_id": "51100639510",
"old_cover_path": "old.jpg",
"collection_scope": "normal_only",
},
)
self.assertTrue(result["collection_skipped"])
self.assertEqual("unlisted", result["product_status"])
self.assertEqual("未上架，按本轮范围略过", result["collection_skip_reason"])
read_title.assert_not_called()
read_cover_src.assert_not_called()
download_cover.assert_not_called()
def test_collect_keeps_reused_product_tab_open(self):
cdp = FakeProductCDP("ws-existing")
cdp.target_id = "target-existing"
tests/test_gui.py
+47 -2
View File
@@ -28,7 +28,7 @@ from app import (
if gui.QT_IMPORT_ERROR is not None:
raise unittest.SkipTest("PySide6 未安装")
from PySide6.QtCore import QItemSelectionModel, QModelIndex, QRect, QSize, Qt
from PySide6.QtCore import QItemSelectionModel, QModelIndex, QRect, QSize, QTimer, Qt
from PySide6.QtGui import QImage, QKeyEvent, QTextCursor
from PySide6.QtWidgets import (
QAbstractItemView,
@@ -38,6 +38,7 @@ from PySide6.QtWidgets import (
QListView,
QPlainTextEdit,
QProgressBar,
QPushButton,
QTableView,
)
@@ -7804,7 +7805,11 @@ class GuiTests(TempDirMixin, unittest.TestCase):
captured["started"] = True
captured = {}
with mock.patch("app.gui.CollectWorker", FakeWorker), mock.patch(
with mock.patch.object(
tab,
"_choose_collect_scope",
return_value="normal_only",
), mock.patch("app.gui.CollectWorker", FakeWorker), mock.patch(
"app.gui.run_worker",
return_value=FakeThread(),
):
@@ -7813,12 +7818,52 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assertTrue(captured["started"])
self.assertEqual(["1001"], [task.item_id for task in captured["tasks"]])
self.assertEqual(cfg["db_path"], captured["kwargs"]["db_path"])
self.assertEqual("normal_only", captured["kwargs"]["collect_scope"])
self.assertFalse(tab.shop_filter.isEnabled())
self.assertFalse(tab.item_filter.isEnabled())
self.assertFalse(tab.status_filter.isEnabled())
self.assert_removed(temp_dir)
def test_collect_scope_dialog_uses_safe_default_and_red_all_status_choice(self):
with self.make_temp_dir() as temp_dir:
tab = CollectTab(config=self.make_config(temp_dir))
self.addCleanup(tab.close)
captured = {}
def click_all_statuses():
box = QApplication.activeModalWidget()
normal_button = box.findChild(QPushButton, "collectNormalOnlyButton")
all_button = box.findChild(QPushButton, "collectAllStatusesButton")
captured["default"] = box.defaultButton().objectName()
captured["all_style"] = all_button.styleSheet()
self.assertIsNotNone(normal_button)
self.assertIsNotNone(all_button)
all_button.click()
QTimer.singleShot(0, click_all_statuses)
scope = tab._choose_collect_scope()
self.assertEqual("all", scope)
self.assertEqual("collectNormalOnlyButton", captured["default"])
self.assertIn("#cf222e", captured["all_style"])
self.assert_removed(temp_dir)
def test_collect_without_candidates_does_not_open_scope_dialog(self):
with self.make_temp_dir() as temp_dir:
statuses = []
tab = CollectTab(config=self.make_config(temp_dir), status_callback=statuses.append)
self.addCleanup(tab.close)
with mock.patch.object(tab, "_choose_collect_scope") as choose_scope:
tab.collect_old_data()
choose_scope.assert_not_called()
self.assertEqual("没有可采集任务", statuses[-1])
self.assert_removed(temp_dir)
def test_collect_activity_tracks_step_resets_each_task_and_freezes_on_stop(self):
with self.make_temp_dir() as temp_dir:
tab = CollectTab(config=self.make_config(temp_dir))
tests/test_workers.py
+161
View File
@@ -24,6 +24,7 @@ from app.gui.workers import (
ProductSuiteAiWriteWorker,
ProductSuiteGenerateWorker,
ProductSuiteHistoryExportWorker,
CollectWorker,
)
@@ -119,6 +120,166 @@ class WorkerTests(unittest.TestCase):
self.assertEqual([(-1, "模拟失败")], failed)
self.assertEqual([{"ok": False, "error": "模拟失败"}], finished)
def test_collect_worker_scope_skips_non_normal_without_overwriting_old_content(self):
with tempfile.TemporaryDirectory() 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": index,
"account_name": "店铺",
"alias": "alias",
"item_id": str(51100639510 + index),
}
for index in range(2, 6)
],
path=db_path,
)
tasks = db.list_tasks(batch_id=batch_id, path=db_path)
connection = db.connect(db_path)
try:
with connection:
connection.execute(
"UPDATE tasks SET old_title = ?, old_cover_path = ? WHERE id = ?",
("历史标题", "history.jpg", tasks[1].id),
)
finally:
connection.close()
status_by_item = {
tasks[0].item_id: "normal",
tasks[1].item_id: "unlisted",
tasks[2].item_id: "reviewing",
tasks[3].item_id: "unknown",
}
account = SimpleNamespace(alias="alias", account_name="店铺", debug_port=9222)
def fake_collect(_account, task, on_step=None):
status = status_by_item[task["item_id"]]
self.assertEqual("normal_only", task["collection_scope"])
on_step("read_product_status")
if status != "normal":
labels = {
"unlisted": "未上架",
"reviewing": "审核中",
"unknown": "状态未知",
}
return {
"product_status": status,
"product_status_note": f"{labels[status]}提示",
"collection_skipped": True,
"collection_skip_reason": f"{labels[status]}，按本轮范围略过",
}
on_step("download_cover")
return {
"product_status": status,
"product_status_note": None,
"old_title": "新采集标题",
"old_cover_path": "new.jpg",
}
with mock.patch(
"app.gui.workers.accounts.list_accounts",
return_value=[account],
), mock.patch(
"app.gui.workers.accounts.detect_login",
return_value={"logged_in": True, "reason": None},
), mock.patch(
"app.gui.workers.editor.collect",
side_effect=fake_collect,
):
summary = CollectWorker(
tasks,
db_path=db_path,
preflight=False,
collect_scope="normal_only",
).execute()
self.assertEqual(1, summary["collected"])
self.assertEqual(3, summary["skipped"])
self.assertEqual(0, summary["failed"])
self.assertEqual(3, summary["status_scope_skipped"])
self.assertEqual(
{"normal": 1, "unlisted": 1, "reviewing": 1, "unknown": 1},
summary["product_status_counts"],
)
refreshed = db.list_tasks(batch_id=batch_id, path=db_path)
by_status = {task.product_status: task for task in refreshed}
self.assertEqual("collected", by_status["normal"].stage)
self.assertEqual("success", by_status["normal"].status)
self.assertEqual("skipped", by_status["unlisted"].status)
self.assertEqual("imported", by_status["unlisted"].stage)
self.assertEqual("历史标题", by_status["unlisted"].old_title)
self.assertEqual("history.jpg", by_status["unlisted"].old_cover_path)
def test_collect_worker_all_scope_collects_every_detected_status(self):
with tempfile.TemporaryDirectory() 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": index,
"account_name": "店铺",
"alias": "alias",
"item_id": str(51100639600 + index),
}
for index in range(2, 6)
],
path=db_path,
)
tasks = db.list_tasks(batch_id=batch_id, path=db_path)
statuses = ["normal", "unlisted", "reviewing", "unknown"]
account = SimpleNamespace(alias="alias", account_name="店铺", debug_port=9222)
def fake_collect(_account, task, on_step=None):
self.assertEqual("all", task["collection_scope"])
on_step("read_product_status")
status = statuses.pop(0)
return {
"product_status": status,
"product_status_note": None,
"old_title": f"标题{status}",
"old_cover_path": f"{status}.jpg",
}
with mock.patch(
"app.gui.workers.accounts.list_accounts",
return_value=[account],
), mock.patch(
"app.gui.workers.accounts.detect_login",
return_value={"logged_in": True, "reason": None},
), mock.patch(
"app.gui.workers.editor.collect",
side_effect=fake_collect,
):
summary = CollectWorker(
tasks,
db_path=db_path,
preflight=False,
collect_scope="all",
).execute()
self.assertEqual(4, summary["collected"])
self.assertEqual(0, summary["skipped"])
self.assertEqual(0, summary["status_scope_skipped"])
self.assertEqual(
{"normal": 1, "unlisted": 1, "reviewing": 1, "unknown": 1},
summary["product_status_counts"],
)
self.assertTrue(
all(task.stage == "collected" for task in db.list_tasks(batch_id=batch_id, path=db_path))
)
def test_run_worker_rejects_plain_object(self):
with self.assertRaises(TypeError):
run_worker(object(), start=False)