v106-v109 童装支持 + 标题模板外部化 + 模板导出增强

- 新增男童/女童检测(gender_from_category 优先判童装)与童装场景图生成
  (configs/kids_features.yaml:模特/场景/服装风格,同一商品固定同一组)
- 童装 SPU 字段映射:kids_type→SPU商品属性-类型、kids_age→适用年龄段、
  target_audience 按性别映射、kids_type_map 女童「上衣」→「针织上衣」
- 标题生成提示词外部化:prompts/title_prompt_{1,2,3}.md + config.yaml 路由表
- 模板多站点匹配:经营站点可配多个,命中任意一个即匹配
- 种草图生成失败自动重试(seed_shot.retries,换场景/模特/风格)
- 模板导出新增 Preview 文件夹(成功产品 _composite.oss.jpg + result.xlsx)
- 修复 SKU 尺码未按从小到大排序(_size_rank 支持单一年龄码 6Y/10Y)
This commit is contained in:
2026-09-01 17:24:23 +08:00
parent 3dc594cf57
commit d68cc3b9e3
20 changed files with 826 additions and 136 deletions
+41 -8
View File
@@ -13,6 +13,7 @@ from pathlib import Path
from typing import Any, Dict, List
import requests
import yaml
# 模型调用一律直连:用户常开 VPN(系统代理),LLM 网关多为国内/自建,走代理会被拦截或变慢。
# 仅请求级 proxies=NO_PROXY 直连,不设置进程级 NO_PROXY 环境变量(避免影响 Google Trends 等外部采集)。
@@ -101,12 +102,14 @@ Rules:
- Return ONLY JSON of shape: {"style_seeds": [...], "related_seeds": [...]}'''
# —— 商品标题生成(多模态:分析服装图片 → 中英双语 SEO 标题)——
# 模板字典按编号存放;TITLE_TEMPLATE_ROUTE 按国家路由到模板编号。
# —— 商品标题生成(多模态:分析服装图片 → SEO 标题)——
# 提示词模板存 prompts/title_prompt_<编号>.md(可自由编辑,优先运行根 exe 旁,回退数据根);
# 国家 → 模板编号路由存 config.yaml 顶层 title_templates.route(可自由编辑)。
# 模板 1:英语市场(US/GB/AU/MX)→ en_title + cn_title
# 模板 2:日本市场(JP)→ en_title + cn_title + ja_title
# 模板 3:西班牙市场(ES)→ es_title + cn_title
TITLE_TEMPLATES: Dict[str, str] = {
# 以下为「内置默认」兜底:md/yaml 缺失或留空时回退,避免标题生成中断。
_TITLE_TEMPLATES_FALLBACK: Dict[str, str] = {
"1": '''# Role
你是一位资深的跨境服装运营专家,精通英语电商的SEO标题逻辑。你的任务是通过分析服装图片,生成高权重的英语-中文商品标题。
@@ -172,8 +175,8 @@ TITLE_TEMPLATES: Dict[str, str] = {
{"es_title": "Título en español", "cn_title": "中文标题"}''',
}
# 国家 → 标题模板编号(JP 路由到模板 2ES 路由到模板 3,其余默认模板 1;后续可按国家新增模板
TITLE_TEMPLATE_ROUTE: Dict[str, str] = {
# 内置默认路由(config.yaml title_templates.route 缺失/留空时回退;JP→模板2ES模板3,其余默认模板1
_TITLE_ROUTE_FALLBACK: Dict[str, str] = {
"US": "1",
"GB": "1",
"JP": "2",
@@ -183,6 +186,33 @@ TITLE_TEMPLATE_ROUTE: Dict[str, str] = {
}
def _read_title_prompt(tpl_no: str) -> str:
"""读取 prompts/title_prompt_<编号>.md:优先运行根(exe 旁,可编辑),回退数据根;找不到/空返回空串。"""
for base in (runtime_root(), project_root()):
p = base / "prompts" / f"title_prompt_{tpl_no}.md"
if p.exists():
text = p.read_text(encoding="utf-8").strip()
if text:
return text
return ""
def _load_title_route() -> Dict[str, str]:
"""从 config.yaml 顶层 title_templates.route 读取国家→模板编号路由;缺失/留空回退内置默认。"""
for base in (runtime_root(), project_root()):
p = base / "config.yaml"
if not p.exists():
continue
try:
data = yaml.safe_load(p.read_text(encoding="utf-8")) or {}
route = (data.get("title_templates") or {}).get("route") or {}
if isinstance(route, dict) and route:
return {str(k): str(v) for k, v in route.items()}
except Exception as e: # noqa: BLE001
print(f"[titles] 读取 config.yaml title_templates.route 失败: {e}")
return dict(_TITLE_ROUTE_FALLBACK)
def _inject_now(prompt: str) -> str:
"""把模板中的 {year}/{month}/{season} 替换为当前时间(用 replace 避免 JSON 花括号冲突)。"""
import datetime
@@ -197,8 +227,10 @@ def _inject_now(prompt: str) -> str:
def resolve_title_prompt(country: str = "") -> str:
"""按国家解析标题生成提示词(自动注入当前时间变量);未知国家/留空回退模板 1。"""
tpl_no = TITLE_TEMPLATE_ROUTE.get(country or "", "1")
return _inject_now(TITLE_TEMPLATES.get(tpl_no, TITLE_TEMPLATES["1"]))
route = _load_title_route()
tpl_no = route.get(country or "", "1")
prompt = _read_title_prompt(tpl_no) or _TITLE_TEMPLATES_FALLBACK.get(tpl_no, _TITLE_TEMPLATES_FALLBACK["1"])
return _inject_now(prompt)
def build_seed_user_prompt(context: Dict[str, Any]) -> str:
@@ -810,7 +842,8 @@ class OpenAICompatBackend(LLMBackend):
def generate_title(self, image_path: str, system_prompt: str = "", country: str = "") -> Dict[str, Any]:
"""多模态:分析服装图片,生成商品标题(按国家路由模板)。
系统提示词:显式传入优先;否则按 country 经 TITLE_TEMPLATE_ROUTE 路由到对应模板。
系统提示词:显式传入优先;否则按 country 经 config.yaml title_templates.route 路由到
prompts/title_prompt_<编号>.md 对应模板(缺失回退内置默认)。
模板 1US/GB/AU/MX)返回 {"en_title","cn_title"}
模板 2(JP)额外返回 {"ja_title"}
模板 3ES)返回 {"es_title","cn_title"}。
+21 -16
View File
@@ -113,18 +113,26 @@ FLAT_LAY_PROMPT = (
)
def _active_prompt(kind: str, prompts: Optional[Dict] = None) -> str:
def _read_prompt_md(filename: str) -> str:
"""读取 prompts/<filename> 提示词文件:优先运行根(exe 旁,可编辑),回退数据根;找不到/空返回空串。"""
for base in (runtime_root(), project_root()):
p = base / "prompts" / filename
if p.exists():
text = p.read_text(encoding="utf-8").strip()
if text:
return text
return ""
def _active_prompt(kind: str) -> str:
"""返回当前图源应使用的合成提示词。
kind="model" → 模特三图提示词(配置覆盖优先生效,否则内置 MODEL_WEAR_PROMPT);
kind="flat" → 平铺三图提示词(配置覆盖优先生效,否则内置 FLAT_LAY_PROMPT)。
kind="model" → 模特三图提示词(优先读 prompts/model_prompt.md,缺失/留空回退内置 MODEL_WEAR_PROMPT);
kind="flat" → 平铺三图提示词(优先读 prompts/flat_prompt.md,缺失/留空回退内置 FLAT_LAY_PROMPT)。
"""
cfgs = prompts or {}
if kind == "flat":
return (str(cfgs.get("flat_prompt") or "").strip()
or FLAT_LAY_PROMPT)
return (str(cfgs.get("model_prompt") or "").strip()
or MODEL_WEAR_PROMPT)
return _read_prompt_md("flat_prompt.md") or FLAT_LAY_PROMPT
return _read_prompt_md("model_prompt.md") or MODEL_WEAR_PROMPT
def _resolve_sku(db_path, basemap_root, spu_code: str, sku_code: str, colors=None) -> Optional[str]:
@@ -187,7 +195,7 @@ def _process_spu(
db_path, basemap_root, material_root, category, prod_dir, brief, ib,
spu, sku_code, pcfg, errors, shared_design=None, title_backend=None, country="",
img_code="", model_img=None, design_size="1024x1024", compose_size="1536x2048",
on_503=None, model_kind="model", prompts=None,
on_503=None, model_kind="model",
) -> Optional[Dict[str, Any]]:
"""处理单个款号:选色 → 底图 → 设计稿 → (mark==1) 模特 → 合成 → 模板导出。
shared_design: compose 节点生成的纯印花设计稿路径(图2);为 None 时回退本节点 generate。
@@ -195,7 +203,6 @@ def _process_spu(
(按货号命名,包含该货号对应的所有图片)。
on_503: 致命图像服务错误(503/账户不可用)回调(供调用方提前终止任务)。
model_kind: 图源类型 "model"(模特图)/ "flat"(平铺图),决定用哪个合成提示词。
prompts: {"model": str, "flat": str} 可配置提示词覆盖(来自 config.product.mark_dirs);缺省用内置。
返回 result dict;内部异常已兜底,不中断。
"""
# 输出目录 = 货号子文件夹(product/DG000/…),该货号所有图片都放这里
@@ -324,8 +331,8 @@ def _process_spu(
elif model_img is not None:
composite_path = str(prod_dir / f"{img_code}_composite.png")
try:
# 三图合成:按图源类型选提示词(可配置覆盖优先,否则内置 MODEL_WEAR/FLAT_LAY
wear_prompt = _active_prompt(model_kind, prompts)
# 三图合成:按图源类型选提示词(优先读 prompts/*.md,否则内置 MODEL_WEAR/FLAT_LAY
wear_prompt = _active_prompt(model_kind)
kind_label = "平铺" if model_kind == "flat" else "模特"
print(f"{tag} {kind_label}三图合成提交中(3 参考图 img2img,网关处理约 2-6 分钟,请耐心等待)…")
t0 = time.time()
@@ -407,7 +414,7 @@ def _process_spu(
continue
cp = str(prod_dir / f"{img_code}_{str(sc).split('-')[-1]}_composite.png")
try:
ib.print(_active_prompt(model_kind, prompts), str(model_img), cp,
ib.print(_active_prompt(model_kind), str(model_img), cp,
brief.get("composite_negative", ""),
extra_images=[design_path, str(bm)], # 图2印花, 图3该色底图
size=compose_size) # 合成图尺寸按 config compose.size
@@ -546,7 +553,6 @@ def product_node(state: Dict[str, Any]) -> Dict[str, Any]:
# → product_node 用对应提示词合成(模特三图 / 平铺三图)
model_assign: Dict[int, Any] = {}
mark_dirs = pcfg.get("mark_dirs") or {}
prompts_cfg = (mark_dirs.get("1") or {})
try:
from graph.product import build_mark_sources
sources = build_mark_sources(material_root, mark_dirs, category)
@@ -559,7 +565,7 @@ def product_node(state: Dict[str, Any]) -> Dict[str, Any]:
if source_pool:
for _i in range(len(worklist)):
img, kind = random.choice(source_pool) # 每任务随机抽一张(含 kind
model_assign[_i] = {"img": img, "kind": kind, "prompts": prompts_cfg}
model_assign[_i] = {"img": img, "kind": kind}
print(f"[product] 任务级图源分配:{len(model_assign)} 个产品任务(mark_dirs 模特/平铺图随机抽,含 kind)")
else:
print("[product] material_library 无任何可用图(模特/平铺均无)→ 跳过合成,仅导出模板")
@@ -628,7 +634,6 @@ def product_node(state: Dict[str, Any]) -> Dict[str, Any]:
country, img_code=img_code,
model_img=_src.get("img"), # 按任务序号取独立随机图源
model_kind=_src.get("kind", "model"),
prompts=_src.get("prompts"),
design_size=str((config.get("compose") or {}).get("design_size") or "1024x1024"),
compose_size=str((config.get("compose") or {}).get("size") or "1536x2048"))
if r:
+4 -2
View File
@@ -103,7 +103,8 @@ def seed_shot_node(state: Dict[str, Any]) -> Dict[str, Any]:
category = read_template_category(tp)
gender = gender_from_category(category)
if gender:
print(f"[seed_shot] 类目「{category[:30]}…」含{'' if gender == 'male' else ''} → 固定 {gender} 模特")
label = {"male": "", "female": "", "boy_kids": "男童", "girl_kids": "女童"}.get(gender, gender)
print(f"[seed_shot] 类目「{category[:30]}…」检测到 {label} → 固定 {gender} 模特")
elif category:
print(f"[seed_shot] 类目「{category[:30]}…」无男/女 → 男女模特随机")
@@ -141,7 +142,8 @@ def seed_shot_node(state: Dict[str, Any]) -> Dict[str, Any]:
continue
generated = generate_seed_shots(ib, base, cn, material, n, str(shot_dir),
r.get("composite_negative", ""),
size=size, prefix=pfx, gender=gender)
size=size, prefix=pfx, gender=gender,
retries=int(ss_cfg.get("retries", 3)))
paths.extend(generated)
if not paths:
return None
+37 -1
View File
@@ -29,6 +29,38 @@ def _template_out_path(prod_dir: Path, tpl_name: str) -> Path:
return prod_dir / f"{tpl_name}_已填写_{int(time.time())}.xlsx"
def _build_preview(output_dir: Path, products: List[Dict[str, Any]]) -> None:
"""生成 Preview 文件夹:仅成功导入模板的产品(template_path 已设置),
复制其 _composite.oss.jpg(压缩版合成图)到 Preview,并生成 result.xlsx
(货号 | 中文标题 | 英文标题 三列)。"""
import shutil
from openpyxl import Workbook
preview_dir = output_dir / "Preview"
preview_dir.mkdir(parents=True, exist_ok=True)
rows: List[tuple] = []
for r in products:
if not r.get("template_path"):
continue
code = str(r.get("img_code") or r.get("oss_code") or "")
comp = r.get("composite_path") or r.get("printed_path")
if comp:
oss_file = Path(comp).with_suffix(".oss.jpg")
if oss_file.exists():
dst = preview_dir / oss_file.name
if not dst.exists():
shutil.copy2(oss_file, dst)
rows.append((code, r.get("cn_title") or "", r.get("en_title") or ""))
if rows:
wb = Workbook()
ws = wb.active
ws.title = "result"
ws.append(["货号", "中文标题", "英文标题"])
for row in rows:
ws.append(list(row))
wb.save(str(preview_dir / "result.xlsx"))
print(f"[template] Preview 已生成({len(rows)} 个成功产品): {preview_dir}")
@with_fallback("template_export")
def template_export_node(state: Dict[str, Any]) -> Dict[str, Any]:
products: List[Dict[str, Any]] = state.get("product") or []
@@ -68,7 +100,8 @@ def template_export_node(state: Dict[str, Any]) -> Dict[str, Any]:
break
from graph.template_export import (export_products, _resolve_component_map,
_resolve_season_map, _resolve_pattern_map)
_resolve_season_map, _resolve_pattern_map,
_resolve_target_audience_map, _resolve_kids_type_map)
tdir = (pcfg.get("template_dir") or "").strip() or str(Path(tp).parent)
prod_dir = output_dir / "product"
prod_dir.mkdir(parents=True, exist_ok=True)
@@ -124,12 +157,15 @@ def template_export_node(state: Dict[str, Any]) -> Dict[str, Any]:
component_map=_resolve_component_map(config),
season_map=_resolve_season_map(config),
pattern_map=_resolve_pattern_map(config),
target_audience_map=_resolve_target_audience_map(config),
kids_type_map=_resolve_kids_type_map(config),
)
for r in products:
if (r.get("composite_path") or r.get("printed_path")) and (r.get("en_title") or "").strip():
r["template_path"] = str(out)
exported.append(str(out))
print(f"[template] 商品上传模板已生成({len(batch)} 个产品一次合并): {out}")
_build_preview(output_dir, products)
except Exception as e: # noqa: BLE001
errors.append({"node": "template_export", "type": type(e).__name__,
"message": f"模板批量导出失败: {e}", "trace": ""})
+6 -6
View File
@@ -105,7 +105,8 @@ class PinterestPipeline:
category = read_template_category(tp)
self._gender = gender_from_category(category)
if self._gender:
print(f"[pinterest_pipeline] 类目「{category[:30]}…」含{'' if self._gender == 'male' else ''} → 固定 {self._gender} 模特")
label = {"male": "", "female": "", "boy_kids": "男童", "girl_kids": "女童"}.get(self._gender, self._gender)
print(f"[pinterest_pipeline] 类目「{category[:30]}…」检测到 {label} → 固定 {self._gender} 模特")
elif category:
print(f"[pinterest_pipeline] 类目「{category[:30]}…」无男/女 → 男女模特随机")
@@ -240,8 +241,7 @@ class PinterestPipeline:
if not pool:
continue
img, kind = _random.choice(pool) # 每任务独立随机抽一张(含 kind)
assign[key] = {"img": img, "kind": kind,
"prompts": (mark_dirs.get(mark) or mark_dirs.get("1") or {})}
assign[key] = {"img": img, "kind": kind}
return assign
def _load_materials(self) -> Dict[str, str]:
@@ -674,8 +674,7 @@ class PinterestPipeline:
if spu is not None and int(spu.get("mark") or 0) != 1:
return {"model_img": img} # 非 mark=1 走旧逻辑(不传 kindproduct_node 按其 mark 自行判定)
return {"model_img": img,
"model_kind": src.get("kind", "model"),
"prompts": src.get("prompts") or {}}
"model_kind": src.get("kind", "model")}
def _process_spu(self, brief: Dict[str, Any], spu, skus: str, img_code: str,
design_path: str, task_idx: int = 0) -> Optional[Dict[str, Any]]:
@@ -769,7 +768,8 @@ class PinterestPipeline:
try:
generated = generate_seed_shots(self._ib, base, cn, material, n, str(shot_dir),
r.get("composite_negative", ""),
size=size, prefix=pfx, gender=self._gender)
size=size, prefix=pfx, gender=self._gender,
retries=int(ss_cfg.get("retries", 3)))
except Exception as e: # noqa: BLE001
if self.is_fatal_503(e):
self.record_503()
+117 -21
View File
@@ -44,9 +44,47 @@ def load_templates() -> List[Dict[str, str]]:
for t in tpls if t.get("prompt")]
_KIDS_FALLBACK = {
"model_features": {
"boy_kids": ["6岁亚洲小男孩,黑色短发,白净肌肤,圆脸大眼睛,活泼可爱"],
"girl_kids": ["6岁亚洲小女孩,黑色长直发,白净肌肤,圆脸大眼睛,甜美可爱"],
},
"scene_features": {
"boy_kids": ["阳光明媚的公园草坪,儿童游乐设施虚化背景,明亮自然光"],
"girl_kids": ["阳光明媚的公园草坪,花朵与秋千虚化背景,明亮自然光"],
},
"style_features": {
"boy_kids": ["童趣休闲风,活泼明快,适合日常玩耍"],
"girl_kids": ["童趣甜美风,活泼可爱,适合日常玩耍"],
},
}
def _load_kids_features(section: str, gender: str) -> List[str]:
"""从 kids_features.yaml 读取指定 sectionmodel_features/scene_features/style_features)的 gender 特征列表。
支持两种条目格式:字符串直接作为特征;dict 取 prompt 字段(场景特征为完整独立提示词时用)。"""
data = _load_yaml("configs/kids_features.yaml")
kf = (data.get("kids_features") or {}).get(section) or {}
feats: List[str] = []
for f in kf.get(gender, []):
if isinstance(f, dict):
p = f.get("prompt")
if p and str(p).strip():
feats.append(str(p).strip())
elif f and str(f).strip():
feats.append(str(f).strip())
return feats
def load_model_features(gender: Optional[str] = None) -> List[str]:
"""模特特征列表(无配置时给内置兜底)。
gender: "male"/"female" 时只返回对应性别;None/其他 返回全部(指定性别组为空时回退全部)。"""
gender: "male"/"female" 时只返回对应性别;"boy_kids"/"girl_kids" 时从 kids_features.yaml 读取童装模特;
None/其他 返回全部(指定性别组为空时回退全部)。"""
if gender in ("boy_kids", "girl_kids"):
feats = _load_kids_features("model_features", gender)
if feats:
return feats
return list(_KIDS_FALLBACK["model_features"].get(gender, []))
data = _load_yaml("configs/model_features.yaml")
mf = data.get("model_features") or []
feats: List[str] = []
@@ -62,6 +100,16 @@ def load_model_features(gender: Optional[str] = None) -> List[str]:
return feats
def load_scene_features(gender: Optional[str] = None) -> List[str]:
"""场景特征列表(童装专用,从 kids_features.yaml 读取;无配置给内置兜底)。"""
if gender in ("boy_kids", "girl_kids"):
feats = _load_kids_features("scene_features", gender)
if feats:
return feats
return list(_KIDS_FALLBACK["scene_features"].get(gender, []))
return []
def read_template_category(template_path: str) -> str:
"""读取模版「类目」表头对应的值(如 服装、鞋靴和珠宝饰品>男士时尚>男装>男装上衣、T恤、衬衫>男装T恤)。
遍历所有 sheet(类目表头可能在「模版」等 sheet),找到即返回其下一行同列值。"""
@@ -85,7 +133,13 @@ def read_template_category(template_path: str) -> str:
def gender_from_category(category: str) -> Optional[str]:
"""类目含「男」→ male;含「女」→ female;都不含 → None(全部随机)。"""
"""类目含「男童」→ boy_kids;含「女童」→ girl_kids;含「男」→ male;含「女」→ female;都不含 → None
顺序必须先判「男童/女童」再判「男/女」,因为「男童」也含「男」、「女童」也含「女」。"""
if "男童" in category:
return "boy_kids"
if "女童" in category:
return "girl_kids"
if "" in category:
return "male"
if "" in category:
@@ -93,8 +147,15 @@ def gender_from_category(category: str) -> Optional[str]:
return None
def load_style_features() -> List[str]:
"""服装风格列表(style_features.yaml,随机取一条替换 [服装风格];无配置时内置兜底)。"""
def load_style_features(gender: Optional[str] = None) -> List[str]:
"""服装风格列表(童装从 kids_features.yaml 读取;无配置回退 style_features.yaml 全量)。"""
if gender in ("boy_kids", "girl_kids"):
feats = _load_kids_features("style_features", gender)
if feats:
return feats
feats = list(_KIDS_FALLBACK["style_features"].get(gender, []))
if feats:
return feats
data = _load_yaml("configs/style_features.yaml")
feats = [str(f) for f in (data.get("style_features") or []) if str(f).strip()]
if not feats:
@@ -104,35 +165,47 @@ def load_style_features() -> List[str]:
def render_prompt(template_prompt: str, cn_title: str, material: str, model_feature: str,
style_feature: str = "") -> str:
"""占位替换:[商品名称]/[材质]/[模特特征]/[服装风格]"""
style_feature: str = "", scene_feature: str = "") -> str:
"""占位替换:[商品名称]/[材质]/[模特特征]/[服装风格]/[场景]
模板含 [场景] 占位符时替换;不含时若提供了场景描述则追加到提示词末尾(童装场景图生效)。"""
out = template_prompt.replace("[商品名称]", (cn_title or "").strip() or "这件衣服")
out = out.replace("[材质]", (material or "").strip() or "面料")
out = out.replace("[模特特征]", (model_feature or "").strip() or "模特")
out = out.replace("[服装风格]", (style_feature or "").strip() or "日常休闲风")
scene = (scene_feature or "").strip()
if "[场景]" in template_prompt:
out = out.replace("[场景]", scene or "明亮干净的室内摄影棚")
elif scene:
out = f"{out} 场景:{scene}"
return out
def generate_seed_shots(image_backend, base_image: str, cn_title: str, material: str,
count: int, out_dir: str, negative: str = "",
size: str = "1536x2048", prefix: str = "",
gender: Optional[str] = None) -> List[str]:
gender: Optional[str] = None, retries: int = 3) -> List[str]:
"""生成 count 张种草图(img2img,图1=合成图)。返回产物路径列表。
size: 种草图统一 1536x2048(与合成图一致)。
prefix: 货号前缀(对应产品货号,命名 {prefix}_{随机4位}.png,不覆盖旧文件)。
gender: "male"/"female" 时只从对应性别模特特征随机;None 全部随机。
占位符 [商品名称]/[材质]/[模特特征]/[服装风格] 均随机组合(模板/模特/服装风格各随机取一条)。"""
gender: "male"/"female" 时只从对应性别模特特征随机;"boy_kids"/"girl_kids" 时从 kids_features.yaml
读取童装模特/场景/服装风格;None 全部随机。
retries: 单张失败重试次数(图像 API 偶发安全拦截 400 时自动换一组场景/模特/风格重试)。
占位符 [商品名称]/[材质]/[模特特征]/[服装风格]/[场景] 均随机组合(模板/模特/服装风格/场景各随机取一条)。
童装(boy_kids/girl_kids):场景特征为完整独立提示词,直接作为模板使用(不套用 seed_shot_templates.yaml),
且同一商品全套图固定同一组 [模特特征]+[场景]+[服装风格],系列图调性统一(失败重试时换一组)。"""
templates = load_templates()
features = load_model_features(gender=gender)
style_features = load_style_features()
style_features = load_style_features(gender=gender)
scene_features = load_scene_features(gender=gender)
is_kids = gender in ("boy_kids", "girl_kids")
out = Path(out_dir)
out.mkdir(parents=True, exist_ok=True)
paths: List[str] = []
# 童装:全套图固定同一组 模特+场景+服装风格(系列图调性统一;失败重试时换一组)
kids_feat = random.choice(features) if is_kids else ""
kids_style = random.choice(style_features) if is_kids else ""
kids_scene = random.choice(scene_features) if (is_kids and scene_features) else ""
for i in range(count):
tpl = random.choice(templates)
feat = random.choice(features)
style_feat = random.choice(style_features)
prompt = render_prompt(tpl["prompt"], cn_title, material, feat, style_feat)
# 命名:{货号}_{随机4位}.png(按货号命名,随机4位避免自增序号/覆盖)
while True:
rand4 = f"{random.randint(0, 9999):04d}"
@@ -140,11 +213,34 @@ def generate_seed_shots(image_backend, base_image: str, cn_title: str, material:
else out / f"seed_{rand4}.png")
if not Path(out_path).exists():
break
try:
image_backend.print(prompt, base_image, out_path, negative, size=size)
paths.append(out_path)
print(f"[seed_shot] 已生成种草图 {i + 1}/{count}: {out_path}"
f"(模板={tpl['name']},模特={feat[:14]}…,风格={style_feat[:14]}…)")
except Exception as e: # noqa: BLE001
print(f"[seed_shot] 种草图 {i + 1} 生成失败: {e}")
last_err: Optional[Exception] = None
for attempt in range(1, retries + 1):
if is_kids and kids_scene:
# 童装:场景完整提示词直接作为模板,仅填占位符,不追加任何其他模板
prompt = render_prompt(kids_scene, cn_title, material, kids_feat, kids_style)
tpl_name = "kids_scene"
feat, style_feat = kids_feat, kids_style
else:
tpl = random.choice(templates)
feat = random.choice(features)
style_feat = random.choice(style_features)
scene_feat = random.choice(scene_features) if scene_features else ""
prompt = render_prompt(tpl["prompt"], cn_title, material, feat, style_feat, scene_feat)
tpl_name = tpl["name"]
try:
image_backend.print(prompt, base_image, out_path, negative, size=size)
paths.append(out_path)
print(f"[seed_shot] 已生成种草图 {i + 1}/{count}: {out_path}"
f"(模板={tpl_name},模特={feat[:14]}…,风格={style_feat[:14]}…)")
break
except Exception as e: # noqa: BLE001
last_err = e
print(f"[seed_shot] 种草图 {i + 1}{attempt}/{retries} 次失败: {e}")
# 重试换一组(童装换场景/模特/风格;成人换模板/特征)
if is_kids and scene_features:
kids_scene = random.choice(scene_features)
kids_feat = random.choice(features)
kids_style = random.choice(style_features)
else:
print(f"[seed_shot] 种草图 {i + 1} 重试 {retries} 次仍失败: {last_err}")
return paths
+248 -6
View File
@@ -137,6 +137,84 @@ def _map_season(value: Any, gender: Optional[str], season_map: Optional[Dict[str
return mapping.get(str(value), value)
# 适用人群值字典映射:db target_audience 值 → 童装模板下拉框选项(男童/女童互映射)。
# 女童模板遇到 target_audience="男童" 时映射为「女童」;男童模板遇到「女童」时映射为「男童」。
# 该字典为「内置默认」,运行时可被 config.yaml 顶层 target_audience_map.kids 覆盖(缺失/留空回退这里)。
_TARGET_AUDIENCE_KIDS_MAP = {
"girl_kids": {"男童": "女童"},
"boy_kids": {"女童": "男童"},
}
def _resolve_target_audience_map(config: Optional[Dict[str, Any]]) -> Dict[str, Any]:
"""从 config 读取可配置适用人群映射(config.target_audience_map.kids),缺失/留空回退内置默认。"""
if not config:
return {g: dict(m) for g, m in _TARGET_AUDIENCE_KIDS_MAP.items()}
kids = None
try:
tm = config.get("target_audience_map") or {}
kids = (tm.get("kids") or {}) if isinstance(tm, dict) else {}
except Exception: # noqa: BLE001
kids = {}
if not isinstance(kids, dict) or not kids:
return {g: dict(m) for g, m in _TARGET_AUDIENCE_KIDS_MAP.items()}
merged = {g: dict(m) for g, m in _TARGET_AUDIENCE_KIDS_MAP.items()}
for g, m in kids.items():
if isinstance(m, dict):
merged.setdefault(str(g), {}).update(
{str(k): v for k, v in m.items() if v not in (None, "")})
return merged
def _map_target_audience(value: Any, gender: Optional[str],
kids_map: Optional[Dict[str, Any]] = None) -> Any:
"""适用人群值按童装性别映射(仅 spu.target_audience 字段):gender 为 boy_kids/girl_kids 时
查适用人群映射表(默认内置、可由 config 覆盖,找不到保留原值),如「男童」→「女童」。"""
if gender not in ("boy_kids", "girl_kids") or value in (None, ""):
return value
mapping = kids_map if kids_map is not None else _TARGET_AUDIENCE_KIDS_MAP
return mapping.get(gender, {}).get(str(value), value)
# kids_type 值字典映射:db kids_type 值 → 模板「SPU商品属性-类型」下拉框选项(仅女童模板映射)。
# 女童模板遇到 kids_type="上衣" 时映射为「针织上衣」;男童模板保留原值。
# 该字典为「内置默认」,运行时可被 config.yaml 顶层 kids_type_map.girl_kids 覆盖(缺失/留空回退这里)。
_KIDS_TYPE_GIRL_MAP = {
"girl_kids": {"上衣": "针织上衣"},
}
def _resolve_kids_type_map(config: Optional[Dict[str, Any]]) -> Dict[str, Any]:
"""从 config 读取可配置 kids_type 映射(config.kids_type_map),缺失/留空回退内置默认。"""
if not config:
return {g: dict(m) for g, m in _KIDS_TYPE_GIRL_MAP.items()}
kids = None
try:
ktm = config.get("kids_type_map") or {}
kids = ktm if isinstance(ktm, dict) else {}
except Exception: # noqa: BLE001
kids = {}
if not isinstance(kids, dict) or not kids:
return {g: dict(m) for g, m in _KIDS_TYPE_GIRL_MAP.items()}
merged = {g: dict(m) for g, m in _KIDS_TYPE_GIRL_MAP.items()}
for g, m in kids.items():
if isinstance(m, dict):
merged.setdefault(str(g), {}).update(
{str(k): v for k, v in m.items() if v not in (None, "")})
return merged
def _map_kids_type(value: Any, gender: Optional[str],
kids_type_map: Optional[Dict[str, Any]] = None) -> Any:
"""kids_type 值按童装性别映射(仅 spu.kids_type 字段):gender 为 girl_kids 时
查 kids_type 映射表(默认内置、可由 config 覆盖,找不到保留原值),如「上衣」→「针织上衣」。
男童/成人保留原值。"""
if gender != "girl_kids" or value in (None, ""):
return value
mapping = kids_type_map if kids_type_map is not None else _KIDS_TYPE_GIRL_MAP
return mapping.get("girl_kids", {}).get(str(value), value)
def _template_gender(template_path: str) -> Optional[str]:
"""通过表头解析模版「类目」判断男装/女装:类目含「男」→male;含「女」→female;都不含→None。
用于成分值映射(女装模板下拉框是「中文+英文」格式,需字典映射;男装保留 db 值)。"""
@@ -170,7 +248,10 @@ def _size_rank(size) -> tuple:
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) # 童装年龄码
m = re.fullmatch(r"(\d+)\s*(?:M|Y|YRS|YR)", s) # 单一年龄码6Y/10Y/18M
if m:
return (3, float(m.group(1)), s)
m = re.fullmatch(r"(\d+)\s*[-/]\s*(\d+)\s*(?:M|Y|YRS|YR)?", s) # 童装年龄码范围:1-2Y/6/7Y
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
@@ -332,13 +413,20 @@ def _build_spu_row(spu: Dict[str, Any], spu_code: str,
component_map: Optional[Dict[str, Any]] = None,
season_map: Optional[Dict[str, Any]] = None,
pattern_map: Optional[Dict[str, Any]] = None,
oss_code: str = "") -> Dict[str, Any]:
target_audience_map: Optional[Dict[str, Any]] = None,
kids_type_map: Optional[Dict[str, Any]] = None,
oss_code: str = "",
model_info: Optional[Dict[str, str]] = None) -> Dict[str, Any]:
"""构造一行 SPU(固定字段:SKC货号=当前生成货号、风格=休闲、商品产地=中国大陆、产地省份=广东省;多颜色时用色值区分)。
fabric 填「面料弹性」列(fabric_headers,如 SPU商品属性-面料弹性,检测到才填 spu.fabric)。
component_1/2/3 按性别映射(gender=female 时查成分映射表 component_map,男装/None 保留原值;
material 字段不经过任何映射,原样透传)。
season 按性别映射(gender=female 时查季节映射表 season_map,如「四季」→「ALL/全球/所有」)。
pattern 按性别映射(gender=female 时空值查图案映射表 pattern_map,如空值→「卡通」)。"""
pattern 按性别映射(gender=female 时空值查图案映射表 pattern_map,如空值→「卡通」)。
target_audience 按童装性别映射(gender=boy_kids/girl_kids 时查适用人群映射表 target_audience_map
如「男童」→「女童」)。
kids_type 按童装性别映射(gender=girl_kids 时查 kids_type 映射表 kids_type_map,如「上衣」→「针织上衣」;
男童保留原值)。"""
row: Dict[str, Any] = {
"基础信息-商品层级": "spu",
"SKC货号": oss_code or spu_code, # SKC货号 = 当前生成货号 oss_code(无则回退 spu_code
@@ -357,12 +445,30 @@ def _build_spu_row(spu: Dict[str, Any], spu_code: str,
v = _map_season(v, gender, season_map)
elif dbk == "pattern":
v = _map_pattern(v, gender, pattern_map)
elif dbk == "target_audience":
v = _map_target_audience(v, gender, target_audience_map)
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
if model_info:
if model_info.get("model"):
row["试穿模特"] = model_info["model"]
if model_info.get("size"):
row["试穿尺码"] = model_info["size"]
if model_info.get("feel"):
row["试穿感受"] = model_info["feel"]
# 童装(男童/女童):kids_type → SPU商品属性-类型(女童查 kids_type 映射表,如「上衣」→「针织上衣」),
# kids_age → 适用年龄段(仅童装模板填)
if gender in ("boy_kids", "girl_kids"):
for dbk, header in (("kids_type", "SPU商品属性-类型"), ("kids_age", "适用年龄段")):
v = spu.get(dbk)
if dbk == "kids_type":
v = _map_kids_type(v, gender, kids_type_map)
if v not in (None, ""):
row[header] = v
return row
@@ -416,6 +522,81 @@ def _read_range_values(ws, ref: str) -> Optional[set]:
return set(opts) if opts else None
def _resolve_named_range(router, values) -> Optional[set]:
"""级联下拉:INDIRECT 引用的单元格值可能是名称引用(如「欧美尺码常规」)。
从 defined names 解析该名称对应的区域(可能跨表,如 ProductSizeSpec!$B$6:$I$6),
读取区域内容作为真实选项(如 S/M/L/XL/XXL/XXXL)。解析不到返回 None。"""
wb = router.ws.parent
resolved = set()
for v in values:
name = str(v).strip()
if name not in wb.defined_names:
continue
ref = wb.defined_names[name].value # 如 ProductSizeSpec!$B$6:$I$6
sheet_name, _, cell_ref = ref.partition("!")
if not cell_ref:
continue
ws = wb[sheet_name] if sheet_name in wb.sheetnames else router.ws
opts = _read_range_values(ws, cell_ref)
if opts:
resolved |= opts
return resolved if resolved else None
def _read_range_values_any(router, ref: str) -> Optional[set]:
"""读取单元格/区域内的非空值集合,支持跨表(Sheet!$A$1:$A$5);解析失败返回 None。"""
sheet_name, _, cell_ref = ref.partition("!")
if cell_ref:
ws = router.ws.parent[sheet_name] if sheet_name in router.ws.parent.sheetnames else router.ws
else:
ws = router.ws
cell_ref = ref
return _read_range_values(ws, cell_ref)
def _read_dropdown_options(router, col: Optional[int]) -> Optional[set]:
"""读取指定列(col)的 dataValidation 下拉选项集合;无下拉/非 list/无值 返回 None。
支持内联列表("S,M,L")、引用区域(=Sheet!$A$1:$A$5)、跨表引用(Sheet!$A$1:$A$5)、
INDIRECT 动态引用(含名称引用级联)。"""
import re
from openpyxl.utils import get_column_letter
if col is None:
return None
lf = get_column_letter(col)
dvs = getattr(router.ws, "data_validations", None)
if dvs is None:
return None
for dv in dvs.dataValidation:
if getattr(dv, "type", "") != "list":
continue
if lf not in str(getattr(dv, "sqref", "")):
continue
f = (getattr(dv, "formula1", "") or "").strip()
if not f:
continue
# 动态引用 INDIRECT(...):尝试解析引用的单元格/区域内容;无法解析 → None
m = re.match(r"^=?INDIRECT\((.+)\)$", f, re.IGNORECASE)
if m:
opts = _read_range_values(router.ws, m.group(1).strip())
if opts:
named = _resolve_named_range(router, opts)
if named:
return named
return opts if opts else None
# 引用区域(可能跨表:Sheet!$A$1:$A$5 或 =Sheet!$A$1:$A$5
ref = f[1:] if f.startswith("=") else f
if ":" in ref:
opts = _read_range_values_any(router, ref)
return opts if opts else None
# 内联列表("S,M,L" / "S;M;L" / "SML"
sep = "" if "" in f else (";" if ";" in f else ",")
opts = [x.strip().strip('"') for x in f.split(sep) if x.strip()]
return set(opts) if opts else None
return None
def _read_size_options(router) -> Optional[set]:
"""读取「尺码」列的 dataValidation 下拉选项集合;无下拉/非 list/无值 返回 None。
@@ -443,6 +624,12 @@ def _read_size_options(router) -> Optional[set]:
m = re.match(r"^=?INDIRECT\((.+)\)$", f, re.IGNORECASE)
if m:
opts = _read_range_values(router.ws, m.group(1).strip())
if opts:
# 级联下拉:INDIRECT 引用的单元格值可能是名称引用(如「欧美尺码常规」),
# 从 defined names 解析对应命名区域(跨表)得到真实尺码选项
named = _resolve_named_range(router, opts)
if named:
return named
return opts if opts else None
# 引用区域(=Sheet!$A$1:$A$5 或 Sheet!$A$1:$A$5
ref = f[1:] if f.startswith("=") else f
@@ -461,6 +648,45 @@ def _find_sku_category_headers(router) -> List[str]:
return [str(k) for k in router.column_map if "SKU分类" in str(k)]
def _read_model_info(router, sku_sizes: Optional[set]) -> Dict[str, str]:
"""读取模板「模特信息」三属性(试穿模特/试穿尺码/试穿感受)的 SPU 行填写值。
优先用模板数据区已有值(校验通过才用):
- 试穿模特:值须在下拉框选项内(如 ModalConfig 表)
- 试穿感受:值须在 [偏小/合身/偏大] 内
- 试穿尺码:值须是商品有效尺码(sku 尺码之一)
模板无值或校验失败 → 默认值:试穿模特=下拉框第一个、试穿感受=合身、试穿尺码=L。"""
col_model = router.column_map.get("试穿模特") or router.column_map.get("模特信息-试穿模特")
col_size = router.column_map.get("试穿尺码") or router.column_map.get("模特信息-试穿尺码")
col_feel = router.column_map.get("试穿感受") or router.column_map.get("模特信息-试穿感受")
model_opts = _read_dropdown_options(router, col_model) if col_model else None
feel_opts = _read_dropdown_options(router, col_feel) if col_feel else None
# 模板数据区(第 6 行起)已有值
tpl_model = tpl_size = tpl_feel = None
ws = router.ws
for r in range(6, min(ws.max_row, 30) + 1):
if col_model and tpl_model is None and ws.cell(r, col_model).value not in (None, ""):
tpl_model = str(ws.cell(r, col_model).value).strip()
if col_size and tpl_size is None and ws.cell(r, col_size).value not in (None, ""):
tpl_size = str(ws.cell(r, col_size).value).strip()
if col_feel and tpl_feel is None and ws.cell(r, col_feel).value not in (None, ""):
tpl_feel = str(ws.cell(r, col_feel).value).strip()
# 试穿模特:模板值在下拉框内才用,否则下拉框第一个
if tpl_model and model_opts and tpl_model in model_opts:
model = tpl_model
elif model_opts:
model = sorted(model_opts)[0]
else:
model = tpl_model or ""
# 试穿感受:模板值在 [偏小/合身/偏大] 内才用,否则「合身」
feel = tpl_feel if tpl_feel and feel_opts and tpl_feel in feel_opts else "合身"
# 试穿尺码:模板值是有效尺码才用,否则「L」
size = tpl_size if tpl_size and (not sku_sizes or tpl_size in sku_sizes) else "L"
return {"model": model, "size": size, "feel": feel}
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)]
@@ -698,6 +924,8 @@ def _insert_product_block(
component_map: Optional[Dict[str, Any]] = None,
season_map: Optional[Dict[str, Any]] = None,
pattern_map: Optional[Dict[str, Any]] = None,
target_audience_map: Optional[Dict[str, Any]] = None,
kids_type_map: Optional[Dict[str, Any]] = None,
suggested_price_ratio: float = 0.0,
suggested_price_headers: Optional[List[str]] = None,
suggested_unit_headers: Optional[List[str]] = None,
@@ -723,12 +951,14 @@ def _insert_product_block(
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))
skus_by_color.append((sc, sorted(skus, key=lambda sk: _size_rank(sk.get("size")))))
images = [str(i) for i in (images or []) if i]
color_col = router.resolve_col("色值(主规格)")
block_rows: List[int] = []
size_options = _read_size_options(router) # 尺码下拉框选项(sku.size 不在其中则跳过该 SKU 行)
sku_sizes = {sk.get("size") for _, skus in skus_by_color for sk in skus if sk.get("size")}
model_info = _read_model_info(router, sku_sizes) # 模特信息三属性(试穿模特/试穿尺码/试穿感受)
if spu_per_color:
# 单 SPU 多色:1 个 SPU 行(无色值,SPU 级信息由 _fill_design_fields 填充)
@@ -737,7 +967,9 @@ def _insert_product_block(
_build_spu_row(spu, spu_code, fabric_headers=fabric_headers,
gender=gender, component_map=component_map,
season_map=season_map, pattern_map=pattern_map,
oss_code=oss_code),
target_audience_map=target_audience_map,
kids_type_map=kids_type_map,
oss_code=oss_code, model_info=model_info),
match="exact",
))
for ci, (sc, skus) in enumerate(skus_by_color):
@@ -778,7 +1010,9 @@ def _insert_product_block(
_build_spu_row(spu, spu_code, fabric_headers=fabric_headers,
gender=gender, component_map=component_map,
season_map=season_map, pattern_map=pattern_map,
oss_code=oss_code), match="exact"))
target_audience_map=target_audience_map,
kids_type_map=kids_type_map,
oss_code=oss_code, model_info=model_info), match="exact"))
multi_variant = len(skus_by_color) > 1
for ci, (sc, skus) in enumerate(skus_by_color):
first = skus[0]
@@ -841,6 +1075,8 @@ def export_product(
component_map: Optional[Dict[str, Any]] = None,
season_map: Optional[Dict[str, Any]] = None,
pattern_map: Optional[Dict[str, Any]] = None,
target_audience_map: Optional[Dict[str, Any]] = None,
kids_type_map: Optional[Dict[str, Any]] = None,
) -> Path:
"""生成商品上传"已填写"模板(支持多产品合并到同一文件)。
@@ -893,6 +1129,8 @@ def export_product(
component_map=component_map,
season_map=season_map,
pattern_map=pattern_map,
target_audience_map=target_audience_map,
kids_type_map=kids_type_map,
suggested_price_ratio=suggested_price_ratio,
suggested_price_headers=suggested_price_headers,
@@ -920,6 +1158,8 @@ def export_products(
component_map: Optional[Dict[str, Any]] = None,
season_map: Optional[Dict[str, Any]] = None,
pattern_map: Optional[Dict[str, Any]] = None,
target_audience_map: Optional[Dict[str, Any]] = None,
kids_type_map: Optional[Dict[str, Any]] = None,
) -> Path:
"""批量合并导出:所有产品一次性写入同一模板,只打开/保存一次。
@@ -967,6 +1207,8 @@ def export_products(
component_map=component_map,
season_map=season_map,
pattern_map=pattern_map,
target_audience_map=target_audience_map,
kids_type_map=kids_type_map,
suggested_price_ratio=suggested_price_ratio,
suggested_price_headers=suggested_price_headers,