feat: 完成Tab①采集旧数据

新增 CollectWorker,通过后台线程逐条采集旧标题和旧封面,成功后立即 db.set_collected 写库。

采集前按别名匹配账号并检测登录态;未匹配或未登录写 skipped,单条采集失败 mark_failed 后继续。

Tab① 增加采集/停止按钮和进度状态回传;补充 GUI 测试覆盖成功采集、未登录略过和未匹配略过;同步任务文档与 progress。
This commit is contained in:
chengma
2026-06-27 11:51:42 +08:00
parent 3bf42dc11e
commit 6e00b1b093
6 changed files with 311 additions and 9 deletions
+79
View File
@@ -19,6 +19,7 @@ from PySide6.QtWidgets import QApplication, QLineEdit, QTableView
from app.gui import (
AccountDialog,
AccountsTab,
CollectWorker,
CollectTab,
MainWindow,
TAB_STYLE,
@@ -35,6 +36,7 @@ class GuiTests(TempDirMixin, unittest.TestCase):
return {
"chrome_path": "chrome.exe",
"user_data_root": os.path.join(temp_dir, "chrome_user_data_dir"),
"image_dir": os.path.join(temp_dir, "images"),
"db_path": os.path.join(temp_dir, "cmshopee.db"),
"debug_port_range": [9222, 9260],
}
@@ -303,6 +305,83 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assert_removed(temp_dir)
def test_collect_worker_collects_success_and_skips_unmatched_or_logged_out(self):
with self.make_temp_dir() as temp_dir:
cfg = self.make_config(temp_dir)
logged = accounts.create_account("主店", "alias-a", debug_port=9222, config=cfg)
logged_out = accounts.create_account("副店", "alias-b", debug_port=9223, config=cfg)
batch_id = db.create_batch(["input.xlsx"], path=cfg["db_path"])
db.insert_tasks(
batch_id,
[
{
"source_file_abs": os.path.join(temp_dir, "input.xlsx"),
"source_sheet": "商品",
"source_row": 2,
"account_name": "Excel主店",
"alias": "alias-a",
"item_id": "51100639510",
},
{
"source_file_abs": os.path.join(temp_dir, "input.xlsx"),
"source_sheet": "商品",
"source_row": 3,
"account_name": "Excel副店",
"alias": "alias-b",
"item_id": "51100639511",
},
{
"source_file_abs": os.path.join(temp_dir, "input.xlsx"),
"source_sheet": "商品",
"source_row": 4,
"account_name": "Excel未知",
"alias": "missing",
"item_id": "51100639512",
},
],
path=cfg["db_path"],
)
tasks = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
def fake_login(account, path=None, config=None):
self.assertEqual(cfg["db_path"], path)
if account.alias == logged.alias:
return {"logged_in": True, "reason": None}
if account.alias == logged_out.alias:
return {"logged_in": False, "reason": "LOGIN_PAGE"}
raise AssertionError(account.alias)
def fake_collect(account, task):
self.assertEqual(logged.alias, account.alias)
self.assertEqual("51100639510", task["item_id"])
self.assertTrue(task["old_cover_path"].endswith(
os.path.join(logged.slug, "51100639510_old.jpg")
))
return {
"old_title": "旧标题",
"old_cover_path": task["old_cover_path"],
}
with mock.patch("app.gui.accounts.detect_login", side_effect=fake_login), \
mock.patch("app.gui.editor.collect", side_effect=fake_collect) as collect:
summary = CollectWorker(tasks, db_path=cfg["db_path"], config=cfg).execute()
self.assertEqual({"ok": True, "total": 3, "done": 3, "collected": 1, "skipped": 2, "failed": 0}, summary)
collect.assert_called_once()
updated = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
by_alias = {task.alias: task for task in updated}
self.assertEqual("collected", by_alias["alias-a"].stage)
self.assertEqual("success", by_alias["alias-a"].status)
self.assertEqual("旧标题", by_alias["alias-a"].old_title)
self.assertTrue(by_alias["alias-a"].old_cover_path.endswith("51100639510_old.jpg"))
self.assertEqual("imported", by_alias["alias-b"].stage)
self.assertEqual("skipped", by_alias["alias-b"].status)
self.assertIn("账号未登录", by_alias["alias-b"].last_error)
self.assertEqual("skipped", by_alias["missing"].status)
self.assertEqual("别名未匹配账号", by_alias["missing"].last_error)
self.assert_removed(temp_dir)
if __name__ == "__main__":
unittest.main()