From 893aa5688193917af7791b049079270b04230b19 Mon Sep 17 00:00:00 2001 From: chengma Date: Sat, 27 Jun 2026 10:59:51 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=8C=E6=88=90Excel=E5=AF=BC?= =?UTF-8?q?=E5=85=A5=E5=85=A5=E5=BA=93?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 实现 app/excel.py 多文件导入，解析账号名、别名、商品id，记录 source_file_abs/source_sheet/source_row/row_key。 导入时支持缺必需列整文件拒绝、脏行逐行跳过统计，并写入 batches/tasks；补充 match_summary 与回写占位。 新增 tests/test_excel.py 覆盖入库、缺列拒绝、多文件容错和别名匹配；同步任务看板、API 合约、当前状态和 progress。 --- app/excel.py | 288 ++++++++++++++++++++++++++++++++++++++++++ docs/06-tasks.md | 2 +- docs/api.md | 32 +++-- docs/current-state.md | 19 +-- progress.md | 9 ++ tests/test_excel.py | 167 ++++++++++++++++++++++++ 6 files changed, 500 insertions(+), 17 deletions(-) create mode 100644 app/excel.py create mode 100644 tests/test_excel.py diff --git a/app/excel.py b/app/excel.py new file mode 100644 index 0000000..5e2844a --- /dev/null +++ b/app/excel.py @@ -0,0 +1,288 @@ +"""Excel import and write-back helpers.""" + +from __future__ import annotations + +import os +import re +from typing import Iterable, Optional + +from . import db + + +try: + from openpyxl import load_workbook + + OPENPYXL_IMPORT_ERROR = None +except ModuleNotFoundError as exc: + load_workbook = None + OPENPYXL_IMPORT_ERROR = exc + + +REQUIRED_FIELDS = ("alias", "item_id") +HEADER_ALIASES = { + "account_name": {"账号名", "账号名称", "店铺", "店铺名", "shop", "accountname"}, + "alias": {"别名", "账号别名", "alias"}, + "item_id": {"商品id", "商品编号", "商品id号", "itemid", "item"}, +} + + +class ExcelError(RuntimeError): + """Raised when Excel import or export cannot complete.""" + + +def _require_openpyxl() -> None: + if OPENPYXL_IMPORT_ERROR is not None: + raise ExcelError("openpyxl 未安装，无法读取 Excel 文件") + + +def _text(value) -> str: + if value is None: + return "" + if isinstance(value, float) and value.is_integer(): + return str(int(value)) + return str(value).strip() + + +def _normalize_header(value) -> str: + return re.sub(r"[\s_]+", "", _text(value).lower()) + + +def _field_for_header(value) -> Optional[str]: + normalized = _normalize_header(value) + for field, names in HEADER_ALIASES.items(): + if normalized in {_normalize_header(name) for name in names}: + return field + return None + + +def _normalize_item_id(value) -> str: + text = _text(value) + if re.fullmatch(r"\d+\.0", text): + text = text[:-2] + return text + + +def _valid_item_id(value) -> bool: + return bool(re.fullmatch(r"\d+", value or "")) + + +def _row_key(batch_id, source_file_abs, source_sheet, source_row) -> str: + return f"{batch_id}:{source_file_abs}:{source_sheet}:{source_row}" + + +def _empty_row(values) -> bool: + return all(_text(value) == "" for value in values) + + +def _headers_for_sheet(sheet): + first_row = next(sheet.iter_rows(min_row=1, max_row=1, values_only=True), ()) + if _empty_row(first_row): + return None + headers = {} + for index, value in enumerate(first_row): + field = _field_for_header(value) + if field and field not in headers: + headers[field] = index + return headers + + +def _stats(): + return { + "files": 0, + "total": 0, + "valid": 0, + "invalid": 0, + "inserted": 0, + "file_errors": [], + "row_errors": [], + } + + +def _file_error(file_path, message, sheet=None, missing_columns=None): + error = { + "file": file_path, + "error": message, + } + if sheet is not None: + error["sheet"] = sheet + if missing_columns: + error["missing_columns"] = list(missing_columns) + return error + + +def _row_error(file_path, sheet, row_number, message): + return { + "file": file_path, + "sheet": sheet, + "row": row_number, + "error": message, + } + + +def _read_file(file_path, stats): + source_file = str(file_path) + source_file_abs = os.path.abspath(source_file) + try: + workbook = load_workbook(source_file_abs, read_only=True, data_only=True) + except Exception as exc: + stats["file_errors"].append(_file_error(source_file, f"读取失败: {exc}")) + return [] + + try: + usable_sheets = [] + sheet_errors = [] + for sheet in workbook.worksheets: + headers = _headers_for_sheet(sheet) + if headers is None: + continue + missing = [ + "别名" if field == "alias" else "商品id" + for field in REQUIRED_FIELDS + if field not in headers + ] + if missing: + sheet_errors.append( + _file_error( + source_file, + "缺少必需列", + sheet=sheet.title, + missing_columns=missing, + ) + ) + else: + usable_sheets.append((sheet, headers)) + + if sheet_errors: + stats["file_errors"].extend(sheet_errors) + return [] + if not usable_sheets: + stats["file_errors"].append(_file_error(source_file, "没有可解析工作表")) + return [] + + rows = [] + for sheet, headers in usable_sheets: + max_column = max(headers.values()) + 1 + for row_number, values in enumerate( + sheet.iter_rows( + min_row=2, + max_col=max_column, + values_only=True, + ), + start=2, + ): + if _empty_row(values): + continue + stats["total"] += 1 + account_name = _text(values[headers["account_name"]]) if "account_name" in headers else "" + alias = _text(values[headers["alias"]]) + item_id = _normalize_item_id(values[headers["item_id"]]) + if not alias: + stats["invalid"] += 1 + stats["row_errors"].append( + _row_error(source_file, sheet.title, row_number, "别名不能为空") + ) + continue + if not item_id: + stats["invalid"] += 1 + stats["row_errors"].append( + _row_error(source_file, sheet.title, row_number, "商品id不能为空") + ) + continue + if not _valid_item_id(item_id): + stats["invalid"] += 1 + stats["row_errors"].append( + _row_error(source_file, sheet.title, row_number, "商品id必须是数字") + ) + continue + + rows.append( + { + "source_file": source_file, + "source_file_abs": source_file_abs, + "source_sheet": sheet.title, + "source_row": row_number, + "account_name": account_name or None, + "alias": alias, + "item_id": item_id, + } + ) + stats["valid"] += 1 + return rows + finally: + workbook.close() + + +def import_tasks(file_paths: Iterable[str], path=None, note=None, write_db=True) -> dict: + """Parse input Excel files and optionally insert rows into SQLite.""" + + _require_openpyxl() + if isinstance(file_paths, (str, os.PathLike)): + paths = [str(file_paths)] + else: + paths = [str(file_path) for file_path in file_paths] + stats = _stats() + stats["files"] = len(paths) + rows = [] + for file_path in paths: + rows.extend(_read_file(file_path, stats)) + + batch_id = None + if write_db and rows: + db.init_db(path) + batch_id = db.create_batch(paths, note=note, path=path) + for row in rows: + row["row_key"] = _row_key( + batch_id, + row["source_file_abs"], + row["source_sheet"], + row["source_row"], + ) + stats["inserted"] = db.insert_tasks(batch_id, rows, path=path) + return { + "batch_id": batch_id, + "rows": rows, + "stats": stats, + } + + +def _account_alias(account) -> str: + if isinstance(account, dict): + return str(account.get("alias") or "").strip() + return str(getattr(account, "alias", "") or "").strip() + + +def match_summary(rows: list[dict], accounts: list) -> dict: + """Summarize whether imported rows match configured account aliases.""" + + known_aliases = {_account_alias(account) for account in accounts if _account_alias(account)} + by_account = {} + unmatched_aliases = [] + matched = 0 + unmatched = 0 + for row in rows: + alias = str(row.get("alias") or "").strip() + if alias in known_aliases: + matched += 1 + by_account[alias] = by_account.get(alias, 0) + 1 + else: + unmatched += 1 + if alias and alias not in unmatched_aliases: + unmatched_aliases.append(alias) + return { + "matched": matched, + "unmatched": unmatched, + "by_account": by_account, + "unmatched_aliases": unmatched_aliases, + } + + +def write_back(batch_id, excel_path=None, path=None) -> dict: + """Placeholder for Excel result write-back, implemented in later tasks.""" + + raise ExcelError("Excel 回写将在 T-204/T-403 实现") + + +def export_copy(batch_id, out_dir_or_path, path=None) -> dict: + """Placeholder for exporting a copy when the original Excel is locked.""" + + raise ExcelError("Excel 另存副本将在 T-204/T-403 实现") diff --git a/docs/06-tasks.md b/docs/06-tasks.md index 09d3089..92e5a6f 100644 --- a/docs/06-tasks.md +++ b/docs/06-tasks.md @@ -47,7 +47,7 @@ | ID | 任务 | 依赖 | 验收要点 | 状态 | | --- | --- | --- | --- | --- | -| T-201 | `app/excel.py` 导入：解析多文件输入列入库 | T-003 | 按模板解析账号名/别名/商品id；记录 source_file_abs/source_sheet/source_row/row_key；缺必需列则拒绝整文件并记 file_errors；脏行逐行跳过计 invalid；写 batches/tasks | TODO | +| T-201 | `app/excel.py` 导入：解析多文件输入列入库 | T-003 | 按模板解析账号名/别名/商品id；记录 source_file_abs/source_sheet/source_row/row_key；缺必需列则拒绝整文件并记 file_errors；脏行逐行跳过计 invalid；写 batches/tasks | DONE | | T-202 | Tab① 任务列表 + 导入按钮 + 别名匹配标记 | T-201, T-105 | `QTableView` 显示账号/别名/商品id/阶段；未匹配标“略过” | TODO | | T-202b | Tab① 导入汇总栏 | T-202 | 导入后显示 文件数/解析行数/有效/无效/匹配(按账号)/未匹配；未匹配可点击筛出 | TODO | | T-203 | 采集旧标题+旧封面（只读），下载图片，立即写库 | T-202, T-001, T-104b | 通过 worker 执行；逐条 set_collected；旧封面下载到 `images//`；未登录/未匹配略过记原因 | TODO | diff --git a/docs/api.md b/docs/api.md index 56e018a..f57b7f6 100644 --- a/docs/api.md +++ b/docs/api.md @@ -88,24 +88,40 @@ SQLite 连接规则： - `connect()` 必须设置 `PRAGMA foreign_keys=ON`、`journal_mode=WAL`、`busy_timeout=5000`、`synchronous=NORMAL`。 - DB 写入短事务、单条提交；Excel 回写失败不回滚 DB。 -## excel 模块（`app/excel.py`，待建，依赖 openpyxl） +## excel 模块（`app/excel.py`，导入已建；回写待 T-204/T-403，依赖 openpyxl） ```python -import_tasks(file_paths: list[str]) -> dict -# 只解析【输入列】：账号名、别名、商品id；并记录 source_file/source_file_abs/source_sheet/source_row/row_key -# -> {"rows": [...], "stats": {"files": int, "total": int, "valid": int, "invalid": int, "file_errors": [...]}} -# 必需列缺失（别名/商品id）= 整个文件拒绝并记录到 file_errors，不导入该文件任何行 -# invalid = 单行缺别名/商品id 或商品id格式错误等脏数据；逐行跳过，不阻塞同文件其他有效行 +class ExcelError(RuntimeError): ... +OPENPYXL_IMPORT_ERROR: Exception | None + +import_tasks(file_paths, path=None, note=None, write_db=True) -> dict +# file_paths 可传单个路径或多个路径。 +# 只解析【输入列】：账号名(可选)、别名、商品id；并记录 +# source_file/source_file_abs/source_sheet/source_row/row_key。 +# 默认 write_db=True：有有效行时创建 batch 并写入 tasks。 +# -> { +# "batch_id": str|None, +# "rows": [...], +# "stats": { +# "files": int, "total": int, "valid": int, "invalid": int, +# "inserted": int, "file_errors": [...], "row_errors": [...] +# } +# } +# 必需列缺失（别名/商品id）= 整个文件拒绝并记录到 file_errors，不导入该文件任何行。 +# invalid = 单行缺别名/商品id 或商品id 格式错误等脏数据；逐行跳过，不阻塞同文件其他有效行。 +# openpyxl 缺失时模块仍可导入，调用 import_tasks 时抛 ExcelError。 match_summary(rows: list[dict], accounts: list) -> dict # 用 accounts 的别名对 rows 做匹配统计（导入汇总栏用） # -> {"matched": int, "unmatched": int, "by_account": {别名: 行数}, "unmatched_aliases": [..]} -write_back(batch_id, excel_path=None) -> dict +write_back(batch_id, excel_path=None, path=None) -> dict # 把【旧标题/旧封面/新标题/新封面/更新状态】批量回写到【原 Excel】 # excel_path 为空则按 source_file_abs 分组回写本批次涉及的所有原文件 # 原文件被占用(锁) → 抛错，调用方提示“请关闭后重试”，或改用 export_copy -export_copy(batch_id, out_dir_or_path) -> dict # 退路：另存新结果文件，不动原文件 +export_copy(batch_id, out_dir_or_path, path=None) -> dict +# 退路：另存新结果文件，不动原文件 +# T-201 仅保证函数存在；实际回写/另存实现留给 T-204/T-403。 ``` 列模板见 [架构 5.3](04-architecture.md)；别名以“别名”列为权威。 diff --git a/docs/current-state.md b/docs/current-state.md index ca33c10..fd086bb 100644 --- a/docs/current-state.md +++ b/docs/current-state.md @@ -6,10 +6,10 @@ ## 当前快照 - 日期：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 账号快捷方式。 +- 阶段：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 导入入库。 - 技术栈：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/config.py` 已实现账号 slug 与 user-data-dir 创建；`app/accounts.py` 已实现账号 CRUD 服务、端口默认分配、启动登录、检测登录、生成快捷方式；`app/chrome.py` 已实现 Chrome 参数拼装、启动、CDP 端口探测、PowerShell `.lnk` 快捷方式生成；`app/gui.py` 已实现 PySide6 `MainWindow`、五 Tab、顶部 Tab 栏防误点样式、④ 账号管理表格/弹窗/按钮/快捷方式与状态栏；`app/workers.py` 已实现 `BaseWorker`、通用 signals、取消标记和 `QThread` 启动包装。 -- 测试：`tests/` 已建立；T-006 后纯逻辑改动必须运行 `python -m unittest discover -s tests`，当前覆盖 appconfig/db/config/accounts/chrome 启动与快捷方式/editor 登录检测/gui 账号管理/worker signal 与线程包装，并对尚未实现的 app.excel/app.prompts 做契约占位 skip；CDP/Shopee 改动仍需测试商品手动验证。 +- 生产代码：已建立 `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 栏防误点样式、④ 账号管理表格/弹窗/按钮/快捷方式与状态栏；`app/workers.py` 已实现 `BaseWorker`、通用 signals、取消标记和 `QThread` 启动包装。 +- 测试：`tests/` 已建立；T-006 后纯逻辑改动必须运行 `python -m unittest discover -s tests`，当前覆盖 appconfig/db/config/accounts/chrome 启动与快捷方式/editor 登录检测/excel 导入/gui 账号管理/worker signal 与线程包装，并对尚未实现的 app.prompts 做契约占位 skip；CDP/Shopee 改动仍需测试商品手动验证。 - 数据：`config.json`、`config/ai_models.json`、`cmshopee.db`、`chrome_user_data_dir/`、`images/` 已由 `.gitignore` 排除；`app/appconfig.py` 首次读取缺失的 `config.json` 时会在本地写默认配置，`app/db.py` 调用 `init_db()` 时会在本地创建 SQLite DB。 ## 既定设计要点（文档已定） @@ -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 产出：stdlib unittest 基座；覆盖 appconfig/db/config/accounts/chrome/editor/gui/workers；excel/prompts 模块契约占位测试 | -| `app/excel.py` | 待建 | Phase 2 产出 | +| `tests/` | 已有 | T-006/T-201 产出：stdlib unittest 基座；覆盖 appconfig/db/config/accounts/chrome/editor/excel/gui/workers；prompts 模块契约占位测试 | +| `app/excel.py` | 已有 | T-201 产出：多文件 Excel 输入列解析、必需列整文件拒绝、脏行逐行跳过、批次/任务入库、匹配统计；回写留给 T-204/T-403 | | `config.json` / `config/ai_models.json` / `cmshopee.db` / `chrome_user_data_dir/` / `images/` | 本地待建，已忽略 | 含配置、密钥、业务、登录态、图片，不提交版本库 | ## 已验证能力（单账号） @@ -56,9 +56,9 @@ 任务状态以 [`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-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-201（`app/excel.py` 导入：解析多文件输入列入库）**。 +- 下一个可领取任务：**T-202（Tab① 任务列表 + 导入按钮 + 别名匹配标记）**。 ## 当前可运行内容 @@ -84,6 +84,9 @@ py -3 -c "import tempfile; from app import config; d=tempfile.TemporaryDirectory # chrome 参数拼装 / 端口探测由 tests/test_chrome.py 覆盖 python -m unittest discover -s tests +# Excel 导入由 tests/test_excel.py 覆盖；默认 python 环境需可导入 openpyxl 才执行真实解析测试 +python -m unittest discover -s tests -p "test_excel.py" + # GUI 骨架与 ④ 账号管理由 tests/test_gui.py 覆盖；worker 基座由 tests/test_workers.py 覆盖 # 默认 python 当前 PySide6=6.5.3，py -3 环境缺 PySide6 时相关测试会 skip @@ -105,7 +108,7 @@ set UPDATE=1 && python prototypes/demo.py 前置条件： - Chrome 已用某 user-data-dir 带 `--remote-debugging-port` + `--remote-allow-origins=*` 启动并登录 Shopee。 -- 已 `pip install websocket-client requests`。 +- 已 `pip install websocket-client requests`；Excel 导入还需要 `openpyxl`，默认 `python` 环境本轮已可执行真实解析测试，`py -3` 环境缺 openpyxl 时 `tests/test_excel.py` 会按设计跳过。 - PySide6 当前环境已可导入（验证版本 6.5.3）；GUI 实现固定使用 PySide6。 - 默认连 `127.0.0.1:9222`（开发期可用 `CDP_HOST` 指向 WSL 转发的 `192.168.0.224:9333`）。 - 本机 `python` 当前指向 `C:\Python37\python.exe`（3.7.9，isolated，不符合 Python 3.10+ 要求且 `python -m app` 不搜索当前目录）；开发/验证优先用 `py -3`（当前 3.14.4）或确认 `python` 已指向 3.10+。 diff --git a/progress.md b/progress.md index 4ec23bf..8ece3d1 100644 --- a/progress.md +++ b/progress.md @@ -367,3 +367,12 @@ - 结论：按文档完成；5 Tab 顺序/降级、worker 6 信号与线程纪律、账号④增删改/打码/不自动登录、`.lnk` 含账号参数均对上；33 测试通过（skipped=4 为 excel/prompts 契约测试）。 - 待办：账号启动登录、快捷方式、登录检测本轮均为 mock；接 Phase 2（T-201/T-203 真连 CDP）前，建议 `editor.py` 真机冒烟 + 账号④走一遍（新增→启动→人工登录→检测→生成快捷方式）一起验。 - 完整报告：[`docs/reviews/2026-06-27-T104-T106-acceptance.md`](docs/reviews/2026-06-27-T104-T106-acceptance.md) + +## 【2026-06-27】T-201 Excel 导入入库 + +- 状态：DONE +- 变更：新增 `app/excel.py`，实现多 Excel 文件导入、输入列解析、必需列校验、脏行跳过、`batches/tasks` 入库、导入统计与 `match_summary()`；新增 `tests/test_excel.py` 覆盖有效/无效行、缺必需列整文件拒绝、DB 写入和别名匹配统计；同步 `docs/06-tasks.md`、`docs/current-state.md`、`docs/api.md`。 +- 细节：输入列只解析 `账号名`（可选）、`别名`、`商品id`；有效行记录 `source_file/source_file_abs/source_sheet/source_row/row_key`；缺 `别名` 或 `商品id` 的非空工作表会拒绝整个文件并写入 `file_errors`；单行缺别名、缺商品 id 或商品 id 非数字会跳过并计入 `invalid/row_errors`。`write_back()` 与 `export_copy()` 本轮只保留可调用占位，实际回写留给 T-204/T-403。 +- 验证：`python -m compileall app main.py tests` 通过；`python -m unittest discover -s tests` 通过（45 tests，skipped=1，默认 python 环境已安装 openpyxl 并执行真实 Excel 解析测试）；`py -3 -m compileall app main.py tests` 通过；`py -3 -m unittest discover -s tests` 通过（34 tests，skipped=4，py -3 环境缺 openpyxl/PySide6，相关测试按设计跳过，`app.excel` 契约测试通过）。 +- 注意：本轮不涉及 Shopee/CDP 页面改动，无需测试商品实跑。目录中未跟踪的本地 Excel 文件未处理、未纳入本次变更。 +- 下一步：按任务看板领取 T-202（Tab① 任务列表 + 导入按钮 + 别名匹配标记）。 diff --git a/tests/test_excel.py b/tests/test_excel.py new file mode 100644 index 0000000..87c7f1e --- /dev/null +++ b/tests/test_excel.py @@ -0,0 +1,167 @@ +import os +import unittest + +from _helpers import TempDirMixin + +try: + from openpyxl import Workbook +except ModuleNotFoundError: + raise unittest.SkipTest("openpyxl 未安装") + +from app import db, excel + + +class ExcelImportTests(TempDirMixin, unittest.TestCase): + def save_workbook(self, path, sheets): + workbook = Workbook() + default = workbook.active + workbook.remove(default) + for title, rows in sheets: + sheet = workbook.create_sheet(title=title) + for row in rows: + sheet.append(row) + workbook.save(path) + 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") + excel_path = os.path.join(temp_dir, "input.xlsx") + self.save_workbook( + excel_path, + [ + ( + "商品", + [ + ["账号名", "别名", "商品id", "旧标题"], + ["主店", "alias-a", 51100639510, ""], + ["副店", "alias-b", "52999999", ""], + ["缺别名", "", "123456", ""], + ["坏商品", "alias-c", "abc123", ""], + [None, None, None, None], + ], + ) + ], + ) + + result = excel.import_tasks([excel_path], path=db_path, note="导入测试") + + self.assertIsNotNone(result["batch_id"]) + self.assertEqual(2, len(result["rows"])) + self.assertEqual(1, result["stats"]["files"]) + self.assertEqual(4, result["stats"]["total"]) + self.assertEqual(2, result["stats"]["valid"]) + self.assertEqual(2, result["stats"]["invalid"]) + self.assertEqual(2, result["stats"]["inserted"]) + self.assertEqual([], result["stats"]["file_errors"]) + self.assertEqual(2, len(result["stats"]["row_errors"])) + self.assertEqual("商品", result["rows"][0]["source_sheet"]) + self.assertEqual(2, result["rows"][0]["source_row"]) + self.assertEqual("51100639510", result["rows"][0]["item_id"]) + self.assertTrue(result["rows"][0]["row_key"].startswith(result["batch_id"] + ":")) + + batch = db.get_batch(result["batch_id"], path=db_path) + tasks = db.list_tasks(batch_id=result["batch_id"], path=db_path) + self.assertEqual("导入测试", batch.note) + self.assertEqual(2, len(tasks)) + self.assertEqual("alias-a", tasks[0].alias) + self.assertEqual(os.path.abspath(excel_path), tasks[0].source_file_abs) + self.assertEqual(2, tasks[0].source_row) + + self.assert_removed(temp_dir) + + def test_missing_required_column_rejects_whole_file(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, "missing_alias.xlsx") + self.save_workbook( + excel_path, + [ + ( + "商品", + [ + ["账号名", "商品id"], + ["主店", "51100639510"], + ], + ) + ], + ) + + result = excel.import_tasks([excel_path], path=db_path) + + self.assertIsNone(result["batch_id"]) + self.assertEqual([], result["rows"]) + self.assertEqual(1, len(result["stats"]["file_errors"])) + self.assertEqual(["别名"], result["stats"]["file_errors"][0]["missing_columns"]) + self.assertFalse(os.path.exists(db_path)) + + self.assert_removed(temp_dir) + + def test_import_tasks_continues_when_another_file_has_column_error(self): + with self.make_temp_dir() as temp_dir: + db_path = os.path.join(temp_dir, "cmshopee.db") + good_path = os.path.join(temp_dir, "good.xlsx") + bad_path = os.path.join(temp_dir, "missing_item.xlsx") + self.save_workbook( + good_path, + [ + ( + "商品", + [ + ["账号名", "别名", "商品id"], + ["主店", "alias-a", "51100639510"], + ], + ) + ], + ) + self.save_workbook( + bad_path, + [ + ( + "商品", + [ + ["账号名", "别名"], + ["主店", "alias-a"], + ], + ) + ], + ) + + result = excel.import_tasks([good_path, bad_path], path=db_path) + + self.assertIsNotNone(result["batch_id"]) + self.assertEqual(2, result["stats"]["files"]) + self.assertEqual(1, result["stats"]["valid"]) + self.assertEqual(1, result["stats"]["inserted"]) + self.assertEqual(1, len(result["stats"]["file_errors"])) + self.assertEqual(["商品id"], result["stats"]["file_errors"][0]["missing_columns"]) + self.assertEqual(os.path.abspath(good_path), result["rows"][0]["source_file_abs"]) + self.assertEqual(1, len(db.list_tasks(batch_id=result["batch_id"], path=db_path))) + + self.assert_removed(temp_dir) + + def test_match_summary_counts_known_and_unknown_aliases(self): + rows = [ + {"alias": "a"}, + {"alias": "b"}, + {"alias": "a"}, + {"alias": "missing"}, + ] + accounts = [ + {"alias": "a"}, + {"alias": "b"}, + ] + + self.assertEqual( + { + "matched": 3, + "unmatched": 1, + "by_account": {"a": 2, "b": 1}, + "unmatched_aliases": ["missing"], + }, + excel.match_summary(rows, accounts), + ) + + +if __name__ == "__main__": + unittest.main()