"""Tab 3: Shopee update UI.""" from __future__ import annotations from ..models import ApplyTaskTableModel from ..widgets import * from ..workers import ApplyWorker as _RealApplyWorker, WriteBackWorker as _RealWriteBackWorker def ApplyWorker(*args, **kwargs): return _call_package_attr("ApplyWorker", _RealApplyWorker, *args, **kwargs) def WriteBackWorker(*args, **kwargs): return _call_package_attr("WriteBackWorker", _RealWriteBackWorker, *args, **kwargs) class ApplyTab(QWidget): """Tab 3: list generated tasks and confirm the update scope.""" STATUS_FILTERS = [ ("已生成", "generated"), ("失败", "failed"), ("已更新", "applied"), ("略过", "skipped"), ("全部状态", "all"), ] def __init__( self, parent=None, db_path=None, config=None, status_callback=None, open_accounts_callback=None, open_settings_callback=None, ): super().__init__(parent) self.config = appconfig.load_config() if config is None else config self.db_path = _database_path(db_path, self.config) self.status_callback = status_callback self.open_accounts_callback = open_accounts_callback self.open_settings_callback = open_settings_callback self.apply_worker = None self.apply_thread = None self.result_write_back_worker = None self.result_write_back_thread = None self.last_apply_summary = None self.batch_filter = QComboBox() self.batch_filter.setObjectName("applyBatchFilter") self.shop_filter = QComboBox() self.shop_filter.setObjectName("applyShopFilter") self.item_filter = QLineEdit() self.item_filter.setObjectName("applyItemFilter") self.item_filter.setPlaceholderText("商品ID") self.status_filter = QComboBox() self.status_filter.setObjectName("applyStatusFilter") for label, value in self.STATUS_FILTERS: self.status_filter.addItem(label, value) self.refresh_button = QPushButton("刷新") filter_layout = QHBoxLayout() filter_layout.addWidget(QLabel("批次")) filter_layout.addWidget(self.batch_filter, 2) filter_layout.addWidget(QLabel("店铺")) filter_layout.addWidget(self.shop_filter, 1) filter_layout.addWidget(QLabel("商品ID")) filter_layout.addWidget(self.item_filter, 1) filter_layout.addWidget(QLabel("状态")) filter_layout.addWidget(self.status_filter, 1) filter_layout.addWidget(self.refresh_button) self.summary_label = QLabel("任务 0 条") self.batch_progress_label = _build_batch_progress_overview("applyBatchProgressOverview") self.risk_label = QLabel("可先点击「检查本轮更新」确认当前筛选范围；点击「开始更新」后会再次确认并按批提交线上。") ( self.empty_state_card, self.empty_state_label, self.empty_state_button, ) = _build_empty_state_card("applyEmptyStateCard") if self.open_accounts_callback is not None: self.empty_state_button.clicked.connect(self.open_accounts_callback) self.task_table = QTableView() self.model = ApplyTaskTableModel(self.task_table) self.task_table.setModel(self.model) self.task_table.setSelectionBehavior(QAbstractItemView.SelectRows) self.task_table.setSelectionMode(QAbstractItemView.SingleSelection) self.task_table.setEditTriggers(QAbstractItemView.NoEditTriggers) self.task_table.setContextMenuPolicy(Qt.CustomContextMenu) self.task_table.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch) self.task_table.verticalHeader().setVisible(False) self.run_log_view = QPlainTextEdit() self.run_log_view.setObjectName("applyRunLogView") self.run_log_view.setReadOnly(True) self.run_log_view.setMaximumHeight(128) self.run_log_view.setPlaceholderText("运行日志") self.preview_update_button = QPushButton("检查本轮更新") self.preview_update_button.setObjectName("previewUpdateButton") self.start_update_button = QPushButton("开始更新") self.start_update_button.setObjectName("startUpdateButton") self.start_update_button.setMinimumWidth(118) self.start_update_button.setStyleSheet( _warning_outline_button_style("startUpdateButton") ) self.stop_update_button = QPushButton("停止") self.reset_update_button = QPushButton("重置更新状态") self.reset_update_button.setObjectName("resetUpdateButton") self.reset_update_button.setVisible(False) self.write_back_button = QPushButton("回写结果到 Excel") self.stop_update_button.setEnabled(False) self.write_back_button.setEnabled(False) action_layout = QHBoxLayout() action_layout.addWidget(self.preview_update_button) action_layout.addWidget(self.start_update_button) action_layout.addWidget(self.stop_update_button) action_layout.addStretch(1) action_layout.addWidget(self.write_back_button) layout = QVBoxLayout(self) layout.setContentsMargins(18, 18, 18, 18) layout.addLayout(filter_layout) layout.addWidget(self.risk_label) layout.addWidget(self.summary_label) layout.addWidget(self.batch_progress_label) layout.addWidget(self.empty_state_card) layout.addWidget(self.task_table, 1) layout.addWidget(QLabel("运行日志")) layout.addWidget(self.run_log_view) layout.addLayout(action_layout) self.batch_filter.currentIndexChanged.connect(self.refresh_tasks) self.shop_filter.currentIndexChanged.connect(self.refresh_tasks) self.item_filter.textChanged.connect(self.refresh_tasks) self.status_filter.currentIndexChanged.connect(self.refresh_tasks) self.refresh_button.clicked.connect(self.refresh_tasks) self.preview_update_button.clicked.connect(self.preview_update) self.start_update_button.clicked.connect(self.start_update) self.stop_update_button.clicked.connect(self.stop_update) self.reset_update_button.clicked.connect(self.reset_apply_status) self.task_table.customContextMenuRequested.connect(self.show_task_context_menu) self.write_back_button.clicked.connect(self.write_back_results) self.refresh_tasks() self._show_current_run_log_empty() def _set_status(self, message): if self.status_callback is not None: self.status_callback(message) def refresh_tasks(self, checked=False): try: db.init_db(self.db_path) batches = db.list_batches(path=self.db_path) account_rows = accounts.list_accounts(path=self.db_path, config=self.config) selected_batch = self.batch_filter.currentData() selected_shop = self.shop_filter.currentData() selected_status = self.status_filter.currentData() or "generated" item_query = self.item_filter.text().strip() self._populate_batch_filter(batches, selected_batch) selected_batch = self.batch_filter.currentData() all_batch_tasks = db.list_tasks(batch_id=selected_batch, path=self.db_path) batch_tasks = [ task for task in all_batch_tasks if self._is_update_task(task) ] self._populate_shop_filter(batch_tasks, account_rows, selected_shop) selected_shop = self.shop_filter.currentData() filtered_tasks = [ task for task in batch_tasks if self._matches_shop(task, selected_shop) and self._matches_item(task, item_query) and self._matches_status(task, selected_status) ] except Exception as exc: self.model.set_tasks([], []) self.summary_label.setText("任务读取失败") _set_batch_progress_overview(self.batch_progress_label, []) _set_empty_state(self.empty_state_card, self.empty_state_label, self.empty_state_button) self._set_status(f"更新任务读取失败：{exc}") return self.model.set_tasks(filtered_tasks, account_rows) self.summary_label.setText( f"任务 {len(filtered_tasks)}/{len(batch_tasks)} 条" ) _set_batch_progress_overview(self.batch_progress_label, all_batch_tasks) self._update_empty_state(batch_tasks, filtered_tasks, account_rows) self._update_write_back_button() def _update_empty_state(self, batch_tasks, filtered_tasks, account_rows): if not account_rows: _set_empty_state( self.empty_state_card, self.empty_state_label, self.empty_state_button, "第一步：前往『④账号管理』配置并登录账号，再回到③更新 Shopee。", self.open_accounts_callback is not None, ) return if not batch_tasks: _set_empty_state( self.empty_state_card, self.empty_state_label, self.empty_state_button, "还没有可更新任务。请先在②AI生成完成新标题或新封面。", ) return if not filtered_tasks: _set_empty_state( self.empty_state_card, self.empty_state_label, self.empty_state_button, "当前筛选没有可更新任务，请调整批次、店铺、商品ID或状态筛选。", ) return _set_empty_state(self.empty_state_card, self.empty_state_label, self.empty_state_button) def start_update(self, checked=False): self._start_update(dry_run=False) def preview_update(self, checked=False): self._start_update(dry_run=True) def _start_update(self, dry_run=False): if self.apply_thread is not None: self._set_status("更新正在进行...") return tasks = [ task for task in self.model.tasks if self._is_actionable_task(task) ] if not tasks: self._set_status("当前筛选结果没有可更新任务") return update_cfg = self._shopee_update_config() dry_run = bool(dry_run) safety_error = self._update_safety_error(tasks, dry_run=dry_run) if safety_error: self._show_update_safety_error(safety_error) self._set_status(safety_error.replace("\n", " ")) return answer = QMessageBox.question( self, "确认检查本轮更新" if dry_run else "确认开始更新", self._confirmation_message(tasks, dry_run=dry_run), QMessageBox.Yes | QMessageBox.No, QMessageBox.No, ) if answer != QMessageBox.Yes: self._set_status("已取消检查本轮更新" if dry_run else "已取消开始更新") return batch_size = max(1, int(update_cfg.get("max_items_per_run", 1) or 1)) worker = ApplyWorker( tasks, db_path=self.db_path, config=self.config, close_success_tab=bool(update_cfg.get("close_success_tab", False)), dry_run=dry_run, parallel_accounts=bool(update_cfg.get("parallel_accounts", False)), max_parallel_accounts=max( 1, int(update_cfg.get("max_parallel_accounts", 1) or 1), ), batch_size=batch_size, diagnostic_log_dir=diagnostics.DEFAULT_LOG_DIR, ) worker.progress.connect(self._on_apply_progress) worker.row_updated.connect(self._on_apply_row_updated) worker.log.connect(self._on_apply_log) worker.failed.connect(self._on_apply_failed) worker.finished.connect(self._on_apply_finished) worker.cancelled.connect(self._on_apply_cancelled) thread = run_worker(worker, thread_name="ApplyWorker", start=False) thread.finished.connect(lambda: self._forget_apply_thread(thread)) self.apply_worker = worker self.apply_thread = thread self._set_apply_running(True) self._reset_run_log(tasks, dry_run=dry_run, batch_size=batch_size) if dry_run: self._set_status(f"开始检查本轮更新：{len(tasks)} 条") else: self._set_status(f"开始更新：{len(tasks)} 条，按每批最多 {batch_size} 条执行") thread.start() def stop_update(self, checked=False): if self.apply_worker is not None: self.apply_worker.cancel() self._set_status("正在停止更新...") def write_back_results(self, checked=False): batch_ids = self._active_batch_ids() if not batch_ids: self._set_status("没有可回写结果的批次") return self._start_result_write_back(batch_ids, auto=False) def show_task_context_menu(self, position): index = self.task_table.indexAt(position) if index.isValid(): self.task_table.setCurrentIndex(index) menu = QMenu(self) reset_action = menu.addAction("重置更新状态") reset_action.setEnabled( self.apply_thread is None and self.result_write_back_thread is None and self._selected_task() is not None ) reset_action.triggered.connect(self.reset_apply_status) menu.exec(self.task_table.viewport().mapToGlobal(position)) def reset_apply_status(self, checked=False): if self.apply_thread is not None or self.result_write_back_thread is not None: self._set_status("更新或回写正在进行，不能重置") return task = self._selected_task() if task is None: self._set_status("请选择要重置更新状态的任务") return if not (getattr(task, "new_title", None) or getattr(task, "new_cover_path", None)): self._set_status("选中任务没有新标题或新封面，不能重置为可更新") return lines = [ "确定重置当前选中任务的本地更新状态吗？", "", f"商品ID：{task.item_id}", f"店铺：{self.model.account_name_for(task)}", "", "将保留新标题和新封面路径，只把本地状态退回可更新。", "不会触碰 Shopee，也不会自动回写 Excel。", ] if getattr(task, "committed", 0): lines.extend([ "", "注意：该记录已经提交过线上。本地重置不会回滚 Shopee，重复更新会再次提交线上。", ]) answer = QMessageBox.question( self, "重置更新状态", "\n".join(lines), QMessageBox.Yes | QMessageBox.No, QMessageBox.No, ) if answer != QMessageBox.Yes: self._set_status("已取消重置更新状态") return try: db.reset_apply_status(task.id, path=self.db_path) message = ( "action=reset_apply_status step=db_write result=success " f"detail=退回可更新 task_id={task.id}" ) run_id = _write_reset_run_log(self.db_path, task, "reset_apply_status", message) except Exception as exc: QMessageBox.warning(self, "重置更新状态", str(exc)) self._set_status(f"重置更新状态失败：{exc}") return self.refresh_tasks() self._append_run_log(message) self._set_status( f"已重置更新状态：商品 {task.item_id}，run_id={run_id}" ) def _selected_task(self): index = self.task_table.currentIndex() if index.isValid(): return self.model.task_at(index.row()) if self.model.rowCount() > 0: return self.model.task_at(0) return None def _is_actionable_task(self, task): return ( getattr(task, "stage", None) == "generated" and getattr(task, "status", None) in {"success", "pending", "failed"} and bool(getattr(task, "new_title", None) or getattr(task, "new_cover_path", None)) ) def _populate_batch_filter(self, batches, selected_batch): previous = selected_batch if selected_batch in {batch.id for batch in batches} else None self.batch_filter.blockSignals(True) self.batch_filter.clear() self.batch_filter.addItem("全部批次", None) for batch in batches: self.batch_filter.addItem(self._batch_label(batch), batch.id) index = self.batch_filter.findData(previous) self.batch_filter.setCurrentIndex(index if index >= 0 else 0) self.batch_filter.blockSignals(False) def _populate_shop_filter(self, tasks, account_rows, selected_shop): account_by_alias = { str(account.alias).strip(): account for account in account_rows if str(account.alias).strip() } aliases = [] for task in tasks: alias = str(task.alias).strip() if alias and alias not in aliases: aliases.append(alias) previous = selected_shop if selected_shop in aliases else None self.shop_filter.blockSignals(True) self.shop_filter.clear() self.shop_filter.addItem("全部店铺", None) for alias in sorted(aliases, key=lambda value: self._shop_label(value, account_by_alias)): self.shop_filter.addItem(self._shop_label(alias, account_by_alias), alias) index = self.shop_filter.findData(previous) self.shop_filter.setCurrentIndex(index if index >= 0 else 0) self.shop_filter.blockSignals(False) def _batch_label(self, batch): source_files = batch.source_files first_file = os.path.basename(source_files[0]) if source_files else batch.id return f"{batch.created_at} · {first_file}" def _shop_label(self, alias, account_by_alias): account = account_by_alias.get(alias) if account is not None: return f"{account.account_name} ({alias})" return alias def _status_label(self): return self.status_filter.currentText() or "已生成" def _batch_filter_label(self): return self.batch_filter.currentText() or "全部批次" def _shop_filter_label(self): return self.shop_filter.currentText() or "全部店铺" def _item_filter_label(self): return self.item_filter.text().strip() or "全部商品" def _is_update_task(self, task): if task.stage in {"generated", "applied"}: return True return bool((task.new_title or task.new_cover_path) and task.status in {"failed", "skipped"}) def _matches_shop(self, task, selected_shop): return selected_shop is None or str(task.alias).strip() == selected_shop def _matches_item(self, task, item_query): if not item_query: return True return item_query in str(getattr(task, "item_id", "")) def _matches_status(self, task, selected_status): if selected_status in (None, "all"): return True if selected_status == "generated": return task.stage == "generated" and task.status in {"success", "pending"} if selected_status == "failed": return task.status == "failed" if selected_status == "applied": return task.stage == "applied" if selected_status == "skipped": return task.status == "skipped" return True def _confirmation_message(self, tasks, dry_run=False): update_cfg = self._shopee_update_config() cover_text = "允许" if update_cfg.get("allow_cover_update") else "不允许" close_text = "是" if update_cfg.get("close_success_tab") else "否" batch_size = max(1, int(update_cfg.get("max_items_per_run", 1) or 1)) batch_count = (len(tasks) + batch_size - 1) // batch_size if tasks else 0 parallel_text = ( f"开启，最多 {update_cfg.get('max_parallel_accounts', 1)} 个账号" if update_cfg.get("parallel_accounts") else "关闭" ) intro = ( "即将检查当前筛选结果。\n\n" if dry_run else "即将按当前筛选结果分批更新 Shopee 线上商品。\n\n" ) return ( intro + f"批次：{self._batch_filter_label()}\n" + f"店铺：{self._shop_filter_label()}\n" + f"商品ID：{self._item_filter_label()}\n" + f"状态：{self._status_label()}\n" + f"任务数：{len(tasks)}\n" + f"预计批次：{batch_count}\n\n" + "安全设置：" + f"封面更新={cover_text}，" + f"每批最大更新条数={batch_size}，" + f"成功后关闭新页={close_text}，" + f"多账号并行={parallel_text}\n\n" + ( "检查只写运行日志，不打开 Shopee、不点击「更新」、不改任务状态。" if dry_run else f"确认后会打开商品编辑页、替换标题/允许时替换封面，并按每批最多 {batch_size} 条点击「更新」提交线上；点击停止后不再开始下一条或下一批。" ) ) def _update_safety_error(self, tasks, dry_run=False): update_cfg = self._shopee_update_config() if dry_run: return None if not update_cfg.get("allow_real_submit", False): return ( "设置未开启「允许真实提交线上商品」，已阻止本次更新。\n" "请到⑤设置 > Shopee 更新安全开启该开关后再开始更新。" ) if not update_cfg.get("allow_cover_update", False): cover_tasks = [ str(getattr(task, "item_id", "")) for task in tasks if getattr(task, "new_cover_path", None) ] if cover_tasks: return ( "设置未开启「允许更新封面」，当前任务包含新封面路径，已阻止本次更新。\n" "请到⑤设置 > Shopee 更新安全开启该开关，或先筛掉含新封面的任务。" ) return None def _show_update_safety_error(self, message): box = QMessageBox(self) box.setIcon(QMessageBox.Warning) box.setWindowTitle("更新安全开关") box.setText(str(message)) settings_button = None if self.open_settings_callback is not None: settings_button = box.addButton("前往设置", QMessageBox.ActionRole) box.addButton(QMessageBox.Ok) box.exec() if settings_button is not None and box.clickedButton() is settings_button: self.open_settings_callback() def _shopee_update_config(self): defaults = appconfig.default_config().get("shopee_update", {}) loaded = self.config.get("shopee_update", {}) if not isinstance(loaded, dict): loaded = {} merged = dict(defaults) merged.update(loaded) return merged def _set_apply_running(self, running): self.preview_update_button.setEnabled(not running) self.start_update_button.setEnabled(not running) self.stop_update_button.setEnabled(running) self.reset_update_button.setEnabled(not running) self.refresh_button.setEnabled(not running) self.batch_filter.setEnabled(not running) self.shop_filter.setEnabled(not running) self.item_filter.setEnabled(not running) self.status_filter.setEnabled(not running) self._update_write_back_button() def _set_result_write_back_running(self, running): self.preview_update_button.setEnabled(not running) self.start_update_button.setEnabled(not running) self.reset_update_button.setEnabled(not running) self.refresh_button.setEnabled(not running) self.batch_filter.setEnabled(not running) self.shop_filter.setEnabled(not running) self.item_filter.setEnabled(not running) self.status_filter.setEnabled(not running) self.write_back_button.setEnabled(False if running else bool(self._active_batch_ids())) def _forget_apply_thread(self, thread): if self.apply_thread is thread: self.apply_thread = None self.apply_worker = None def _forget_result_write_back_thread(self, thread): if self.result_write_back_thread is thread: self.result_write_back_thread = None self.result_write_back_worker = None self._update_write_back_button() def _on_apply_progress(self, payload): self._set_status("更新进度：" + self._apply_progress_text(payload)) def _on_apply_log(self, message): self._append_run_log(message) self._set_status(message) def _on_apply_row_updated(self, task_id, fields): self.refresh_tasks() def _on_apply_failed(self, task_id, error): self._set_status(f"任务 {task_id} 更新失败：{error}") def _on_apply_finished(self, payload): self._set_apply_running(False) self.refresh_tasks() if payload.get("blocked"): self._show_apply_blocked(payload) return self.last_apply_summary = dict(payload) prefix = "检查本轮更新完成：" if payload.get("dry_run") else "更新完成：" message = prefix + self._apply_progress_text(payload) batch_ids = payload.get("batch_ids") or self._active_batch_ids() if (not payload.get("dry_run")) and payload.get("done", 0) > 0 and batch_ids: if self._start_result_write_back( batch_ids, auto=True, apply_summary=payload, ): self._set_status(f"{message}，正在自动回写结果到 Excel...") return self._set_status(message) self._show_apply_summary(payload) def _on_apply_cancelled(self, payload): self._set_apply_running(False) self.refresh_tasks() self._set_status("更新已停止：" + self._apply_progress_text(payload)) def _apply_progress_text(self, payload): success_label = "可更新" if payload.get("dry_run") else "成功" return "完成{done}/{total}，{success_label}{applied}，略过{skipped}，失败{failed}".format( done=payload.get("done", 0), total=payload.get("total", 0), success_label=success_label, applied=payload.get("applied", 0), skipped=payload.get("skipped", 0), failed=payload.get("failed", 0), ) def _append_run_log(self, message): self.run_log_view.appendPlainText(str(message)) scroll_bar = self.run_log_view.verticalScrollBar() scroll_bar.setValue(scroll_bar.maximum()) def _show_current_run_log_empty(self): self.run_log_view.setPlainText("本轮日志会在开始运行后显示") def _reset_run_log(self, tasks, dry_run=False, batch_size=1): self.run_log_view.clear() action = "检查" if dry_run else "更新" self._append_run_log( f"本轮{action}开始：任务 {len(tasks)} 条，每批 {batch_size} 条" ) def _load_latest_run_log(self): try: logs = db.list_run_logs(limit=1, run_type="apply", path=self.db_path) if not logs: return events = db.list_run_log_events(logs[0].id, limit=30, path=self.db_path) except Exception: return lines = [ f"{event.created_at} [{event.level}] {event.message}" for event in reversed(events) ] self.run_log_view.setPlainText("\n".join(lines)) scroll_bar = self.run_log_view.verticalScrollBar() scroll_bar.setValue(scroll_bar.maximum()) def _show_apply_blocked(self, payload): lines = ["更新前检查未通过。"] if payload.get("no_accounts"): lines.append("当前没有配置账号。") duplicate_ports = payload.get("duplicate_ports") or [] if duplicate_ports: for item in duplicate_ports: lines.append( "以下账号调试端口冲突：端口 {port} -> {aliases}".format( port=item.get("debug_port"), aliases="、".join(item.get("aliases") or []), ) ) not_running = payload.get("not_running") or [] if not_running: lines.append( "以下账号 Chrome 未启动或调试端口不可访问：" + "、".join(self._account_label(item) for item in not_running) ) logged_out = payload.get("logged_out") or [] if logged_out: lines.append( "以下账号未登录 Shopee：" + "、".join(self._account_label(item) for item in logged_out) ) self._show_account_guide("\n".join(lines)) def _show_account_guide(self, message): full_message = ( f"{message}\n\n" "本轮更新已中止，不会自动打开账号 Chrome，也不会提交任何商品。\n" "请先到「④ 账号管理」配置账号、启动对应账号 Chrome，并确认已人工登录 Shopee。" ) QMessageBox.warning(self, "账号未就绪", full_message) self._set_status(full_message.replace("\n", " ")) if self.open_accounts_callback is not None: self.open_accounts_callback() def _account_label(self, item): if isinstance(item, dict): name = item.get("account_name") or item.get("alias") or "" alias = item.get("alias") or "" reason = item.get("reason") else: name = getattr(item, "account_name", "") or getattr(item, "alias", "") alias = getattr(item, "alias", "") reason = getattr(item, "reason", None) label = f"{name}({alias})" if alias and name != alias else (name or alias) return f"{label}: {reason}" if reason else label def _active_batch_ids(self): selected_batch = self.batch_filter.currentData() if selected_batch: return [selected_batch] batch_ids = [] for task in self.model.tasks: batch_id = getattr(task, "batch_id", None) if batch_id and batch_id not in batch_ids: batch_ids.append(batch_id) return batch_ids def _update_write_back_button(self): if getattr(self, "write_back_button", None) is None: return enabled = ( self.apply_thread is None and self.result_write_back_thread is None and bool(self._active_batch_ids()) ) self.write_back_button.setEnabled(enabled) def _start_result_write_back(self, batch_ids, auto=False, apply_summary=None): if self.result_write_back_thread is not None: self._set_status("Excel 结果回写正在进行...") return False worker = WriteBackWorker( batch_ids, db_path=self.db_path, mode="results", diagnostic_log_dir=diagnostics.DEFAULT_LOG_DIR, ) worker.failed.connect( lambda task_id, error, auto=auto, apply_summary=apply_summary: self._on_result_write_back_failed( task_id, error, auto=auto, apply_summary=apply_summary, ) ) worker.finished.connect( lambda payload, auto=auto, apply_summary=apply_summary: self._on_result_write_back_finished( payload, auto=auto, apply_summary=apply_summary, ) ) thread = run_worker(worker, thread_name="ResultWriteBackWorker", start=False) thread.finished.connect(lambda: self._forget_result_write_back_thread(thread)) self.result_write_back_worker = worker self.result_write_back_thread = thread self._set_result_write_back_running(True) self._set_status("正在自动回写更新结果到 Excel..." if auto else "正在回写更新结果到 Excel...") thread.start() return True def _on_result_write_back_failed(self, task_id, error, auto=False, apply_summary=None): message = f"Excel {'自动' if auto else ''}回写更新结果失败：{error}" if "被占用" in str(error): message += "\n请关闭原 Excel 后点击「回写结果到 Excel」手动重试；SQLite 已保留更新结果。" if auto and apply_summary: message = self._apply_summary_message(apply_summary, error=message) QMessageBox.warning(self, "回写结果到 Excel", message) self._set_status(message.replace("\n", " ")) def _on_result_write_back_finished(self, payload, auto=False, apply_summary=None): self._set_result_write_back_running(False) self.refresh_tasks() if payload.get("ok") is False: error = payload.get("error") or "未知错误" retry_hint = "，可点击「回写结果到 Excel」手动重试" if auto else "" self._set_status(f"Excel {'自动' if auto else ''}回写更新结果失败：{error}{retry_hint}") return self._set_status( "Excel {prefix}回写更新结果完成：文件{files}，行{rows}".format( prefix="自动" if auto else "", files=payload.get("files", 0), rows=payload.get("rows", 0), ) ) if auto and apply_summary: self._show_apply_summary(apply_summary, write_back_payload=payload) elif not auto: QMessageBox.information( self, "回写结果到 Excel", "结果回写完成：文件{files}，行{rows}".format( files=payload.get("files", 0), rows=payload.get("rows", 0), ), ) def _show_apply_summary(self, apply_summary, write_back_payload=None): QMessageBox.information( self, "检查本轮更新完成" if apply_summary.get("dry_run") else "更新完成", self._apply_summary_message(apply_summary, write_back_payload), ) def _apply_summary_message(self, apply_summary, write_back_payload=None, error=None): dry_run = bool(apply_summary.get("dry_run")) lines = [ "检查本轮更新完成，未打开 Shopee、未提交线上、未改任务状态。" if dry_run else "更新完成。", "{success_label}：{applied}，失败：{failed}，略过：{skipped}".format( success_label="可更新" if dry_run else "成功", applied=apply_summary.get("applied", 0), failed=apply_summary.get("failed", 0), skipped=apply_summary.get("skipped", 0), ), ] if write_back_payload: lines.append( "Excel 回写：文件{files}，行{rows}".format( files=write_back_payload.get("files", 0), rows=write_back_payload.get("rows", 0), ) ) if error: lines.append(str(error)) return "\n".join(lines)