# -*- coding: utf-8 -*- """ 商品上传模版 —— 模版表解析 + 行路由 + 数据插入 ================================================ 解析目标：当前文件夹下「商品上传模版 (1).xlsx」的【模版】工作表。 表结构（【模版】sheet）： 第 1 行 元信息表头（经营站点 / 发货仓 / 类目 / 运费模版 ...） 第 2 行 元信息值 第 3 行 分组行（基础信息（SKU/SPU均必填）、SPU商品属性、商品规格、尺码表 ...） 第 4 行 列名行（商品层级、SPU货号、商品名称 ...） 第 5 行 填写说明 第 6 行 空行（含数据验证下拉）——数据区起始行，由本模块负责路由 + 插入 完整表头 = 「分组-列名」，例如「基础信息-商品层级」；无分组的列直接用列名， 例如「SPU货号」。这一组合后的表头就是路由与插入数据的键。 路由规则（核心函数 route）： - 按「基础信息-商品层级」（A列，取值 spu / sku / 单sku商品）路由： * 不传 spu_code：返回数据区第一个空行（新行位置） * 传 spu_code（SPU货号）：返回该 SPU 所在行**下方紧邻的空行**， 用于把 SKU 变体插到所属 SPU 行后面，实现 SPU-SKU 关联 - match 参数控制匹配模式： * "exact"（默认）：精准匹配 —— 货号/层级去空格后完全相等（不区分大小写） * "fuzzy" ：模糊匹配 —— 货号/层级包含关键字即命中（不区分大小写） * 兼容中文别名：精准/精确 -> exact，模糊/包含 -> fuzzy - 若该层级/货号已存在，可指定 update=True 直接覆盖其行 用法示例： from template_router import TemplateRouter r = TemplateRouter(r"商品上传模版 (1).xlsx") # 1) 按层级路由：拿到应写入的行号 row = r.route("spu") # -> 数据区第一个空行 row = r.route("sku", spu_code="A001") # -> SPU 货号 A001 所在行之后（精准） row = r.route("sku", spu_code="A0", match="fuzzy") # -> 模糊匹配，命中 A001 之后 row = r.route("sku", spu_code="a00", match="模糊") # 中文别名等价写法 # 2) 直接插入一行（自动路由；insert 同样支持 match 参数） r.insert({ "基础信息-商品层级": "spu", "SPU货号": "A001", "商品名称": "纯棉白T恤", "商品产地": "中国", "产地省份": "广东省", }) r.insert({ "商品层级": "sku", # 列名也可直接作键 "SPU货号": "A001", "SKU货号": "A001-白色-M", "商品名称": "纯棉白T恤 白色 M", "尺码": "M", "申报价格-日本站": 1290, "发货仓1库存": 100, }, spu_code="A001") # 自动插到 A001 后面 r.insert({"商品层级": "sku", "SPU货号": "A001", "SKU货号": "A001-白色-L", ...}, match="fuzzy") # 模糊匹配货号路由 r.save() # 默认输出到 <原名>_已填写.xlsx """ from __future__ import annotations import re from pathlib import Path from openpyxl import load_workbook from openpyxl.utils import column_index_from_string, get_column_letter # -------------------------------------------------------------------------- # 常量：模版表关键行列 # -------------------------------------------------------------------------- SHEET_NAME = "模版" GROUP_ROW = 3 # 分组行 HEADER_ROW = 4 # 列名行 DATA_START = 6 # 数据区起始行（第 5 行为说明，第 6 行为空行，数据默认从第 6 行开始填） LEVEL_COL = 1 # A 列：基础信息-商品层级 SPU_CODE_COL = 2 # B 列：SPU货号 VALID_LEVELS = ("spu", "sku", "单sku商品") _GROUP_BRACKET = re.compile(r"（[^）]*）|\([^)]*\)") # 去掉分组名里的括号说明 _UNIT_SUFFIX = re.compile(r"[（(](?:cm|g|kg|mm|m²|m2)[）)]\s*$", re.IGNORECASE) # 列名尾部单位（肩宽（cm）→肩宽） class TemplateRouter: """解析【模版】表，并提供 路由 + 插入 能力。""" def __init__(self, path: str | Path, sheet: str = SHEET_NAME): self.path = Path(path) self.wb = load_workbook(self.path, data_only=False) if sheet not in self.wb.sheetnames: raise ValueError(f"工作簿中不存在工作表 {sheet!r}，现有: {self.wb.sheetnames}") self.ws = self.wb[sheet] # 布局行号（自动探测，失败回退模块级常量，兼容旧版模版） self.group_row = GROUP_ROW self.header_row = HEADER_ROW self.data_start = DATA_START self._detect_layout() # 完整表头 -> 列号 （如 "基础信息-商品层级" -> 1，含括号全名别名） self.header_map: dict[str, int] = {} # 列名(去分组) -> 列号 （如 "商品层级" -> 1） self.column_map: dict[str, int] = {} # 列号 -> 完整表头 self.col_headers: dict[int, str] = {} self._build_headers() # 关键列按列名解析（不依赖固定列号，列名缺失时回退模块级常量） self.level_col = self.column_map.get("商品层级", LEVEL_COL) self.spu_code_col = self.column_map.get("SPU货号", SPU_CODE_COL) self._used_rows: set[int] | None = None # 已占用行缓存，见 _refresh() # ------------------------------------------------------------------ # 布局自动探测 # ------------------------------------------------------------------ def _detect_layout(self) -> None: """按固定分组名定位布局（不依赖固定行列）。 分组名（如「基础信息（SKU/SPU均必填）」「SPU商品属性」）是模版固定不变的， 以其为锚点： 1. 分组行 = 含「基础信息」的行（去括号说明后精确匹配，兼容含括号全名）； 2. 列名行 = 分组行正下方的列名行（校验含「商品层级」+「SPU货号」）； 3. 数据起始行 = 列名行之后第一个整行空行（跳过说明行）。 探测失败时保持模块级常量默认值。 """ max_col = min(self.ws.max_column, 30) def row_texts(row: int) -> list[str]: return [str(self.ws.cell(row, c).value or "").strip() for c in range(1, max_col + 1)] # 1) 分组行：按固定分组名「基础信息」匹配 group_row = None for row in range(1, min(self.ws.max_row, 30) + 1): for v in row_texts(row): if _GROUP_BRACKET.sub("", v).strip() == "基础信息" or "基础信息" in v: group_row = row break if group_row: break # 2) 列名行：固定位于分组行正下方（校验含「商品层级」+「SPU货号」； # 列名可能变化，校验失败时直接用分组行正下方一行） header_row = None if group_row: for row in range(group_row + 1, min(self.ws.max_row, group_row + 4) + 1): vals = row_texts(row) if any("商品层级" in v for v in vals) and any("SPU货号" in v for v in vals): header_row = row break if header_row is None: header_row = group_row + 1 # 3) 数据起始行：列名行正下方第2行（列名行→说明行→数据起始行）。 # 不用「第一个空行」——已填写的模版首个空行在数据之后，会漏扫已有数据。 if header_row: data_start = header_row + 2 if group_row: self.group_row = group_row if header_row: self.header_row = header_row if data_start: self.data_start = data_start # ------------------------------------------------------------------ # 表头解析 # ------------------------------------------------------------------ def _build_headers(self) -> None: for col in range(1, self.ws.max_column + 1): raw_group = self.ws.cell(self.group_row, col).value raw_name = self.ws.cell(self.header_row, col).value group_full = str(raw_group).strip() if raw_group else "" group = _GROUP_BRACKET.sub("", group_full).strip() name = str(raw_name).strip() if raw_name else "" if not name: continue header = f"{group}-{name}" if group else name self.header_map[header] = col if group_full and group_full != group: # 完整分组名（含括号说明）别名，如「基础信息（SKU/SPU均必填）-商品层级」 self.header_map[f"{group_full}-{name}"] = col self.column_map[name] = col self.col_headers[col] = header # ------------------------------------------------------------------ # 内部工具 # ------------------------------------------------------------------ @staticmethod def _normalize_match(match: str) -> str: """把 match 参数归一化为 exact / fuzzy，兼容中文别名。""" m = str(match).strip().lower() alias = {"精准": "exact", "精确": "exact", "完全": "exact", "模糊": "fuzzy", "包含": "fuzzy", "部分": "fuzzy"} m = alias.get(m, m) if m not in ("exact", "fuzzy"): raise ValueError(f"match 必须是 exact(精准) 或 fuzzy(模糊)，收到: {match!r}") return m def _refresh_used_rows(self) -> None: """扫描数据区，收集商品层级列（按列名解析）非空的行号集合。""" used: set[int] = set() for row in range(self.data_start, self.ws.max_row + 1): v = self.ws.cell(row, self.level_col).value if v is not None and str(v).strip(): used.add(row) self._used_rows = used def first_empty_row(self, start: int | None = None) -> int: """返回数据区第一个空行（A 列为空）。""" if self._used_rows is None: self._refresh_used_rows() row = max(start or self.data_start, self.data_start) while row in self._used_rows: row += 1 return row def find_spu_rows(self, spu_code: str, match: str = "exact") -> list[int]: """ 按 SPU货号（B列）查找所有行号。 match="exact"（默认）：精准匹配，去首尾空格后完全相等（不区分大小写）； match="fuzzy" ：模糊匹配，货号包含关键字即命中（不区分大小写）。 """ spu_code = str(spu_code).strip().lower() match = self._normalize_match(match) hits = [] for row in range(self.data_start, self.ws.max_row + 1): v = self.ws.cell(row, self.spu_code_col).value if v is None: continue v = str(v).strip().lower() if (match == "exact" and v == spu_code) or (match == "fuzzy" and spu_code in v): hits.append(row) return hits def find_sku_rows(self, spu_code: str, match: str = "exact") -> list[int]: """按 SPU货号（按列名解析）+ 层级=sku（商品层级列）查找所有 SKU 行号（单 SPU 多色时无 SPU 行，SKU 行即全部）。""" spu_code = str(spu_code).strip().lower() match = self._normalize_match(match) hits = [] for row in range(self.data_start, self.ws.max_row + 1): lv = self.ws.cell(row, self.level_col).value if lv is None or str(lv).strip().lower() != "sku": continue v = self.ws.cell(row, self.spu_code_col).value if v is None: continue v = str(v).strip().lower() if (match == "exact" and v == spu_code) or (match == "fuzzy" and spu_code in v): hits.append(row) return hits def find_level_rows(self, level: str, match: str = "exact") -> list[int]: """ 按 商品层级（按列名解析）查找所有行号。 match="exact"（默认）：精准匹配，去首尾空格后完全相等（不区分大小写）； match="fuzzy" ：模糊匹配，层级包含关键字即命中（不区分大小写）。 """ level = str(level).strip().lower() match = self._normalize_match(match) hits = [] for row in range(self.data_start, self.ws.max_row + 1): v = self.ws.cell(row, self.level_col).value if v is None: continue v = str(v).strip().lower() if (match == "exact" and v == level) or (match == "fuzzy" and level in v): hits.append(row) return hits # ------------------------------------------------------------------ # 核心路由函数 # ------------------------------------------------------------------ def route(self, level: str | None = None, spu_code: str | None = None, update: bool = False, match: str = "exact") -> int: """ 根据「基础信息-商品层级」(A列) 路由到模版表中的一行，返回行号。 参数 ---- level : 商品层级，取值 spu / sku / 单sku商品（不区分大小写）。 仅用于未提供 spu_code 时定位「该层级最后一行之后」的空行。 spu_code : SPU货号。提供时路由到该货号所在行： - update=False（默认）：返回其下方第一个空行（插入新行）； - update=True ：返回该货号最后一行（原地覆盖更新）。 未找到匹配行会抛 KeyError。 update : 是否原地更新已存在的行。 match : 匹配模式（对 spu_code 与 level 均生效）： - "exact"（默认）：精准匹配，去首尾空格后完全相等（不区分大小写）； - "fuzzy" ：模糊匹配，包含关键字即命中（不区分大小写）。 兼容中文别名：精准/精确/完全 -> exact，模糊/包含/部分 -> fuzzy。 模糊匹配命中多行时，插入定位取最后命中行，更新取最后命中行。 返回 ---- int 行号（从 7 开始）。 """ match = self._normalize_match(match) if spu_code is not None: rows = self.find_spu_rows(spu_code, match) if not rows: raise KeyError(f"未找到 SPU货号 = {spu_code!r} 的{('模糊' if match == 'fuzzy' else '精准')}匹配行") last = max(rows) if update: return last # 从该 SPU 行之后找第一个空行（保证 SPU 与 SKU 相邻） return self.first_empty_row(last + 1) if level is not None: level_l = str(level).strip().lower() if match == "exact" and level_l not in VALID_LEVELS: raise ValueError(f"商品层级取值必须是 {VALID_LEVELS} 之一，收到: {level!r}") if update: rows = self.find_level_rows(level_l, match) if not rows: raise KeyError(f"未找到商品层级 = {level!r} 的{('模糊' if match == 'fuzzy' else '精准')}匹配行") return max(rows) rows = self.find_level_rows(level_l, match) if rows: return self.first_empty_row(max(rows) + 1) return self.first_empty_row() # ------------------------------------------------------------------ # 数据插入 # ------------------------------------------------------------------ def resolve_col(self, key: str | int) -> int: """把键解析为列号：完整表头 / 列名 / 列号 / Excel 列字母。 精确匹配失败时做单位归一化匹配（去掉尾部（cm）/(g)…）， 兼容「肩宽（cm）」↔「肩宽」这类带单位/不带单位的列名差异。 """ if isinstance(key, int): return key key_s = str(key).strip() if key_s in self.header_map: return self.header_map[key_s] if key_s in self.column_map: return self.column_map[key_s] # 兼容「SPU商品属性-袖型」↔「袖型」列名差异：不同模板对同一属性列命名不同（带/不带分组前缀） if key_s.startswith("SPU商品属性-"): bare = key_s[len("SPU商品属性-"):] if bare in self.column_map: return self.column_map[bare] if bare in self.header_map: return self.header_map[bare] if re.fullmatch(r"[A-Za-z]{1,3}", key_s): return column_index_from_string(key_s.upper()) norm = _UNIT_SUFFIX.sub("", key_s).strip() if norm and norm != key_s: if norm in self.column_map: return self.column_map[norm] if norm in self.header_map: return self.header_map[norm] for name, col in self.column_map.items(): if _UNIT_SUFFIX.sub("", str(name)).strip() == norm: return col raise KeyError(f"无法识别的表头: {key!r}（可用完整表头/列名/列号/列字母）") def _value_by_col(self, data: dict, name: str): """从 data 中按列名取第一个匹配的值（键可为完整表头/列名/列号/列字母）。""" col = self.column_map.get(name) if col is None: return None for k, v in data.items(): try: if self.resolve_col(k) == col: return v except KeyError: continue return None def insert(self, data: dict, spu_code: str | None = None, level: str | None = None, update: bool = False, match: str = "exact") -> int: """ 插入一行数据到模版表。data 键支持：完整表头（推荐）、列名、列号、列字母。 - 若 data 中带「商品层级」（任意键形式，如「基础信息（SKU/SPU均必填）-商品层级」/ 「基础信息-商品层级」/「商品层级」）则自动取其值路由； - 传 spu_code 会把该行路由到对应 SPU 行之后（SKU 关联）； - update=True 时覆盖已存在的目标行，否则写入新空行； - match 控制货号/层级的匹配模式：exact(精准，默认) / fuzzy(模糊)，见 route()。 返回写入的行号。 """ match = self._normalize_match(match) # 1) 解析层级与 spu_code（按列名匹配，不依赖固定键名/列号） lv = level if level is not None else self._value_by_col(data, "商品层级") lv_s = str(lv).strip().lower() if lv is not None else "" # 只有「SKU 层」或未声明层级时才按 SPU货号 关联路由；SPU/单sku商品 层走新行路由 spu = spu_code use_spu_route = spu is not None if spu is None and (lv_s == "sku" or lv is None): spu = self._value_by_col(data, "SPU货号") if spu is not None: use_spu_route = True if use_spu_route and spu is not None and not self.find_spu_rows(spu, match): use_spu_route = False # 无 SPU 行（单 SPU 多色，不填 SPU 行）→ 按层级末尾追加 spu = None # 2) 路由到目标行 row = self.route(lv, spu, update=update, match=match) # 3) 清空目标行（仅新建行，update 覆盖模式保留模板原有内容）： # 模板数据区可能预填了示例/默认值（如 SKU销售信息-建议售价单位=CNY、 # SPU商品属性-印花类型=定位印花），不清理会把模板预填值误带到生成结果里。 # 只清空本行已写过的数据单元格，保留样式/数据验证下拉框。 if not update: for c in range(1, self.ws.max_column + 1): cell = self.ws.cell(row, c) v = cell.value if v is not None and str(v).strip() != "": cell.value = None # 4) 写入 for key, value in data.items(): if value is None: continue try: col = self.resolve_col(key) except KeyError: continue # 未知键静默跳过，避免整行失败 self.ws.cell(row, col, value) if self._used_rows is not None: self._used_rows.add(row) return row # ------------------------------------------------------------------ # 保存 # ------------------------------------------------------------------ def save(self, path: str | Path | None = None) -> Path: """ 保存工作簿。默认输出到 <源文件名>_已填写.xlsx，不覆盖原始模版； 显式传 path（或 path 与源文件相同时）则按指定路径保存。 """ out = Path(path) if path else self.path.with_name(f"{self.path.stem}_已填写.xlsx") out = out.with_suffix(".xlsx") if out.suffix.lower() != ".xlsx" else out self.wb.save(out) return out def close(self) -> None: self.wb.close() # -------------------------------------------------------------------------- # 演示：路由 + 插入 一个 SPU 与两个 SKU # -------------------------------------------------------------------------- def demo(src: str | Path = r"商品上传模版 (1).xlsx") -> Path: r = TemplateRouter(src) # 1) 按层级路由 —— spu 层应落在数据区第一行（行 7） row = r.route("spu") print(f"[route] spu -> 第 {row} 行") assert row == DATA_START # 2) 插入 SPU 层 r.insert({ "基础信息-商品层级": "spu", "SPU货号": "A001", "商品名称": "纯棉圆领白T恤 基础款", "英文名称": "Cotton Crew Neck White T-Shirt", "日语名称": "コットン クルーネック 白Tシャツ", "商品产地": "中国", "产地省份": "广东省", "品牌名": "Inkreach", "季节": "四季", "风格": "休闲", }) # 3) 按 SPU货号 路由 —— sku 应插到 A001 之后（行 8） row = r.route("sku", spu_code="A001") print(f"[route] sku(A001) -> 第 {row} 行") assert row == DATA_START + 1 # 4) 插入 SKU 变体 1 / 2（自动关联到 A001 下方） r.insert({ "基础信息-商品层级": "sku", "SPU货号": "A001", "商品名称": "纯棉圆领白T恤 白色 M", "SKU货号": "A001-W-M", "色值(主规格)": "白色", "尺码": "M", "申报价格-日本站": 1290, "币种": "JPY", "发货仓1": "名古屋仓", "发货仓1库存": 100, "重量（g）": 180, }, spu_code="A001") r.insert({ "商品层级": "sku", "SPU货号": "A001", "商品名称": "纯棉圆领白T恤 白色 L", "SKU货号": "A001-W-L", "色值(主规格)": "白色", "尺码": "L", "申报价格-日本站": 1290, "币种": "JPY", "发货仓1": "名古屋仓", "发货仓1库存": 120, "重量（g）": 200, }, spu_code="A001") # 5) 验证 assert r.ws.cell(DATA_START, 1).value == "spu" and r.ws.cell(DATA_START, 2).value == "A001" assert r.ws.cell(DATA_START + 1, 1).value == "sku" and r.ws.cell(DATA_START + 1, 2).value == "A001" assert r.ws.cell(DATA_START + 2, 1).value == "sku" and r.ws.cell(DATA_START + 2, 2).value == "A001" print(f"[ok] 行 {DATA_START} = SPU(A001)，行 {DATA_START+1}/{DATA_START+2} = 其下 SKU 变体，SPU-SKU 已关联") # 6) 匹配模式演示 # 精准匹配：完整货号 A001 -> 其后第一个空行 row = r.route("sku", spu_code="A001", match="exact") print(f"[route] sku(A001, exact) -> 第 {row} 行") assert row == DATA_START + 3 # 模糊匹配：货号片段 A0 -> 命中 A001 所有行，取最后一行之后 row = r.route("sku", spu_code="A0", match="fuzzy") print(f"[route] sku(A0, fuzzy) -> 第 {row} 行") assert row == DATA_START + 3 # 模糊匹配 + 中文别名 + update：覆盖 A001 最后一行 row = r.route("sku", spu_code="a00", match="模糊", update=True) print(f"[route] sku(a00, 模糊, update) -> 第 {row} 行") assert row == DATA_START + 2 # 层级模糊匹配：关键字 s 命中所有行 row = r.route("s", match="fuzzy", update=True) print(f"[route] level(s, fuzzy, update) -> 第 {row} 行") assert row == DATA_START + 2 print("[ok] 精准/模糊两种匹配模式均工作正常（含中文别名）") out = r.save() r.close() print(f"[save] 已写出 -> {out}") return out if __name__ == "__main__": demo()