3218485270
71a48e4ed5
- 图源映射统一：热点采集与 Pinterest 模式均走 config.product.mark_dirs 配置，按任务序号随机抽模特图/平铺图 - 商品产地固定：统一为「中国大陆」+「产地省份=广东省」（不再读站点/字典映射） - 模板 SKU 字段检测：按建议售价同一套路检测 SKU分类/SKU数量/SKU数量单位，必填时填入单品/1/件 - 多模态分析提示词可配置：prompts/pinterest_analyze_system.md + user.md，支持国家覆盖，不丢文件回退内置 - 自定义图片模式：新增 pinterest_custom_load_node，图片数量硬校验，选品清单 ≤ 有效图片数 - 模板导出优化：写入前按货号末 3 位升序排序，不再产生空白 xlsx - 修复 v90 project review 10 项（503 致命终止、线程安全、原子写入等）
814 lines
39 KiB
Python
814 lines
39 KiB
Python
"""商品上传模板导出：从 db 读 SPU/SKU → 调 template_router 路由填入上传模板 Excel。
|
||
|
||
流程（product_node 生成产品图后调用）：
|
||
1. 从 spu_sku.db 读 SPU（款号）+ 该款选定颜色的全部尺码 SKU；
|
||
2. 用 model/template_router.py 的 TemplateRouter：
|
||
- insert 一行 SPU（SPU货号 + 商品属性字段）
|
||
- 每个尺码 insert 一行 SKU（路由到 SPU 行下方，SKU货号 = 款号-颜色编码-尺码，填尺码表）
|
||
- 商品轮播图1~N 填生成的产品图路径（底图/印花/模特/合成）
|
||
3. save 输出 <模板名>_已填写.xlsx 到指定目录。
|
||
|
||
字段映射：db 字段名 → 上传模板列名（见 SPU_MAP / SKU_MAP）。
|
||
"""
|
||
import re
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
from graph.product import _connect
|
||
|
||
# 商品轮播图列名关键词（模板存在中/英/日变体，如 商品轮播图1 / Product Carousel Image 1 / 商品カルーセル画像1）
|
||
_CAROUSEL_KW = ("轮播", "carousel", "カルーセル")
|
||
|
||
# 成分值字典映射：db 成分值 → 女装模板下拉框选项（男装模板选项与 db 值一致，直接保留）。
|
||
# 女装模板（如 SatVoy 沙特）成分下拉框是「中文+英文」格式（棉Cotton），db 存中文（棉），需映射。
|
||
_COMPONENT_FEMALE_MAP = {
|
||
"棉": "棉Cotton",
|
||
"聚酯纤维": "聚酯纤维(涤纶)Polyester",
|
||
"氨纶": "氨纶Elastane",
|
||
"锦纶": "锦纶(尼龙)Polyamide",
|
||
"再生聚酯纤维": "聚酯纤维(涤纶)Polyester", # 女装无再生聚酯纤维，归入聚酯纤维
|
||
"棉Cotton": "棉Cotton", # 已是女装格式，保持
|
||
}
|
||
|
||
|
||
def _map_component(value: Any, gender: Optional[str]) -> Any:
|
||
"""成分值按性别映射：male/None 保留原值；female 查女装字典（找不到保留原值）。"""
|
||
if gender != "female" or value in (None, ""):
|
||
return value
|
||
return _COMPONENT_FEMALE_MAP.get(value, value)
|
||
|
||
|
||
def _template_gender(template_path: str) -> Optional[str]:
|
||
"""通过表头解析模版「类目」判断男装/女装：类目含「男」→male；含「女」→female；都不含→None。
|
||
用于成分值映射（女装模板下拉框是「中文+英文」格式，需字典映射；男装保留 db 值）。"""
|
||
try:
|
||
from graph.seed_shot import gender_from_category, read_template_category
|
||
return gender_from_category(read_template_category(template_path))
|
||
except Exception: # noqa: BLE001
|
||
return None
|
||
|
||
|
||
def _size_rank(size) -> tuple:
|
||
"""把尺码字符串转成可排序 rank（从小到大）。
|
||
|
||
优先级：数字码(90/100...) < 字母码(XS/S/M/L/XL/XXL/3XL...) < 童装年龄码(0/3M、1-2Y...)
|
||
< 尺寸规格(30*40...) < 未知(字典序兜底) < 均码/Onesize(最后)。
|
||
"""
|
||
s = str(size or "").strip().upper()
|
||
if not s:
|
||
return (9, 0, "")
|
||
if s in ("ONESIZE", "ONE SIZE", "FREESIZE", "FREE SIZE", "均码"):
|
||
return (8, 0, s)
|
||
if s == "XS":
|
||
return (2, 0, s)
|
||
if s in ("S", "M", "L"):
|
||
return (2, {"S": 1, "M": 2, "L": 3}[s], s)
|
||
m = re.fullmatch(r"(X+)(L)", s) # XL/XXL/XXXL/XXXXL/XXXXXL
|
||
if m:
|
||
return (2, 3 + len(m.group(1)), s)
|
||
m = re.fullmatch(r"(\d+)XL", s) # 3XL/4XL/5XL
|
||
if m:
|
||
return (2, 3 + int(m.group(1)), s)
|
||
if s.isdigit(): # 数字码：90/100/110...
|
||
return (1, int(s), s)
|
||
m = re.fullmatch(r"(\d+)\s*[-/]\s*(\d+)\s*(?:M|Y|YRS|YR)?", s) # 童装年龄码
|
||
if m:
|
||
return (3, (int(m.group(1)) + int(m.group(2))) / 2.0, s)
|
||
m = re.search(r"(\d+(?:\.\d+)?)\s*(?:CM)?\s*[X*]", s) # 尺寸规格：30*40/12CM X 12CM
|
||
if m:
|
||
return (4, float(m.group(1)), s)
|
||
return (5, 0, s) # 未知格式：字典序兜底
|
||
|
||
|
||
def _carousel_col(router, idx: int) -> Optional[int]:
|
||
"""定位「商品轮播图{idx}」列号：先精确匹配（中文列名），失败则按中/英/日关键词模糊匹配序号。"""
|
||
try:
|
||
return router.resolve_col(f"商品轮播图{idx}")
|
||
except KeyError:
|
||
pass
|
||
for name, col in router.column_map.items():
|
||
low = str(name).lower().replace(" ", "").replace(" ", "")
|
||
if not any(k in low for k in _CAROUSEL_KW):
|
||
continue
|
||
m = re.search(r"(\d+)$", low)
|
||
if m and int(m.group(1)) == idx:
|
||
return col
|
||
return None
|
||
|
||
|
||
def _detail_col(router) -> Optional[int]:
|
||
"""定位「详情图文」列：优先英语（详情图文-英语），回退日语（详情图文-日语），再回退任意详情图文。"""
|
||
for name in ("详情图文-英语", "详情图文-英文", "详情图文-EN"):
|
||
try:
|
||
return router.resolve_col(name)
|
||
except KeyError:
|
||
pass
|
||
try:
|
||
return router.resolve_col("详情图文-日语")
|
||
except KeyError:
|
||
pass
|
||
for name, col in router.column_map.items():
|
||
if "详情图文" in str(name):
|
||
return col
|
||
return None
|
||
|
||
|
||
def _ja_col(router) -> Optional[int]:
|
||
"""定位「日语名称」列（基础信息组，如 日语名称/日语标题）。"""
|
||
try:
|
||
return router.resolve_col("日语名称")
|
||
except KeyError:
|
||
pass
|
||
for name, col in router.column_map.items():
|
||
low = str(name)
|
||
if "日语" in low and "详情图文" not in low and "轮播图" not in low and "名称" in low:
|
||
return col
|
||
return None
|
||
|
||
|
||
def _es_col(router) -> Optional[int]:
|
||
"""定位「西语名称」列（基础信息组，如 西语名称/西班牙语名称）。"""
|
||
try:
|
||
return router.resolve_col("西语名称")
|
||
except KeyError:
|
||
pass
|
||
for name, col in router.column_map.items():
|
||
low = str(name)
|
||
if ("西语" in low or "西班牙" in low) and "详情图文" not in low and "轮播图" not in low and "名称" in low:
|
||
return col
|
||
return None
|
||
|
||
|
||
def _fill_design_fields(router, spu_code: str, oss_code: str, cn_title: str, en_title: str,
|
||
ja_title: str, es_title: str, composite_by_sku: Dict[str, Any],
|
||
all_composite_urls: List[str], seed_shot_urls: List[str],
|
||
only_rows: Optional[List[int]] = None) -> None:
|
||
"""按用户要求填充设计联动字段：
|
||
- SPU 行：SPU货号=设计货号、SKU货号=设计货号、商品名称=en_title、英文名称=en_title、
|
||
日语名称=ja_title、西语名称=es_title、商品轮播图1=随机一张三合一主图、
|
||
详情图文=全部主图+种草图 链接 | 分割（不含 img_url_2）
|
||
- SKU 行：SPU货号=设计货号、SKU货号=该颜色货号、商品轮播图1=该颜色三合一链接、
|
||
商品名称/英文名称/日语名称/西语名称 与 SPU 一致
|
||
only_rows：合并模式下只填充本产品块的行（None=该款全部行）
|
||
"""
|
||
import random
|
||
try:
|
||
color_col = router.resolve_col("色值(主规格)")
|
||
except KeyError:
|
||
color_col = None
|
||
try:
|
||
spu_col = router.resolve_col("SPU货号")
|
||
except KeyError:
|
||
spu_col = None
|
||
try:
|
||
name_col = router.resolve_col("商品名称")
|
||
except KeyError:
|
||
name_col = None
|
||
try:
|
||
en_col = router.resolve_col("英文名称")
|
||
except KeyError:
|
||
en_col = None
|
||
try:
|
||
ja_col = _ja_col(router)
|
||
except KeyError:
|
||
ja_col = None
|
||
try:
|
||
es_col = _es_col(router)
|
||
except KeyError:
|
||
es_col = None
|
||
try:
|
||
sku_code_col = router.resolve_col("SKU货号")
|
||
except KeyError:
|
||
sku_code_col = None
|
||
car1 = _carousel_col(router, 1)
|
||
detail_col = _detail_col(router)
|
||
|
||
for row in router.find_spu_rows(spu_code):
|
||
if only_rows is not None and row not in only_rows:
|
||
continue # 合并模式：只填本产品块的行
|
||
lvl = str(router.ws.cell(row, router.level_col).value or "").strip().lower()
|
||
color = str(router.ws.cell(row, color_col).value or "").strip() if color_col else ""
|
||
if lvl == "spu":
|
||
if spu_col and oss_code:
|
||
router.ws.cell(row, spu_col, oss_code)
|
||
if sku_code_col and oss_code:
|
||
router.ws.cell(row, sku_code_col, oss_code)
|
||
if name_col and en_title:
|
||
router.ws.cell(row, name_col, en_title) # 商品名称统一用 en_title
|
||
if en_col and en_title:
|
||
router.ws.cell(row, en_col, en_title)
|
||
if ja_col and ja_title:
|
||
router.ws.cell(row, ja_col, ja_title)
|
||
if es_col and es_title:
|
||
router.ws.cell(row, es_col, es_title)
|
||
if car1 is not None and all_composite_urls:
|
||
router.ws.cell(row, car1, random.choice(all_composite_urls)) # SPU 轮播图1 随机
|
||
if detail_col is not None:
|
||
links = [u for u in (all_composite_urls + list(seed_shot_urls or [])) if u]
|
||
if links:
|
||
router.ws.cell(row, detail_col, "|".join(links)) # 详情图文 | 分割
|
||
else:
|
||
if spu_col and oss_code:
|
||
router.ws.cell(row, spu_col, oss_code)
|
||
# SKU 行与 SPU 一致：商品名称/英文名称/日语名称/西语名称
|
||
if name_col and en_title:
|
||
router.ws.cell(row, name_col, en_title) # 商品名称统一用 en_title
|
||
if en_col and en_title:
|
||
router.ws.cell(row, en_col, en_title)
|
||
if ja_col and ja_title:
|
||
router.ws.cell(row, ja_col, ja_title)
|
||
if es_col and es_title:
|
||
router.ws.cell(row, es_col, es_title)
|
||
cc = composite_by_sku.get(color) or composite_by_sku.get("") # 按色值匹配该颜色主图
|
||
if sku_code_col and oss_code:
|
||
router.ws.cell(row, sku_code_col, oss_code) # SKU货号=SPU货号（同一货号）
|
||
if car1 is not None and cc and cc.get("url"):
|
||
router.ws.cell(row, car1, cc["url"]) # 该颜色轮播图1
|
||
|
||
|
||
def _build_spu_row(spu: Dict[str, Any], spu_code: str,
|
||
color: Optional[str] = None,
|
||
fabric_headers: Optional[List[str]] = None,
|
||
gender: Optional[str] = None) -> Dict[str, Any]:
|
||
"""构造一行 SPU（固定字段：SKC货号=code、风格=休闲、商品产地=中国大陆、产地省份=广东省；多颜色时用色值列区分）。
|
||
fabric 填「面料弹性」列（fabric_headers，如 SPU商品属性-面料弹性，检测到才填 spu.fabric）。
|
||
component_1/2/3 按性别映射（gender=female 时查 _COMPONENT_FEMALE_MAP，男装/None 保留原值）。"""
|
||
row: Dict[str, Any] = {
|
||
"基础信息-商品层级": "spu",
|
||
"SKC货号": spu_code, # code 路由为 SKC货号（用户要求）
|
||
"风格": "休闲", # style 路由为"休闲"（用户要求）
|
||
"商品产地": "中国大陆", # 所有国家统一「中国大陆」，不读经营站点/不做字典匹配（用户要求）
|
||
"产地省份": "广东省", # 新增：精确匹配「产地省份」列，统一填「广东省」（用户要求）
|
||
"款式来源": "现货款", # SPU商品属性-款式来源 统一填「现货款」（用户要求）
|
||
}
|
||
if color:
|
||
row["色值(主规格)"] = color
|
||
for dbk, header in SPU_MAP.items():
|
||
v = spu.get(dbk)
|
||
if dbk in ("component_1", "component_2", "component_3"):
|
||
v = _map_component(v, gender)
|
||
if v not in (None, ""):
|
||
row[header] = v
|
||
fabric = spu.get("fabric")
|
||
if fabric not in (None, "") and fabric_headers:
|
||
for h in fabric_headers:
|
||
row[h] = fabric
|
||
return row
|
||
|
||
|
||
def _find_price_headers(router) -> List[str]:
|
||
"""定位所有「申报价格」列（美站/日站/英站…模糊匹配）；匹配到多个时全部返回，统一填加价后价格。"""
|
||
hits = [str(k) for k in router.column_map if "申报价格" in str(k)]
|
||
return hits or ["申报价格-日本站"]
|
||
|
||
|
||
def _find_bust_headers(router) -> List[str]:
|
||
"""定位所有「胸围」列（基码表-胸围（cm）/胸围全围（cm）…模糊匹配）；匹配到多个时全部填 sku.bust。"""
|
||
hits = [str(k) for k in router.column_map if "胸围" in str(k)]
|
||
return hits or ["胸围全围（cm）"]
|
||
|
||
|
||
def _find_fabric_headers(router) -> List[str]:
|
||
"""定位所有「面料弹性」列（SPU商品属性-面料弹性…模糊匹配）；匹配到多个时全部填 spu.fabric。"""
|
||
return [str(k) for k in router.column_map if "面料弹性" in str(k)]
|
||
|
||
|
||
def _find_sa_size_headers(router) -> List[str]:
|
||
"""定位所有「沙特阿拉伯码」列（尺码表-沙特阿拉伯码…模糊匹配）；匹配到多个时全部填 sku.size。"""
|
||
return [str(k) for k in router.column_map if "沙特阿拉伯码" in str(k)]
|
||
|
||
|
||
def _find_suggested_price_headers(router) -> List[str]:
|
||
"""定位「建议售价」列（不含「单位」）；匹配到多个时全部返回。"""
|
||
return [str(k) for k in router.column_map if "建议售价" in str(k) and "单位" not in str(k)]
|
||
|
||
|
||
def _find_suggested_unit_headers(router) -> List[str]:
|
||
"""定位「建议售价单位」列。"""
|
||
return [str(k) for k in router.column_map if "建议售价单位" in str(k)]
|
||
|
||
|
||
def _read_suggested_required(router, col: int) -> bool:
|
||
"""读取「建议售价」列下方注意事项，判断是否必填。
|
||
结合「非必填/必填」判断（不能只看「必填」两字）：含「非必填」→非必填；否则含「必填」→必填。"""
|
||
note = str(router.ws.cell(router.header_row + 1, col).value or "")
|
||
if "非必填" in note:
|
||
return False
|
||
return "必填" in note
|
||
|
||
|
||
def _find_sku_category_headers(router) -> List[str]:
|
||
"""定位「SKU分类」列（模糊匹配，同建议售价检测方式）；匹配到多个时全部返回。"""
|
||
return [str(k) for k in router.column_map if "SKU分类" in str(k)]
|
||
|
||
|
||
def _find_sku_qty_headers(router) -> List[str]:
|
||
"""定位「SKU数量」列（排除「SKU数量单位」列）；匹配到多个时全部返回。"""
|
||
return [str(k) for k in router.column_map if "SKU数量" in str(k) and "单位" not in str(k)]
|
||
|
||
|
||
def _find_sku_qty_unit_headers(router) -> List[str]:
|
||
"""定位「SKU数量单位」列。"""
|
||
return [str(k) for k in router.column_map if "SKU数量单位" in str(k)]
|
||
|
||
|
||
def _required_headers(router, headers) -> List[str]:
|
||
"""过滤「非必填」列：读列头下一行备注（与建议售价同套路），备注显式写「非必填」才跳过。
|
||
|
||
SKU分类/SKU数量/SKU数量单位在本模板族属必填项（SKU分类备注「必填」，数量/单位为按规则填写），
|
||
因此默认视为需填写，仅在列备注写明「非必填」时跳过。"""
|
||
keep: List[str] = []
|
||
for h in headers:
|
||
col = router.column_map.get(h)
|
||
if col is not None:
|
||
note = str(router.ws.cell(router.header_row + 1, col).value or "")
|
||
if "非必填" in note:
|
||
continue
|
||
keep.append(h)
|
||
return keep
|
||
|
||
|
||
def _build_sku_row(spu_code: str, sc: str, sk: Dict[str, Any], size: str, color: str,
|
||
warehouses: List[str], markup_percent: float = 0.0,
|
||
multi: bool = True, price_header: str = "申报价格-日本站",
|
||
bust_headers: Optional[List[str]] = None,
|
||
price_headers: Optional[List[str]] = None,
|
||
sa_size_headers: Optional[List[str]] = None,
|
||
suggested_price_ratio: float = 0.0,
|
||
suggested_price_headers: Optional[List[str]] = None,
|
||
suggested_required: Optional[Dict[str, bool]] = None,
|
||
suggested_unit_headers: Optional[List[str]] = None,
|
||
sku_category_headers: Optional[List[str]] = None,
|
||
sku_qty_headers: Optional[List[str]] = None,
|
||
sku_qty_unit_headers: Optional[List[str]] = None) -> Dict[str, Any]:
|
||
"""构造一行 SKU（固定字段：SPU货号、SKC货号=sku.code、规格类型2、币种 CNY、发货仓1~N 及库存 200）。
|
||
价格（price_headers 列，如 申报价格-美国站/日本站，模糊匹配到多个时全部填）= SKU.price × (1+markup/100)，
|
||
预先填好。建议售价（suggested_price_headers 列，模板「建议售价」必填时才填）= 申报价格 × (1+suggested_price_ratio/100)，
|
||
填了建议售价同时填「建议售价单位」=CNY。bust 填所有「胸围」列（bust_headers，如 基码表-胸围（cm）/胸围全围（cm），检测到才填）。
|
||
size 填「尺码」列 + 所有「沙特阿拉伯码」列（sa_size_headers，如 尺码表-沙特阿拉伯码，检测到才填）。
|
||
规格类型2 统一填「尺码」两个字（不是 size 参数值）。"""
|
||
row: Dict[str, Any] = {
|
||
"基础信息-商品层级": "sku",
|
||
"SPU货号": spu_code,
|
||
"SKC货号": sk.get("code") or sc, # SKC货号 = SKU 的 code（款号-颜色编码）
|
||
"色值(主规格)": color,
|
||
"规格类型2": "尺码", # 规格类型2 统一填「尺码」（不填 size 值）
|
||
"币种": "CNY",
|
||
}
|
||
if price_headers is None:
|
||
price_headers = [price_header]
|
||
if bust_headers is None:
|
||
bust_headers = ["胸围全围（cm）"]
|
||
if sa_size_headers is None:
|
||
sa_size_headers = []
|
||
for j, w in enumerate(warehouses, start=1):
|
||
row[f"发货仓{j}"] = w
|
||
row[f"发货仓{j}库存"] = 200
|
||
for dbk, header in SKU_MAP.items():
|
||
if dbk == "color":
|
||
continue
|
||
v = sk.get(dbk)
|
||
if dbk == "bust":
|
||
if v in (None, ""):
|
||
continue
|
||
for h in bust_headers:
|
||
row[h] = v
|
||
continue
|
||
if dbk == "price":
|
||
if v in (None, ""):
|
||
continue
|
||
v = round(float(v) * (1 + markup_percent / 100), 2) # 申报价格 = price × (1+加价%)
|
||
for h in price_headers:
|
||
row[h] = v
|
||
# 建议售价 = 申报价格 × (1+建议售价比例%)；模板「建议售价」必填时才填，填了同时填单位 CNY
|
||
if suggested_price_headers and suggested_price_ratio > 0:
|
||
suggested = round(v * (1 + suggested_price_ratio / 100), 2)
|
||
for h in suggested_price_headers:
|
||
if (suggested_required or {}).get(h, True):
|
||
row[h] = suggested
|
||
for uh in suggested_unit_headers:
|
||
row[uh] = "CNY"
|
||
continue
|
||
if dbk == "size":
|
||
if v in (None, ""):
|
||
continue
|
||
row[header] = v
|
||
for h in sa_size_headers:
|
||
row[h] = v
|
||
continue
|
||
if v not in (None, ""):
|
||
row[header] = v
|
||
# SKU 分类/数量/数量单位：模板检测到这些列（且非「非必填」）时统一填固定值（单品 / 1 / 件）
|
||
for h in (sku_category_headers or []):
|
||
row[h] = "单品"
|
||
for h in (sku_qty_headers or []):
|
||
row[h] = 1
|
||
for h in (sku_qty_unit_headers or []):
|
||
row[h] = "件"
|
||
return row
|
||
|
||
|
||
def _fill_sku_carousel(router, spu_code: str, color: str, color_col: int,
|
||
first: Dict[str, Any], sku_imgs: List[str]) -> None:
|
||
"""SKU 行商品轮播图2~5：db img_url_2~5 优先，无则回退生成图；按色值列匹配所属 SKU 行。"""
|
||
for j in range(2, 6):
|
||
url = first.get(f"img_url_{j}") # db CDN url（该颜色 SKU 的 img_url_2~5）
|
||
img = url if url not in (None, "") else (sku_imgs[j - 2] if j - 2 < len(sku_imgs) else None)
|
||
if not img:
|
||
continue
|
||
col = _carousel_col(router, j)
|
||
if col is None:
|
||
continue
|
||
for row in router.find_spu_rows(spu_code):
|
||
if (str(router.ws.cell(row, router.level_col).value or "").strip().lower() == "sku"
|
||
and str(router.ws.cell(row, color_col).value or "").strip() == color):
|
||
router.ws.cell(row, col, str(img))
|
||
|
||
# db SPU 字段 -> 上传模板列名
|
||
SPU_MAP: Dict[str, str] = {
|
||
"code": "SPU货号",
|
||
"material": "材质",
|
||
"component_1": "成分1",
|
||
"component_proportion_1": "成分1成分比例",
|
||
"component_2": "成分2",
|
||
"component_proportion_2": "成分2成分比例",
|
||
"component_3": "成分3",
|
||
"component_proportion_3": "成分3成分比例",
|
||
"pattern": "图案",
|
||
"details": "细节",
|
||
"collar_style": "领型",
|
||
"care_Instructions": "护理说明",
|
||
"fabric": "面料",
|
||
"target_audience": "适用人群",
|
||
"season": "季节",
|
||
"is_transparent": "是否透明",
|
||
"layout": "版型",
|
||
"weaving_method": "织造方式",
|
||
"printing_type": "印花类型",
|
||
"fabric_texture_1": "面料纹理1",
|
||
"fabric_weight_1": "面料克重1（g/m²)",
|
||
"fabric_weight_unit_1": "面料克重1（g/m²)单位",
|
||
"lining_texture": "里料纹理",
|
||
}
|
||
|
||
# db SKU 字段 -> 上传模板列名
|
||
SKU_MAP: Dict[str, str] = {
|
||
"color": "色值(主规格)",
|
||
"size": "尺码",
|
||
"size_group": "尺码组别",
|
||
"size_type": "尺码类型",
|
||
"shoulder_width": "肩宽（cm）",
|
||
"bust": "胸围全围（cm）",
|
||
"clothing_length": "衣长（cm）",
|
||
"sleeve_length": "袖长（cm）",
|
||
"longest_side": "最长边（cm）",
|
||
"secondary_long_side": "次长边（cm）",
|
||
"shortest_side": "最短边（cm）",
|
||
"package_weight": "重量（g）", # 包装重量：从 db 提取
|
||
"price": "申报价格-日本站",
|
||
}
|
||
|
||
|
||
def _read_spu(db_path, spu_code: str) -> Optional[Dict[str, Any]]:
|
||
conn = _connect(db_path)
|
||
row = conn.execute("SELECT * FROM SPU WHERE code = ?", (spu_code,)).fetchone()
|
||
conn.close()
|
||
return dict(row) if row else None
|
||
|
||
|
||
def _read_meta(router) -> tuple:
|
||
"""读模板顶头元信息：发货仓（按标签名精准定位，不依赖固定行列）。
|
||
|
||
返回 warehouses：发货仓按「、」分隔的列表（如「名古屋仓、inkreach——东京」→ 2 个）。
|
||
商品产地/产地省份已固定为「中国大陆/广东省」写死在 _build_spu_row，不再读取经营站点。"""
|
||
ws = router.ws
|
||
top = max(1, router.group_row - 1) # 元信息区位于分组行之前
|
||
|
||
def _val(label: str) -> str:
|
||
for row in range(1, top + 1):
|
||
for col in range(1, min(ws.max_column, 30) + 1):
|
||
if str(ws.cell(row, col).value or "").strip() == label:
|
||
return str(ws.cell(row + 1, col).value or "").strip()
|
||
return ""
|
||
|
||
raw = _val("发货仓")
|
||
if not raw:
|
||
# 回退：旧版固定位置（第2行第2列=发货仓）
|
||
raw = str(ws.cell(2, 2).value or "").strip()
|
||
warehouses = [w.strip() for w in raw.split("、") if w.strip()]
|
||
return warehouses
|
||
|
||
|
||
def _read_skus(db_path, spu_code: str, sku_code: str) -> List[Dict[str, Any]]:
|
||
"""该款该颜色的全部尺码 SKU，按尺码从小到大排序（字母码/数字码/童装码/规格码）。"""
|
||
conn = _connect(db_path)
|
||
rows = conn.execute(
|
||
"""SELECT s.*, p.code AS spu_code FROM SKU s
|
||
JOIN SPU p ON s.spu_id = p.id
|
||
WHERE p.code = ? AND s.code = ?""", (spu_code, sku_code)).fetchall()
|
||
conn.close()
|
||
skus = [dict(r) for r in rows]
|
||
skus.sort(key=lambda sk: _size_rank(sk.get("size")))
|
||
return skus
|
||
|
||
|
||
def _import_router(template_dir: str) -> None:
|
||
"""import template_router（优先 config 的 template_dir；打包后回退 _MEIPASS/model；再兜底项目自带 templates/）。"""
|
||
tdir = Path(template_dir)
|
||
candidates = [tdir]
|
||
meipass = getattr(sys, "_MEIPASS", None)
|
||
if meipass:
|
||
candidates.append(Path(meipass) / "model")
|
||
# 用户上传的模板可能在任意目录（无 template_router.py），兜底项目自带 templates/
|
||
from graph.paths import project_root as _proj_root
|
||
candidates.append(_proj_root() / "templates")
|
||
for d in candidates:
|
||
if d.exists() and str(d) not in sys.path:
|
||
sys.path.insert(0, str(d))
|
||
|
||
|
||
def _insert_product_block(
|
||
router, db_path, spu_code, sku_code,
|
||
warehouses, price_headers,
|
||
markup_percent: float = 0.0, images: Optional[List[str]] = None,
|
||
spu_per_color: bool = True,
|
||
oss_code: str = "", cn_title: str = "", en_title: str = "", ja_title: str = "",
|
||
es_title: str = "",
|
||
composite_urls: Optional[List[Dict[str, Any]]] = None,
|
||
seed_shot_urls: Optional[List[str]] = None,
|
||
bust_headers: Optional[List[str]] = None,
|
||
fabric_headers: Optional[List[str]] = None,
|
||
sa_size_headers: Optional[List[str]] = None,
|
||
gender: Optional[str] = None,
|
||
suggested_price_ratio: float = 0.0,
|
||
suggested_price_headers: Optional[List[str]] = None,
|
||
suggested_required: Optional[Dict[str, bool]] = None,
|
||
suggested_unit_headers: Optional[List[str]] = None,
|
||
sku_category_headers: Optional[List[str]] = None,
|
||
sku_qty_headers: Optional[List[str]] = None,
|
||
sku_qty_unit_headers: Optional[List[str]] = None,
|
||
) -> List[int]:
|
||
"""在已打开的 router 中插入一个产品的 SPU+SKU 块并填充设计字段，返回本块行号。
|
||
|
||
供单产品 export_product 与批量 export_products 复用（批量时只打开/保存一次）。
|
||
"""
|
||
spu = _read_spu(db_path, spu_code)
|
||
if spu is None:
|
||
raise ValueError(f"SPU {spu_code} 不存在于 db")
|
||
if isinstance(sku_code, str) and "," in sku_code:
|
||
sku_codes = [s.strip() for s in sku_code.split(",") if s.strip()]
|
||
elif isinstance(sku_code, (list, tuple)):
|
||
sku_codes = list(sku_code)
|
||
else:
|
||
sku_codes = [sku_code]
|
||
skus_by_color: List[tuple] = []
|
||
for sc in sku_codes:
|
||
skus = _read_skus(db_path, spu_code, sc)
|
||
if not skus:
|
||
raise ValueError(f"SKU {sc} 不存在于 db（款号 {spu_code}）")
|
||
skus_by_color.append((sc, skus))
|
||
images = [str(i) for i in (images or []) if i]
|
||
|
||
color_col = router.resolve_col("色值(主规格)")
|
||
block_rows: List[int] = []
|
||
|
||
if spu_per_color:
|
||
# 单 SPU 多色：1 个 SPU 行（无色值，SPU 级信息由 _fill_design_fields 填充）
|
||
# + 全部颜色尺码 SKU 行（色值在 SKU 行区分）
|
||
block_rows.append(router.insert(
|
||
_build_spu_row(spu, spu_code, fabric_headers=fabric_headers,
|
||
gender=gender),
|
||
match="exact",
|
||
))
|
||
for ci, (sc, skus) in enumerate(skus_by_color):
|
||
first = skus[0]
|
||
color = first.get("color") or sc
|
||
|
||
# 该颜色全部尺码 SKU（SKU 行 SPU货号/SKU货号=spu_code，色值区分）
|
||
for i, sk in enumerate(skus):
|
||
size = sk.get("size") or f"{i+1}"
|
||
block_rows.append(router.insert(
|
||
_build_sku_row(spu_code, sc, sk, size, color, warehouses,
|
||
markup_percent=markup_percent, multi=True,
|
||
bust_headers=bust_headers, price_headers=price_headers,
|
||
sa_size_headers=sa_size_headers,
|
||
suggested_price_ratio=suggested_price_ratio,
|
||
suggested_price_headers=suggested_price_headers,
|
||
suggested_required=suggested_required,
|
||
suggested_unit_headers=suggested_unit_headers,
|
||
sku_category_headers=sku_category_headers,
|
||
sku_qty_headers=sku_qty_headers,
|
||
sku_qty_unit_headers=sku_qty_unit_headers),
|
||
spu_code=spu_code, match="exact",
|
||
))
|
||
|
||
# 轮播图：首色 SKU 行轮播图1 = 生成首图；SKU 行按色值填 db url/生成图
|
||
if ci == 0 and images:
|
||
col1 = _carousel_col(router, 1)
|
||
if col1 is not None:
|
||
sku_rows = router.find_sku_rows(spu_code)
|
||
if sku_rows:
|
||
router.ws.cell(min(sku_rows), col1, str(images[0]))
|
||
sku_imgs = [x for x in (images[1:] + images[:1]) if x][:4] if (ci == 0 and images) else []
|
||
_fill_sku_carousel(router, spu_code, color, color_col, first, sku_imgs)
|
||
else:
|
||
# 单 SPU + 多颜色变体：1 个 SPU 行（无色值）+ 所有颜色所有尺码 SKU 行（色值区分）
|
||
block_rows.append(router.insert(
|
||
_build_spu_row(spu, spu_code, fabric_headers=fabric_headers,
|
||
gender=gender), match="exact"))
|
||
multi_variant = len(skus_by_color) > 1
|
||
for ci, (sc, skus) in enumerate(skus_by_color):
|
||
first = skus[0]
|
||
color = first.get("color") or sc
|
||
for i, sk in enumerate(skus):
|
||
size = sk.get("size") or f"{i+1}"
|
||
block_rows.append(router.insert(
|
||
_build_sku_row(spu_code, sc, sk, size, color, warehouses,
|
||
markup_percent=markup_percent, multi=multi_variant,
|
||
bust_headers=bust_headers, price_headers=price_headers,
|
||
sa_size_headers=sa_size_headers,
|
||
suggested_price_ratio=suggested_price_ratio,
|
||
suggested_price_headers=suggested_price_headers,
|
||
suggested_required=suggested_required,
|
||
suggested_unit_headers=suggested_unit_headers,
|
||
sku_category_headers=sku_category_headers,
|
||
sku_qty_headers=sku_qty_headers,
|
||
sku_qty_unit_headers=sku_qty_unit_headers),
|
||
))
|
||
sku_imgs = [x for x in (images[1:] + images[:1]) if x][:4] if (ci == 0 and images) else []
|
||
_fill_sku_carousel(router, spu_code, color, color_col, first, sku_imgs)
|
||
|
||
# 设计联动字段：货号/标题/轮播图路由/详情图文（图床链接，| 分割）
|
||
if oss_code or cn_title or en_title or ja_title or es_title or composite_urls:
|
||
by_sku: Dict[str, Any] = {}
|
||
all_urls: List[str] = []
|
||
for cc in (composite_urls or []):
|
||
if cc.get("color"):
|
||
by_sku[str(cc["color"]).strip()] = cc
|
||
if cc.get("url"):
|
||
all_urls.append(str(cc["url"]))
|
||
_fill_design_fields(router, spu_code, oss_code, cn_title, en_title, ja_title,
|
||
es_title, by_sku, all_urls, seed_shot_urls or [],
|
||
only_rows=block_rows)
|
||
|
||
return block_rows
|
||
|
||
|
||
def export_product(
|
||
db_path,
|
||
spu_code: str,
|
||
sku_code, # str | List[str]：单颜色或多个颜色
|
||
template_dir: str,
|
||
template_path: str,
|
||
out_path: str,
|
||
images: Optional[List[str]] = None,
|
||
spu_per_color: bool = True,
|
||
oss_code: str = "",
|
||
cn_title: str = "",
|
||
en_title: str = "",
|
||
ja_title: str = "",
|
||
es_title: str = "",
|
||
composite_urls: Optional[List[Dict[str, Any]]] = None,
|
||
seed_shot_urls: Optional[List[str]] = None,
|
||
append_to: str = "",
|
||
markup_percent: float = 0.0,
|
||
suggested_price_ratio: float = 0.0,
|
||
) -> Path:
|
||
"""生成商品上传"已填写"模板（支持多产品合并到同一文件）。
|
||
|
||
sku_code ：SKU 颜色编码，支持单个 str 或多个（list/tuple/逗号分隔字符串）。
|
||
spu_per_color ：True（默认）= 每个颜色导出一个 SPU 块；False = 单 SPU 下挂所有颜色 SKU 变体。
|
||
template_dir ：template_router.py 所在目录（用于 import）
|
||
template_path ：商品上传模版 xlsx 路径
|
||
images ：生成的产品图路径列表（仅作用于第一个颜色块：SPU 行轮播图1 + SKU 行回退）
|
||
oss_code ：设计货号（前缀+3位计数），SPU货号/SKU货号 列均填它
|
||
cn_title ：商品名称（中文标题）
|
||
en_title ：英文名称（英文标题）
|
||
ja_title ：日语名称（日语标题，JP 模板生成）
|
||
es_title ：西语名称（西班牙语标题，ES 模板生成）
|
||
composite_urls：[{"sku_code","color","url","code"}] 每色三合一主图（含图床链接与货号）
|
||
seed_shot_urls ：种草图图床链接列表（详情图文 | 拼接用）
|
||
append_to ：已有输出文件路径；提供则在其基础上追加本产品块（一次任务多产品合并一个模板）
|
||
markup_percent ：加价百分比，申报价格 = SKU.price × (1+markup/100) 预填
|
||
返回输出文件路径。
|
||
"""
|
||
_import_router(template_dir)
|
||
from template_router import TemplateRouter # noqa: E402
|
||
|
||
# append_to：合并模式从已有输出文件继续追加（一次任务多产品填一个模板）
|
||
router = TemplateRouter(append_to if append_to else template_path)
|
||
try:
|
||
warehouses = _read_meta(router)
|
||
price_headers = _find_price_headers(router) # 申报价格列（美站/日站/英站…模糊匹配，多个全填）
|
||
bust_headers = _find_bust_headers(router) # 胸围列（基码表-胸围（cm）/胸围全围（cm）…多个全填）
|
||
fabric_headers = _find_fabric_headers(router) # 面料弹性列（SPU商品属性-面料弹性…检测到才填）
|
||
sa_size_headers = _find_sa_size_headers(router) # 沙特阿拉伯码列（尺码表-沙特阿拉伯码…检测到才填）
|
||
gender = _template_gender(template_path) # 类目含「男」→male /「女」→female（成分值映射用）
|
||
suggested_price_headers = _find_suggested_price_headers(router) # 建议售价列（不含单位）
|
||
suggested_unit_headers = _find_suggested_unit_headers(router) # 建议售价单位列
|
||
suggested_required = {h: _read_suggested_required(router, router.column_map[h])
|
||
for h in suggested_price_headers} # 各建议售价列是否必填（结合非必填/必填判断）
|
||
sku_category_headers = _required_headers(router, _find_sku_category_headers(router)) # SKU分类列（非必填才跳过）
|
||
sku_qty_headers = _required_headers(router, _find_sku_qty_headers(router)) # SKU数量列
|
||
sku_qty_unit_headers = _required_headers(router, _find_sku_qty_unit_headers(router)) # SKU数量单位列
|
||
_insert_product_block(router, db_path, spu_code, sku_code,
|
||
warehouses, price_headers,
|
||
markup_percent=markup_percent, images=images,
|
||
spu_per_color=spu_per_color,
|
||
oss_code=oss_code, cn_title=cn_title, en_title=en_title,
|
||
ja_title=ja_title, es_title=es_title,
|
||
composite_urls=composite_urls,
|
||
seed_shot_urls=seed_shot_urls,
|
||
bust_headers=bust_headers,
|
||
fabric_headers=fabric_headers,
|
||
sa_size_headers=sa_size_headers,
|
||
gender=gender,
|
||
suggested_price_ratio=suggested_price_ratio,
|
||
suggested_price_headers=suggested_price_headers,
|
||
suggested_required=suggested_required,
|
||
suggested_unit_headers=suggested_unit_headers,
|
||
sku_category_headers=sku_category_headers,
|
||
sku_qty_headers=sku_qty_headers,
|
||
sku_qty_unit_headers=sku_qty_unit_headers)
|
||
out = router.save(out_path)
|
||
return Path(out)
|
||
finally:
|
||
try:
|
||
router.close()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def export_products(
|
||
db_path,
|
||
products: List[Dict[str, Any]],
|
||
template_dir: str,
|
||
template_path: str,
|
||
out_path: str,
|
||
markup_percent: float = 0.0,
|
||
suggested_price_ratio: float = 0.0,
|
||
) -> Path:
|
||
"""批量合并导出：所有产品一次性写入同一模板，只打开/保存一次。
|
||
|
||
products 每项字段：spu_code / sku_codes（str 或 list）/ oss_code / cn_title / en_title /
|
||
ja_title / es_title / composite_urls / seed_shot_urls / images / spu_per_color。
|
||
相比逐产品调用 export_product（每次全量读写工作簿），批量模式显著提速。
|
||
"""
|
||
_import_router(template_dir)
|
||
from template_router import TemplateRouter # noqa: E402
|
||
|
||
router = TemplateRouter(template_path)
|
||
try:
|
||
warehouses = _read_meta(router)
|
||
price_headers = _find_price_headers(router) # 申报价格列（美站/日站/英站…模糊匹配，多个全填）
|
||
bust_headers = _find_bust_headers(router) # 胸围列（基码表-胸围（cm）/胸围全围（cm）…多个全填）
|
||
fabric_headers = _find_fabric_headers(router) # 面料弹性列（SPU商品属性-面料弹性…检测到才填）
|
||
sa_size_headers = _find_sa_size_headers(router) # 沙特阿拉伯码列（尺码表-沙特阿拉伯码…检测到才填）
|
||
gender = _template_gender(template_path) # 类目含「男」→male /「女」→female（成分值映射用）
|
||
suggested_price_headers = _find_suggested_price_headers(router) # 建议售价列（不含单位）
|
||
suggested_unit_headers = _find_suggested_unit_headers(router) # 建议售价单位列
|
||
suggested_required = {h: _read_suggested_required(router, router.column_map[h])
|
||
for h in suggested_price_headers} # 各建议售价列是否必填（结合非必填/必填判断）
|
||
sku_category_headers = _required_headers(router, _find_sku_category_headers(router)) # SKU分类列（非必填才跳过）
|
||
sku_qty_headers = _required_headers(router, _find_sku_qty_headers(router)) # SKU数量列
|
||
sku_qty_unit_headers = _required_headers(router, _find_sku_qty_unit_headers(router)) # SKU数量单位列
|
||
for r in products:
|
||
try:
|
||
_insert_product_block(
|
||
router, db_path, r.get("spu_code", ""),
|
||
r.get("sku_codes") or r.get("sku_code") or "",
|
||
warehouses, price_headers,
|
||
markup_percent=markup_percent,
|
||
images=r.get("images"),
|
||
spu_per_color=bool(r.get("spu_per_color", True)),
|
||
oss_code=r.get("oss_code", ""),
|
||
cn_title=r.get("cn_title", ""),
|
||
en_title=r.get("en_title", ""),
|
||
ja_title=r.get("ja_title", ""),
|
||
es_title=r.get("es_title", ""),
|
||
composite_urls=r.get("composite_urls"),
|
||
seed_shot_urls=r.get("seed_shot_urls"),
|
||
bust_headers=bust_headers,
|
||
fabric_headers=fabric_headers,
|
||
sa_size_headers=sa_size_headers,
|
||
gender=gender,
|
||
suggested_price_ratio=suggested_price_ratio,
|
||
suggested_price_headers=suggested_price_headers,
|
||
suggested_required=suggested_required,
|
||
suggested_unit_headers=suggested_unit_headers,
|
||
sku_category_headers=sku_category_headers,
|
||
sku_qty_headers=sku_qty_headers,
|
||
sku_qty_unit_headers=sku_qty_unit_headers,
|
||
)
|
||
except Exception as e: # noqa: BLE001
|
||
print(f"[template_export] 产品 {r.get('spu_code')} 写入失败，跳过: {e}")
|
||
continue
|
||
out = router.save(out_path)
|
||
return Path(out)
|
||
finally:
|
||
try:
|
||
router.close()
|
||
except Exception:
|
||
pass
|