"""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"}, } OUTPUT_HEADERS = { "old_title": "旧标题", "old_cover_path": "旧封面图片路径", "new_title": "新标题", "new_cover_path": "新封面图片路径", "update_status": "更新状态", } OLD_WRITE_BACK_FIELDS = ("old_title", "old_cover_path") RESULT_WRITE_BACK_FIELDS = ("new_title", "new_cover_path", "update_status") WRITEABLE_STAGES = {"collected", "generated", "applied"} RESULT_WRITEABLE_STAGES = {"generated", "applied"} RESULT_WRITEABLE_STATUSES = {"failed", "skipped", "cancelled"} LOCK_WINERRORS = {5, 32, 33} LOCK_ERRNOS = {13, 16} 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 _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 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 _is_result_fields(fields) -> bool: return any(field in RESULT_WRITE_BACK_FIELDS for field in fields) def _has_write_back_data(task, fields) -> bool: if _is_result_fields(fields): if getattr(task, "stage", None) in RESULT_WRITEABLE_STAGES: return True if getattr(task, "status", None) in RESULT_WRITEABLE_STATUSES: return True return any( getattr(task, field, None) not in (None, "") for field in ("new_title", "new_cover_path") if field in fields ) 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 _update_status(task) -> str: error = getattr(task, "last_error", None) status = getattr(task, "status", None) stage = getattr(task, "stage", None) committed = bool(getattr(task, "committed", 0)) if stage == "applied" and committed: return "成功" if status == "failed": return f"失败：{error}" if error else "失败" if status == "skipped": return f"略过：{error}" if error else "略过" if status == "running": return "处理中" if status == "cancelled": return f"已取消：{error}" if error else "已取消" if stage == "generated": return "待更新" return "" def _field_value(task, field): if field == "update_status": return _update_status(task) return getattr(task, field, None) or "" 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 = _field_value(task, field) 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 write_back_results(batch_id, excel_path=None, path=None) -> dict: """Write generated title/cover and update status back to the original Excel file.""" _require_openpyxl() fields = RESULT_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: """Export a write-back copy without touching the original Excel file.""" _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)