POD 趋势感知 Agent：缓存热点模式 + 三图合成 + 热点去重/风格去重 + review 兜底
- 缓存热点批量流程（有采集缓存不触发 Google） - 简报不足直接从采集缓存生成（轻量补齐） - 三图合成（模特/印花/底图）+ 底图压缩 <2MB - 热点去重→风格去重自动切换 + 不适合类目 review 兜底 - 透明背景（background=transparent）+ 提示词清洗（敏感词/背景描述） - 任务前 basemap 校验 + 模板国家校验 + 模特任务级分配
This commit is contained in:
@@ -0,0 +1,442 @@
|
||||
# -*- 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"（[^）]*）|\([^)]*\)") # 去掉分组名里的括号说明
|
||||
|
||||
|
||||
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]
|
||||
|
||||
# 完整表头 -> 列号 （如 "基础信息-商品层级" -> 1）
|
||||
self.header_map: dict[str, int] = {}
|
||||
# 列名(去分组) -> 列号 （如 "商品层级" -> 1）
|
||||
self.column_map: dict[str, int] = {}
|
||||
# 列号 -> 完整表头
|
||||
self.col_headers: dict[int, str] = {}
|
||||
self._build_headers()
|
||||
|
||||
self._used_rows: set[int] | None = None # 已占用行缓存，见 _refresh()
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# 表头解析
|
||||
# ------------------------------------------------------------------
|
||||
def _build_headers(self) -> None:
|
||||
for col in range(1, self.ws.max_column + 1):
|
||||
raw_group = self.ws.cell(GROUP_ROW, col).value
|
||||
raw_name = self.ws.cell(HEADER_ROW, col).value
|
||||
group = _GROUP_BRACKET.sub("", str(raw_group)).strip() if raw_group else ""
|
||||
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
|
||||
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:
|
||||
"""扫描数据区，收集 A 列（商品层级）非空的行号集合。"""
|
||||
used: set[int] = set()
|
||||
for row in range(DATA_START, self.ws.max_row + 1):
|
||||
v = self.ws.cell(row, 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 = DATA_START) -> int:
|
||||
"""返回数据区第一个空行（A 列为空）。"""
|
||||
if self._used_rows is None:
|
||||
self._refresh_used_rows()
|
||||
row = max(start, 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(DATA_START, self.ws.max_row + 1):
|
||||
v = self.ws.cell(row, 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货号（B列）+ 层级=sku（A列）查找所有 SKU 行号（单 SPU 多色时无 SPU 行，SKU 行即全部）。"""
|
||||
spu_code = str(spu_code).strip().lower()
|
||||
match = self._normalize_match(match)
|
||||
hits = []
|
||||
for row in range(DATA_START, self.ws.max_row + 1):
|
||||
lv = self.ws.cell(row, LEVEL_COL).value
|
||||
if lv is None or str(lv).strip().lower() != "sku":
|
||||
continue
|
||||
v = self.ws.cell(row, 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]:
|
||||
"""
|
||||
按 商品层级（A列）查找所有行号。
|
||||
|
||||
match="exact"（默认）：精准匹配，去首尾空格后完全相等（不区分大小写）；
|
||||
match="fuzzy" ：模糊匹配，层级包含关键字即命中（不区分大小写）。
|
||||
"""
|
||||
level = str(level).strip().lower()
|
||||
match = self._normalize_match(match)
|
||||
hits = []
|
||||
for row in range(DATA_START, self.ws.max_row + 1):
|
||||
v = self.ws.cell(row, 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 列字母。"""
|
||||
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())
|
||||
raise KeyError(f"无法识别的表头: {key!r}（可用完整表头/列名/列号/列字母）")
|
||||
|
||||
def insert(self, data: dict, spu_code: str | None = None,
|
||||
level: str | None = None, update: bool = False,
|
||||
match: str = "exact") -> int:
|
||||
"""
|
||||
插入一行数据到模版表。data 键支持：完整表头（推荐）、列名、列号、列字母。
|
||||
|
||||
- 若 data 中带「商品层级/基础信息-商品层级」则自动取其值路由；
|
||||
- 传 spu_code 会把该行路由到对应 SPU 行之后（SKU 关联）；
|
||||
- update=True 时覆盖已存在的目标行，否则写入新空行；
|
||||
- match 控制货号/层级的匹配模式：exact(精准，默认) / fuzzy(模糊)，见 route()。
|
||||
|
||||
返回写入的行号。
|
||||
"""
|
||||
match = self._normalize_match(match)
|
||||
|
||||
# 1) 解析层级与 spu_code
|
||||
lv = level
|
||||
if lv is None:
|
||||
for k in ("基础信息-商品层级", "商品层级", LEVEL_COL):
|
||||
if k in data:
|
||||
lv = data[k]
|
||||
break
|
||||
|
||||
# 只有「SKU 层」才需要按 SPU货号 关联路由；SPU/单sku商品 层走新行路由
|
||||
use_spu_route = spu_code is not None
|
||||
if spu_code is None:
|
||||
if lv is not None and str(lv).strip().lower() == "sku" and "SPU货号" in data:
|
||||
use_spu_route = True # sku 层 -> 插到所属 SPU 行之后
|
||||
elif lv is None and "SPU货号" in data and self.find_spu_rows(data["SPU货号"], match):
|
||||
use_spu_route = True # 无层级但货号已存在 -> 视为该 SPU 的变体行
|
||||
spu = spu_code if use_spu_route else None
|
||||
if use_spu_route and spu is None:
|
||||
spu = data["SPU货号"]
|
||||
if 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()
|
||||
Reference in New Issue
Block a user