from __future__ import annotations import logging import re import secrets import string from datetime import datetime, timezone from decimal import Decimal from typing import Any import bcrypt from sqlalchemy import delete as sa_delete, insert, select from sqlalchemy.ext.asyncio import AsyncSession from app.core.models.brand import Brand from app.core.models.category import Category from app.core.models.customer import Customer from app.core.models.customer_address import CustomerAddress from app.core.models.product import Product, ProductImage, ProductTierPrice, ProductVariant, product_categories from app.plugins.cin7_import.models import Cin7EntityMap, Cin7SyncLog, Cin7SyncState logger = logging.getLogger("uvicorn.error") def _slugify(text: str) -> str: slug = re.sub(r"[^a-z0-9]+", "-", text.lower()).strip("-") return slug[:200] or "item" def _generate_password(length: int = 10) -> str: alphabet = string.ascii_letters + string.digits while True: pwd = "".join(secrets.choice(alphabet) for _ in range(length)) if any(c.isupper() for c in pwd) and any(c.islower() for c in pwd) and any(c.isdigit() for c in pwd): return pwd def _hash_pwd(pwd: str) -> str: return bcrypt.hashpw(pwd.encode(), bcrypt.gensalt()).decode() def _d(val, default="0") -> Decimal: try: return Decimal(str(val)).quantize(Decimal("0.01")) except Exception: return Decimal(default) def _int(val, default=0) -> int: try: return int(val) except (ValueError, TypeError): return default def _str(val) -> str: return str(val).strip() if val else "" class Cin7ImportService: def __init__(self, db: AsyncSession, tenant_id: int, client: Any, sync_options: dict | None = None): self.db = db self.tenant_id = tenant_id self.client = client opts = sync_options or {} self.sync_price = opts.get("sync_price", True) self.auto_delist_zero_stock = opts.get("auto_delist_zero_stock", True) self.sync_status = opts.get("sync_status", True) self.publish_field = (opts.get("publish_field") or "").strip() self.publish_field_value = (opts.get("publish_field_value") or "").strip() self.price_col_default = (opts.get("price_col_default") or "").strip() self.price_col_cost = (opts.get("price_col_cost") or "").strip() self.price_col_market = (opts.get("price_col_market") or "").strip() self.level_mapping: dict[str, int] = {} raw_lm = opts.get("level_mapping", {}) for col_key, level_id in raw_lm.items(): try: self.level_mapping[col_key] = int(level_id) except (ValueError, TypeError): pass logger.info("[cin7_import] level_mapping loaded: %s", self.level_mapping) def _matches_publish_field(self, sp: dict) -> bool: if not self.publish_field: return False value = sp.get(self.publish_field) if self.publish_field_value: return str(value) == self.publish_field_value return bool(value and value not in (0, False)) def _is_publishable(self, sp: dict) -> bool: if _str(sp.get("status")).casefold() != "public": return False return not self.publish_field or self._matches_publish_field(sp) # ── helpers ────────────────────────────────────────────────── async def _get_sync_state(self, resource: str) -> Cin7SyncState | None: result = await self.db.execute( select(Cin7SyncState).where( Cin7SyncState.tenant_id == self.tenant_id, Cin7SyncState.resource == resource, ) ) return result.scalar_one_or_none() async def _update_sync_state(self, resource: str): state = await self._get_sync_state(resource) if not state: state = Cin7SyncState(tenant_id=self.tenant_id, resource=resource) self.db.add(state) state.last_success_at = datetime.now(timezone.utc) async def _write_log(self, resource: str, mode: str, status: str, created: int, updated: int, skipped: int, failed: int, message: str | None = None): log = Cin7SyncLog( tenant_id=self.tenant_id, resource=resource, mode=mode, status=status, created_count=created, updated_count=updated, skipped_count=skipped, failed_count=failed, message=message, ) self.db.add(log) await self.db.flush() async def _find_map(self, entity_type: str, cin7_id: str) -> Cin7EntityMap | None: result = await self.db.execute( select(Cin7EntityMap).where( Cin7EntityMap.tenant_id == self.tenant_id, Cin7EntityMap.entity_type == entity_type, Cin7EntityMap.cin7_id == cin7_id, ) ) return result.scalar_one_or_none() async def _create_map(self, entity_type: str, cin7_id: str, local_id: int): em = Cin7EntityMap( tenant_id=self.tenant_id, entity_type=entity_type, cin7_id=str(cin7_id), local_id=local_id, cin7_updated_at=datetime.now(timezone.utc), ) self.db.add(em) async def _delist_missing_products(self, cin7_ids: set[str]) -> int: result = await self.db.execute( select(Cin7EntityMap, Product) .join(Product, Product.id == Cin7EntityMap.local_id) .where( Cin7EntityMap.tenant_id == self.tenant_id, Cin7EntityMap.entity_type == "product", Product.tenant_id == self.tenant_id, ) ) delisted = 0 for product_map, product in result.all(): if product_map.cin7_id not in cin7_ids and product.status != "draft": product.status = "draft" delisted += 1 return delisted def _from_col(self, col: str, price_cols: dict, sp: dict, options: list) -> Decimal: """从 priceColumns 或顶层字段取价格，priceColumns 优先；值为 0 时再从 options 取最小非零值。""" if not col: return Decimal("0") pc_lower = {k.lower(): v for k, v in price_cols.items()} v = price_cols.get(col) or pc_lower.get(col.lower()) or sp.get(col) if not v and options: vals = [] for o in options: opc = o.get("priceColumns") or {} opc_lower = {k.lower(): vv for k, vv in opc.items()} ov = opc.get(col) or opc_lower.get(col.lower()) or o.get(col) d = _d(ov) if d > 0: vals.append(d) return min(vals) if vals else Decimal("0") return _d(v) def _resolve_prices(self, price_cols: dict, sp: dict, options: list) -> tuple[Decimal, Decimal | None, Decimal | None]: """根据 price_col_* 配置解析 base_price / cost_price / market_price。""" if self.price_col_default: base_price = self._from_col(self.price_col_default, price_cols, sp, options) if base_price <= 0: base_price = _d(sp.get("retailPrice") or sp.get("sellingPrice")) else: retail_aud = _d(price_cols.get("retailAUD")) retail = _d(sp.get("retailPrice") or sp.get("sellingPrice")) base_price = retail_aud if retail_aud > 0 else retail if base_price <= 0 and options: base_price = self._from_col("retailPrice", {}, sp, options) if self.price_col_market: v = self._from_col(self.price_col_market, price_cols, sp, options) market_price = v if v > 0 else None else: wholesale = _d(sp.get("wholesalePrice")) if wholesale <= 0 and options: wholesale = self._from_col("wholesalePrice", {}, sp, options) market_price = wholesale if wholesale > 0 else None if self.price_col_cost: v = self._from_col(self.price_col_cost, price_cols, sp, options) cost_price = v if v > 0 else None else: cost_aud = _d(price_cols.get("costAUD")) if cost_aud <= 0 and options: cost_aud = self._from_col("costPrice", {}, sp, options) vip = _d(sp.get("vipPrice")) cost_price = cost_aud if cost_aud > 0 else (vip if vip > 0 else None) return base_price, cost_price, market_price async def _resolve_categories(self, sp: dict) -> list[int]: """从 categoryIdArray 解析本地分类 ID 列表，找不到时按名字回退。""" category_ids: list[int] = [] for cin7_cat_id in (sp.get("categoryIdArray") or []): cat_map = await self._find_map("category", str(cin7_cat_id)) if cat_map: category_ids.append(cat_map.local_id) if not category_ids: SKIP_CATS = {"unassigned", "n/a", "none", ""} sub_cat_name = _str(sp.get("subCategory")) or _str(sp.get("category")) if sub_cat_name and sub_cat_name.lower() not in SKIP_CATS: cat_r = await self.db.execute( select(Category).where( Category.tenant_id == self.tenant_id, Category.name == sub_cat_name, ).limit(1) ) cat_obj = cat_r.scalar_one_or_none() if cat_obj: category_ids.append(cat_obj.id) return category_ids async def _sync_product_categories(self, product_id: int, category_ids: list[int]): """清空旧分类关联，重新写入所有分类。""" await self.db.execute( sa_delete(product_categories).where(product_categories.c.product_id == product_id) ) for cat_id in category_ids: await self.db.execute( insert(product_categories).values(product_id=product_id, category_id=cat_id).prefix_with("IGNORE") ) # ── 分类同步 ───────────────────────────────────────────────── async def sync_categories(self, mode: str = "manual") -> dict: logger.info("[cin7_import] sync_categories start, mode=%s", mode) try: raw_categories = await self.client.get_categories() except Exception as e: await self._write_log("categories", mode, "failed", 0, 0, 0, 0, str(e)[:500]) await self.db.commit() return {"resource": "categories", "created": 0, "updated": 0, "skipped": 0, "failed": 0, "error": str(e)} by_id = {c["id"]: c for c in raw_categories} def get_depth(cid, visited=None): if visited is None: visited = set() if cid in visited or cid not in by_id: return 0 visited.add(cid) pid = by_id[cid].get("parentId") if not pid or pid not in by_id: return 0 return 1 + get_depth(pid, visited) sorted_cats = sorted(raw_categories, key=lambda c: get_depth(c["id"])) created = updated = skipped = failed = 0 for cat in sorted_cats: try: cin7_id = str(cat.get("id")) name = _str(cat.get("name")) if not name or not cin7_id: skipped += 1 continue parent_cin7_id = cat.get("parentId") local_parent_id = None if parent_cin7_id: parent_map = await self._find_map("category", str(parent_cin7_id)) if parent_map: local_parent_id = parent_map.local_id existing_map = await self._find_map("category", cin7_id) cin7_is_active = 1 if cat.get("isActive", True) else 0 if existing_map: result = await self.db.execute( select(Category).where(Category.id == existing_map.local_id) ) category = result.scalar_one_or_none() if category: category.name = name category.parent_id = local_parent_id if self.sync_status: category.is_active = cin7_is_active updated += 1 else: skipped += 1 else: slug = _slugify(name) dup = await self.db.execute( select(Category).where( Category.tenant_id == self.tenant_id, Category.slug == slug, ) ) if dup.scalar_one_or_none(): slug = f"{slug}-{cin7_id}" category = Category( tenant_id=self.tenant_id, name=name, slug=slug, parent_id=local_parent_id, is_active=cin7_is_active if self.sync_status else 1, ) self.db.add(category) await self.db.flush() await self._create_map("category", cin7_id, category.id) created += 1 await self.db.flush() except Exception as e: logger.warning("[cin7_import] category sync failed cin7_id=%s: %s", cat.get("id"), e) await self.db.rollback() failed += 1 await self._update_sync_state("categories") await self._write_log("categories", mode, "success", created, updated, skipped, failed) await self.db.commit() logger.info("[cin7_import] categories done: created=%d updated=%d skipped=%d failed=%d", created, updated, skipped, failed) return {"resource": "categories", "created": created, "updated": updated, "skipped": skipped, "failed": failed} # ── 品牌同步 ───────────────────────────────────────────────── async def sync_brands(self, mode: str = "manual") -> dict: logger.info("[cin7_import] sync_brands start, mode=%s", mode) try: raw_brands = await self.client.get_brands() except Exception as e: await self._write_log("brands", mode, "failed", 0, 0, 0, 0, str(e)[:500]) await self.db.commit() return {"resource": "brands", "created": 0, "updated": 0, "skipped": 0, "failed": 0, "error": str(e)} created = updated = skipped = failed = 0 for b in raw_brands: try: cin7_id = str(b.get("id")) name = _str(b.get("company")) if not name or not cin7_id: skipped += 1 continue logo_url = _str(b.get("logoUrl")) cin7_is_active = 1 if b.get("isActive", True) else 0 existing_map = await self._find_map("brand", cin7_id) if existing_map: result = await self.db.execute( select(Brand).where(Brand.id == existing_map.local_id) ) brand = result.scalar_one_or_none() if brand: brand.name = name if self.sync_status: brand.is_active = cin7_is_active prev_cin7_logo = (existing_map.extra_data or {}).get("logo_url") if logo_url and (not brand.logo_url or brand.logo_url == prev_cin7_logo): brand.logo_url = logo_url existing_map.extra_data = {**(existing_map.extra_data or {}), "logo_url": logo_url} updated += 1 else: skipped += 1 else: slug = _slugify(name) dup = await self.db.execute( select(Brand).where( Brand.tenant_id == self.tenant_id, Brand.slug == slug, ) ) if dup.scalar_one_or_none(): slug = f"{slug}-{cin7_id}" brand = Brand( tenant_id=self.tenant_id, name=name, slug=slug, logo_url=logo_url, is_active=cin7_is_active if self.sync_status else 1, ) self.db.add(brand) await self.db.flush() await self._create_map("brand", cin7_id, brand.id) brand_map = await self._find_map("brand", cin7_id) if brand_map: brand_map.extra_data = {"logo_url": logo_url} created += 1 await self.db.flush() except Exception as e: logger.warning("[cin7_import] brand sync failed cin7_id=%s: %s", b.get("id"), e) await self.db.rollback() failed += 1 await self._update_sync_state("brands") await self._write_log("brands", mode, "success", created, updated, skipped, failed) await self.db.commit() logger.info("[cin7_import] brands done: created=%d updated=%d skipped=%d failed=%d", created, updated, skipped, failed) return {"resource": "brands", "created": created, "updated": updated, "skipped": skipped, "failed": failed} # ── 商品同步 ───────────────────────────────────────────────── async def sync_products(self, mode: str = "manual") -> dict: logger.info("[cin7_import] sync_products start, mode=%s", mode) try: raw_products = await self.client.get_products() except Exception as e: await self._write_log("products", mode, "failed", 0, 0, 0, 0, str(e)[:500]) await self.db.commit() return {"resource": "products", "created": 0, "updated": 0, "skipped": 0, "failed": 0, "error": str(e)} logger.info("[cin7_import] fetched %d products from Cin7", len(raw_products)) created = updated = skipped = failed = 0 cin7_ids = {str(sp["id"]) for sp in raw_products if sp.get("id") is not None} for sp in raw_products: try: cin7_id = str(sp.get("id")) name = _str(sp.get("name")) if not name or not cin7_id: skipped += 1 continue result = await self._sync_one_product(sp) if result == "created": created += 1 elif result == "updated": updated += 1 else: skipped += 1 await self.db.flush() except Exception as e: logger.error("[cin7_import] product sync failed cin7_id=%s name=%s: %s", sp.get("id"), sp.get("name"), e, exc_info=True) await self.db.rollback() failed += 1 delisted = await self._delist_missing_products(cin7_ids) if delisted: logger.info("[cin7_import] products delisted because absent from Cin7: %d", delisted) await self._update_sync_state("products") await self._write_log("products", mode, "success", created, updated, skipped, failed) await self.db.commit() logger.info("[cin7_import] products done: created=%d updated=%d skipped=%d failed=%d", created, updated, skipped, failed) return {"resource": "products", "created": created, "updated": updated, "skipped": skipped, "failed": failed} async def _sync_one_product(self, sp: dict) -> str: cin7_id = str(sp["id"]) name = _str(sp.get("name")) description = _str(sp.get("description")) style_code = _str(sp.get("styleCode")) sku_code = style_code or f"cin7-{cin7_id}" create_only = not self._is_publishable(sp) existing_map = await self._find_map("product", cin7_id) product = None if existing_map: result = await self.db.execute( select(Product).where(Product.id == existing_map.local_id) ) product = result.scalar_one_or_none() if product and create_only: if self.sync_status: product.status = "draft" existing_map.cin7_updated_at = datetime.now(timezone.utc) return "updated" return "skipped" if not product: await self.db.delete(existing_map) await self.db.flush() existing_map = None if create_only and not existing_map: result = await self.db.execute( select(Product).where( Product.tenant_id == self.tenant_id, Product.sku == sku_code, ).limit(1) ) product = result.scalar_one_or_none() if product: await self._create_map("product", cin7_id, product.id) return "skipped" images = sp.get("images") or [] cover_url = "" if images: cover_url = images[0].get("link", "") if isinstance(images[0], dict) else str(images[0]) options = sp.get("productOptions") or [] cin7_status = "active" if not create_only else "draft" # 价格 price_cols = sp.get("priceColumns") or {} base_price, cost_price, market_price = self._resolve_prices(price_cols, sp, options) total_stock = _int(sp.get("stockAvailable")) weight = _d(sp.get("weight"), "0") # 品牌：按名字匹配（Cin7 brand 字段是字符串品牌名） brand_name = _str(sp.get("brand")) brand_id = None if brand_name: brand_map_r = await self.db.execute( select(Cin7EntityMap).where( Cin7EntityMap.tenant_id == self.tenant_id, Cin7EntityMap.entity_type == "brand", ).join(Brand, Brand.id == Cin7EntityMap.local_id).where( Brand.name == brand_name, ) ) brand_map = brand_map_r.scalar_one_or_none() if brand_map: brand_id = brand_map.local_id else: brand_r = await self.db.execute( select(Brand).where( Brand.tenant_id == self.tenant_id, Brand.name == brand_name, ).limit(1) ) brand_obj = brand_r.scalar_one_or_none() if brand_obj: brand_id = brand_obj.id # 分类：从 categoryIdArray 解析所有分类 category_ids = await self._resolve_categories(sp) category_id = category_ids[0] if category_ids else None existing_map = await self._find_map("product", cin7_id) if existing_map: result = await self.db.execute( select(Product).where(Product.id == existing_map.local_id) ) product = result.scalar_one_or_none() if not product: # 商品已被删除，清掉旧 map，走新建流程 await self.db.delete(existing_map) await self.db.flush() existing_map = None if existing_map: # 更新已有商品 product.name = name product.description = description # 价格：直接覆盖（Cin7 为准） if self.sync_price: product.base_price = base_price product.market_price = market_price product.cost_price = cost_price # 品牌：直接覆盖（找不到则置 None） product.brand_id = brand_id # 分类主字段：直接覆盖 product.category_id = category_id # 分类关联表：清空后重新写入 await self._sync_product_categories(product.id, category_ids) # 封面图：仅在本地未手动修改时覆盖 prev_cin7_cover = (existing_map.extra_data or {}).get("cover_url") if cover_url and (not product.cover_url or product.cover_url == prev_cin7_cover): product.cover_url = cover_url # 同步更新 product_images 第一张主图 img_r = await self.db.execute( select(ProductImage).where( ProductImage.product_id == product.id, ProductImage.is_primary == 1, ).limit(1) ) primary_img = img_r.scalar_one_or_none() if not primary_img: img_r = await self.db.execute( select(ProductImage).where( ProductImage.product_id == product.id, ).order_by(ProductImage.sort_order).limit(1) ) primary_img = img_r.scalar_one_or_none() if primary_img: if primary_img.url != cover_url: primary_img.url = cover_url else: self.db.add(ProductImage( product_id=product.id, tenant_id=self.tenant_id, url=cover_url, sort_order=0, is_primary=1, )) product.set_attribute("cin7", {"product_id": cin7_id}) if self.sync_status: product.status = cin7_status if not options: product.stock_qty = total_stock product.weight = weight if weight > 0 else product.weight if not self.publish_field: if self.auto_delist_zero_stock and total_stock == 0: product.status = "draft" elif product.status == "draft" and total_stock > 0: product.status = "active" existing_map.cin7_updated_at = datetime.now(timezone.utc) existing_map.extra_data = {**(existing_map.extra_data or {}), "cover_url": cover_url} # SPU 阶梯价：清空后重新写入 spu_price_cols = dict(price_cols) if sp.get("vipPrice"): spu_price_cols.setdefault("vipPrice", sp["vipPrice"]) await self._sync_tier_prices(product, spu_price_cols, variant_id=None) if options: await self._sync_variants(product, options, sp) return "updated" else: # 新建商品 slug = _slugify(name) dup = await self.db.execute( select(Product).where( Product.tenant_id == self.tenant_id, Product.slug == slug, ) ) if dup.scalar_one_or_none(): slug = f"{slug}-{cin7_id}" sku_check = await self.db.execute( select(Product).where( Product.tenant_id == self.tenant_id, Product.sku == sku_code, ) ) if sku_check.scalar_one_or_none(): sku_code = f"{sku_code}-{cin7_id}" if self.auto_delist_zero_stock and total_stock == 0 and not options: cin7_status = "draft" product = Product( tenant_id=self.tenant_id, name=name, slug=slug, sku=sku_code, description=description, base_price=base_price if self.sync_price else Decimal("0"), market_price=market_price if self.sync_price else None, cost_price=cost_price if self.sync_price else None, stock_qty=total_stock if not options else 0, weight=weight if weight > 0 else None, status=cin7_status, brand_id=brand_id, category_id=category_id, cover_url=cover_url, extra_attributes={"cin7": {"product_id": cin7_id}}, ) self.db.add(product) await self.db.flush() # 图片 for idx, img in enumerate(images[:15]): img_url = img.get("link", "") if isinstance(img, dict) else str(img) if img_url: pi = ProductImage( product_id=product.id, tenant_id=self.tenant_id, url=img_url, sort_order=idx, is_primary=1 if idx == 0 else 0, ) self.db.add(pi) # 分类关联表 await self._sync_product_categories(product.id, category_ids) await self._create_map("product", cin7_id, product.id) prod_map = await self._find_map("product", cin7_id) if prod_map: prod_map.extra_data = {"cover_url": cover_url} # SPU 阶梯价 spu_price_cols = dict(price_cols) if sp.get("vipPrice"): spu_price_cols.setdefault("vipPrice", sp["vipPrice"]) await self._sync_tier_prices(product, spu_price_cols, variant_id=None) if options: await self._sync_variants(product, options, sp) await self._refresh_product_stock(product) return "created" async def _sync_variants(self, product: Product, options: list[dict], sp: dict): for opt in options: opt_id = str(opt.get("id")) if not opt_id: continue opt_code = _str(opt.get("code") or opt.get("productOptionCode")) barcode = _str(opt.get("productOptionBarcode") or opt.get("barCode")) sku_str = opt_code or barcode or f"{product.sku}-{opt_id}" attrs = {} for n in range(1, 4): val = _str(opt.get(f"option{n}")) if val: label = _str(opt.get(f"optionLabel{n}")) or f"option{n}" attrs[label] = val opt_price_cols = opt.get("priceColumns") or {} # 变体售价：按 price_col_default 取，计算与 SPU 的差价 if self.price_col_default: pc_lower = {k.lower(): v for k, v in opt_price_cols.items()} retail = _d( opt_price_cols.get(self.price_col_default) or pc_lower.get(self.price_col_default.lower()) or opt.get(self.price_col_default) ) if retail <= 0: retail = _d(opt.get("retailPrice") or sp.get("retailPrice") or sp.get("sellingPrice")) else: retail = _d(opt.get("retailPrice") or sp.get("retailPrice") or sp.get("sellingPrice")) price_modifier = retail - product.base_price if self.sync_price and retail > 0 else Decimal("0") stock = _int(opt.get("stockAvailable")) weight = _d(opt.get("optionWeight"), "0") pic_url = "" img_link = opt.get("image") if img_link: pic_url = img_link.get("link", "") if isinstance(img_link, dict) else str(img_link) existing_map = await self._find_map("variant", opt_id) # 合并 SPU 价格列到变体（vipPrice 等 SPU 级别字段） var_price_cols = dict(opt_price_cols) for k in ("vipPrice",): if sp.get(k) and k not in var_price_cols: var_price_cols[k] = sp[k] if existing_map: result = await self.db.execute( select(ProductVariant).where(ProductVariant.id == existing_map.local_id) ) variant = result.scalar_one_or_none() if variant: variant.stock_qty = stock if self.sync_price: variant.price_modifier = price_modifier variant.barcode = barcode or variant.barcode if attrs: variant.attributes = attrs if weight > 0: variant.weight = weight if pic_url: variant.image_url = pic_url existing_map.cin7_updated_at = datetime.now(timezone.utc) await self._sync_tier_prices(product, var_price_cols, variant_id=variant.id) else: sku_check = await self.db.execute( select(ProductVariant).where( ProductVariant.tenant_id == self.tenant_id, ProductVariant.sku == sku_str, ) ) if sku_check.scalar_one_or_none(): sku_str = f"{sku_str}-{opt_id}" variant = ProductVariant( product_id=product.id, tenant_id=self.tenant_id, sku=sku_str, barcode=barcode, price_modifier=price_modifier if self.sync_price else Decimal("0"), stock_qty=stock, weight=weight if weight > 0 else None, image_url=pic_url, attributes=attrs or {"default": "default"}, is_active=1, ) self.db.add(variant) await self.db.flush() await self._create_map("variant", opt_id, variant.id) await self._sync_tier_prices(product, var_price_cols, variant_id=variant.id) await self.db.flush() await self._refresh_product_stock(product) async def _sync_tier_prices(self, product: Product, price_columns: dict, variant_id: int | None = None): """清空后重新写入阶梯价（Cin7 为准）。""" if not self.level_mapping or not self.sync_price: return # 清空该 product + variant 组合的旧阶梯价 await self.db.execute( sa_delete(ProductTierPrice).where( ProductTierPrice.product_id == product.id, ProductTierPrice.variant_id == variant_id, ) ) pc_lower = {k.lower(): v for k, v in price_columns.items()} for col_key, level_id in self.level_mapping.items(): raw_val = price_columns.get(col_key) or pc_lower.get(col_key.lower()) if raw_val is None: continue price = _d(raw_val) if price <= 0: continue tp = ProductTierPrice( tenant_id=self.tenant_id, product_id=product.id, variant_id=variant_id, member_level_id=level_id, price=price, ) self.db.add(tp) async def _refresh_product_stock(self, product: Product): from sqlalchemy import func result = await self.db.execute( select(func.sum(ProductVariant.stock_qty)).where( ProductVariant.product_id == product.id, ProductVariant.is_active == 1, ) ) total = result.scalar() or 0 product.stock_qty = int(total) if not self.publish_field: if self.auto_delist_zero_stock and total == 0: product.status = "draft" elif product.status == "draft" and total > 0: product.status = "active" # ── 客户同步 ───────────────────────────────────────────────── async def sync_contacts(self, mode: str = "manual") -> dict: logger.info("[cin7_import] sync_contacts start, mode=%s", mode) try: raw_contacts = await self.client.get_contacts() except Exception as e: await self._write_log("contacts", mode, "failed", 0, 0, 0, 0, str(e)[:500]) await self.db.commit() return {"resource": "contacts", "created": 0, "updated": 0, "skipped": 0, "failed": 0, "error": str(e)} logger.info("[cin7_import] fetched %d Customer contacts from Cin7", len(raw_contacts)) created = updated = skipped = failed = 0 for contact in raw_contacts: try: cin7_id = str(contact.get("id")) if not cin7_id: skipped += 1 continue email = _str(contact.get("email")) if not email: email = f"cin7-{cin7_id}@placeholder.local" first = _str(contact.get("firstName")) last = _str(contact.get("lastName")) name = f"{first} {last}".strip() or email.split("@")[0] phone = _str(contact.get("mobile") or contact.get("phone")) company = _str(contact.get("company")) price_column = _str(contact.get("priceColumn")) # 大小写不敏感匹配 level_mapping（Cin7 返回的大小写与配置可能不一致） level_id = None if price_column: lm_lower = {k.lower(): v for k, v in self.level_mapping.items()} level_id = lm_lower.get(price_column.lower()) existing_map = await self._find_map("customer", cin7_id) if existing_map: result = await self.db.execute( select(Customer).where(Customer.id == existing_map.local_id) ) customer = result.scalar_one_or_none() if not customer: # 客户已被删除，清掉旧 map，走新建流程 await self.db.delete(existing_map) await self.db.flush() existing_map = None if existing_map: customer.name = name customer.phone = phone or customer.phone if level_id: customer.member_level_id = level_id if not email.endswith("@placeholder.local"): old_placeholder = (customer.email or "").endswith("@placeholder.local") if old_placeholder or customer.email != email: customer.email = email customer.set_attribute("cin7", { "contact_id": cin7_id, "company": company, "type": _str(contact.get("type")), "priceColumn": price_column, }) if company: customer.set_attribute("company", company) existing_map.cin7_updated_at = datetime.now(timezone.utc) updated += 1 else: cust_result = await self.db.execute( select(Customer).where( Customer.tenant_id == self.tenant_id, Customer.email == email, ) ) customer = cust_result.scalar_one_or_none() if customer: customer.name = name customer.phone = phone or customer.phone if level_id: customer.member_level_id = level_id customer.set_attribute("cin7", { "contact_id": cin7_id, "company": company, "priceColumn": price_column, }) if company: customer.set_attribute("company", company) updated += 1 else: has_real_email = not email.endswith("@placeholder.local") extra = {"cin7": {"contact_id": cin7_id, "company": company}} if company: extra["company"] = company if has_real_email: raw_password = _generate_password() extra["password_hash"] = _hash_pwd(raw_password) customer = Customer( tenant_id=self.tenant_id, email=email, name=name, phone=phone, member_level_id=level_id, extra_data=extra, ) self.db.add(customer) await self.db.flush() created += 1 await self._create_map("customer", cin7_id, customer.id) await self.db.flush() # 同步地址 addr1 = _str(contact.get("address1")) addr2 = _str(contact.get("address2")) cin7_city = _str(contact.get("city")) cin7_state = _str(contact.get("state")) cin7_postcode = _str(contact.get("postCode")) cin7_country = _str(contact.get("country") or contact.get("countryCode")) or "NZ" if len(cin7_country) > 2: cin7_country = cin7_country[:2].upper() street = " ".join(filter(None, [addr1, addr2])) if street or cin7_city: # 找已有的 cin7 同步地址（extra_fields 含 cin7_synced） ar = await self.db.execute( select(CustomerAddress).where( CustomerAddress.customer_id == customer.id ) ) existing_addr = next( (a for a in ar.scalars().all() if isinstance(a.extra_fields, dict) and a.extra_fields.get("cin7_synced")), None, ) ef = {"cin7_synced": True} if company: ef["company"] = company if existing_addr: existing_addr.name = name existing_addr.phone = phone or existing_addr.phone or "" existing_addr.street = street or existing_addr.street existing_addr.city = cin7_city or existing_addr.city existing_addr.province = cin7_state or existing_addr.province existing_addr.zip_code = cin7_postcode or existing_addr.zip_code existing_addr.country = cin7_country existing_addr.extra_fields = ef else: new_addr = CustomerAddress( tenant_id=self.tenant_id, customer_id=customer.id, name=name, phone=phone or "", street=street, city=cin7_city, province=cin7_state, zip_code=cin7_postcode, country=cin7_country, is_default=1, extra_fields=ef, ) self.db.add(new_addr) await self.db.flush() except Exception as e: logger.warning("[cin7_import] contact sync failed cin7_id=%s: %s", contact.get("id"), e) await self.db.rollback() failed += 1 await self._update_sync_state("contacts") await self._write_log("contacts", mode, "success", created, updated, skipped, failed) await self.db.commit() logger.info("[cin7_import] contacts done: created=%d updated=%d skipped=%d failed=%d", created, updated, skipped, failed) return {"resource": "contacts", "created": created, "updated": updated, "skipped": skipped, "failed": failed} # ── 全量同步 ───────────────────────────────────────────────── async def sync_all(self, mode: str = "manual") -> dict: results = {} results["categories"] = await self.sync_categories(mode) results["brands"] = await self.sync_brands(mode) results["products"] = await self.sync_products(mode) results["contacts"] = await self.sync_contacts(mode) return results