From b9d84e71ec73e7132f965f0d44ee659eee0df942 Mon Sep 17 00:00:00 2001 From: chengma Date: Sat, 27 Jun 2026 14:37:58 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=8C=E6=88=90T-204=E6=97=A7?= =?UTF-8?q?=E5=AD=97=E6=AE=B5=E5=9B=9E=E5=86=99?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 实现 Excel 旧标题与旧封面路径回写原文件，按 source_file_abs/source_sheet/source_row 定位行，支持输出列自动追加、xlsm 保留 VBA、文件占用中文错误提示与 export_copy 另存副本。 Tab① 增加回写旧数据按钮与 WriteBackWorker，后台调用 excel.write_back，锁文件时提示关闭后重试；补充 Excel 与 GUI 单测覆盖回写、锁文件、另存副本和 worker 调用。 同步 harness 文档：T-204 标记完成，新增 T-204b 记录采集完成自动回写缺口，扩展 T-205 覆盖 Chrome 未启动/未登录引导保护，并更新 current-state、routes、api、architecture、requirements 与 progress。 --- app/excel.py | 219 +++++++++++++++++++++++++++++++++++++++- app/gui.py | 91 +++++++++++++++++ docs/02-requirements.md | 2 +- docs/04-architecture.md | 3 +- docs/06-tasks.md | 5 +- docs/api.md | 17 ++-- docs/current-state.md | 21 ++-- docs/routes.md | 11 +- progress.md | 17 ++++ tests/test_excel.py | 127 ++++++++++++++++++++++- tests/test_gui.py | 19 ++++ 11 files changed, 503 insertions(+), 29 deletions(-) diff --git a/app/excel.py b/app/excel.py index 5e2844a..6acd8a9 100644 --- a/app/excel.py +++ b/app/excel.py @@ -24,6 +24,17 @@ HEADER_ALIASES = { "alias": {"别名", "账号别名", "alias"}, "item_id": {"商品id", "商品编号", "商品id号", "itemid", "item"}, } +OUTPUT_HEADERS = { + "old_title": "旧标题", + "old_cover_path": "旧封面图片路径", + "new_title": "新标题", + "new_cover_path": "新封面图片路径", + "update_status": "更新状态", +} +OLD_WRITE_BACK_FIELDS = ("old_title", "old_cover_path") +WRITEABLE_STAGES = {"collected", "generated", "applied"} +LOCK_WINERRORS = {5, 32, 33} +LOCK_ERRNOS = {13, 16} class ExcelError(RuntimeError): @@ -276,13 +287,211 @@ def match_summary(rows: list[dict], accounts: list) -> dict: } -def write_back(batch_id, excel_path=None, path=None) -> dict: - """Placeholder for Excel result write-back, implemented in later tasks.""" +def _batch_tasks(batch_id, path=None): + if not batch_id: + raise ExcelError("缺少批次 ID，无法回写 Excel") + db.init_db(path) + tasks = db.list_tasks(batch_id=batch_id, path=path) + if not tasks: + raise ExcelError(f"批次没有可回写任务: {batch_id}") + return tasks - raise ExcelError("Excel 回写将在 T-204/T-403 实现") + +def _filter_tasks_by_excel_path(tasks, excel_path): + if excel_path is None: + return list(tasks) + target = os.path.abspath(str(excel_path)) + filtered = [ + task for task in tasks + if os.path.abspath(task.source_file_abs) == target + ] + if not filtered: + raise ExcelError(f"批次中没有来自该 Excel 的任务: {target}") + return filtered + + +def _has_write_back_data(task, fields) -> bool: + if getattr(task, "stage", None) in WRITEABLE_STAGES: + return True + return any(getattr(task, field, None) not in (None, "") for field in fields) + + +def _group_by_source_file(tasks, fields): + groups = {} + for task in tasks: + if not _has_write_back_data(task, fields): + continue + source_file_abs = os.path.abspath(task.source_file_abs) + groups.setdefault(source_file_abs, []).append(task) + return groups + + +def _is_locked_error(exc) -> bool: + return ( + isinstance(exc, PermissionError) + or getattr(exc, "winerror", None) in LOCK_WINERRORS + or getattr(exc, "errno", None) in LOCK_ERRNOS + ) + + +def _excel_suffix(path_value) -> str: + return os.path.splitext(str(path_value))[1].lower() + + +def _load_write_workbook(source_path): + kwargs = {} + if _excel_suffix(source_path) == ".xlsm": + kwargs["keep_vba"] = True + return load_workbook(source_path, **kwargs) + + +def _ensure_output_columns(sheet, fields): + header_by_normalized = {} + for column in range(1, sheet.max_column + 1): + header = _text(sheet.cell(row=1, column=column).value) + if header: + header_by_normalized.setdefault(_normalize_header(header), column) + + columns = {} + next_column = sheet.max_column + 1 + for field in fields: + header = OUTPUT_HEADERS[field] + normalized = _normalize_header(header) + column = header_by_normalized.get(normalized) + if column is None: + column = next_column + sheet.cell(row=1, column=column).value = header + header_by_normalized[normalized] = column + next_column += 1 + columns[field] = column + return columns + + +def _write_tasks_to_workbook(workbook, tasks, fields) -> int: + rows = 0 + missing_sheets = [] + by_sheet = {} + for task in tasks: + by_sheet.setdefault(task.source_sheet, []).append(task) + + for sheet_name, sheet_tasks in by_sheet.items(): + if sheet_name not in workbook.sheetnames: + missing_sheets.append(sheet_name) + continue + sheet = workbook[sheet_name] + columns = _ensure_output_columns(sheet, fields) + for task in sheet_tasks: + if not _has_write_back_data(task, fields): + continue + for field in fields: + sheet.cell( + row=int(task.source_row), + column=columns[field], + ).value = getattr(task, field, None) or "" + rows += 1 + + if missing_sheets: + names = ", ".join(sorted(set(missing_sheets))) + raise ExcelError(f"原 Excel 缺少导入时的工作表: {names}") + return rows + + +def _write_source_to_path(source_path, tasks, target_path, fields): + workbook = None + try: + workbook = _load_write_workbook(source_path) + except Exception as exc: + if _is_locked_error(exc): + raise ExcelError(f"Excel 文件被占用，请关闭后重试: {source_path}") from exc + raise ExcelError(f"读取 Excel 失败: {source_path}: {exc}") from exc + + try: + rows = _write_tasks_to_workbook(workbook, tasks, fields) + try: + workbook.save(target_path) + except Exception as exc: + if _is_locked_error(exc): + raise ExcelError(f"Excel 文件被占用，请关闭后重试: {target_path}") from exc + raise ExcelError(f"保存 Excel 失败: {target_path}: {exc}") from exc + return rows + finally: + if workbook is not None: + workbook.close() + + +def _write_result(batch_id, groups, target_for_source, fields): + rows = 0 + written_files = [] + for source_path, tasks in groups.items(): + target_path = target_for_source(source_path) + rows += _write_source_to_path(source_path, tasks, target_path, fields) + written_files.append(os.path.abspath(target_path)) + return { + "ok": True, + "batch_id": batch_id, + "files": len(written_files), + "rows": rows, + "written_files": written_files, + } + + +def write_back(batch_id, excel_path=None, path=None) -> dict: + """Write collected old-title/old-cover fields back to the original Excel file.""" + + _require_openpyxl() + fields = OLD_WRITE_BACK_FIELDS + tasks = _filter_tasks_by_excel_path(_batch_tasks(batch_id, path=path), excel_path) + groups = _group_by_source_file(tasks, fields) + return _write_result( + batch_id, + groups, + lambda source_path: source_path, + fields, + ) + + +def _is_excel_output_path(path_value) -> bool: + return _excel_suffix(path_value) in {".xlsx", ".xlsm"} + + +def _unique_copy_path(out_dir, source_path): + base_name = os.path.basename(source_path) + stem, suffix = os.path.splitext(base_name) + candidate = os.path.join(out_dir, f"{stem}_cmshopee回写{suffix}") + index = 1 + while os.path.exists(candidate): + candidate = os.path.join(out_dir, f"{stem}_cmshopee回写_{index}{suffix}") + index += 1 + return candidate def export_copy(batch_id, out_dir_or_path, path=None) -> dict: - """Placeholder for exporting a copy when the original Excel is locked.""" + """Export a write-back copy without touching the original Excel file.""" - raise ExcelError("Excel 另存副本将在 T-204/T-403 实现") + _require_openpyxl() + if not out_dir_or_path: + raise ExcelError("缺少另存路径，无法导出 Excel 副本") + + fields = OLD_WRITE_BACK_FIELDS + tasks = _batch_tasks(batch_id, path=path) + groups = _group_by_source_file(tasks, fields) + output_path = os.path.abspath(str(out_dir_or_path)) + output_is_file = _is_excel_output_path(output_path) + if output_is_file and len(groups) > 1: + raise ExcelError("多个源 Excel 另存副本时，目标必须是目录") + if output_is_file: + output_dir = os.path.dirname(output_path) + if output_dir: + os.makedirs(output_dir, exist_ok=True) + + def target_for_source(source_path): + if os.path.abspath(source_path) == output_path: + raise ExcelError("另存副本路径不能与原 Excel 相同") + return output_path + else: + os.makedirs(output_path, exist_ok=True) + + def target_for_source(source_path): + return _unique_copy_path(output_path, source_path) + + return _write_result(batch_id, groups, target_for_source, fields) diff --git a/app/gui.py b/app/gui.py index 36fc4ee..d398cfc 100644 --- a/app/gui.py +++ b/app/gui.py @@ -206,11 +206,14 @@ if QT_IMPORT_ERROR is None: self.last_import_stats = None self.collect_worker = None self.collect_thread = None + self.write_back_worker = None + self.write_back_thread = None self.import_button = QPushButton("导入 Excel...") self.refresh_button = QPushButton("刷新") self.collect_button = QPushButton("采集旧标题/旧封面") self.stop_collect_button = QPushButton("停止") + self.write_back_button = QPushButton("回写旧数据到 Excel") self.stop_collect_button.setEnabled(False) toolbar = QHBoxLayout() @@ -218,6 +221,7 @@ if QT_IMPORT_ERROR is None: toolbar.addWidget(self.refresh_button) toolbar.addWidget(self.collect_button) toolbar.addWidget(self.stop_collect_button) + toolbar.addWidget(self.write_back_button) toolbar.addStretch(1) self.summary_label = QLabel("未导入任务") @@ -254,6 +258,7 @@ if QT_IMPORT_ERROR is None: self.refresh_button.clicked.connect(self.refresh_tasks) self.collect_button.clicked.connect(self.collect_old_data) self.stop_collect_button.clicked.connect(self.stop_collect) + self.write_back_button.clicked.connect(self.write_back_old_data) self.show_all_button.clicked.connect(self.show_all_tasks) self.show_unmatched_button.clicked.connect(self.show_unmatched_tasks) @@ -339,17 +344,57 @@ if QT_IMPORT_ERROR is None: self.collect_worker.cancel() self._set_status("正在停止采集...") + def write_back_old_data(self, checked=False): + batch_id = self._active_batch_id() + if not batch_id: + self._set_status("没有可回写批次") + return + worker = WriteBackWorker(batch_id, db_path=self.db_path) + worker.failed.connect(self._on_write_back_failed) + worker.finished.connect(self._on_write_back_finished) + thread = run_worker(worker, thread_name="WriteBackWorker", start=False) + thread.finished.connect(lambda: self._forget_write_back_thread(thread)) + self.write_back_worker = worker + self.write_back_thread = thread + self._set_write_back_running(True) + self._set_status("正在回写旧数据到 Excel...") + thread.start() + + def _active_batch_id(self): + if self.current_batch_id: + return self.current_batch_id + batch_ids = { + task.batch_id + for task in self.model.all_tasks + if getattr(task, "batch_id", None) + } + if len(batch_ids) == 1: + return next(iter(batch_ids)) + return None + def _set_collect_running(self, running): self.import_button.setEnabled(not running) self.refresh_button.setEnabled(not running) self.collect_button.setEnabled(not running) + self.write_back_button.setEnabled(not running) self.stop_collect_button.setEnabled(running) + def _set_write_back_running(self, running): + self.import_button.setEnabled(not running) + self.refresh_button.setEnabled(not running) + self.collect_button.setEnabled(not running) + self.write_back_button.setEnabled(not running) + def _forget_collect_thread(self, thread): if self.collect_thread is thread: self.collect_thread = None self.collect_worker = None + def _forget_write_back_thread(self, thread): + if self.write_back_thread is thread: + self.write_back_thread = None + self.write_back_worker = None + def _on_collect_progress(self, payload): self._set_status( "采集进度：{done}/{total}，成功{collected}，略过{skipped}，失败{failed}".format( @@ -388,6 +433,27 @@ if QT_IMPORT_ERROR is None: ) ) + def _on_write_back_failed(self, task_id, error): + message = f"Excel 回写失败：{error}" + if "被占用" in str(error): + message += "\n请关闭原 Excel 后重试；SQLite 已保留采集结果，也可另存副本。" + QMessageBox.warning(self, "回写旧数据", message) + self._set_status(message.replace("\n", " ")) + + def _on_write_back_finished(self, payload): + self._set_write_back_running(False) + if payload.get("ok") is False: + error = payload.get("error") or "未知错误" + self._set_status(f"Excel 回写失败：{error}") + return + self.refresh_tasks() + self._set_status( + "Excel 回写完成：文件{files}，行{rows}".format( + files=payload.get("files", 0), + rows=payload.get("rows", 0), + ) + ) + def show_all_tasks(self, checked=False): self.model.set_filter_mode("all") self._update_empty_label(len(self.model.all_tasks)) @@ -669,6 +735,31 @@ if QT_IMPORT_ERROR is None: ) + class WriteBackWorker(BaseWorker): + """Write collected old fields back to Excel in a background thread.""" + + def __init__(self, batch_id, db_path=None, excel_path=None): + super().__init__() + self.batch_id = batch_id + self.db_path = db_path + self.excel_path = excel_path + + def execute(self): + result = excel.write_back( + self.batch_id, + excel_path=self.excel_path, + path=self.db_path, + ) + self.progress.emit( + { + "done": result.get("rows", 0), + "total": result.get("rows", 0), + "files": result.get("files", 0), + } + ) + return result + + class AccountLoginCheckWorker(BaseWorker): def __init__(self, account, db_path=None, config=None, timeout=8): super().__init__() diff --git a/docs/02-requirements.md b/docs/02-requirements.md index 8e056c4..fe0aa34 100644 --- a/docs/02-requirements.md +++ b/docs/02-requirements.md @@ -48,7 +48,7 @@ | 更新 shopee（③） | 按批次/店铺/状态筛选；点击「开始更新」后弹窗确认，确认后对当前筛选出的已生成任务打开编辑页换标题+封面，并逐条点「更新」提交线上；可按状态=失败重试 | P0 | | 结果存储与回写 | 各阶段结果实时存 SQLite；该文件全部完成后把旧/新数据+状态批量回写原 Excel | P0 | | 设置（⑤） | AI 模型管理（下拉+新增/删除/详情/测试连接，至少各一个文本+图像模型）；标题/图片大模型角色选择；分辨率(512/1k/2k/4k，返回超时随分辨率自动)；并发/重试/jpg质量；图片目录/Chrome 路径/端口 | P0 | -| 首次引导保护 | 未配账号/未登录时，① ③ 执行按钮禁用并提示去④ | P0 | +| 首次引导保护 | 未配账号、对应账号 Chrome 未启动或未登录时，① ③ 执行按钮禁用/执行前拦截并提示去④ | P0 | ### 后续迭代 diff --git a/docs/04-architecture.md b/docs/04-architecture.md index 53748da..0ed92a4 100644 --- a/docs/04-architecture.md +++ b/docs/04-architecture.md @@ -59,7 +59,7 @@ imported → collected → generated → applied ## 三、职责划分 -**GUI（5 Tab）**：见 [routes.md](routes.md)。只做交互与预览，不写业务逻辑；耗时操作走 PySide6 `QObject` worker + `QThread`，用 signal 回主线程刷新 UI。**首次未配账号/未登录时，① ③ 执行按钮禁用并提示去 ④。** +**GUI（5 Tab）**：见 [routes.md](routes.md)。只做交互与预览，不写业务逻辑；耗时操作走 PySide6 `QObject` worker + `QThread`，用 signal 回主线程刷新 UI。**首次未配账号、对应账号 Chrome 未启动或未登录时，① ③ 执行按钮禁用或执行前拦截，并提示去 ④。** **核心模块** @@ -288,6 +288,7 @@ images//_new. # AI 生成的新封面 - 用账号 Chrome 打开商品页，等就绪，读旧标题（标题输入框 value）。 - 旧封面：取第一张 itembox 的 `img.src`（CDN 链接），下载到 `images//_old`。 - 写 `old_title/old_cover_path`、stage=collected；批量回写 Excel 旧字段。 +- T-204b 目标：采集任务结束后自动触发当前批次旧字段回写；原 Excel 被锁时不影响 SQLite 结果，提示关闭后重试，并保留手动「回写旧数据到 Excel」入口。 ### 6.2 AI 生成（② Tab） diff --git a/docs/06-tasks.md b/docs/06-tasks.md index 5a591ff..9a2267d 100644 --- a/docs/06-tasks.md +++ b/docs/06-tasks.md @@ -51,8 +51,9 @@ | T-202 | Tab① 任务列表 + 导入按钮 + 别名匹配标记 | T-201, T-105 | `QTableView` 显示账号/别名/商品id/阶段；未匹配标“略过” | DONE | | T-202b | Tab① 导入汇总栏 | T-202 | 导入后显示 文件数/解析行数/有效/无效/匹配(按账号)/未匹配；未匹配可点击筛出 | DONE | | T-203 | 采集旧标题+旧封面（只读），下载图片，立即写库 | T-202, T-001, T-104b | 通过 worker 执行；逐条 set_collected；旧封面下载到 `images//`；未登录/未匹配略过记原因 | DONE | -| T-204 | 回写旧字段到原 Excel（含文件锁处理） | T-203, T-201 | 旧标题/旧封面回写原文件；被锁提示重试或 export_copy | TODO | -| T-205 | 首次未配账号/未登录的引导保护 | T-105, T-203 | 无账号/未登录时 ① 执行按钮禁用并提示去④ | TODO | +| T-204 | 回写旧字段到原 Excel（含文件锁处理） | T-203, T-201 | `excel.write_back()` 按源文件/工作表/行号回写旧标题、旧封面；`export_copy()` 另存副本；Tab① 用 `WriteBackWorker` 后台回写，文件被占用时提示关闭后重试 | DONE | +| T-204b | 采集完成后自动回写旧字段到 Excel | T-204 | `CollectWorker` 完成后自动触发 `excel.write_back()` 回写当前批次旧字段；成功时状态栏/日志提示“已回写”；原文件被锁时不影响 SQLite，提示关闭后点「回写旧数据到 Excel」手动重试或另存副本 | TODO | +| T-205 | 首次未配账号 / Chrome 未启动 / 未登录的引导保护 | T-105, T-203 | 无账号、匹配账号未启动 CDP 端口或未登录时，① 执行按钮禁用或采集前汇总提示，并引导去④；可提供“打开账号管理/启动登录”入口，但不无提示批量启动所有账号 Chrome | TODO | ## Phase 3 · AI 生成（②） diff --git a/docs/api.md b/docs/api.md index 464b719..927636f 100644 --- a/docs/api.md +++ b/docs/api.md @@ -88,7 +88,7 @@ SQLite 连接规则： - `connect()` 必须设置 `PRAGMA foreign_keys=ON`、`journal_mode=WAL`、`busy_timeout=5000`、`synchronous=NORMAL`。 - DB 写入短事务、单条提交；Excel 回写失败不回滚 DB。 -## excel 模块（`app/excel.py`，导入已建；回写待 T-204/T-403，依赖 openpyxl） +## excel 模块（`app/excel.py`，导入与旧字段回写已建；结果回写待 T-403，依赖 openpyxl） ```python class ExcelError(RuntimeError): ... @@ -116,12 +116,14 @@ match_summary(rows: list[dict], accounts: list) -> dict # -> {"matched": int, "unmatched": int, "by_account": {别名: 行数}, "unmatched_aliases": [..]} write_back(batch_id, excel_path=None, path=None) -> dict -# 把【旧标题/旧封面/新标题/新封面/更新状态】批量回写到【原 Excel】 +# 把【旧标题/旧封面图片路径】批量回写到【原 Excel】 # excel_path 为空则按 source_file_abs 分组回写本批次涉及的所有原文件 -# 原文件被占用(锁) → 抛错，调用方提示“请关闭后重试”，或改用 export_copy +# 只写 stage 已到 collected/generated/applied 或已有旧字段值的任务；无可写任务时返回 rows=0 +# 原文件被占用(锁) → 抛 ExcelError，调用方提示“请关闭后重试”，或改用 export_copy +# -> {"ok": True, "batch_id": str, "files": int, "rows": int, "written_files": [abs_path, ...]} export_copy(batch_id, out_dir_or_path, path=None) -> dict -# 退路：另存新结果文件，不动原文件 -# T-201 仅保证函数存在；实际回写/另存实现留给 T-204/T-403。 +# 退路：另存带旧字段的副本，不动原文件；目录输出时生成 *_cmshopee回写.xlsx +# T-403 会在此基础上扩展新标题/新封面/更新状态回写。 ``` 列模板见 [架构 5.3](04-architecture.md)；别名以“别名”列为权威，必须与 ④ 账号管理中的账号别名一致。`shopee待处理任务模板.xlsx` 是可提交的标准空模板；运营填写后的 Excel 副本属于业务数据，不提交。 @@ -273,6 +275,7 @@ main() -> int # 创建 QApplication + MainWindow class MainWindow(QMainWindow) # QTabWidget: ①②③④⑤；支持注入 db_path/config 便于测试 class CollectTab(QWidget) # ① 导入采集：导入 Excel + 汇总栏 + QTableView 任务列表 + 未匹配略过标记 class CollectWorker(BaseWorker) # ① 后台采集：登录检测 -> editor.collect -> db.set_collected/mark_skipped/mark_failed +class WriteBackWorker(BaseWorker) # ① 后台回写：excel.write_back(batch_id) 写旧标题/旧封面到原 Excel class TaskTableModel(QAbstractTableModel) # 任务表格模型：账号/别名/商品ID/阶段；未匹配别名显示“略过” class AccountsTab(QWidget) # ④ 账号管理：表格 + 增删改 + 启动登录 + 检测登录 + 快捷方式 class AccountDialog(QDialog) # 账号编辑弹窗；密码 QLineEdit.Password @@ -298,8 +301,10 @@ TAB_STYLE: str # 顶层 Tab 栏防误点样式： - 任务列表使用 `QTableView + TaskTableModel`，列为：账号、别名、商品ID、阶段。 - 账号列优先显示匹配到的 `accounts.account_name`；未匹配账号时保留 Excel 输入账号名。 - 别名未匹配 `accounts.alias` 时列表阶段列显示“略过”；点击「采集旧标题/旧封面」后由 `CollectWorker` 写库为 `skipped`，原因 `别名未匹配账号`。 -- 「采集旧标题/旧封面」通过 `CollectWorker` 后台执行，只处理 `stage=imported` 的任务；每条先检测登录，未登录写 `mark_skipped`，已登录则下载旧封面到 `image_dir//_old.jpg` 并 `db.set_collected()`；单条失败 `mark_failed(..., "collect", error)` 后继续。 +- 「采集旧标题/旧封面」通过 `CollectWorker` 后台执行，只处理 `stage=imported` 的任务；每条先检测登录，未登录写 `mark_skipped`，已登录则下载旧封面到 `image_dir//_old.jpg` 并 `db.set_collected()`；单条失败 `mark_failed(..., "collect", error)` 后继续。T-204b 后采集完成应自动触发当前批次旧字段回写，锁文件失败时只提示，不回滚 SQLite。 - 「停止」调用 worker 的协作式 `cancel()`，已开始的单条跑到安全边界后结束。 +- 「回写旧数据到 Excel」通过 `WriteBackWorker` 后台调用 `excel.write_back()`，把已采集旧标题/旧封面路径按原 Excel 行定位写回；T-204b 后该按钮主要作为自动回写失败后的手动重试入口。原文件被占用时弹窗提示关闭后重试，SQLite 采集结果不回滚。 +- T-205 后，若无账号、当前批次匹配账号未启动 CDP 端口或未登录，① 的采集执行应禁用或在执行前汇总拦截并提示去④账号管理；可以提供跳转/启动登录入口，但不无提示批量启动所有账号 Chrome。 ## workers 模块（`app/workers.py`，已建，PySide6） diff --git a/docs/current-state.md b/docs/current-state.md index 2d0b751..8802d1f 100644 --- a/docs/current-state.md +++ b/docs/current-state.md @@ -6,9 +6,9 @@ ## 当前快照 - 日期：2026-06-27 -- 阶段：V0 单账号 CDP 流程已验证；V1 已完成 T-000 正式代码包结构、T-001 `app/editor.py` 模块化、T-002 `app/appconfig.py` 应用配置、T-003 SQLite 持久化地基、T-004 本地数据忽略规则、T-005 AI 模型清单后端、T-006 单元测试基座、T-101 账号 user-data-dir 工具、T-102 Chrome 启动器、T-103 登录保活与检测、T-104 PySide6 主窗口骨架、T-104b PySide6 worker 基座、T-105 Tab④ 账号管理、T-106 账号快捷方式、T-201 Excel 导入入库、T-202 Tab① 任务列表与导入按钮、T-202b Tab① 导入汇总栏、T-203 采集旧标题旧封面。 +- 阶段：V0 单账号 CDP 流程已验证；V1 已完成 T-000 正式代码包结构、T-001 `app/editor.py` 模块化、T-002 `app/appconfig.py` 应用配置、T-003 SQLite 持久化地基、T-004 本地数据忽略规则、T-005 AI 模型清单后端、T-006 单元测试基座、T-101 账号 user-data-dir 工具、T-102 Chrome 启动器、T-103 登录保活与检测、T-104 PySide6 主窗口骨架、T-104b PySide6 worker 基座、T-105 Tab④ 账号管理、T-106 账号快捷方式、T-201 Excel 导入入库、T-202 Tab① 任务列表与导入按钮、T-202b Tab① 导入汇总栏、T-203 采集旧标题旧封面、T-204 回写旧字段到原 Excel。 - 技术栈：Python 3.10+，自研 CDP（websocket-client + requests），SQLite（sqlite3）+ `config.json` + openpyxl + AI（服务商待定），GUI PySide6 5 Tab（已定）。 -- 生产代码：已建立 `app/` 包 + 根入口 `main.py`；`app/cdp.py` 为已验证 CDP 底座；`app/editor.py` 已封装登录状态检测、标题/封面/采集/更新按钮能力；`app/appconfig.py` 已实现 `config.json` 默认值/读写/更新、AI 参数与端口读取，以及 `config/ai_models.json` 模型清单 CRUD/过滤/打码/测试连接；`app/db.py` 已实现 SQLite schema、连接 PRAGMA、批次/账号/任务与阶段写库函数；`app/excel.py` 已实现多 Excel 输入列解析、整文件列校验、脏行统计跳过、导入批次与任务入库、别名匹配统计；`app/config.py` 已实现账号 slug 与 user-data-dir 创建；`app/accounts.py` 已实现账号 CRUD 服务、端口默认分配、启动登录、检测登录、生成快捷方式；`app/chrome.py` 已实现 Chrome 参数拼装、启动、CDP 端口探测、PowerShell `.lnk` 快捷方式生成；`app/gui.py` 已实现 PySide6 `MainWindow`、五 Tab、顶部 Tab 栏防误点样式、① 导入采集的 Excel 导入按钮/导入汇总栏/QTableView 任务列表/未匹配筛选与略过标记/采集旧标题旧封面 worker、④ 账号管理表格/弹窗/按钮/快捷方式与状态栏；`app/workers.py` 已实现 `BaseWorker`、通用 signals、取消标记和 `QThread` 启动包装。 +- 生产代码：已建立 `app/` 包 + 根入口 `main.py`；`app/cdp.py` 为已验证 CDP 底座；`app/editor.py` 已封装登录状态检测、标题/封面/采集/更新按钮能力；`app/appconfig.py` 已实现 `config.json` 默认值/读写/更新、AI 参数与端口读取，以及 `config/ai_models.json` 模型清单 CRUD/过滤/打码/测试连接；`app/db.py` 已实现 SQLite schema、连接 PRAGMA、批次/账号/任务与阶段写库函数；`app/excel.py` 已实现多 Excel 输入列解析、整文件列校验、脏行统计跳过、导入批次与任务入库、别名匹配统计、旧标题/旧封面路径回写原 Excel 与另存副本；`app/config.py` 已实现账号 slug 与 user-data-dir 创建；`app/accounts.py` 已实现账号 CRUD 服务、端口默认分配、启动登录、检测登录、生成快捷方式；`app/chrome.py` 已实现 Chrome 参数拼装、启动、CDP 端口探测、PowerShell `.lnk` 快捷方式生成；`app/gui.py` 已实现 PySide6 `MainWindow`、五 Tab、顶部 Tab 栏防误点样式、① 导入采集的 Excel 导入按钮/导入汇总栏/QTableView 任务列表/未匹配筛选与略过标记/采集旧标题旧封面 worker/旧数据回写按钮与 worker、④ 账号管理表格/弹窗/按钮/快捷方式与状态栏；`app/workers.py` 已实现 `BaseWorker`、通用 signals、取消标记和 `QThread` 启动包装。 - 测试：`tests/` 已建立；T-006 后纯逻辑改动必须运行 `python -m unittest discover -s tests`，当前覆盖 appconfig/db/config/accounts/chrome 启动与快捷方式/editor 登录检测/excel 导入/gui ① 导入采集/gui ④ 账号管理/worker signal 与线程包装，并对尚未实现的 app.prompts 做契约占位 skip；CDP/Shopee 改动仍需测试商品手动验证。 - 数据：`config.json`、`config/ai_models.json`、`cmshopee.db`、`chrome_user_data_dir/`、`images/` 已由 `.gitignore` 排除；运营填写后的 Excel 业务文件默认忽略，标准空模板 `shopee待处理任务模板.xlsx` 可提交；`app/appconfig.py` 首次读取缺失的 `config.json` 时会在本地写默认配置，`app/db.py` 调用 `init_db()` 时会在本地创建 SQLite DB。 @@ -21,7 +21,7 @@ - 账号↔任务绑定：以 Excel“别名”列为权威；未匹配略过，结束弹窗汇总。 - 执行：多账号串行、单条失败继续；③ 点击「开始更新」后弹窗确认当前筛选范围和任务数量，确认后逐条点「更新」提交线上。 - AI：服务商待定；生成内容直接用于更新，本地留档+回写 Excel 供追溯。 -- 登录：人工登录 + 程序检测，不自动登录；无 Shopee tab 时检测入口为 `https:///`（默认 `https://seller.shopee.tw/`）；首次未配账号/未登录时 ① ③ 禁用并引导去④。 +- 登录：人工登录 + 程序检测，不自动登录；无 Shopee tab 时检测入口为 `https:///`（默认 `https://seller.shopee.tw/`）；首次未配账号、对应账号 Chrome 未启动或未登录时，① ③ 应禁用/提示并引导去④。 ## 当前目录要点 @@ -32,7 +32,7 @@ | `prototypes/` | 已有 | 已验证原型/探查脚本（demo/set_title/set_cover/get_title/cookies/inspect_images/grab/1.py），保留作人工回归与探查参考；见 `prototypes/README.md` | | `chrome-remote-debug-lan.md` | 已有 | WSL→Windows CDP 转发排查记录 | | `app/__init__.py` / `app/__main__.py` / `main.py` | 已有 | 正式包与启动入口；`python main.py` / `python -m app` 可运行占位入口 | -| `app/gui.py` | 已有 | T-104/T-105/T-106/T-202/T-202b/T-203 产出：PySide6 `QMainWindow` + 五 Tab；顶部 Tab 栏防误点样式；① 导入采集导入按钮、导入汇总栏、`QTableView` 任务列表、未匹配筛选与略过标记、采集旧标题旧封面 worker；④ 账号管理表格、账号弹窗、启动登录、检测登录、快捷方式 | +| `app/gui.py` | 已有 | T-104/T-105/T-106/T-202/T-202b/T-203/T-204 产出：PySide6 `QMainWindow` + 五 Tab；顶部 Tab 栏防误点样式；① 导入采集导入按钮、导入汇总栏、`QTableView` 任务列表、未匹配筛选与略过标记、采集旧标题旧封面 worker、旧数据回写 worker；④ 账号管理表格、账号弹窗、启动登录、检测登录、快捷方式 | | `app/workers.py` | 已有 | T-104b 产出：`BaseWorker` + 通用 signals + 取消标记 + `run_worker()` QThread 包装 | | `app/accounts.py` | 已有 | T-105/T-106 产出：账号 CRUD 服务、目录创建、端口分配、启动登录、检测登录、快捷方式 | | `app/editor.py` | 已有 | T-001/T-103 产出：登录状态检测、打开商品页、读/写标题、读/下载封面、上传拖封面、更新按钮、apply_task | @@ -40,8 +40,8 @@ | `app/db.py` | 已有 | T-003 产出：batches/accounts/tasks schema；WAL/busy_timeout/foreign_keys；账号/批次/任务与 set_* 阶段写库 | | `app/config.py` | 已有 | T-101 产出：别名→稳定 slug；创建并返回绝对 user-data-dir | | `app/chrome.py` | 已有 | T-102/T-106 产出：Chrome 启动参数、`subprocess.Popen` 启动、`/json/version` 端口探测、PowerShell `.lnk` 快捷方式 | -| `tests/` | 已有 | T-006/T-201/T-202/T-202b/T-203 产出：stdlib unittest 基座；覆盖 appconfig/db/config/accounts/chrome/editor/excel/gui/workers；prompts 模块契约占位测试 | -| `app/excel.py` | 已有 | T-201 产出：多文件 Excel 输入列解析、必需列整文件拒绝、脏行逐行跳过、批次/任务入库、匹配统计；回写留给 T-204/T-403 | +| `tests/` | 已有 | T-006/T-201/T-202/T-202b/T-203/T-204 产出：stdlib unittest 基座；覆盖 appconfig/db/config/accounts/chrome/editor/excel/gui/workers；prompts 模块契约占位测试 | +| `app/excel.py` | 已有 | T-201/T-204 产出：多文件 Excel 输入列解析、必需列整文件拒绝、脏行逐行跳过、批次/任务入库、匹配统计；按源文件/工作表/行号回写旧标题与旧封面路径；支持原文件被占用时另存副本 | | `shopee待处理任务模板.xlsx` | 已有，待提交 | 标准空 Excel 模板；单工作表 `待处理任务`，表头 `账号名 | 别名 | 商品id | 旧标题 | 旧封面图片路径 | 新标题 | 新封面图片路径 | 更新状态`；运营复制后填写，填写副本不提交 | | `config.json` / `config/ai_models.json` / `cmshopee.db` / `chrome_user_data_dir/` / `images/` | 本地待建，已忽略 | 含配置、密钥、业务、登录态、图片，不提交版本库 | @@ -57,9 +57,14 @@ 任务状态以 [`06-tasks.md`](06-tasks.md) 为准，历史记录见 [`../progress.md`](../progress.md)。 -- 已完成：T-000（正式代码包结构）、T-001（`app/editor.py` 模块化）、T-002（`app/appconfig.py` + `config.json`）、T-003（`app/db.py` + SQLite 建表）、T-004（本地数据 gitignore）、T-005（AI 模型清单后端）、T-006（单元测试基座）、T-101（账号 slug/user-data-dir）、T-102（Chrome 启动器）、T-103（登录保活与检测）、T-104（PySide6 五 Tab 主窗口骨架）、T-104b（PySide6 worker 基座）、T-105（Tab④ 账号管理）、T-106（账号快捷方式）、T-201（Excel 导入：解析多文件输入列入库）、T-202（Tab① 任务列表 + 导入按钮 + 别名匹配标记）、T-202b（Tab① 导入汇总栏）、T-203（采集旧标题+旧封面）。 +- 已完成：T-000（正式代码包结构）、T-001（`app/editor.py` 模块化）、T-002（`app/appconfig.py` + `config.json`）、T-003（SQLite 建表）、T-004（本地数据 gitignore）、T-005（AI 模型清单后端）、T-006（单元测试基座）、T-101（账号 slug/user-data-dir）、T-102（Chrome 启动器）、T-103（登录保活与检测）、T-104（PySide6 五 Tab 主窗口骨架）、T-104b（PySide6 worker 基座）、T-105（Tab④ 账号管理）、T-106（账号快捷方式）、T-201（Excel 导入：解析多文件输入列入库）、T-202（Tab① 任务列表 + 导入按钮 + 别名匹配标记）、T-202b（Tab① 导入汇总栏）、T-203（采集旧标题+旧封面）、T-204（回写旧字段到原 Excel）。 - 正在进行：无。 -- 下一个可领取任务：**T-204（回写旧字段到原 Excel，含文件锁处理）**。 +- 下一个可领取任务：**T-204b（采集完成后自动回写旧字段到 Excel）**。 + +## 当前已发现待修体验问题 + +- T-204 已提供手动「回写旧数据到 Excel」，但当前采集完成后不会自动写回原 Excel；已拆为 T-204b 修复，目标是采集完成自动回写，失败时保留手动重试。 +- ① 采集依赖对应账号 Chrome 已用专属 user-data-dir 和 CDP 端口启动并登录；若未启动或未登录，当前会采集失败/略过。已在 T-205 明确处理：禁用或提示，并引导去④账号管理启动登录，不无提示批量启动所有账号 Chrome。 ## 当前可运行内容 diff --git a/docs/routes.md b/docs/routes.md index c1f7ebd..920e7b2 100644 --- a/docs/routes.md +++ b/docs/routes.md @@ -30,7 +30,8 @@ ## 首次使用引导保护 - ① 导入采集 与 ③ 更新shopee 都依赖**账号已配置且已登录**（在 ④ 账号管理）。 -- 当无账号 / 账号未登录时：相关执行按钮**禁用**，并提示「请先到『账号管理』配置账号并登录」。 +- 当无账号 / 对应账号 Chrome 未启动 / 账号未登录时：相关执行按钮**禁用或在执行前汇总拦截**，并提示「请先到『账号管理』配置账号并登录」。 +- 可以提供「打开账号管理」或「启动登录」入口辅助用户处理当前账号；不要无提示批量启动所有账号 Chrome，避免开错账号或启动过多浏览器进程。 - 老用户账号已就绪则无感。 ## ① 导入采集 @@ -51,8 +52,8 @@ - 导入：openpyxl 解析**输入列**（账号名/别名/商品id）入 SQLite。 - **导入汇总栏**（导入后即时刷新，跑采集前的校验关口）：显示 文件数、解析行数（原始数据量）、有效/无效行、匹配账号行数（按账号细分）、未匹配行数。未匹配/无效数字标红可点，点击在列表筛出便于定位纠错。 -- 采集：用该账号 Chrome 只读打开商品页，读旧标题、下载旧封面到本地图片目录，写 `old_title/old_cover_path`，stage=collected。 -- 回写：采集完把旧标题/旧封面路径批量回写原 Excel（原文件被锁→提示重试/另存）。 +- 采集：用该账号已启动并登录的 Chrome 只读打开商品页，读旧标题、下载旧封面到本地图片目录，写 `old_title/old_cover_path`，stage=collected。 +- 回写：采集完成后自动把旧标题/旧封面路径批量回写原 Excel；保留「回写旧数据到 Excel」作为手动重试入口（原文件被锁→提示关闭后重试/另存）。 - 别名未匹配账号 / 账号未登录 → 该行 skipped 并记原因。 ## ② AI生成 @@ -130,14 +131,14 @@ ```text ④ 账号管理：配账号 + 启动登录（首次必做） │ -① 导入采集：导入 Excel → 采集旧标题/旧封面 → 回写旧字段 +① 导入采集：导入 Excel → 采集旧标题/旧封面 → 自动回写旧字段（失败可手动重试） │ ② AI生成：提示词 → 生成新标题/新封面（无逐条审核） │ ③ 更新shopee：对已生成任务点击开始更新 → 弹窗确认 → 换标题+封面 → 点「更新」提交 → 回写结果 ``` -- 未配账号/未登录：① ③ 的执行按钮禁用并提示去 ④。 +- 未配账号 / Chrome 未启动 / 未登录：① ③ 的执行按钮禁用或执行前提示去 ④。 - 已生成的任务即可进 ③；③ 用户确认批量弹窗后提交线上，无常驻提交开关。 - 任意步骤失败：记入该任务、日志标明，不影响其他任务。 diff --git a/progress.md b/progress.md index 05d6b8c..d738935 100644 --- a/progress.md +++ b/progress.md @@ -417,3 +417,20 @@ - 变更：`app/editor.py` 将无 Shopee tab 时的登录检测入口从 `https:///portal/` 改为 `https:///`，默认即 `https://seller.shopee.tw/`；更新 `tests/test_editor_login.py`、`docs/api.md`、`docs/04-architecture.md`、`docs/current-state.md`。 - 决策：商品编辑页 URL 仍保持 `/portal/product/?pageEntry=product_list&ignore-html-cache=1`，本轮只改登录检测/人工登录入口。 - 验证：`python -m unittest discover -s tests -p "test_editor_login.py"` 通过（5 tests）；`python -m unittest discover -s tests -p "test_accounts.py"` 通过（8 tests）；`python -m compileall app main.py tests` 通过；`python -m unittest discover -s tests` 通过（49 tests，skipped=1）；`py -3 -m compileall app main.py tests` 通过；`py -3 -m unittest discover -s tests` 通过（34 tests，skipped=4）。 + +## 【2026-06-27】T-204 回写旧字段到原 Excel + +- 状态：DONE +- 变更：`app/excel.py` 实现 `write_back()` 与 `export_copy()`：按 `tasks.source_file_abs/source_sheet/source_row` 定位原 Excel 行，回写 `旧标题` 与 `旧封面图片路径`；输出列缺失时自动追加；`.xlsm` 写入时保留 VBA；原文件被占用或无权限时抛中文 `ExcelError`，SQLite 采集结果不回滚。`app/gui.py` 在 Tab① 增加「回写旧数据到 Excel」按钮与 `WriteBackWorker`，后台调用 `excel.write_back()`，锁文件时弹窗提示关闭后重试或另存副本。更新 `tests/test_excel.py`、`tests/test_gui.py`、`docs/06-tasks.md`、`docs/current-state.md`、`docs/api.md`。 +- 细节：`write_back(batch_id, excel_path=None, path=None)` 默认按批次涉及的源文件分组回写；只写已到 `collected/generated/applied` 阶段或已有旧字段值的任务；无可写行时返回 `rows=0`。`export_copy(batch_id, out_dir_or_path, path=None)` 不修改原文件，目录输出时生成 `*_cmshopee回写.xlsx`，同名自动加序号。 +- 验证：`python -m unittest discover -s tests -p "test_excel.py"` 通过（7 tests）；`python -m unittest discover -s tests -p "test_gui.py"` 通过（11 tests）；`python -m compileall app main.py tests` 通过；`python -m unittest discover -s tests` 通过（53 tests，skipped=1）；`py -3 -m compileall app main.py tests` 通过；`py -3 -m unittest discover -s tests` 通过（34 tests，skipped=4，py -3 环境缺 openpyxl/PySide6，相关测试按设计跳过）。 +- 注意：本轮只做 Excel 回写与 GUI 后台接入，不涉及 Shopee/CDP 页面操作，无需测试商品实跑。`export_copy()` 已有后端能力，Tab① 当前只在锁文件提示中说明可另存副本，尚未提供另存按钮。 +- 下一步：按任务看板领取 T-205（首次未配账号/未登录的引导保护）。 + +## 【2026-06-27】文档补充 · 采集自动回写与 Chrome 未启动引导 + +- 状态：DONE（文档调整） +- 背景：运行最新代码后发现两个体验缺口：① 采集旧标题/旧封面后仍需手动点「回写旧数据到 Excel」；② 未在④账号管理启动对应账号 Chrome 时无法采集。 +- 变更：`docs/06-tasks.md` 新增 T-204b（采集完成后自动回写旧字段到 Excel），并把 T-205 扩展为“未配账号 / Chrome 未启动 / 未登录”的引导保护；同步 `docs/current-state.md` 下一个可领取任务为 T-204b，并补充当前已发现待修体验问题；同步 `docs/routes.md`、`docs/api.md`、`docs/04-architecture.md`、`docs/02-requirements.md`。 +- 决策：T-204b 先解决采集闭环自动回写，保留手动回写作为锁文件失败后的重试入口；T-205 解决 Chrome 未启动/未登录的禁用、提示和跳转④，不做无提示批量启动所有账号 Chrome。 +- 验证：文档-only 更新，未运行单元测试。 diff --git a/tests/test_excel.py b/tests/test_excel.py index 87c7f1e..6215458 100644 --- a/tests/test_excel.py +++ b/tests/test_excel.py @@ -1,10 +1,11 @@ import os import unittest +from unittest import mock from _helpers import TempDirMixin try: - from openpyxl import Workbook + from openpyxl import Workbook, load_workbook except ModuleNotFoundError: raise unittest.SkipTest("openpyxl 未安装") @@ -23,6 +24,32 @@ class ExcelImportTests(TempDirMixin, unittest.TestCase): workbook.save(path) workbook.close() + def row_values_by_header(self, path, row_number=2): + workbook = load_workbook(path, data_only=True) + try: + sheet = workbook.active + headers = { + sheet.cell(row=1, column=column).value: column + for column in range(1, sheet.max_column + 1) + } + return { + header: sheet.cell(row=row_number, column=column).value + for header, column in headers.items() + } + finally: + workbook.close() + + def headers(self, path): + workbook = load_workbook(path, data_only=True) + try: + sheet = workbook.active + return [ + sheet.cell(row=1, column=column).value + for column in range(1, sheet.max_column + 1) + ] + finally: + workbook.close() + def test_import_tasks_parses_rows_and_writes_batch_tasks(self): with self.make_temp_dir() as temp_dir: db_path = os.path.join(temp_dir, "cmshopee.db") @@ -162,6 +189,104 @@ class ExcelImportTests(TempDirMixin, unittest.TestCase): excel.match_summary(rows, accounts), ) + def test_write_back_writes_old_fields_to_original_excel(self): + with self.make_temp_dir() as temp_dir: + db_path = os.path.join(temp_dir, "cmshopee.db") + excel_path = os.path.join(temp_dir, "input.xlsx") + self.save_workbook( + excel_path, + [ + ( + "待处理任务", + [ + ["账号名", "别名", "商品id"], + ["主店", "alias-a", "51100639510"], + ], + ) + ], + ) + result = excel.import_tasks([excel_path], path=db_path) + task = db.list_tasks(batch_id=result["batch_id"], path=db_path)[0] + db.set_collected( + task.id, + "原始商品标题", + r"D:\images\51100639510_old.jpg", + path=db_path, + ) + + summary = excel.write_back(result["batch_id"], path=db_path) + + self.assertEqual(True, summary["ok"]) + self.assertEqual(1, summary["files"]) + self.assertEqual(1, summary["rows"]) + self.assertEqual([os.path.abspath(excel_path)], summary["written_files"]) + values = self.row_values_by_header(excel_path) + self.assertEqual("原始商品标题", values["旧标题"]) + self.assertEqual(r"D:\images\51100639510_old.jpg", values["旧封面图片路径"]) + + self.assert_removed(temp_dir) + + def test_write_back_locked_file_raises_clear_error(self): + with self.make_temp_dir() as temp_dir: + db_path = os.path.join(temp_dir, "cmshopee.db") + excel_path = os.path.join(temp_dir, "input.xlsx") + self.save_workbook( + excel_path, + [ + ( + "待处理任务", + [ + ["账号名", "别名", "商品id"], + ["主店", "alias-a", "51100639510"], + ], + ) + ], + ) + result = excel.import_tasks([excel_path], path=db_path) + task = db.list_tasks(batch_id=result["batch_id"], path=db_path)[0] + db.set_collected(task.id, "原始商品标题", "old.jpg", path=db_path) + + with mock.patch("app.excel.load_workbook", side_effect=PermissionError("locked")): + with self.assertRaisesRegex(excel.ExcelError, "Excel 文件被占用"): + excel.write_back(result["batch_id"], path=db_path) + + self.assert_removed(temp_dir) + + def test_export_copy_writes_copy_without_touching_original(self): + with self.make_temp_dir() as temp_dir: + db_path = os.path.join(temp_dir, "cmshopee.db") + excel_path = os.path.join(temp_dir, "input.xlsx") + out_dir = os.path.join(temp_dir, "out") + self.save_workbook( + excel_path, + [ + ( + "待处理任务", + [ + ["账号名", "别名", "商品id"], + ["主店", "alias-a", "51100639510"], + ], + ) + ], + ) + result = excel.import_tasks([excel_path], path=db_path) + task = db.list_tasks(batch_id=result["batch_id"], path=db_path)[0] + db.set_collected(task.id, "原始商品标题", "old.jpg", path=db_path) + + summary = excel.export_copy(result["batch_id"], out_dir, path=db_path) + + self.assertEqual(1, summary["files"]) + self.assertEqual(1, summary["rows"]) + copy_path = summary["written_files"][0] + self.assertTrue(os.path.exists(copy_path)) + self.assertIn("input_cmshopee回写", os.path.basename(copy_path)) + self.assertNotIn("旧标题", self.headers(excel_path)) + values = self.row_values_by_header(copy_path) + self.assertEqual("原始商品标题", values["旧标题"]) + self.assertEqual("old.jpg", values["旧封面图片路径"]) + + self.assert_removed(temp_dir) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_gui.py b/tests/test_gui.py index c714e30..2057416 100644 --- a/tests/test_gui.py +++ b/tests/test_gui.py @@ -24,6 +24,7 @@ from app.gui import ( MainWindow, TAB_STYLE, TAB_TITLES, + WriteBackWorker, ) @@ -57,6 +58,10 @@ class GuiTests(TempDirMixin, unittest.TestCase): self.assertIn("padding: 8px 18px", window.tabs.styleSheet()) self.assertIn("margin-right: 8px", window.tabs.styleSheet()) self.assertIsInstance(window.tabs.widget(0), CollectTab) + self.assertEqual( + "回写旧数据到 Excel", + window.tabs.widget(0).write_back_button.text(), + ) self.assert_removed(temp_dir) @@ -382,6 +387,20 @@ class GuiTests(TempDirMixin, unittest.TestCase): self.assert_removed(temp_dir) + def test_write_back_worker_calls_excel_write_back(self): + with mock.patch( + "app.gui.excel.write_back", + return_value={"ok": True, "batch_id": "batch-1", "files": 1, "rows": 2}, + ) as write_back: + summary = WriteBackWorker("batch-1", db_path="db.sqlite").execute() + + self.assertEqual({"ok": True, "batch_id": "batch-1", "files": 1, "rows": 2}, summary) + write_back.assert_called_once_with( + "batch-1", + excel_path=None, + path="db.sqlite", + ) + if __name__ == "__main__": unittest.main()