gdm-Servo-model/scripts/import_shop_catalog.py

#!/usr/bin/env python3
"""Import the ecommerce catalog into the site's catalog data.

Usage:
    python3 scripts/import_shop_catalog.py \
      "/Users/jacky/Desktop/美好智玩/02_客户项目/高徳乐/商城产品资料_20260806"

The source export contains product families with SKU variants. The website
keeps one card per source product family and records all source variant codes
inside the product's specifications.
"""

from __future__ import annotations

import json
import re
import sys
from collections import Counter
from pathlib import Path


PROJECT_ROOT = Path(__file__).resolve().parents[1]
OUTPUT_PATH = PROJECT_ROOT / "app" / "catalog-data.ts"


def clean(value: object) -> str:
    if value is None:
        return ""
    return str(value).strip()


def category_for(product: dict[str, object]) -> str:
    top_category = clean(product.get("top_cat"))
    sub_category = clean(product.get("sub_cat"))
    if top_category == "伺服舵机":
        return "舵盘配件" if sub_category == "舵盘" else "标准舵机"
    if top_category == "电动积木系列":
        return "积木舵机"
    if top_category == "标准齿轮箱":
        return "标准齿轮箱"
    return "精密齿轮配件"


def normalize_spec_label(label: str) -> str:
    if label.startswith("空载速度"):
        return "空载速度"
    if label.startswith("堵转扭矩") or label.startswith("堵转扭力"):
        return "堵转扭矩"
    if label.startswith("最大砝码负载"):
        return "最大砝码负载"
    if label.startswith("产品尺寸") or label.startswith("外形尺寸"):
        return "产品尺寸"
    if label.startswith("额定电压"):
        return "额定电压"
    if label.startswith("电压范围") or label.startswith("工作电压范围"):
        return "电压范围"
    if label in {"控制操作角度", "可操作角度", "操作角度范围"}:
        return "可操作角度"
    if label.startswith("端子连接线") or label.startswith("连接线类型"):
        return "端子连接线"
    if label == "齿轮类型":
        return "齿轮材质"
    if label.startswith("单重"):
        return "重量"
    if label == "空载速度(rpm)":
        return "空载速度"
    if label == "堵转扭力(Kgf.cm)":
        return "堵转扭矩"
    if label == "外形尺寸(mm)":
        return "产品尺寸"
    return label


def merged_specs(product: dict[str, object]) -> tuple[dict[str, str], list[str]]:
    specs: dict[str, str] = {}
    for source in (product.get("specs") or {}, product.get("sku_specs") or {}):
        if not isinstance(source, dict):
            continue
        for raw_label, raw_value in source.items():
            value = clean(raw_value)
            if not value:
                continue
            label = normalize_spec_label(clean(raw_label))
            specs[label] = value

    variant_codes: list[str] = []
    attrs = product.get("attrs") or {}
    if isinstance(attrs, dict):
        for values in attrs.values():
            if isinstance(values, list):
                variant_codes.extend(clean(value) for value in values if clean(value))
    variant_codes = list(dict.fromkeys(variant_codes))
    if variant_codes:
        specs["可选编码"] = "、".join(variant_codes)

    price = clean(product.get("price"))
    if price:
        specs["参考单价"] = f"¥{price}"
    return specs, variant_codes


def parse_number(value: str) -> float | None:
    match = re.search(r"\d+(?:\.\d+)?", value)
    return float(match.group()) if match else None


def main() -> None:
    if len(sys.argv) != 2:
        raise SystemExit("请提供商城产品资料目录路径")

    source_root = Path(sys.argv[1]).expanduser().resolve()
    parsed_path = source_root / "data" / "parsed_products.json"
    manifest_path = source_root / "data" / "image_manifest.json"
    products_by_id = json.loads(parsed_path.read_text(encoding="utf-8"))
    manifest_by_id = json.loads(manifest_path.read_text(encoding="utf-8"))

    products: list[dict[str, object]] = []
    category_counts: Counter[str] = Counter()
    for source_id, source in sorted(products_by_id.items(), key=lambda item: int(item[0])):
        manifest = manifest_by_id[source_id]
        main_image = next(file for file in manifest["files"] if file["tag"] == "main")
        specs, variant_codes = merged_specs(source)
        category = category_for(source)
        code = clean(source.get("code")) or (variant_codes[0] if variant_codes else source_id)
        price = parse_number(clean(source.get("price")))
        product = {
            "id": f"shop-{source_id}",
            "sourceId": int(source_id),
            "sourceCategory": clean(source.get("top_cat")),
            "subCategory": clean(source.get("sub_cat")),
            "category": category,
            "name": clean(source.get("name")) or code,
            "code": code,
            "image": f"/catalog-products/source-main/{source_id}.jpg",
            "specs": specs,
            "torqueValue": parse_number(specs.get("堵转扭矩", "")),
            "ratedVoltage": parse_number(specs.get("额定电压", "")),
            "price": price,
            "variantCodes": variant_codes,
            "hasDescription": bool(source.get("has_desc")),
        }
        products.append(product)
        category_counts[category] += 1

    payload = json.dumps(products, ensure_ascii=False, indent=2)
    counts = json.dumps(dict(category_counts), ensure_ascii=False, indent=2)
    OUTPUT_PATH.write_text(
        "// Generated from 商城产品资料_20260806/data/parsed_products.json.\n"
        "// Regenerate with scripts/import_shop_catalog.py <商城产品资料目录>.\n\n"
        "export type CatalogProduct = {\n"
        "  id: string;\n"
        "  sourceId: number;\n"
        "  sourceCategory: string;\n"
        "  subCategory: string;\n"
        "  category: string;\n"
        "  name: string;\n"
        "  code: string;\n"
        "  image: string;\n"
        "  specs: Record<string, string>;\n"
        "  torqueValue: number | null;\n"
        "  ratedVoltage: number | null;\n"
        "  price: number | null;\n"
        "  variantCodes: string[];\n"
        "  hasDescription: boolean;\n"
        "};\n\n"
        f"export const CATALOG_PRODUCTS: CatalogProduct[] = {payload};\n\n"
        f"export const CATALOG_COUNTS: Record<string, number> = {counts};\n",
        encoding="utf-8",
    )
    print(json.dumps({"total": len(products), "categories": dict(category_counts)}, ensure_ascii=False))


if __name__ == "__main__":
    main()