"""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 实现")