scripts/amazon_xlsx_to_shoplazza_xlsx.py

#!/usr/bin/env python3
"""
Convert Amazon-format Excel exports (with Parent/Child ASIN structure) into
Shoplazza (店匠) product import Excel format based on `docs/商品导入模板.xlsx`.
Data source:
- Directory with multiple `*.xlsx` files under `products_data/`.
- Each file contains a main sheet + "Notes" sheet.
- Column meanings (sample):
  - ASIN: variant id (sku_id)
  - 父ASIN: parent product id (spu_id)
Output:
- For each 父ASIN group:
  - If only 1 ASIN: generate one "S" row
  - Else: generate one "M" row + multiple "P" rows
Multi-variant (M/P) key point:
- Variant dimensions are parsed primarily from the `SKU` column, e.g.
  "Size: One Size | Color: Black", and mapped into 款式1/2/3.
"""
# NOTE: This file is intentionally the same implementation as
# `competitor_xlsx_to_shoplazza_xlsx.py`, but renamed to reflect the correct
# data source (Amazon-format exports). Keep the logic in sync.
import os
import re
import sys
import argparse
import random
from datetime import datetime
from collections import defaultdict, Counter
from pathlib import Path
from openpyxl import load_workbook
# Allow running as `python scripts/xxx.py` without installing as a package
sys.path.insert(0, str(Path(__file__).resolve().parent))
from shoplazza_excel_template import create_excel_from_template_fast
PREFERRED_OPTION_KEYS = [
    "Size", "Color", "Style", "Pattern", "Material", "Flavor", "Scent",
    "Pack", "Pack of", "Number of Items", "Count", "Capacity", "Length",
    "Width", "Height", "Model", "Configuration",
]
def clean_str(v):
    if v is None:
        return ""
    return str(v).strip()
def html_escape(s):
    s = clean_str(s)
    return (s.replace("&", "&amp;")
             .replace("<", "&lt;")
             .replace(">", "&gt;"))
def generate_handle(title):
    """
    Generate URL-friendly handle from title (ASCII only).
    Keep consistent with existing scripts.
    """
    handle = clean_str(title).lower()
    handle = re.sub(r"[^a-z0-9\\s-]", "", handle)
    handle = re.sub(r"[-\\s]+", "-", handle).strip("-")
    if len(handle) > 255:
        handle = handle[:255]
    return handle or "product"
def parse_date_to_template(dt_value):
    """
    Template expects: YYYY-MM-DD HH:MM:SS
    Input could be "2018-05-09" or datetime/date.
    """
    if dt_value is None or dt_value == "":
        return ""
    if isinstance(dt_value, datetime):
        return dt_value.strftime("%Y-%m-%d %H:%M:%S")
    s = clean_str(dt_value)
    for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%Y-%m-%d %H:%M:%S", "%Y/%m/%d %H:%M:%S"):
        try:
            d = datetime.strptime(s, fmt)
            return d.strftime("%Y-%m-%d %H:%M:%S")
        except Exception:
            pass
    return ""
def parse_weight(weight_conv, weight_raw):
    """
    Return (weight_value, unit) where unit in {kg, lb, g, oz}.
    Prefer '商品重量（单位换算）' like '68.04 g'.
    Fallback to '商品重量' like '0.15 pounds'.
    """
    s = clean_str(weight_conv) or clean_str(weight_raw)
    if not s:
        return ("", "")
    m = re.search(r"([0-9]+(?:\\.[0-9]+)?)\\s*([a-zA-Z]+)", s)
    if not m:
        return ("", "")
    val = float(m.group(1))
    unit = m.group(2).lower()
    if unit in ("g", "gram", "grams"):
        return (val, "g")
    if unit in ("kg", "kilogram", "kilograms"):
        return (val, "kg")
    if unit in ("lb", "lbs", "pound", "pounds"):
        return (val, "lb")
    if unit in ("oz", "ounce", "ounces"):
        return (val, "oz")
    return ("", "")
def parse_dimensions_inches(dim_raw):
    """
    Template '尺寸信息': 'L,W,H' in inches.
    Input example: '7.9 x 7.9 x 2 inches'
    """
    s = clean_str(dim_raw)
    if not s:
        return ""
    nums = re.findall(r"([0-9]+(?:\\.[0-9]+)?)", s)
    if len(nums) < 3:
        return ""
    return "{},{},{}".format(nums[0], nums[1], nums[2])
def parse_sku_options(sku_text):
    """
    Parse 'SKU' column into {key: value}.
    Example:
      'Size: One Size | Color: Black' -> {'Size':'One Size','Color':'Black'}
    """
    s = clean_str(sku_text)
    if not s:
        return {}
    parts = [p.strip() for p in s.split("|") if p.strip()]
    out = {}
    for p in parts:
        if ":" not in p:
            continue
        k, v = p.split(":", 1)
        k = clean_str(k)
        v = clean_str(v)
        if k and v:
            out[k] = v
    return out
def choose_option_keys(variant_dicts, max_keys=3):
    freq = Counter()
    for d in variant_dicts:
        for k, v in d.items():
            if v:
                freq[k] += 1
    if not freq:
        return []
    preferred_rank = {k: i for i, k in enumerate(PREFERRED_OPTION_KEYS)}
    def key_sort(k):
        return (preferred_rank.get(k, 10 ** 6), -freq[k], k.lower())
    keys = sorted(freq.keys(), key=key_sort)
    return keys[:max_keys]
def build_description_html(title, details, product_url):
    parts = []
    if title:
        parts.append("<p>{}</p>".format(html_escape(title)))
    detail_items = [x.strip() for x in clean_str(details).split("|") if x.strip()]
    if detail_items:
        li = "".join(["<li>{}</li>".format(html_escape(x)) for x in detail_items[:30]])
        parts.append("<ul>{}</ul>".format(li))
    if product_url:
        parts.append('<p>Source: <a href="{0}">{0}</a></p>'.format(html_escape(product_url)))
    return "".join(parts)
def read_amazon_rows_from_file(xlsx_path, max_rows=None):
    wb = load_workbook(xlsx_path, read_only=True, data_only=True)
    sheet_name = None
    for name in wb.sheetnames:
        if str(name).lower() == "notes":
            continue
        sheet_name = name
        break
    if sheet_name is None:
        return []
    ws = wb[sheet_name]
    # Build header index from first row
    header = next(ws.iter_rows(min_row=1, max_row=1, values_only=True))
    idx = {clean_str(v): i for i, v in enumerate(header) if v is not None and clean_str(v)}
    required = ["ASIN", "父ASIN", "商品标题", "商品主图", "SKU", "详细参数", "价格($)", "prime价格($)",
                "上架时间", "类目路径", "大类目", "小类目", "品牌", "品牌链接", "商品详情页链接",
                "商品重量（单位换算）", "商品重量", "商品尺寸"]
    for k in required:
        if k not in idx:
            raise RuntimeError("Missing column '{}' in {} sheet {}".format(k, xlsx_path, sheet_name))
    # OPT: use iter_rows(values_only=True) instead of ws.cell() per field.
    # openpyxl cell access is relatively expensive; values_only is much faster.
    pos = {k: idx[k] for k in required}  # 0-based positions in row tuple
    rows = []
    end_row = ws.max_row
    if max_rows is not None:
        end_row = min(end_row, 1 + int(max_rows))
    for tup in ws.iter_rows(min_row=2, max_row=end_row, values_only=True):
        asin = clean_str(tup[pos["ASIN"]])
        if not asin:
            continue
        parent = clean_str(tup[pos["父ASIN"]]) or asin
        rows.append({
            "ASIN": asin,
            "父ASIN": parent,
            "SKU": clean_str(tup[pos["SKU"]]),
            "详细参数": clean_str(tup[pos["详细参数"]]),
            "商品标题": clean_str(tup[pos["商品标题"]]),
            "商品主图": clean_str(tup[pos["商品主图"]]),
            "价格($)": tup[pos["价格($)"]],
            "prime价格($)": tup[pos["prime价格($)"]],
            "上架时间": clean_str(tup[pos["上架时间"]]),
            "类目路径": clean_str(tup[pos["类目路径"]]),
            "大类目": clean_str(tup[pos["大类目"]]),
            "小类目": clean_str(tup[pos["小类目"]]),
            "品牌": clean_str(tup[pos["品牌"]]),
            "品牌链接": clean_str(tup[pos["品牌链接"]]),
            "商品详情页链接": clean_str(tup[pos["商品详情页链接"]]),
            "商品重量（单位换算）": clean_str(tup[pos["商品重量（单位换算）"]]),
            "商品重量": clean_str(tup[pos["商品重量"]]),
            "商品尺寸": clean_str(tup[pos["商品尺寸"]]),
        })
    return rows
def to_price(v):
    if v is None or v == "":
        return None
    try:
        return float(v)
    except Exception:
        s = clean_str(v)
        m = re.search(r"([0-9]+(?:\\.[0-9]+)?)", s)
        return float(m.group(1)) if m else None
def build_common_fields(base_row, spu_id):
    title = base_row.get("商品标题") or "Product"
    brand = base_row.get("品牌") or ""
    big_cat = base_row.get("大类目") or ""
    small_cat = base_row.get("小类目") or ""
    cat_path = base_row.get("类目路径") or ""
    handle = generate_handle(title)
    if handle and not handle.startswith("products/"):
        handle = "products/{}".format(handle)
    seo_title = title
    seo_desc_parts = [x for x in [brand, title, big_cat] if x]
    seo_description = " ".join(seo_desc_parts)[:5000]
    seo_keywords = ",".join([x for x in [title, brand, big_cat, small_cat] if x])[:5000]
    tags = ",".join([x for x in [brand, big_cat, small_cat] if x])
    created_at = parse_date_to_template(base_row.get("上架时间"))
    description = build_description_html(title, base_row.get("详细参数"), base_row.get("商品详情页链接"))
    inventory_qty = 100
    weight_val, weight_unit = parse_weight(base_row.get("商品重量（单位换算）"), base_row.get("商品重量"))
    size_info = parse_dimensions_inches(base_row.get("商品尺寸"))
    album = big_cat or (cat_path.split(":")[0] if cat_path else "")
    return {
        "商品ID": "",
        "创建时间": created_at,
        "商品标题*": title[:255],
        "商品副标题": "{} {}".format(brand, big_cat).strip()[:600],
        "商品描述": description,
        "SEO标题": seo_title[:5000],
        "SEO描述": seo_description,
        "SEO URL Handle": handle,
        "SEO URL 重定向": "N",
        "SEO关键词": seo_keywords,
        "商品上架": "Y",
        "需要物流": "Y",
        "商品收税": "N",
        "商品spu": spu_id[:100],
        "启用虚拟销量": "N",
        "虚拟销量值": "",
        "跟踪库存": "Y",
        "库存规则*": "1",
        "专辑名称": album,
        "标签": tags,
        "供应商名称": "Amazon",
        "供应商URL": base_row.get("商品详情页链接") or base_row.get("品牌链接") or "",
        "商品重量": weight_val if weight_val != "" else "",
        "重量单位": weight_unit,
        "商品库存": inventory_qty,
        "尺寸信息": size_info,
        "原产地国别": "",
        "HS（协调制度）代码": "",
        "商品备注": "ASIN:{}; ParentASIN:{}; CategoryPath:{}".format(
            base_row.get("ASIN", ""), spu_id, (cat_path[:200] if cat_path else "")
        )[:500],
        "款式备注": "",
    }
def build_s_row(base_row):
    spu_id = base_row.get("父ASIN") or base_row.get("ASIN")
    common = build_common_fields(base_row, spu_id=spu_id)
    price = to_price(base_row.get("prime价格($)")) or to_price(base_row.get("价格($)")) or 9.99
    image = base_row.get("商品主图") or ""
    row = {}
    row.update(common)
    row.update({
        "商品属性*": "S",
        "款式1": "",
        "款式2": "",
        "款式3": "",
        "商品售价*": price,
        "商品原价": price,
        "成本价": "",
        "商品SKU": base_row.get("ASIN") or "",
        "商品条形码": "",
        "商品图片*": image,
        "商品主图": image,
    })
    return row
def build_m_p_rows(variant_rows):
    base = variant_rows[0]
    spu_id = base.get("父ASIN") or base.get("ASIN")
    common = build_common_fields(base, spu_id=spu_id)
    option_dicts = [parse_sku_options(v.get("SKU")) for v in variant_rows]
    option_keys = choose_option_keys(option_dicts, max_keys=3) or ["Variant"]
    m = {}
    m.update(common)
    m.update({
        "商品属性*": "M",
        "款式1": option_keys[0] if len(option_keys) > 0 else "",
        "款式2": option_keys[1] if len(option_keys) > 1 else "",
        "款式3": option_keys[2] if len(option_keys) > 2 else "",
        "商品售价*": "",
        "商品原价": "",
        "成本价": "",
        "商品SKU": "",
        "商品条形码": "",
        "商品图片*": base.get("商品主图") or "",
        "商品主图": base.get("商品主图") or "",
    })
    m["商品重量"] = ""
    m["重量单位"] = ""
    m["商品库存"] = ""
    m["尺寸信息"] = ""
    rows = [m]
    for v in variant_rows:
        v_common = build_common_fields(v, spu_id=spu_id)
        v_common.update({
            "商品副标题": "",
            "商品描述": "",
            "SEO标题": "",
            "SEO描述": "",
            "SEO URL Handle": "",
            "SEO URL 重定向": "",
            "SEO关键词": "",
            "专辑名称": "",
            "标签": "",
            "供应商名称": "",
            "供应商URL": "",
            "商品备注": "",
        })
        opt = parse_sku_options(v.get("SKU"))
        opt_vals = [v.get("ASIN")] if option_keys == ["Variant"] else [opt.get(k, "") for k in option_keys]
        price = to_price(v.get("prime价格($)")) or to_price(v.get("价格($)")) or 9.99
        image = v.get("商品主图") or ""
        p = {}
        p.update(v_common)
        p.update({
            "商品属性*": "P",
            "款式1": opt_vals[0] if len(opt_vals) > 0 else "",
            "款式2": opt_vals[1] if len(opt_vals) > 1 else "",
            "款式3": opt_vals[2] if len(opt_vals) > 2 else "",
            "商品售价*": price,
            "商品原价": price,
            "成本价": "",
            "商品SKU": v.get("ASIN") or "",
            "商品条形码": "",
            "商品图片*": image,
            "商品主图": "",
        })
        rows.append(p)
    return rows
def main():
    parser = argparse.ArgumentParser(description="Convert Amazon-format xlsx files to Shoplazza import xlsx")
    parser.add_argument("--input-dir", default="data/mai_jia_jing_ling/products_data", help="Directory containing Amazon-format xlsx files")
    parser.add_argument("--template", default="docs/商品导入模板.xlsx", help="Shoplazza import template xlsx")
    parser.add_argument("--output", default="amazon_shoplazza_import.xlsx", help="Output xlsx file path (or prefix when split)")
    parser.add_argument("--max-files", type=int, default=None, help="Limit number of xlsx files to read (for testing)")
    parser.add_argument("--max-rows-per-output", type=int, default=40000, help="Max total Excel rows per output file (including模板头部行，默认40000)")
    parser.add_argument("--max-products", type=int, default=None, help="Limit number of SPU groups to output (for testing)")
    args = parser.parse_args()
    if not os.path.isdir(args.input_dir):
        raise RuntimeError("input-dir not found: {}".format(args.input_dir))
    if not os.path.exists(args.template):
        raise RuntimeError("template not found: {}".format(args.template))
    files = [os.path.join(args.input_dir, f) for f in os.listdir(args.input_dir) if f.lower().endswith(".xlsx")]
    files.sort()
    if args.max_files is not None:
        files = files[: int(args.max_files)]
    print("Reading Amazon-format files: {} (from {})".format(len(files), args.input_dir), flush=True)
    groups = defaultdict(list)
    seen_asin = set()
    for fp in files:
        print("  - loading: {}".format(fp), flush=True)
        try:
            rows = read_amazon_rows_from_file(fp)
        except Exception as e:
            print("WARN: failed to read {}: {}".format(fp, e))
            continue
        print("    loaded rows: {}".format(len(rows)), flush=True)
        for r in rows:
            asin = r.get("ASIN")
            if asin in seen_asin:
                continue
            seen_asin.add(asin)
            spu_id = r.get("父ASIN") or asin
            groups[spu_id].append(r)
    print("Collected variants: {}, SPU groups: {}".format(len(seen_asin), len(groups)), flush=True)
    # 先按 SPU 构造每个组的行，方便做“按最大行数拆分但不拆组”
    group_rows_list = []  # List[List[dict]]
    spu_count = 0
    next_product_id = 1  # 用于填充商品ID，全局自增
    # 将SPU顺序打乱，避免过于依赖输入文件的顺序
    spu_items = list(groups.items())
    random.shuffle(spu_items)
    for spu_id, variants in spu_items:
        if not variants:
            continue
        # 确保父ASIN对应的变体在列表最前面
        parent_variant = None
        other_variants = []
        for v in variants:
            if v.get("ASIN") == spu_id:
                parent_variant = v
            else:
                other_variants.append(v)
        
        # 重新排序：父ASIN在前，其他在后
        if parent_variant:
            variants = [parent_variant] + other_variants
        else:
            # 如果找不到父ASIN对应的变体，打印错误日志
            print(
                f"WARN: Parent ASIN not found in variants: SPU={spu_id}, "
                f"variant_count={len(variants)}, first_ASIN={variants[0].get('ASIN') if variants else 'N/A'}",
                flush=True,
            )
        # 统一变体标题：如果与主商品不一致，改为与主商品一致
        main_title = variants[0].get("商品标题") or ""
        for v in variants:
            title = v.get("商品标题") or ""
            if main_title and title and title != main_title:
                print(
                    f"FIX variant title mismatch: SPU={spu_id}, ASIN={v.get('ASIN')}, "
                    f"main_title='{main_title}', variant_title='{title}' -> using main_title",
                    flush=True,
                )
                v["商品标题"] = main_title  # 统一为主商品标题
        spu_count += 1
        if args.max_products is not None and spu_count > int(args.max_products):
            break
        if len(variants) == 1:
            rows = [build_s_row(variants[0])]
        else:
            rows = build_m_p_rows(variants)
        # 填充商品ID（从1开始全局递增）
        for r in rows:
            r["商品ID"] = next_product_id
            next_product_id += 1
        group_rows_list.append(rows)
    # 按最大行数拆成多个文件（注意：同一 SPU 不拆分）
    data_start_row = 4  # 与模板/写入工具保持一致
    header_rows = data_start_row - 1  # 包含标题行+说明行
    max_total_rows = args.max_rows_per_output or 0
    if max_total_rows and max_total_rows > header_rows:
        max_data_rows = max_total_rows - header_rows
    else:
        max_data_rows = None  # 不限制
    chunks = []
    current_chunk = []
    current_count = 0
    if max_data_rows is None:
        # 不做分片，直接一个 chunk
        for gr in group_rows_list:
            current_chunk.extend(gr)
        if current_chunk:
            chunks.append(current_chunk)
    else:
        for gr in group_rows_list:
            gsize = len(gr)
            # 如果单个 SPU 本身就超过阈值，只能独占一个文件
            if gsize > max_data_rows:
                if current_chunk:
                    chunks.append(current_chunk)
                    current_chunk = []
                    current_count = 0
                chunks.append(gr)
                continue
            # 如果放不下当前 chunk，则先封一个，再开新 chunk
            if current_count + gsize > max_data_rows:
                if current_chunk:
                    chunks.append(current_chunk)
                current_chunk = list(gr)
                current_count = gsize
            else:
                current_chunk.extend(gr)
                current_count += gsize
        if current_chunk:
            chunks.append(current_chunk)
    total_rows = sum(len(c) for c in chunks)
    print(
        "Generated Excel data rows: {} (SPU groups output: {}, files: {})".format(
            total_rows, len(group_rows_list), len(chunks)
        ),
        flush=True,
    )
    # 输出多个文件：如果只一个 chunk，直接用指定 output；多个则加 _partN 后缀
    base = Path(args.output)
    stem = base.stem
    suffix = base.suffix or ".xlsx"
    for idx, chunk in enumerate(chunks, start=1):
        out_path = str(base) if len(chunks) == 1 else str(base.with_name(f"{stem}_part{idx}{suffix}"))
        print(f"Writing file {idx}/{len(chunks)}: {out_path} (rows: {len(chunk)})", flush=True)
        create_excel_from_template_fast(args.template, out_path, chunk, data_start_row=data_start_row)
if __name__ == "__main__":
    main()