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）
555 lines
22 KiB
Python
555 lines
22 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
从 inkreach PostgreSQL 同步指定国家货盘商品数据到 SQLite (spu_sku.db)
|
||
|
||
支持多国累积同步（保留已存在数据）：
|
||
- sync(plate_ids, country, clear_first)
|
||
clear_first=True -> 先 DELETE 整库再插入（重置该次目标）
|
||
clear_first=False -> 仅追加，跳过已存在的 SPU code 与 SKU 组合
|
||
|
||
国家货盘 plate_id 映射：
|
||
- 美国 : (2, 3, 4, 5) country="US"
|
||
- 墨西哥: (6,) country="MX"
|
||
- 日本 : (7,) country="JP"
|
||
- 韩国 : (8,) country="KR"
|
||
- 沙特 : (9,) country="SA"
|
||
- 巴西 : (10,) country="BR"
|
||
- 英国 : (11,) country="GB"
|
||
- 加拿大: (12,) country="CA"
|
||
- 波兰 : (13,) country="PL"
|
||
- 西班牙: (14,) country="ES"
|
||
- 德国 : (15,) country="DE"
|
||
- 澳洲 : (16,) country="AU"
|
||
- 意大利: (17,) country="IT"
|
||
|
||
SPU = categories（code 去重）
|
||
SKU = 颜色 × 尺码 粒度：
|
||
- code = colors.code 颜色映射编码（脏数据兜底为 款号-颜色的英文名，英文名取自 color_detail_images.color_name_web）
|
||
- color = colors.name 颜色名
|
||
- size = product_extra.product_size JSON 的尺码列（缺 JSON 回退 size_chart）
|
||
- 肩宽/胸围/衣长/袖长 = product_size JSON 对应列（动态识别表头）
|
||
- 最长边/次长边/最短边 = packaging_spec JSON "包装尺寸（cm）" 列 长*宽*高 拆分排序
|
||
- package_weight = packaging_spec JSON "含包装重量（g）" 列（单位: 克）
|
||
- price = 该品类 prices 首条价格（颜色尺码粒度无独立价格）
|
||
- img_url_2~5 = 该品类首颜色 web_sku 的图（seq 2~5）
|
||
"""
|
||
import sqlite3
|
||
import psycopg2
|
||
import re
|
||
import json
|
||
|
||
PG = dict(host="localhost", port=5432, user="postgres", password="inkreach", dbname="inkreach")
|
||
SQLITE = r"C:\Users\Admin\Desktop\test模版\design_agent\pod_trend_agent\db\spu_sku.db"
|
||
|
||
# 尺码体系按国家区分：亚洲市场(JP/KR)用亚洲尺码，其余默认 欧美尺码/欧美尺码常规
|
||
# （库内约定：除 JP/KR 外所有国家 size_group 一律"欧美尺码"，与下方默认一致）
|
||
SIZE_SYSTEM = {
|
||
"JP": ("亚洲尺码", "亚洲尺码亚洲常规"),
|
||
"KR": ("亚洲尺码", "亚洲尺码亚洲常规"),
|
||
}
|
||
|
||
|
||
def clean(v):
|
||
if v is None:
|
||
return None
|
||
s = str(v).strip()
|
||
return s if s else None
|
||
|
||
|
||
def format_pct(v):
|
||
"""成分比例转百分比形式：'100'/'100.0' -> '100%'，空值保持 None"""
|
||
s = clean(v)
|
||
if s is None:
|
||
return None
|
||
try:
|
||
num = float(s)
|
||
if num == int(num):
|
||
return f"{int(num)}%"
|
||
return f"{num}%"
|
||
except ValueError:
|
||
return s
|
||
|
||
|
||
# 中文颜色名 -> 英文（color_name_web 缺失时的兜底）
|
||
ZH_TO_EN = {
|
||
"黑": "Black", "黑色": "Black",
|
||
"白": "White", "白色": "White",
|
||
"灰": "Gray", "灰色": "Gray",
|
||
"杏": "Apricot", "杏色": "Apricot",
|
||
"咖色": "Coffee", "咖": "Coffee",
|
||
"藏青": "Navy", "藏青色": "Navy",
|
||
"蓝": "Blue", "蓝色": "Blue", "海蓝": "Blue", "海蓝色": "Ocean Blue",
|
||
"翠绿": "Green", "翠绿色": "Green", "绿": "Green", "绿色": "Green",
|
||
"紫": "Purple", "紫色": "Purple",
|
||
"玫红": "Rose", "玫红色": "Rose", "粉": "Pink", "粉色": "Pink", "粉红": "Pink",
|
||
"红": "Red", "红色": "Red",
|
||
"黄": "Yellow", "黄色": "Yellow",
|
||
"金": "Gold", "金色": "Gold",
|
||
"银": "Silver", "银色": "Silver",
|
||
"棕": "Brown", "棕色": "Brown",
|
||
"米": "Beige", "米色": "Beige", "卡其": "Khaki", "卡其色": "Khaki",
|
||
"橙": "Orange", "橙色": "Orange",
|
||
}
|
||
|
||
|
||
def clean_color_name(name):
|
||
"""清洗颜色名：
|
||
'S-3XL黑色' -> '黑色'（去开头尺码前缀）
|
||
'灰色（S~XXL）' -> '灰色'（去结尾尺码区间括号）
|
||
'（M~3XL）' -> None（纯尺码无颜色）
|
||
保留中文颜色核心部分。"""
|
||
s = clean(name)
|
||
if not s:
|
||
return None
|
||
# 开头若全是非中文（尺码前缀如 S-3XL / S（4-5）y），取其后中文部分
|
||
m = re.match(r"^[^一-鿿]+([一-鿿].*)$", s)
|
||
if m:
|
||
s = m.group(1).strip()
|
||
# 去掉结尾的尺码区间括号，如 （S~XXL）/（M~5XL）/（XS~3XL）
|
||
s = re.sub(r"[（(][^一-鿿]*[）)]$", "", s).strip()
|
||
# 去掉残留的纯尺码/空白字符
|
||
s = s.strip(" ()（）~ ")
|
||
return s if s else None
|
||
|
||
|
||
def is_clean_code(ccode, catcode):
|
||
"""判断颜色 code 是否规范：必须以 品类code- 开头，
|
||
且后缀为纯字母数字（如 BL01 / ESPRESSO），不含中文、~ 或括号。
|
||
纯尺码区间（如 JPHM009-（S~XXXL））与多色拼接脏数据会被判为非规范。"""
|
||
if not ccode or not catcode:
|
||
return False
|
||
prefix = catcode + "-"
|
||
if not ccode.startswith(prefix):
|
||
return False
|
||
suffix = ccode[len(prefix):]
|
||
if not suffix or re.search(r"[一-鿿~（(）)]", suffix):
|
||
return False
|
||
return True
|
||
|
||
|
||
def first_color_en(name):
|
||
"""从可能含多色的脏名称中取首个可识别颜色的英文名，否则原样返回。"""
|
||
if not name:
|
||
return None
|
||
for zh, en in ZH_TO_EN.items():
|
||
if zh in name:
|
||
return en
|
||
return name
|
||
|
||
|
||
def extract_weight(name):
|
||
if not name:
|
||
return None, None
|
||
m = re.search(r"(\d+(?:\.\d+)?)\s*(?:G|g|克)", name)
|
||
if m:
|
||
return m.group(1), "G"
|
||
return None, None
|
||
|
||
|
||
def parse_size_json(raw):
|
||
"""product_size JSON -> [{size, shoulder, bust, length, sleeve}...]"""
|
||
if not raw:
|
||
return []
|
||
try:
|
||
data = json.loads(raw)
|
||
except Exception:
|
||
return []
|
||
if not isinstance(data, list) or len(data) < 2:
|
||
return []
|
||
header = [c.get("content", "") for c in data[0]]
|
||
size_i = next((i for i, h in enumerate(header) if "尺码" in h), None)
|
||
sh_i = next((i for i, h in enumerate(header) if "肩宽" in h), None)
|
||
bu_i = next((i for i, h in enumerate(header) if "胸围" in h), None)
|
||
le_i = next((i for i, h in enumerate(header) if "衣长" in h), None)
|
||
sl_i = next((i for i, h in enumerate(header) if "袖长" in h), None)
|
||
if size_i is None:
|
||
return []
|
||
rows = []
|
||
for r in data[1:]:
|
||
if not isinstance(r, list):
|
||
continue
|
||
def cell(i):
|
||
if i is None or i >= len(r):
|
||
return None
|
||
return clean(r[i].get("content", "") if isinstance(r[i], dict) else r[i])
|
||
size = cell(size_i)
|
||
if not size or size == "尺码":
|
||
continue
|
||
rows.append(dict(size=size, shoulder=cell(sh_i), bust=cell(bu_i),
|
||
length=cell(le_i), sleeve=cell(sl_i)))
|
||
return rows
|
||
|
||
|
||
def parse_pkg_json(raw):
|
||
"""packaging_spec JSON -> {size: (longest, second, shortest)} 按 cm 列拆分排序"""
|
||
if not raw:
|
||
return {}
|
||
try:
|
||
data = json.loads(raw)
|
||
except Exception:
|
||
return {}
|
||
if not isinstance(data, list) or len(data) < 2:
|
||
return {}
|
||
header = data[0]
|
||
size_i = next((i for i, c in enumerate(header) if "尺码" in c.get("content", "")), None)
|
||
cm_i = next((i for i, c in enumerate(header)
|
||
if "包装尺寸" in c.get("content", "") and "cm" in c.get("content", "")), None)
|
||
if size_i is None or cm_i is None:
|
||
return {}
|
||
out = {}
|
||
for r in data[1:]:
|
||
if not isinstance(r, list) or cm_i >= len(r):
|
||
continue
|
||
size = clean(r[size_i].get("content", "")) if isinstance(r[size_i], dict) else clean(r[size_i])
|
||
val = clean(r[cm_i].get("content", "")) if isinstance(r[cm_i], dict) else clean(r[cm_i])
|
||
if not size or not val:
|
||
continue
|
||
parts = re.findall(r"\d+(?:\.\d+)?", val)
|
||
if len(parts) >= 3:
|
||
nums = sorted((float(p) for p in parts[:3]), reverse=True)
|
||
out[size] = (nums[0], nums[1], nums[2])
|
||
return out
|
||
|
||
|
||
def parse_weight_json(raw):
|
||
"""packaging_spec JSON -> {size: 含包装重量(g)}，取"含包装重量（g）"列"""
|
||
if not raw:
|
||
return {}
|
||
try:
|
||
data = json.loads(raw)
|
||
except Exception:
|
||
return {}
|
||
if not isinstance(data, list) or len(data) < 2:
|
||
return {}
|
||
header = data[0]
|
||
size_i = next((i for i, c in enumerate(header) if "尺码" in c.get("content", "")), None)
|
||
w_i = next((i for i, c in enumerate(header)
|
||
if "重量" in c.get("content", "") and "（g）" in c.get("content", "")), None)
|
||
if size_i is None or w_i is None:
|
||
return {}
|
||
out = {}
|
||
for r in data[1:]:
|
||
if not isinstance(r, list) or w_i >= len(r):
|
||
continue
|
||
size = clean(r[size_i].get("content", "")) if isinstance(r[size_i], dict) else clean(r[size_i])
|
||
val = clean(r[w_i].get("content", "")) if isinstance(r[w_i], dict) else clean(r[w_i])
|
||
if not size or not val:
|
||
continue
|
||
out[size] = val
|
||
return out
|
||
|
||
|
||
def sync(plate_ids, country, clear_first=True):
|
||
print(f"\n==== 同步 {country} (plate_ids={plate_ids}, clear_first={clear_first}) ====")
|
||
pg = psycopg2.connect(**PG)
|
||
pg.autocommit = True
|
||
pc = pg.cursor()
|
||
|
||
# ---------- SPU 数据 ----------
|
||
pc.execute("""
|
||
SELECT c.id, c.plate_id, c.code, c.name, c.sub_category, c.fabric, c.composition, p.display_name
|
||
FROM categories c
|
||
LEFT JOIN plates p ON c.plate_id = p.id
|
||
WHERE c.plate_id IN %s AND c.code IS NOT NULL AND c.code <> ''
|
||
ORDER BY c.plate_id, c.id
|
||
""", (plate_ids,))
|
||
cat_rows = pc.fetchall()
|
||
|
||
spu_map = {}
|
||
for cid, plate_id, code, name, sub, fabric, comp, display in cat_rows:
|
||
if code in spu_map:
|
||
prev = spu_map[code]
|
||
prev_score = (0 if prev["sub"] == "组合款" else 1, -prev["plate_id"], -prev["cid"])
|
||
new_score = (0 if sub == "组合款" else 1, -plate_id, -cid)
|
||
if new_score > prev_score:
|
||
spu_map[code] = dict(cid=cid, plate_id=plate_id, name=name, sub=sub,
|
||
fabric=fabric, comp_raw=comp, display=display)
|
||
else:
|
||
spu_map[code] = dict(cid=cid, plate_id=plate_id, name=name, sub=sub,
|
||
fabric=fabric, comp_raw=comp, display=display)
|
||
|
||
pc.execute("""
|
||
SELECT category_id, fabric, comp1, comp1_pct, comp2, comp2_pct, comp3, comp3_pct
|
||
FROM composition
|
||
""")
|
||
comp_by_cat = {}
|
||
for cid, fabric, c1, p1, c2, p2, c3, p3 in pc.fetchall():
|
||
comp_by_cat[cid] = dict(c1=c1, p1=p1, c2=c2, p2=p2, c3=c3, p3=p3)
|
||
|
||
pc.execute("""
|
||
SELECT code, english_name, washing_instructions, design_explanation, texture,
|
||
product_size, packaging_spec
|
||
FROM product_extra
|
||
""")
|
||
extra_by_code = {}
|
||
for code, en, wash, design, texture, psize, pkg in pc.fetchall():
|
||
extra_by_code[code] = dict(en=en, wash=wash, design=design, texture=texture,
|
||
psize=psize, pkg=pkg)
|
||
|
||
# ---------- SKU 数据 ----------
|
||
# 颜色
|
||
pc.execute("""
|
||
SELECT id, category_id, name, code, size_range
|
||
FROM colors
|
||
WHERE category_id IN (SELECT id FROM categories WHERE plate_id IN %s)
|
||
ORDER BY category_id, seq
|
||
""", (plate_ids,))
|
||
colors_by_cat = {}
|
||
for col_id, cid, name, code, sr in pc.fetchall():
|
||
colors_by_cat.setdefault(cid, []).append(dict(id=col_id, name=name, code=code, sr=sr))
|
||
|
||
# 颜色英文名（color_detail_images.color_name_web，按 color_id）
|
||
pc.execute("""
|
||
SELECT DISTINCT color_id, color_name_web
|
||
FROM color_detail_images
|
||
WHERE color_id IN (
|
||
SELECT id FROM colors
|
||
WHERE category_id IN (SELECT id FROM categories WHERE plate_id IN %s)
|
||
)
|
||
""", (plate_ids,))
|
||
color_en_by_id = {}
|
||
for col_id, en in pc.fetchall():
|
||
en = clean(en)
|
||
if en:
|
||
color_en_by_id[col_id] = en
|
||
|
||
# size_chart 回退
|
||
pc.execute(f"""
|
||
SELECT category_id, size, shoulder, bust, length, sleeve
|
||
FROM size_chart
|
||
WHERE category_id IN (SELECT id FROM categories WHERE plate_id IN %s)
|
||
ORDER BY category_id, seq
|
||
""", (plate_ids,))
|
||
sizechart_by_cat = {}
|
||
for cid, size, sh, bu, le, sl in pc.fetchall():
|
||
sizechart_by_cat.setdefault(cid, []).append(dict(size=size, shoulder=sh, bust=bu,
|
||
length=le, sleeve=sl))
|
||
|
||
# prices（首条价格）
|
||
pc.execute("""
|
||
SELECT code, price FROM prices
|
||
WHERE code IN (SELECT DISTINCT code FROM categories WHERE plate_id IN %s)
|
||
ORDER BY code, web_product_id
|
||
""", (plate_ids,))
|
||
price_by_code = {}
|
||
for code, price in pc.fetchall():
|
||
if code not in price_by_code:
|
||
price_by_code[code] = price
|
||
|
||
# 图片（按 code，取首个 web_sku 的图）
|
||
pc.execute("""
|
||
SELECT cd.code, cd.web_sku, cd.seq, cd.image_url
|
||
FROM color_detail_images cd
|
||
WHERE cd.code IN (SELECT DISTINCT code FROM categories WHERE plate_id IN %s)
|
||
ORDER BY cd.code, cd.web_sku, cd.seq
|
||
""", (plate_ids,))
|
||
imgs_by_code = {}
|
||
for code, wsku, seq, url in pc.fetchall():
|
||
imgs_by_code.setdefault(code, []).append((wsku, seq, url))
|
||
|
||
pg.close()
|
||
|
||
# ---------- 写入 SQLite ----------
|
||
db = sqlite3.connect(SQLITE)
|
||
cur = db.cursor()
|
||
|
||
if clear_first:
|
||
cur.execute("DELETE FROM SKU")
|
||
cur.execute("DELETE FROM SPU")
|
||
db.commit()
|
||
print("[clear] 已清空 SPU/SKU 旧数据")
|
||
else:
|
||
print("[append] 保留现有数据，仅追加新国家")
|
||
|
||
# append 模式：收集已存在的 SPU code 与 SKU 组合，避免重复插入
|
||
existing_spu_codes = set()
|
||
existing_sku_keys = set()
|
||
if not clear_first:
|
||
existing_spu_codes = {r[0] for r in cur.execute("SELECT code FROM SPU")}
|
||
existing_sku_keys = {(r[0], r[1], r[2])
|
||
for r in cur.execute("SELECT spu_id, code, size FROM SKU")}
|
||
print(f"[append] 已有 SPU {len(existing_spu_codes)} 条, SKU {len(existing_sku_keys)} 条")
|
||
|
||
# SPU
|
||
spu_id_by_code = {}
|
||
n_spu = 0
|
||
n_spu_skip = 0
|
||
for code, s in sorted(spu_map.items()):
|
||
# append 模式：若该 SPU code 已存在则跳过（保留首次写入的数据）
|
||
if code in existing_spu_codes:
|
||
spu_id = cur.execute("SELECT id FROM SPU WHERE code=?", (code,)).fetchone()[0]
|
||
spu_id_by_code[code] = spu_id
|
||
n_spu_skip += 1
|
||
continue
|
||
comp = comp_by_cat.get(s["cid"], {}) or {}
|
||
weight, wunit = extract_weight(s["name"])
|
||
# 业务模板固定值（按用户要求，不随 PG 数据变化）
|
||
fixed = dict(
|
||
pattern="印花", details="无", collar_style="圆领", style="休闲",
|
||
care_Instructions="数码印花类可机洗且不可干洗", fabric="微弹",
|
||
target_audience="成人", season="四季", is_transparent="否", layout="常规",
|
||
weaving_method="针织(含钩织、毛织面料)", printing_type="定位印花",
|
||
fabric_texture_1="光面", fabric_weight_unit_1="g/㎡",
|
||
lining_texture="无里料/无内衬", mark="1",
|
||
)
|
||
cur.execute("""
|
||
INSERT INTO SPU (
|
||
code, material, component_1, component_2, component_3,
|
||
component_proportion_1, component_proportion_2, component_proportion_3,
|
||
pattern, details, collar_style, style, care_Instructions, fabric,
|
||
target_audience, season, is_transparent, layout, weaving_method,
|
||
printing_type, fabric_texture_1, fabric_weight_1, fabric_weight_unit_1,
|
||
lining_texture, country, mark
|
||
) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
|
||
""", (
|
||
code, clean(s["fabric"]),
|
||
clean(comp.get("c1")), clean(comp.get("c2")), clean(comp.get("c3")),
|
||
format_pct(comp.get("p1")), format_pct(comp.get("p2")), format_pct(comp.get("p3")),
|
||
fixed["pattern"], fixed["details"], fixed["collar_style"], fixed["style"],
|
||
fixed["care_Instructions"], fixed["fabric"], fixed["target_audience"],
|
||
fixed["season"], fixed["is_transparent"], fixed["layout"], fixed["weaving_method"],
|
||
fixed["printing_type"], fixed["fabric_texture_1"], weight, fixed["fabric_weight_unit_1"],
|
||
fixed["lining_texture"],
|
||
country, fixed["mark"],
|
||
))
|
||
spu_id_by_code[code] = cur.lastrowid
|
||
n_spu += 1
|
||
|
||
# SKU（颜色 × 尺码）
|
||
n_sku = 0
|
||
n_sku_skip = 0
|
||
detail = dict(no_color=0, no_size=0, no_pkg=0, no_img=0, color_sku=0)
|
||
for code, s in sorted(spu_map.items()):
|
||
spu_id = spu_id_by_code[code]
|
||
cat_id = s["cid"]
|
||
extra = extra_by_code.get(code, {})
|
||
|
||
# 颜色列表
|
||
color_list = colors_by_cat.get(cat_id, [])
|
||
if not color_list:
|
||
color_list = [dict(name=None, code=code, sr=None)] # 兜底
|
||
detail["no_color"] += 1
|
||
# 品类尺码范围（取首个非空）
|
||
sr = next((c["sr"] for c in color_list if c["sr"]), None)
|
||
if sr is None:
|
||
sr = None
|
||
|
||
# 尺码表：product_size JSON 优先，回退 size_chart
|
||
size_rows = parse_size_json(extra.get("psize"))
|
||
if not size_rows:
|
||
size_rows = sizechart_by_cat.get(cat_id, [])
|
||
if not size_rows:
|
||
detail["no_size"] += 1
|
||
pkg_map = parse_pkg_json(extra.get("pkg"))
|
||
weight_map = parse_weight_json(extra.get("pkg"))
|
||
# 归一化兜底：尺码列可能带英寸后缀（如 '30*40/76.2*101.6' vs '30*40'）
|
||
pkg_map_norm = {k.split("/")[0]: v for k, v in pkg_map.items()}
|
||
weight_map_norm = {k.split("/")[0]: v for k, v in weight_map.items()}
|
||
|
||
def lookup_pkg(size):
|
||
v = pkg_map.get(size) or pkg_map_norm.get(size.split("/")[0])
|
||
if v is None and ("one" in size.lower() or "均码" in size):
|
||
v = pkg_map.get("均码") or pkg_map.get("Onesize") or pkg_map.get("OneSize")
|
||
return v
|
||
|
||
def lookup_weight(size):
|
||
v = weight_map.get(size) or weight_map_norm.get(size.split("/")[0])
|
||
if v is None and ("one" in size.lower() or "均码" in size):
|
||
v = weight_map.get("均码") or weight_map.get("Onesize") or weight_map.get("OneSize")
|
||
return v
|
||
|
||
# 图片
|
||
imgs = imgs_by_code.get(code, [])
|
||
if imgs:
|
||
first_sku = imgs[0][0]
|
||
urls = [u for ws, _, u in imgs if ws == first_sku]
|
||
else:
|
||
urls = []
|
||
if not urls:
|
||
detail["no_img"] += 1
|
||
img2 = urls[1] if len(urls) > 1 else None
|
||
img3 = urls[2] if len(urls) > 2 else None
|
||
img4 = urls[3] if len(urls) > 3 else None
|
||
img5 = urls[4] if len(urls) > 4 else None
|
||
|
||
price = price_by_code.get(code)
|
||
# 尺码体系按国家区分：亚洲市场(JP/KR)=亚洲尺码，其余默认 欧美尺码/欧美尺码常规
|
||
sgroup, stype = SIZE_SYSTEM.get(country, ("欧美尺码", "欧美尺码常规"))
|
||
|
||
for col in color_list:
|
||
raw_code = col["code"]
|
||
raw_name = col["name"]
|
||
col_id = col["id"]
|
||
cname = clean_color_name(raw_name) # 清洗颜色名（去尺码前缀/区间）
|
||
# 颜色英文名：优先 PG color_name_web，否则用中文名查兜底字典/取首个颜色
|
||
en = None
|
||
if cname:
|
||
en = color_en_by_id.get(col_id) or first_color_en(cname)
|
||
# 命名规则：
|
||
# 规范 code（品类code-英文/缩写，无中文，如 JPHM009-BL01 / JPTM007-ESPRESSO）
|
||
# -> 直接使用
|
||
# 缺失/脏 code（含中文或尺码区间）-> 款号(品类code)-颜色的英文名
|
||
if not is_clean_code(raw_code, code):
|
||
ccode = f"{code}-{en}" if en else code
|
||
else:
|
||
ccode = raw_code
|
||
for srow in size_rows:
|
||
size = clean(srow["size"])
|
||
if not size:
|
||
continue
|
||
# append 模式：跳过已存在的 SKU 组合
|
||
if (spu_id, ccode, size) in existing_sku_keys:
|
||
n_sku_skip += 1
|
||
continue
|
||
pkg = lookup_pkg(size)
|
||
pkg_weight = lookup_weight(size)
|
||
cur.execute("""
|
||
INSERT INTO SKU (
|
||
spu_id, code, price, color, size, size_group, size_type,
|
||
shoulder_width, bust, clothing_length, sleeve_length,
|
||
longest_side, secondary_long_side, shortest_side,
|
||
package_weight, img_url_2, img_url_3, img_url_4, img_url_5
|
||
) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
|
||
""", (
|
||
spu_id, ccode, price, cname, size, sgroup, stype,
|
||
clean(srow.get("shoulder")), clean(srow.get("bust")),
|
||
clean(srow.get("length")), clean(srow.get("sleeve")),
|
||
str(pkg[0]) if pkg else None,
|
||
str(pkg[1]) if pkg else None,
|
||
str(pkg[2]) if pkg else None,
|
||
pkg_weight,
|
||
img2, img3, img4, img5,
|
||
))
|
||
n_sku += 1
|
||
|
||
db.commit()
|
||
|
||
# ---------- 验证 ----------
|
||
print(f" 本次新增 SPU: {n_spu} (跳过已存在 {n_spu_skip})")
|
||
print(f" 本次新增 SKU: {n_sku} (跳过已存在 {n_sku_skip}) (颜色×尺码展开)")
|
||
print(f" 无颜色品类: {detail['no_color']}, 无尺码品类: {detail['no_size']}, 无图品类: {detail['no_img']}")
|
||
print(f" 库内 SPU 总数: {cur.execute('SELECT COUNT(*) FROM SPU').fetchone()[0]}")
|
||
print(f" 库内 SKU 总数: {cur.execute('SELECT COUNT(*) FROM SKU').fetchone()[0]}")
|
||
print(f" country 分布: {cur.execute('SELECT country, COUNT(*) FROM SPU GROUP BY country').fetchall()}")
|
||
db.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# 追加模式（clear_first=False）：保留现有 US/MX/JP/GB 及手动修改，仅补齐其余 9 国
|
||
# 任何国家若已存在会被跳过（SPU code / SKU 组合去重），不会覆盖手动编辑
|
||
sync((2, 3, 4, 5), "US", clear_first=False)
|
||
sync((6,), "MX", clear_first=False)
|
||
sync((7,), "JP", clear_first=False)
|
||
sync((11,), "GB", clear_first=False)
|
||
# 新增国家
|
||
sync((8,), "KR", clear_first=False)
|
||
sync((9,), "SA", clear_first=False)
|
||
sync((10,), "BR", clear_first=False)
|
||
sync((12,), "CA", clear_first=False)
|
||
sync((13,), "PL", clear_first=False)
|
||
sync((14,), "ES", clear_first=False)
|
||
sync((15,), "DE", clear_first=False)
|
||
sync((16,), "AU", clear_first=False)
|
||
sync((17,), "IT", clear_first=False)
|
||
print("\n✅ 同步完成：13 国数据已包含（US/MX/JP/GB 保留，KR/SA/BR/CA/PL/ES/DE/AU/IT 已追加）")
|