From 0abd7ffdff2e62b97796f8117bbbb77e001fa903 Mon Sep 17 00:00:00 2001 From: chengma Date: Tue, 14 Jul 2026 11:10:36 +0800 Subject: [PATCH] feat(collect): show per-item collection activity timer --- app/gui/tabs/collect.py | 219 +++++++++++++++++++++++++++++++++++++++- app/gui/workers.py | 116 ++++++++++++++++++++- docs/tasks/T-628.md | 9 +- tests/test_gui.py | 159 +++++++++++++++++++++++++++-- 4 files changed, 494 insertions(+), 9 deletions(-) diff --git a/app/gui/tabs/collect.py b/app/gui/tabs/collect.py index e65b852..cdc137b 100644 --- a/app/gui/tabs/collect.py +++ b/app/gui/tabs/collect.py @@ -20,6 +20,35 @@ def CollectWorker(*args, **kwargs): def WriteBackWorker(*args, **kwargs): return _call_package_attr("WriteBackWorker", _RealWriteBackWorker, *args, **kwargs) + +COLLECT_ACTIVITY_STEP_LABELS = { + "preflight": "检查账号", + "match_account": "匹配账号", + "check_login": "检查账号登录", + "prepare_task": "准备采集", + "open_product": "打开商品页", + "wait_ready": "等待商品页加载", + "read_title": "读取标题", + "read_cover": "读取封面", + "download_cover": "下载封面", + "save_result": "保存采集结果", + "db_write": "保存采集结果", +} + + +def _collect_activity_step_label(step): + return COLLECT_ACTIVITY_STEP_LABELS.get(str(step or ""), "处理当前商品") + + +def _format_collect_elapsed(seconds): + total = max(0, int(seconds or 0)) + hours, remainder = divmod(total, 3600) + minutes, seconds = divmod(remainder, 60) + if hours: + return f"{hours}:{minutes:02d}:{seconds:02d}" + return f"{minutes:02d}:{seconds:02d}" + + class CollectTab(QWidget): """Tab 1: import Excel files and list imported tasks.""" @@ -56,6 +85,15 @@ class CollectTab(QWidget): self.write_back_worker = None self.write_back_thread = None self.last_collect_run_id = None + self._collect_run_started_at = None + self._collect_task_started_at = None + self._collect_task_elapsed_seconds = 0 + self._collect_activity_payload = {} + self._collect_terminal_text = "" + self._collect_stop_requested = False + self._collect_elapsed_timer = QTimer(self) + self._collect_elapsed_timer.setInterval(1000) + self._collect_elapsed_timer.timeout.connect(self._refresh_collect_activity) self.import_button = QPushButton("导入 Excel...") self.refresh_button = QPushButton("刷新") @@ -101,6 +139,14 @@ class CollectTab(QWidget): self.summary_label = QLabel("未导入任务") self.summary_label.setTextFormat(Qt.RichText) self.batch_progress_label = _build_batch_progress_overview("collectBatchProgressOverview") + self.collect_activity_label = QLabel("") + self.collect_activity_label.setObjectName("collectActivityLabel") + self.collect_activity_label.setAlignment(Qt.AlignRight | Qt.AlignVCenter) + activity_sample = "正在采集 999/999 · 等待商品页加载 · 本条 99:59" + activity_width = self.collect_activity_label.fontMetrics().horizontalAdvance(activity_sample) + 24 + self.collect_activity_label.setFixedWidth(activity_width) + self.collect_activity_label.setVisible(False) + self._set_collect_activity_style("muted") self.match_detail_label = QLabel("") self.show_all_button = QPushButton("全部") self.show_unmatched_button = QPushButton("未匹配(0)") @@ -112,6 +158,12 @@ class CollectTab(QWidget): summary_layout.addWidget(self.show_all_button) summary_layout.addWidget(self.show_unmatched_button) + batch_progress_layout = QHBoxLayout() + batch_progress_layout.setContentsMargins(0, 0, 0, 0) + batch_progress_layout.setSpacing(8) + batch_progress_layout.addWidget(self.batch_progress_label, 1) + batch_progress_layout.addWidget(self.collect_activity_label) + self.table = QTableView() self.model = TaskTableModel(self.table) self.table.setModel(self.model) @@ -142,7 +194,7 @@ class CollectTab(QWidget): layout.addLayout(filter_layout) layout.addLayout(summary_layout) layout.addWidget(self.match_detail_label) - layout.addWidget(self.batch_progress_label) + layout.addLayout(batch_progress_layout) layout.addWidget(self.empty_state_card) layout.addWidget(self.table, 1) layout.addWidget(QLabel("采集运行日志")) @@ -168,6 +220,156 @@ class CollectTab(QWidget): def _set_status(self, message, level=None): _emit_status(self.status_callback, message, level=level) + def _set_collect_activity_style(self, level): + color = _status_level_color(level) + self.collect_activity_label.setStyleSheet( + "QLabel#collectActivityLabel { " + "background: #f6f8fa; border: 1px solid #d0d7de; " + "border-radius: 6px; padding: 8px 10px; " + f"color: {color}; font-weight: 600; " + "}" + ) + + def _start_collect_activity(self): + now = time.monotonic() + self._collect_run_started_at = now + self._collect_task_started_at = None + self._collect_task_elapsed_seconds = 0 + self._collect_activity_payload = { + "state": "preflight_started", + "total": 0, + "step": "preflight", + } + self._collect_terminal_text = "" + self._collect_stop_requested = False + self._refresh_collect_activity(now=now) + if not self._collect_elapsed_timer.isActive(): + self._collect_elapsed_timer.start() + + def _on_collect_activity(self, payload): + event = dict(payload or {}) + state = str(event.get("state") or "") + now = time.monotonic() + if self._collect_run_started_at is None: + self._collect_run_started_at = now + self._collect_terminal_text = "" + if not self._collect_elapsed_timer.isActive(): + self._collect_elapsed_timer.start() + + if state == "preflight_started": + self._collect_activity_payload = event + elif state == "task_started": + self._collect_task_started_at = now + self._collect_task_elapsed_seconds = 0 + self._collect_activity_payload = event + elif state in {"task_step", "task_finished"}: + current = dict(self._collect_activity_payload) + current.update(event) + self._collect_activity_payload = current + if state == "task_finished" and self._collect_task_started_at is not None: + self._collect_task_elapsed_seconds = max( + 0, + int(now - self._collect_task_started_at), + ) + self._collect_task_started_at = None + self._refresh_collect_activity(now=now) + + def _refresh_collect_activity(self, now=None): + if self._collect_terminal_text: + self.collect_activity_label.setVisible(True) + self.collect_activity_label.setText(self._collect_terminal_text) + return + if self._collect_run_started_at is None: + self.collect_activity_label.setVisible(False) + return + + current_time = time.monotonic() if now is None else now + run_elapsed = max(0, int(current_time - self._collect_run_started_at)) + if self._collect_task_started_at is not None: + task_elapsed = max(0, int(current_time - self._collect_task_started_at)) + else: + task_elapsed = self._collect_task_elapsed_seconds + event = self._collect_activity_payload + state = str(event.get("state") or "") + index = int(event.get("index") or 0) + total = int(event.get("total") or 0) + progress = f"{index}/{total}" if index and total else str(index or total or "") + step_label = _collect_activity_step_label(event.get("step")) + + if self._collect_stop_requested: + if index: + text = f"正在停止 · 本条 {_format_collect_elapsed(task_elapsed)}" + else: + text = f"正在停止 · 已等待 {_format_collect_elapsed(run_elapsed)}" + level = "warning" + elif state in {"task_started", "task_step"}: + text = ( + f"正在采集 {progress} · {step_label} · " + f"本条 {_format_collect_elapsed(task_elapsed)}" + ) + level = "info" + elif state == "task_finished": + result_label = { + "success": "已完成", + "failed": "失败", + "skipped": "略过", + }.get(str(event.get("result") or ""), "已结束") + text = ( + f"第 {progress} 条 · {result_label} · " + f"本条 {_format_collect_elapsed(task_elapsed)}" + ) + level = "danger" if event.get("result") == "failed" else "muted" + else: + text = f"正在检查账号 · {_format_collect_elapsed(run_elapsed)}" + level = "info" + + tooltip_parts = [] + if event.get("item_id"): + tooltip_parts.append(f"商品ID：{event.get('item_id')}") + if event.get("alias"): + tooltip_parts.append(f"账号别名：{event.get('alias')}") + if state in {"task_started", "task_step"}: + tooltip_parts.append(f"当前阶段：{step_label}") + self.collect_activity_label.setToolTip("\n".join(tooltip_parts)) + self._set_collect_activity_style(level) + self.collect_activity_label.setText(text) + self.collect_activity_label.setVisible(True) + + def _finish_collect_activity(self, outcome, payload=None): + now = time.monotonic() + if self._collect_run_started_at is None: + total_elapsed = 0 + else: + total_elapsed = max(0, int(now - self._collect_run_started_at)) + self._collect_elapsed_timer.stop() + self._collect_run_started_at = None + self._collect_task_started_at = None + self._collect_task_elapsed_seconds = 0 + self._collect_activity_payload = {} + self._collect_stop_requested = False + + if outcome == "blocked": + text = "采集未开始 · 检查未通过" + level = "warning" + tooltip = "采集前检查未通过，请按弹窗提示处理" + elif outcome == "cancelled": + text = f"采集已停止 · 总用时 {_format_collect_elapsed(total_elapsed)}" + level = "warning" + tooltip = "本轮采集已停止" + elif outcome == "error": + text = "采集已结束 · 请查看运行日志" + level = "danger" + tooltip = "采集异常结束，请查看下方采集运行日志" + else: + text = f"采集完成 · 总用时 {_format_collect_elapsed(total_elapsed)}" + level = "success" + tooltip = "本轮采集已经完成" + self._collect_terminal_text = text + self.collect_activity_label.setToolTip(tooltip) + self._set_collect_activity_style(level) + self.collect_activity_label.setText(text) + self.collect_activity_label.setVisible(True) + def _on_collect_log(self, message): self._append_collect_log(message) self._set_status(message) @@ -522,6 +724,9 @@ class CollectTab(QWidget): config=self.config, diagnostic_log_dir=diagnostics.DEFAULT_LOG_DIR, ) + activity_signal = getattr(worker, "activity", None) + if activity_signal is not None: + activity_signal.connect(self._on_collect_activity) worker.progress.connect(self._on_collect_progress) worker.row_updated.connect(self._on_collect_row_updated) worker.log.connect(self._on_collect_log) @@ -534,11 +739,14 @@ class CollectTab(QWidget): self.collect_worker = worker self.collect_thread = thread self._set_collect_running(True) + self._start_collect_activity() thread.start() def stop_collect(self, checked=False): if self.collect_worker is not None: self.collect_worker.cancel() + self._collect_stop_requested = True + self._refresh_collect_activity() self._set_status("正在停止采集...") def write_back_old_data(self, checked=False): @@ -616,6 +824,8 @@ class CollectTab(QWidget): def _forget_collect_thread(self, thread): if self.collect_thread is thread: + if self._collect_elapsed_timer.isActive(): + self._finish_collect_activity("error") self.collect_thread = None self.collect_worker = None @@ -643,6 +853,12 @@ class CollectTab(QWidget): def _on_collect_finished(self, payload): self._set_collect_running(False) + if payload.get("blocked"): + self._finish_collect_activity("blocked", payload) + elif payload.get("error"): + self._finish_collect_activity("error", payload) + else: + self._finish_collect_activity("finished", payload) self.last_collect_run_id = payload.get("run_id") or self.last_collect_run_id self.refresh_tasks() self._load_latest_collect_run_log() @@ -747,6 +963,7 @@ class CollectTab(QWidget): def _on_collect_cancelled(self, payload): self._set_collect_running(False) + self._finish_collect_activity("cancelled", payload) self.refresh_tasks() self._set_status( "采集已停止：完成{done}/{total}".format( diff --git a/app/gui/workers.py b/app/gui/workers.py index f00e9d5..c2426de 100644 --- a/app/gui/workers.py +++ b/app/gui/workers.py @@ -7,6 +7,11 @@ import re import threading import time +try: + from PySide6.QtCore import Signal +except ModuleNotFoundError: # pragma: no cover - GUI import guard + Signal = None + from .. import ai, image_studio, image_studio_export, image_studio_generation, image_studio_images from ..collect_skip import ALIAS_UNMATCHED, LOGIN_REQUIRED, empty_skip_reason_counts from .widgets import * @@ -1743,6 +1748,9 @@ class ApplyWorker(BaseWorker): class CollectWorker(BaseWorker): """Collect old title and cover for imported tasks.""" + if Signal is not None: + activity = Signal(dict) + LOGIN_CHECK_ATTEMPTS = 3 LOGIN_CHECK_RETRY_DELAY_SECONDS = 2.0 @@ -1785,6 +1793,11 @@ class CollectWorker(BaseWorker): skip_reason_counts = empty_skip_reason_counts() self._run_id = self._create_run_log(eligible, batch_ids) + self._emit_activity( + "preflight_started", + total=total, + step="preflight", + ) self._log_run_event( f"step=preflight result=start detail=采集运行开始 total={total}" ) @@ -1822,9 +1835,16 @@ class CollectWorker(BaseWorker): level="warning", ) - for task in eligible: + for index, task in enumerate(eligible, start=1): if self.should_cancel(): break + self._emit_activity( + "task_started", + task=task, + index=index, + total=total, + step="match_account", + ) account = account_by_alias.get(str(task.alias).strip()) if account is None: skipped += 1 @@ -1842,6 +1862,14 @@ class CollectWorker(BaseWorker): task=task, level="warning", ) + self._emit_activity( + "task_finished", + task=task, + index=index, + total=total, + step="match_account", + result="skipped", + ) self._emit_progress(done, total, collected, skipped, failed) continue @@ -1862,9 +1890,24 @@ class CollectWorker(BaseWorker): task=task, level="warning", ) + self._emit_activity( + "task_finished", + task=task, + index=index, + total=total, + step="check_login", + result="skipped", + ) self._emit_progress(done, total, collected, skipped, failed) continue + self._emit_activity( + "task_step", + task=task, + index=index, + total=total, + step="check_login", + ) status = self._confirmed_login_status(account, context="midrun", task=task) if self._is_definitive_logged_out(status): alias = str(task.alias).strip() @@ -1885,6 +1928,14 @@ class CollectWorker(BaseWorker): task=task, level="warning", ) + self._emit_activity( + "task_finished", + task=task, + index=index, + total=total, + step="check_login", + result="skipped", + ) self._emit_progress(done, total, collected, skipped, failed) continue if not status.get("logged_in"): @@ -1900,10 +1951,18 @@ class CollectWorker(BaseWorker): started = time.monotonic() current_step = "db_write" + activity_result = "success" def on_step(step): nonlocal current_step current_step = str(step) + self._emit_activity( + "task_step", + task=task, + index=index, + total=total, + step=current_step, + ) self._log_run_event( "step={step} result=start detail=任务 {task_id} 商品 {item_id}".format( step=current_step, @@ -1914,6 +1973,13 @@ class CollectWorker(BaseWorker): ) try: + self._emit_activity( + "task_step", + task=task, + index=index, + total=total, + step="prepare_task", + ) self._log_run_event( "step=db_write result=start detail=任务 {task_id} 商品 {item_id} 标记采集运行".format( task_id=task.id, @@ -1932,6 +1998,13 @@ class CollectWorker(BaseWorker): on_step=on_step, ) current_step = "db_write" + self._emit_activity( + "task_step", + task=task, + index=index, + total=total, + step="save_result", + ) self._log_run_event( "step=db_write result=start detail=任务 {task_id} 商品 {item_id} 保存采集结果".format( task_id=task.id, @@ -1965,6 +2038,7 @@ class CollectWorker(BaseWorker): task=task, ) except Exception as exc: + activity_result = "failed" failed += 1 error = str(exc) or exc.__class__.__name__ safe_error = diagnostics.redact_log_text(error) @@ -1993,6 +2067,14 @@ class CollectWorker(BaseWorker): ) finally: done += 1 + self._emit_activity( + "task_finished", + task=task, + index=index, + total=total, + step=current_step, + result=activity_result, + ) self._emit_progress(done, total, collected, skipped, failed) summary = self._summary( @@ -2115,6 +2197,38 @@ class CollectWorker(BaseWorker): payload["reason"] = reason return payload + def _emit_activity( + self, + state, + *, + task=None, + index=0, + total=0, + step=None, + result=None, + ): + signal = getattr(self, "activity", None) + if signal is None: + return + payload = { + "state": str(state), + "index": int(index or 0), + "total": int(total or 0), + } + if task is not None: + payload.update( + { + "task_id": getattr(task, "id", None), + "item_id": str(getattr(task, "item_id", "") or ""), + "alias": str(getattr(task, "alias", "") or ""), + } + ) + if step: + payload["step"] = str(step) + if result: + payload["result"] = str(result) + signal.emit(payload) + def _emit_progress(self, done, total, collected, skipped, failed): self.progress.emit( { diff --git a/docs/tasks/T-628.md b/docs/tasks/T-628.md index 9c8274d..c6c947f 100644 --- a/docs/tasks/T-628.md +++ b/docs/tasks/T-628.md @@ -3,7 +3,7 @@ id: T-628 title: 采集当前记录阶段与单条耗时指示器 phase: 2 deps: [T-620] -status: TODO +status: DONE created: 2026-07-14 --- @@ -86,3 +86,10 @@ created: 2026-07-14 - 不修改任务 `stage/status`、SQLite schema、Excel schema、回写流程或批次统计口径。 - 不把计时器放进 Worker 线程，不让 GUI 每秒查询 SQLite，也不把每秒 tick 写入运行日志。 - 不改变②AI生成、③更新蝦皮及⑥商品套图的计时显示。 + +## 执行记录 + +- 2026-07-14：`CollectWorker` 新增独立 `activity` 结构化信号，在前置检查、每条记录开始、采集步骤变化及单条结束时发送稳定 payload；每条记录进入循环即发 `task_started`，登录检查和 `editor.collect()` 步骤只更新阶段，不改变原有进度、日志、状态写库和终态信号。 +- 2026-07-14：①「导入采集」在批次进度同行右侧新增固定宽度运行指示器，显示前置检查、`N/总数`、当前中文阶段和本条 `MM:SS`；基于 `time.monotonic()` 每秒刷新，每条开始重置，步骤变化不重置，停止请求期间继续显示实际等待，完成、阻断、停止和异常后冻结中文终态。 +- 2026-07-14：新增 Worker 活动事件顺序和 GUI 计时生命周期测试，覆盖三条记录的成功/略过事件、步骤不重置、下一条归零、停止等待、终态冻结、小时格式及中文 tooltip；用户可见终态不展示底层英文异常。 +- 验证：当前工作区全套 GUI 测试除任务开始前已存在的默认封面提示词 `papa1.txt` 改名造成的 2 项旧断言外，采集相关测试全部通过；在仅覆盖 T-628 四个文件的隔离 worktree 中运行 `python -m ruff check app tests main.py`、`py -3.10 -m compileall app main.py`、`py -3.10 -m unittest discover -s tests`（461 项通过）及 `git diff --check`，结果全部通过。 diff --git a/tests/test_gui.py b/tests/test_gui.py index ddf278e..d9189f1 100644 --- a/tests/test_gui.py +++ b/tests/test_gui.py @@ -65,6 +65,7 @@ from app.gui import ( ) from app.gui import file_manager import app.gui.workers as gui_workers +from app.gui.tabs.collect import _format_collect_elapsed from app.gui.tabs.generate import CoverGalleryDialog, OriginalImageDialog from app.gui.main_window import _fit_and_center_window @@ -7391,6 +7392,130 @@ class GuiTests(TempDirMixin, unittest.TestCase): 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)) + self.addCleanup(tab.close) + + self.assertEqual("collectActivityLabel", tab.collect_activity_label.objectName()) + self.assertTrue(tab.collect_activity_label.isHidden()) + self.assertEqual("1:01:01", _format_collect_elapsed(3661)) + + with mock.patch("app.gui.tabs.collect.time.monotonic", return_value=100.0): + tab._start_collect_activity() + self.assertTrue(tab._collect_elapsed_timer.isActive()) + self.assertIn("正在检查账号", tab.collect_activity_label.text()) + self.assertIn("00:00", tab.collect_activity_label.text()) + + with mock.patch("app.gui.tabs.collect.time.monotonic", return_value=105.0): + tab._on_collect_activity( + { + "state": "preflight_started", + "total": 2, + "step": "preflight", + } + ) + self.assertIn("正在检查账号", tab.collect_activity_label.text()) + self.assertIn("00:05", tab.collect_activity_label.text()) + + with mock.patch("app.gui.tabs.collect.time.monotonic", return_value=110.0): + tab._on_collect_activity( + { + "state": "task_started", + "index": 1, + "total": 2, + "task_id": 1, + "item_id": "51100639510", + "alias": "alias-a", + "step": "match_account", + } + ) + self.assertEqual( + "正在采集 1/2 · 匹配账号 · 本条 00:00", + tab.collect_activity_label.text(), + ) + + with mock.patch("app.gui.tabs.collect.time.monotonic", return_value=117.0): + tab._on_collect_activity( + { + "state": "task_step", + "index": 1, + "total": 2, + "task_id": 1, + "item_id": "51100639510", + "alias": "alias-a", + "step": "wait_ready", + } + ) + self.assertEqual( + "正在采集 1/2 · 等待商品页加载 · 本条 00:07", + tab.collect_activity_label.text(), + ) + self.assertIn("商品ID：51100639510", tab.collect_activity_label.toolTip()) + self.assertIn("当前阶段：等待商品页加载", tab.collect_activity_label.toolTip()) + + with mock.patch("app.gui.tabs.collect.time.monotonic", return_value=120.0): + tab._on_collect_activity( + { + "state": "task_started", + "index": 2, + "total": 2, + "task_id": 2, + "item_id": "51100639511", + "alias": "alias-a", + "step": "check_login", + } + ) + self.assertEqual( + "正在采集 2/2 · 检查账号登录 · 本条 00:00", + tab.collect_activity_label.text(), + ) + + with mock.patch("app.gui.tabs.collect.time.monotonic", return_value=125.0): + tab._collect_stop_requested = True + tab._refresh_collect_activity() + self.assertEqual("正在停止 · 本条 00:05", tab.collect_activity_label.text()) + + with mock.patch("app.gui.tabs.collect.time.monotonic", return_value=130.0): + tab._finish_collect_activity("cancelled") + self.assertEqual("采集已停止 · 总用时 00:30", tab.collect_activity_label.text()) + self.assertFalse(tab._collect_elapsed_timer.isActive()) + + frozen_text = tab.collect_activity_label.text() + with mock.patch("app.gui.tabs.collect.time.monotonic", return_value=999.0): + tab._refresh_collect_activity() + self.assertEqual(frozen_text, tab.collect_activity_label.text()) + + self.assert_removed(temp_dir) + + def test_collect_activity_terminal_states_use_chinese_semantics(self): + with self.make_temp_dir() as temp_dir: + tab = CollectTab(config=self.make_config(temp_dir)) + self.addCleanup(tab.close) + + with mock.patch("app.gui.tabs.collect.time.monotonic", return_value=10.0): + tab._start_collect_activity() + with mock.patch("app.gui.tabs.collect.time.monotonic", return_value=15.0): + tab._finish_collect_activity("blocked") + self.assertEqual("采集未开始 · 检查未通过", tab.collect_activity_label.text()) + self.assertFalse(tab._collect_elapsed_timer.isActive()) + + with mock.patch("app.gui.tabs.collect.time.monotonic", return_value=20.0): + tab._start_collect_activity() + with mock.patch("app.gui.tabs.collect.time.monotonic", return_value=25.0): + tab._finish_collect_activity("error", {"error": "后台线程异常"}) + self.assertEqual("采集已结束 · 请查看运行日志", tab.collect_activity_label.text()) + self.assertEqual("采集异常结束，请查看下方采集运行日志", tab.collect_activity_label.toolTip()) + + with mock.patch("app.gui.tabs.collect.time.monotonic", return_value=30.0): + tab._start_collect_activity() + with mock.patch("app.gui.tabs.collect.time.monotonic", return_value=40.0): + tab._finish_collect_activity("finished") + self.assertEqual("采集完成 · 总用时 00:10", tab.collect_activity_label.text()) + self.assertFalse(tab._collect_elapsed_timer.isActive()) + + self.assert_removed(temp_dir) + def test_collect_tab_soft_deletes_batch_and_refreshes_workflow_tabs(self): with self.make_temp_dir() as temp_dir: cfg = self.make_config(temp_dir) @@ -7706,14 +7831,17 @@ class GuiTests(TempDirMixin, unittest.TestCase): "old_cover_path": task["old_cover_path"], } + worker = CollectWorker( + tasks, + db_path=cfg["db_path"], + config=cfg, + preflight=False, + ) + activity_events = [] + worker.activity.connect(activity_events.append) 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, - preflight=False, - ).execute() + summary = worker.execute() self.assertTrue(summary["ok"]) self.assertEqual(3, summary["total"]) @@ -7728,6 +7856,25 @@ class GuiTests(TempDirMixin, unittest.TestCase): self.assertEqual([batch_id], summary["batch_ids"]) self.assertIsInstance(summary["run_id"], int) collect.assert_called_once() + self.assertEqual("preflight_started", activity_events[0]["state"]) + self.assertEqual(3, activity_events[0]["total"]) + task_starts = [event for event in activity_events if event["state"] == "task_started"] + self.assertEqual([1, 2, 3], [event["index"] for event in task_starts]) + self.assertEqual( + ["51100639510", "51100639511", "51100639512"], + [event["item_id"] for event in task_starts], + ) + self.assertTrue( + any( + event["state"] == "task_step" and event.get("step") == "download_cover" + for event in activity_events + ) + ) + task_finishes = [event for event in activity_events if event["state"] == "task_finished"] + self.assertEqual( + ["success", "skipped", "skipped"], + [event["result"] for event in task_finishes], + ) 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)