""" Sync products (SPU + SKU) from Cin7 to yudao SPU <-> SKU relationship in yudao: - product_spu.id = Cin7 product.id (NOT AUTO_INCREMENT) - product_sku.id = Cin7 productOptions[].id (NOT AUTO_INCREMENT) - product_sku.spu_id = Cin7 product.id (direct, no foreign key needed) - product_spu.price = min(product_sku.price) - product_spu.stock = sum(product_sku.stock) Cin7 product fields: id, name, description, brand, category, images, ... Cin7 productOptions fields: id, productId, code, productOptionBarcode, retailPrice, wholesalePrice, stockOnHand, option1/2/3, optionWeight, image, priceColumns{costAUD}, ... """ import json import os import threading from datetime import datetime from concurrent.futures import ThreadPoolExecutor, as_completed from db import query_one, query_all, execute_one, insert_and_get_id from sync_brand import BRAND_NAME_TO_ID_MAP from sync_category import CATEGORY_NAME_TO_ID_MAP from config import TENANT_ID _PROP_LOCK = threading.Lock() _PROP_NAME_TO_ID = {} _PROPVAL_NAME_TO_ID = {} def _log_error(cin7_id, product_name, exc, sp=None): """Append one error record to sync_errors.jsonl.""" record = { "ts": datetime.now().isoformat(), "cin7_id": cin7_id, "name": product_name, "error": str(exc), } if sp: record["product_data"] = sp with open(_ERROR_LOG, "a", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\n") # Level id cache: name (lowercase) -> yudao member_level.id _LEVEL_NAME_TO_ID = {} _LEVEL_LOADED = False _ERROR_LOG = os.path.join(os.path.dirname(__file__), 'sync_errors.jsonl') def _float(val, default=0.0): try: return float(val) except (ValueError, TypeError): return default def _int(val, default=0): try: return int(val) except (ValueError, TypeError): return default def _str(val): if val is None: return "" return str(val).strip() def _sync_spu(sp, tenant_id=TENANT_ID): """ Sync one Cin7 product as one SPU + its SKUs. Returns (cin7_id, created_bool) All field defaults mirror the Java ProductSpuConvert logic exactly. """ cin7_id = sp["id"] # product_spu.id = Cin7 product id options = sp.get("productOptions") or [] brand_name = _str(sp.get("brand")) brand_id = BRAND_NAME_TO_ID_MAP.get(brand_name, 0) # Java logic: try subCategory first, then categoryIdArray[0] sub_cat_name = _str(sp.get("subCategory")) category_id = 0 if sub_cat_name: category_id = CATEGORY_NAME_TO_ID_MAP.get(sub_cat_name, 0) if category_id == 0: cat_id_array = sp.get("categoryIdArray") or [] if cat_id_array: category_id = int(cat_id_array[0]) # category_ids: comma-separated string (Java logic: categoryIdArray + categoryId + parentId, deduplicated) cat_id_array = sp.get("categoryIdArray") or [] if category_id and category_id not in cat_id_array: cat_id_array.append(category_id) cat_name = _str(sp.get("category")) if cat_name and cat_name in CATEGORY_NAME_TO_ID_MAP: parent_id = CATEGORY_NAME_TO_ID_MAP.get(cat_name, 0) if parent_id and parent_id not in cat_id_array: cat_id_array.append(parent_id) cat_ids_str = ",".join(str(int(x)) for x in cat_id_array) if cat_id_array else "" channels = _str(sp.get("channels") or "") status = 0 if channels == "Magento 2" else 1 description = _str(sp.get("description") or "") introduction = description images = sp.get("images") or [] pic_url = images[0].get("link", "") if images else "" slider_pics = json.dumps([img.get("link", "") for img in images[:15] if img.get("link")]) spec_type = 1 if options else 0 unit = 1 # Java hardcodes unit=1 sort = 0 delivery_template_id = 1 # Java hardcodes 1L browse_count = 0 # Java hardcodes 0 keyword = _str(sp.get("styleCode"))[:256] recommendation = { "recommend_hot": False, "recommend_benefit": False, "recommend_best": False, "recommend_new": False, "recommend_good": False, } give_integral = 0 virtual_sales_count = 1 activity_orders = "[0]" # Java hardcodes "[0]" sub_commission_type = False existing_row = query_one( "SELECT id FROM product_spu WHERE id=%s AND deleted=0 LIMIT 1", (cin7_id,) ) if not existing_row: sql = """ INSERT INTO product_spu (id, name, category_id, category_ids, brand_id, pic_url, slider_pic_urls, slider_bottom_pic_urls, status, description, unit, sort, spec_type, price, market_price, cost_price, stock, delivery_template_id, keyword, introduction, recommend_hot, recommend_benefit, recommend_best, recommend_new, recommend_good, give_integral, virtual_sales_count, browse_count, activity_orders, sub_commission_type, creator, create_time, updater, tenant_id) VALUES (%s, %s, %s, %s, %s, %s, %s, NULL, %s, %s, %s, %s, %s, -1, -1, -1, 0, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 'cin7_sync', NOW(), 'cin7_sync', %s) """ execute_one(sql, ( cin7_id, _str(sp.get("name")), category_id, cat_ids_str, brand_id or 0, pic_url, slider_pics, status, description, unit, sort, spec_type, keyword, introduction, recommendation["recommend_hot"], recommendation["recommend_benefit"], recommendation["recommend_best"], recommendation["recommend_new"], recommendation["recommend_good"], give_integral, virtual_sales_count, browse_count, activity_orders, sub_commission_type, delivery_template_id, tenant_id )) created = True print(f" [spu] INSERT id={cin7_id} name='{_str(sp.get('name'))}'") else: sql = """ UPDATE product_spu SET name=%s, category_id=%s, category_ids=%s, brand_id=%s, status=%s, description=%s, spec_type=%s, keyword=%s, introduction=%s, recommend_hot=%s, recommend_benefit=%s, recommend_best=%s, recommend_new=%s, recommend_good=%s, update_time=NOW(), updater='cin7_sync' WHERE id=%s """ execute_one(sql, ( _str(sp.get("name")), category_id, cat_ids_str, brand_id or 0, status, description, spec_type, keyword, introduction, recommendation["recommend_hot"], recommendation["recommend_benefit"], recommendation["recommend_best"], recommendation["recommend_new"], recommendation["recommend_good"], cin7_id )) created = False print(f" [spu] UPDATE id={cin7_id} name='{_str(sp.get('name'))}'") _sync_skus(cin7_id, sp, tenant_id) _update_spu_stats(cin7_id) return cin7_id, created def _sync_skus(spu_id, sp, tenant_id): """ Sync all productOptions as SKUs under the given SPU (by cin7 product id). Batch-optimized: collects all new SKUs and all updated SKUs, then fires a single executemany per SPU instead of one DB round-trip per SKU. Matching logic (mirrors Java insertOrUpdateBatch by id): - CINA7 option.id == yudao product_sku.id (exact match by id) - For existing SKU: only update price/stock/levels/bar_code (leave properties/pic_url intact) - For new SKU: full INSERT - CINA7 options NOT in yudao -> insert new SKU - yudao SKUs NOT in CINA7 options -> NOT deleted (preserve manual variants) """ from db import insert_many, update_many options = sp.get("productOptions") or [] if not options: _upsert_single_sku(spu_id, sp, None, tenant_id) return # Build id->row map for existing SKUs (match by id, mirrors Java insertOrUpdateBatch) existing_skus = query_all( "SELECT id, bar_code FROM product_sku WHERE spu_id=%s AND deleted=0", (spu_id,) ) existing_by_id = {es["id"]: es for es in existing_skus} # Collect rows for batch operations new_rows = [] # list of tuples for INSERT upd_rows = [] # list of tuples for UPDATE (bar_code, price, market, cost, stock, id) upd_ids = [] # SKU ids that need levels update for opt in options: cin7_sku_id = opt["id"] existing_sku = existing_by_id.get(cin7_sku_id) row_data = _build_sku_row(spu_id, opt, sp, tenant_id) if existing_sku: # existing: UPDATE price/stock/bar_code/market_price, update levels separately upd_rows.append(( row_data["bar_code"], row_data["price_fen"], row_data["market_fen"], row_data["cost_fen"], row_data["stock"], cin7_sku_id, )) upd_ids.append(cin7_sku_id) _update_sku_props_fast(cin7_sku_id, row_data["properties"]) _update_sku_levels(cin7_sku_id, sp) # existing SKU: update levels on UPDATE too else: # new: full INSERT row (spu_id prepended at insert time, id is AUTO_INCREMENT) new_rows.append(( spu_id, row_data["bar_code"], row_data["properties"], row_data["pic_url"], row_data["price_fen"], row_data["market_fen"], row_data["cost_fen"], row_data["stock"], row_data["weight"], row_data["volume"], tenant_id, )) _ensure_levels_loaded() try: _update_sku_levels(cin7_sku_id, sp) except Exception as e: print(f" [sku] levels skip (product_member_level table?): {e}") # Batch INSERT new SKUs if new_rows: INSERT_SKU_SQL = ( "INSERT INTO product_sku (spu_id, bar_code, properties, pic_url, price, market_price, cost_price, " "stock, weight, volume, creator, create_time, updater, tenant_id) " "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 'cin7_sync', NOW(), 'cin7_sync', %s)" ) inserted = insert_many(INSERT_SKU_SQL, new_rows) print(f" [sku] bulk INSERT {inserted} SKUs for spu_id={spu_id}") # Batch UPDATE existing SKUs if upd_rows: UPDATE_SKU_SQL = ( "UPDATE product_sku SET bar_code=%s, price=%s, market_price=%s, " "cost_price=%s, stock=%s, update_time=NOW() WHERE id=%s" ) updated = update_many(UPDATE_SKU_SQL, upd_rows) print(f" [sku] bulk UPDATE {updated} SKUs for spu_id={spu_id}") # NOTE: do NOT auto-delete yudao SKUs not in CINA7 options. # yudao may have manual variants (15804-15815) that don't exist in CINA7. # Mirrors Java: only CINA7-provided SKUs are touched. def _upsert_single_sku(spu_id, sp, existing_sku, tenant_id): """Single-spec SKU — no productOptions, derive from product-level fields""" code = _str(sp.get("styleCode")) barcode = _str(sp.get("barCode") or sp.get("barcode")) key = barcode if barcode else (code if code else "") # yudao single-SKU mapping: price=retailPrice, market=wholesalePrice, cost=vipPrice price_cols = sp.get("priceColumns") or {} retail_a = _float(price_cols.get("retailAUD")) tier1 = _float(price_cols.get("tier1AUD")) cost_a = _float(price_cols.get("costAUD")) retail = _float(sp.get("retailPrice") or sp.get("sellingPrice")) wholesale = _float(sp.get("wholesalePrice")) vip = _float(sp.get("vipPrice")) price_fen = int((retail_a if retail_a > 0 else retail) * 100) market_fen = int((tier1 if tier1 > 0 else wholesale) * 100) cost_fen = int((cost_a if cost_a > 0 else vip) * 100) stock = _int(sp.get("stockOnHand")) weight = _float(sp.get("weight")) volume = _float(sp.get("volume")) pic_url = "" images = sp.get("images") or [] if images and isinstance(images[0], dict): pic_url = images[0].get("link", "") elif images and isinstance(images[0], str): pic_url = images[0] properties = json.dumps([{"propertyId": 0, "propertyName": "默认", "valueId": 0, "valueName": "默认"}]) if existing_sku: _update_sku_props(existing_sku["id"], properties) execute_one( "UPDATE product_sku SET bar_code=%s, price=%s, market_price=%s, cost_price=%s, stock=%s, " "update_time=NOW() WHERE id=%s", (key, price_fen, market_fen, cost_fen, stock, existing_sku["id"]) ) _ensure_levels_loaded() _update_sku_levels(existing_sku["id"], sp) else: sku_id = _int(sp.get("id")) execute_one( "INSERT INTO product_sku (id, spu_id, bar_code, properties, pic_url, price, market_price, " "cost_price, stock, weight, volume, creator, create_time, updater, tenant_id) " "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 'cin7_sync', NOW(), 'cin7_sync', %s)", (sku_id, spu_id, key, properties, pic_url, price_fen, market_fen, cost_fen, stock, weight, 0.0, tenant_id) ) def _build_sku_row(spu_id, opt, sp, tenant_id): """ Build a dict of all computed SKU fields from one Cin7 productOption. Shared by both INSERT and UPDATE paths. """ price_cols = opt.get("priceColumns") or {} cost_a = _float(price_cols.get("costAUD")) retail = _float(opt.get("retailPrice") or sp.get("retailPrice") or sp.get("sellingPrice")) wholesale = _float(opt.get("wholesalePrice") or sp.get("wholesalePrice")) vip = _float(opt.get("vipPrice") or sp.get("vipPrice")) price_fen = int(retail * 100) market_fen = int(wholesale * 100) cost_fen = int((cost_a if cost_a > 0 else vip) * 100) stock = _int(opt.get("stockOnHand")) weight = _float(opt.get("optionWeight")) barcode = _str(opt.get("productOptionBarcode") or opt.get("barCode") or opt.get("barcode")) # Properties from option1/2/3 props = [] for opt_n, opt_key in enumerate(["option1", "option2", "option3"], start=1): val = _str(opt.get(opt_key)) if val: prop_name = _str(opt.get(f"optionLabel{opt_n}")) or f"属性{opt_n}" with _PROP_LOCK: prop_id = _PROP_NAME_TO_ID.get(prop_name) if prop_id is None: row = query_one("SELECT id FROM product_property WHERE name=%s AND tenant_id=%s", (prop_name, tenant_id)) prop_id = row["id"] if row else None if prop_id: _PROP_NAME_TO_ID[prop_name] = prop_id if prop_id: with _PROP_LOCK: val_id = _PROPVAL_NAME_TO_ID.get((prop_id, val)) if val_id is None: row = query_one("SELECT id FROM product_property_value WHERE property_id=%s AND name=%s", (prop_id, val)) val_id = row["id"] if row else None if val_id: _PROPVAL_NAME_TO_ID[(prop_id, val)] = val_id if val_id is None and prop_id: val_id = insert_and_get_id( "INSERT INTO product_property_value (property_id, name, tenant_id, creator, create_time, updater) VALUES (%s, %s, %s, 'cin7_sync', NOW(), 'cin7_sync')", (prop_id, val, tenant_id) ) with _PROP_LOCK: _PROPVAL_NAME_TO_ID[(prop_id, val)] = val_id if val_id: props.append({"propertyId": prop_id, "propertyName": prop_name, "valueId": val_id, "valueName": val}) if not props: props = [{"propertyId": 0, "propertyName": "默认", "valueId": 0, "valueName": "默认"}] # Image pic_url = "" img_link = opt.get("image") if img_link: pic_url = img_link.get("link") if isinstance(img_link, dict) else img_link if not pic_url and sp.get("images"): pic_url = sp["images"][0].get("link", "") return { "bar_code": barcode, "price_fen": price_fen, "market_fen": market_fen, "cost_fen": cost_fen, "stock": stock, "weight": weight, "volume": 0.0, "properties": json.dumps(props, ensure_ascii=False), "pic_url": pic_url, } def _update_sku_props_fast(sku_id, properties): """Fast single-field update for existing SKU properties (no lock, direct call).""" if properties: execute_one("UPDATE product_sku SET properties=%s WHERE id=%s", (properties, sku_id)) def _upsert_sku(spu_id, opt, sp, existing_sku, tenant_id): """ Insert or update a single SKU (one Cin7 productOption). existing_sku: row dict or None NOTE: this function is kept for single-spec (no-options) path. Multi-option path uses _sync_skus batch logic instead. """ opt_id = opt["id"] price_cols = opt.get("priceColumns") or {} cost_a = _float(price_cols.get("costAUD")) retail = _float(opt.get("retailPrice") or sp.get("retailPrice") or sp.get("sellingPrice")) wholesale = _float(opt.get("wholesalePrice") or sp.get("wholesalePrice")) vip = _float(opt.get("vipPrice") or sp.get("vipPrice")) tier1 = _float(price_cols.get("tier1AUD")) price_fen = int(retail * 100) market_fen = int(wholesale * 100) cost_fen = int((cost_a if cost_a > 0 else vip) * 100) stock = _int(opt.get("stockOnHand")) weight = _float(opt.get("optionWeight")) barcode = _str(opt.get("productOptionBarcode") or opt.get("barCode") or opt.get("barcode")) # Properties from option1/2/3 props = [] for opt_n, opt_key in enumerate(["option1", "option2", "option3"], start=1): val = _str(opt.get(opt_key)) if val: prop_name = _str(opt.get(f"optionLabel{opt_n}")) or f"属性{opt_n}" with _PROP_LOCK: prop_id = _PROP_NAME_TO_ID.get(prop_name) if prop_id is None: row = query_one("SELECT id FROM product_property WHERE name=%s AND tenant_id=%s", (prop_name, tenant_id)) prop_id = row["id"] if row else None if prop_id: _PROP_NAME_TO_ID[prop_name] = prop_id if prop_id: with _PROP_LOCK: val_id = _PROPVAL_NAME_TO_ID.get((prop_id, val)) if val_id is None: row = query_one("SELECT id FROM product_property_value WHERE property_id=%s AND name=%s", (prop_id, val)) val_id = row["id"] if row else None if val_id: _PROPVAL_NAME_TO_ID[(prop_id, val)] = val_id if val_id is None and prop_id: val_id = insert_and_get_id( "INSERT INTO product_property_value (property_id, name, tenant_id, creator, create_time, updater) VALUES (%s, %s, %s, 'cin7_sync', NOW(), 'cin7_sync')", (prop_id, val, tenant_id) ) with _PROP_LOCK: _PROPVAL_NAME_TO_ID[(prop_id, val)] = val_id if val_id: props.append({"propertyId": prop_id, "propertyName": prop_name, "valueId": val_id, "valueName": val}) if not props: props = [{"propertyId": 0, "propertyName": "默认", "valueId": 0, "valueName": "默认"}] properties = json.dumps(props, ensure_ascii=False) # Image pic_url = "" img_link = opt.get("image") if img_link: pic_url = img_link.get("link") if isinstance(img_link, dict) else img_link if not pic_url and sp.get("images"): pic_url = sp["images"][0].get("link", "") if existing_sku: _update_sku_props(existing_sku["id"], properties) try: _update_sku_levels(opt_id, sp) except Exception as e: print(f" [sku] levels skip (product_member_level table?): {e}") execute_one( "UPDATE product_sku SET bar_code=%s, price=%s, market_price=%s, cost_price=%s, stock=%s, " "update_time=NOW() WHERE id=%s", (barcode, price_fen, market_fen, cost_fen, stock, opt_id) ) print(f" [sku] UPDATE id={opt_id} price={price_fen} stock={stock}") else: _ensure_levels_loaded() try: _update_sku_levels(opt_id, sp) except Exception as e: print(f" [sku] levels skip (product_member_level table?): {e}") execute_one( "INSERT INTO product_sku (id, spu_id, bar_code, properties, pic_url, price, market_price, " "cost_price, stock, weight, volume, creator, create_time, updater, tenant_id) " "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 'cin7_sync', NOW(), 'cin7_sync', %s)", (opt_id, spu_id, barcode, properties, pic_url, price_fen, market_fen, cost_fen, stock, weight, 0.0, tenant_id) ) print(f" [sku] INSERT id={opt_id} price={price_fen} stock={stock}") def _update_sku_props(sku_id, properties): """Update properties only for existing SKU (Java: preserve pic_url/user-set fields).""" if properties: execute_one("UPDATE product_sku SET properties=%s WHERE id=%s", (properties, sku_id)) def _ensure_levels_loaded(): """Load level id cache from yudao member_level table. Gracefully skips if table doesn't exist.""" global _LEVEL_NAME_TO_ID, _LEVEL_LOADED if _LEVEL_LOADED: return try: rows = query_all("SELECT id, name FROM product_member_level WHERE deleted=0") for r in rows: _LEVEL_NAME_TO_ID[r["name"].lower()] = r["id"] _LEVEL_LOADED = True print(f" [levels] loaded {len(rows)} levels: {list(_LEVEL_NAME_TO_ID.keys())}") except Exception as e: if "product_member_level" in str(e) or "doesn't exist" in str(e): print(f" [levels] table not found, skipping level sync: {e}") _LEVEL_LOADED = True # mark loaded to avoid repeated attempts else: raise def _update_sku_levels(sku_id, sp): """ Build and persist the levels JSON array for one SKU. Mirrors the Java member-level sync logic (ProductSkuConvert.fillLevels). Each level object: {"id": , "name": , "icon": "", "value": } """ _ensure_levels_loaded() options = sp.get("productOptions") or [] opt = None for o in options: if o.get("id") == sku_id: opt = o break price_cols = (opt or {}).get("priceColumns") or sp.get("priceColumns") or {} def get_level_price(dollar_val): if dollar_val is None or dollar_val == "": return 0 try: return int(float(dollar_val) * 100) except (ValueError, TypeError): return 0 levels = [] for name_key, name_db in [ ("retailAUD", "retailAUD"), ("vipPrice", "vipaud"), # Cin7 vipPrice field -> vipaud level ("wholesale65AUD","wholesale65aud"), ("wholesale60AUD","wholesale60aud"), ("partnerAUD", "partneraud"), ("costAUD", "costaud"), ("specialPrice", "specialprice"), ("tier1AUD", "tier1aud"), ("tier2AUD", "tier2aud"), ("tier3AUD", "tier3aud"), ]: lid = _LEVEL_NAME_TO_ID.get(name_db.lower()) if not lid: continue # vipPrice comes from opt/sp root, others from priceCols if name_key == "vipPrice": val = (opt or {}).get("vipPrice") or sp.get("vipPrice") else: val = price_cols.get(name_key) levels.append({"id": lid, "name": name_db, "icon": "", "value": get_level_price(val)}) if levels: execute_one( "UPDATE product_sku SET levels=%s, update_time=NOW() WHERE id=%s", (json.dumps(levels), sku_id) ) def _update_spu_stats(spu_id): """Update SPU price=min(sku.price) and stock=sum(sku.stock) after SKU sync.""" row = query_one( "SELECT MIN(price) as min_price, MIN(market_price) as min_market_price, SUM(stock) as total_stock FROM product_sku WHERE spu_id=%s AND deleted=0", (spu_id,) ) if row: execute_one( "UPDATE product_spu SET price=%s, market_price=%s, stock=%s, update_time=NOW() WHERE id=%s", (row["min_price"] or 0, row["min_market_price"] or 0, row["total_stock"] or 0, spu_id) ) def sync_spus(cin7_api, tenant_id=TENANT_ID): """Alias for sync_all_products (matches main.py import name).""" return sync_all_products(cin7_api, tenant_id) def sync_all_products(cin7_api, tenant_id=TENANT_ID): """Fetch all Cin7 products and sync each as SPU + SKUs.""" print("\n=== Sync All Products ===") raw = cin7_api.get_products() products = raw.get("d", []) if isinstance(raw, dict) else raw print(f"[products] fetched {len(products)} products") success = fail = 0 with ThreadPoolExecutor(max_workers=16) as executor: futures = {executor.submit(_sync_spu, sp, tenant_id): sp for sp in products} for future in as_completed(futures): sp = futures[future] try: cid, created = future.result() success += 1 except Exception as e: fail += 1 print(f" [ERROR] product {sp.get('id','?')}: {e}") _log_error(sp.get('id'), sp.get('name'), e, sp) print(f"\n[products] done: {success} synced, {fail} failed") return success, fail