Files
pod_trend_agent/templates/template_router.py
T
3218485270 b7f429db89 模板导出增强 + 模特性别分组 + 三合一提示词精简
1) 模板导出:识别「基码表-胸围」填 sku.bust(多个胸围列都填);申报价格模糊匹配多列统一按加价后价格填写;详情图文不再拼接 img_url_2;SPU 款式来源统一填「现货款」;商品产地国家简称映射(沙特→沙特阿拉伯)
2) 模特性别分组:model_features 按男女分组,按模板类目含男/女固定取对应性别模特(含 Pinterest 模式 pipeline)
3) 三合一提示词:去掉 DESIGN CONTENT 四要素描述(设计已由设计稿提供)
4) 生图尺寸:全部改为读 config 不再硬编码(设计图 compose.design_size / 合成图 compose.size / 种草图 seed_shot.size)
2026-08-26 18:05:11 +08:00

531 lines
24 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- 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_codeSPU货号):返回该 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]
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) 写入
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()