"""Map product snapshot / normalized row to Magento REST payload.""" from __future__ import annotations import re from typing import Any, Dict, List, Optional, Set from normalize import _parse_kv_blob, _relation_sku _NON_AXIS_ATTRIBUTE_CODES = frozenset({"variation_option_label", "item_size"}) # Stale Variation Builder / SEO overlays that must never become Magento product names. _POLLUTED_TITLE_TOKENS = ( "variation_option_label", "variation option label", "item_size", ) _DIMENSION_AXIS_CODES = frozenset( { "width", "height", "depth", "variation_width_in", "variation_height_in", "variation_depth_in", "width_in", "height_in", "depth_in", } ) _AXIS_DISPLAY_LABELS = { "width": "Width", "height": "Height", "depth": "Depth", "angle": "Angle", "variation_width_in": "Width", "variation_height_in": "Height", "variation_depth_in": "Depth", "variation_angle": "Angle", "width_in": "Width", "height_in": "Height", "depth_in": "Depth", } def get_axis_attribute_codes(row: Dict[str, Any]) -> Set[str]: """ Derive configurable axis attribute codes for a parent product. From configurable_attributes if present, else inferred from configurable_variations. Returns set of attribute codes (lowercase). """ codes: set[str] = set() attrs_raw = str(row.get("configurable_attributes", "")).strip() if attrs_raw: for part in attrs_raw.split(","): part = part.strip() if "=" in part: codes.add(part.split("=", 1)[0].strip().lower()) elif part: codes.add(part.lower()) if not codes: codes = _infer_axis_codes_from_variations(row) return {c for c in codes if c and c not in _NON_AXIS_ATTRIBUTE_CODES} def _infer_axis_codes_from_variations(row: Dict[str, Any]) -> set[str]: """Infer axis codes from configurable_variations keys (excluding sku).""" cv = str(row.get("configurable_variations", "")).strip() if not cv or "sku=" not in cv.lower(): return set() codes: set[str] = set() for seg in cv.split("|"): for pair in seg.split(","): if "=" in pair: k = pair.split("=", 1)[0].strip().lower() if k and k != "sku": codes.add(k) return codes # Magento visibility values VISIBILITY_NOT_VISIBLE = 1 VISIBILITY_CATALOG = 2 VISIBILITY_SEARCH = 3 VISIBILITY_CATALOG_SEARCH = 4 # Status STATUS_DISABLED = 0 STATUS_ENABLED = 1 _TOP_LEVEL_MAGENTO_PRODUCT_FIELDS = frozenset( { "name", "status", "visibility", "price", "weight", "attribute_set_id", "type_id", } ) # Never send these as custom_attributes — Magento expects them as top-level ints # (status 1/2, visibility 1–4). Master catalog "active" must not leak here. _CUSTOM_ATTR_DENYLIST = frozenset( { "status", "visibility", "sku", "name", "type_id", "attribute_set_id", "attribute_set_code", "price", "weight", } ) def _slugify(text: str) -> str: """Simple slug: lowercase, replace spaces/special with hyphen.""" s = text.lower().strip() s = re.sub(r"[^a-z0-9]+", "-", s) s = re.sub(r"-+", "-", s).strip("-") return s or "product" def _parse_categories(categories_raw: Optional[str]) -> List[str]: """Parse categories: / or comma separated paths.""" if not categories_raw or not str(categories_raw).strip(): return [] raw = str(categories_raw).strip() # Support both "Cat1/Cat2" and "Cat1, Cat2" parts = re.split(r"[/,]", raw) return [p.strip() for p in parts if p.strip()] def _dimension_axis_codes(axis_codes: Set[str]) -> Set[str]: return {code for code in axis_codes if code in _DIMENSION_AXIS_CODES} def _axis_display_label(code: str) -> str: return _AXIS_DISPLAY_LABELS.get(code, code.replace("_", " ").title()) def is_polluted_product_title(text: Optional[str]) -> bool: """True when a title/slug still contains Variation Builder template noise.""" lower = str(text or "").strip().lower() if not lower: return False return any(token in lower for token in _POLLUTED_TITLE_TOKENS) def resolve_product_display_name(row: Dict[str, Any], *, fallback: str = "") -> str: """ Magento product name source of truth: master catalog name/title. Stale meta_title/seo_title overlays must not win — they often still hold pre-repair Variation Builder templates or old Plytix SEO strings. """ for key in ("name", "title", "display_name"): value = str(row.get(key) or "").strip() if value and not is_polluted_product_title(value): return value for key in ("seo_title", "meta_title"): value = str(row.get(key) or "").strip() if value and not is_polluted_product_title(value): return value for key in ("name", "title", "display_name", "seo_title", "meta_title"): value = str(row.get(key) or "").strip() if value: # Last resort: strip polluted suffix before first '('. if is_polluted_product_title(value) and "(" in value: base = value.split("(", 1)[0].strip() if base and not is_polluted_product_title(base): return base return value return str(fallback or "").strip() def _compute_parent_neutral_name( row: Dict[str, Any], axis_codes: Set[str], rows_by_sku: Optional[Dict[str, Dict[str, Any]]] = None ) -> Optional[str]: """ For configurable parent: optional compact range suffix (e.g. 'Height 30–42'). Never lists composite variation_option_label values. Returns None to keep row name. """ from magento.sync_hashes import _parse_children_from_row name = resolve_product_display_name(row) dim_axes = _dimension_axis_codes(axis_codes - _NON_AXIS_ATTRIBUTE_CODES) if not name or not dim_axes or not rows_by_sku: return None children, mappings, _ = _parse_children_from_row(row) if not children or not mappings: return None axis_values: Dict[str, set[str]] = {code: set() for code in dim_axes} for mapping in mappings: for attr, val in mapping.items(): if attr in axis_values and val: axis_values[attr].add(str(val).strip()) ranges: List[str] = [] for attr in sorted(dim_axes): vals = sorted(axis_values.get(attr, [])) if len(vals) <= 1: continue try: nums = [float(v.replace(",", ".")) for v in vals] label = _axis_display_label(attr) lo = min(nums) hi = max(nums) ranges.append( f"{label} {int(lo) if lo.is_integer() else lo:.0f}–{int(hi) if hi.is_integer() else hi:.0f}" ) except (ValueError, TypeError): continue if not ranges: return None base = name.split("(")[0].strip() or name if is_polluted_product_title(base): return None return f"{base} ({', '.join(ranges)})" def snapshot_to_magento_product( row: Dict[str, Any], *, attribute_set_id: int = 4, default_visibility: int = VISIBILITY_CATALOG_SEARCH, default_status: int = STATUS_ENABLED, is_new: bool = True, include_stock: bool = False, price_override: Optional[float] = None, axis_exclude: Optional[Set[str]] = None, parent_name_override: Optional[str] = None, rows_by_sku: Optional[Dict[str, Dict[str, Any]]] = None, selected_attribute_codes: Optional[Set[str]] = None, ) -> Dict[str, Any]: """ Convert a product snapshot / normalized row to Magento product payload. row: dict with keys like sku, name, description, short_description, price, product_type, attribute_set_code, categories, visibility, product_online, base_image, and custom attrs. attribute_set_id: Magento attribute set ID (4 = Default commonly). url_key: never included — Magento auto-generates on create, keeps existing on update. include_stock: if False (default), never set extension_attributes.stock_item. """ sku = str(row.get("sku", "")).strip() if not sku: raise ValueError("sku is required") # Configurable parent: avoid single-axis name; use override or range; exclude axis attrs is_configurable_parent = "configurable" in str(row.get("product_type", "")).strip().lower() axis_codes: Set[str] = set() attrs_to_exclude: Set[str] = set() if is_configurable_parent: axis_codes = get_axis_attribute_codes(row) attrs_to_exclude = axis_codes | (axis_exclude or set()) name = resolve_product_display_name(row, fallback=sku) or sku if parent_name_override and not is_polluted_product_title(parent_name_override): name = parent_name_override elif is_configurable_parent and axis_codes and rows_by_sku: neutral = _compute_parent_neutral_name(row, axis_codes, rows_by_sku) if neutral: name = neutral from channel.url_canonical import build_pdp_slug # Always derive url_key from the Magento name we send. Preserving stale # product_url_slug/meta_title-based slugs left Magento PDPs on old URLs. if is_configurable_parent: slug_base = name.split("(")[0].strip() or name resolved_url_key = build_pdp_slug(slug_base, sku) else: resolved_url_key = build_pdp_slug(name, sku) product_type = str(row.get("product_type", "simple")).strip().lower() if product_type not in ("simple", "configurable", "virtual", "downloadable", "bundle"): # Plytix may send "configurable product" or "Configurable" etc. product_type = "configurable" if "configurable" in product_type else "simple" product_online = str(row.get("product_online", "1")).strip() variant_of = _relation_sku(row.get("variant_of")) # Children: enabled but not visible individually. Parents: catalog + search. is_configurable_child = bool(variant_of) visibility_raw = str(row.get("visibility", "")).strip() visibility = default_visibility if is_configurable_child: visibility = VISIBILITY_NOT_VISIBLE elif is_configurable_parent: visibility = VISIBILITY_CATALOG_SEARCH elif "catalog" in visibility_raw.lower() and "search" in visibility_raw.lower(): visibility = VISIBILITY_CATALOG_SEARCH elif "catalog" in visibility_raw.lower(): visibility = VISIBILITY_CATALOG elif "search" in visibility_raw.lower(): visibility = VISIBILITY_SEARCH elif "not visible" in visibility_raw.lower(): visibility = VISIBILITY_NOT_VISIBLE # Configurable parents (shell products) must be enabled or product page won't show # A1: When MAGENTO_ALWAYS_ENABLED_IGNORE_FEED=true, ignore product_online; always enabled. # Only disable via magento_product_override (force.status=0). try: from settings import magento_always_enabled_ignore_feed always_enabled = magento_always_enabled_ignore_feed() except Exception: always_enabled = True if always_enabled: status = STATUS_ENABLED else: status = ( STATUS_DISABLED if (not is_configurable_child and not is_configurable_parent and product_online in ("0", "false", "no")) else STATUS_ENABLED ) # Price: when price_override is set (new product or Magento has $0), include it. # Otherwise Magento manages price; we do not overwrite. # Stock is managed in Magento; DO NOT write extension_attributes.stock_item. payload: Dict[str, Any] = { "sku": sku, "name": name, "type_id": product_type, "attribute_set_id": attribute_set_id, "status": status, "visibility": visibility, "weight": "1", # Magento requires weight for simple } # url_key: never sent. Magento auto-generates on create; keeps existing on update. # Description / short_description as custom_attributes custom_attrs: List[Dict[str, Any]] = [] desc = row.get("description") if desc is not None and str(desc).strip(): custom_attrs.append({"attribute_code": "description", "value": str(desc)}) short_desc = row.get("short_description") if short_desc is not None and str(short_desc).strip(): custom_attrs.append({"attribute_code": "short_description", "value": str(short_desc)}) meta_title_raw = str(row.get("meta_title") or row.get("seo_title") or "").strip() meta_title = meta_title_raw if meta_title_raw and not is_polluted_product_title(meta_title_raw) else name if meta_title: custom_attrs.append({"attribute_code": "meta_title", "value": str(meta_title)[:70]}) url_slug = resolved_url_key if url_slug: custom_attrs.append({"attribute_code": "url_key", "value": str(url_slug).strip()}) # Attributes that Magento expects as int[] (not string) int_array_attrs = {"store_ids", "store_id"} # Pack other attributes from row (excluding core fields) # configurable_variations, configurable_attributes are relation data, not product attrs # category_ids: injected via extension_attributes.category_links, never in custom_attributes # additional_attributes: expanded to individual attrs below; never sent as single blob core_keys = { "sku", "name", "description", "short_description", "price", "special_price", "tier_prices", "tier_price", "product_type", "attribute_set_code", "attribute_set_id", "visibility", "product_online", "status", "categories", "categories_raw", "category_ids", # Use extension_attributes.category_links only "base_image", "additional_images", "variant_of", "variant_list", "configurable_variations", "configurable_attributes", "meta_title", "url_key", "additional_attributes", # expanded to individual attrs } def _to_int_array(val: object, attr: str) -> Optional[List[int]]: if val is None: return None if isinstance(val, list): return [int(x) for x in val if str(x).strip().isdigit()] s = str(val).strip().strip("[]") if not s: return None return [int(x.strip()) for x in s.split(",") if x.strip().isdigit()] exclude_attrs = attrs_to_exclude # Expand additional_attributes blob (key=val,key2=val2) into individual attrs. # All values come from Plytix; Magento value_index resolution happens at sync time. # Never inject attribute_set_code/attribute_set_id from feed - always use Default via param. ATTRIBUTE_SET_KEYS = {"attribute_set_code", "attribute_set_id"} added_from_blob: Set[str] = set() add_attrs_raw = row.get("additional_attributes") if add_attrs_raw is not None and str(add_attrs_raw).strip(): parsed = _parse_kv_blob(add_attrs_raw) for k, v in parsed.items(): if not k or v is None or str(v).strip() == "": continue code = k.strip().lower() if ( code in exclude_attrs or code in ATTRIBUTE_SET_KEYS or code in _CUSTOM_ATTR_DENYLIST or code in core_keys ): continue custom_attrs.append({"attribute_code": code, "value": str(v).strip()}) added_from_blob.add(code) for k, v in row.items(): if v is None: continue code = k.strip().lower() if isinstance(k, str) else str(k).strip().lower() if ( k in core_keys or code in core_keys or code in exclude_attrs or code in added_from_blob or code in _CUSTOM_ATTR_DENYLIST ): continue if k in int_array_attrs or code in int_array_attrs: ids = _to_int_array(v, k) if ids: custom_attrs.append({"attribute_code": code, "value": ids}) continue if isinstance(v, list): vs = v # Preserve list for other int[] attributes else: vs = str(v).strip() if vs == "" or (isinstance(vs, str) and vs.lower() in ("nan", "none", "null")): continue custom_attrs.append({"attribute_code": code, "value": vs}) if custom_attrs: payload["custom_attributes"] = custom_attrs # When price_override set (new product or Magento $0 fix), include price if price_override is not None and price_override >= 0: payload["price"] = str(round(price_override, 2)) # Strip Magento-managed fields (never push from Plytix) unless we override if price_override is None: payload.pop("price", None) payload.pop("special_price", None) payload.pop("tier_prices", None) payload.pop("url_key", None) if payload.get("custom_attributes"): excluded_attrs = { "price", "special_price", "tier_price", "tier_prices", "category_ids", # Use extension_attributes.category_links only *_CUSTOM_ATTR_DENYLIST, } if not resolved_url_key: excluded_attrs.add("url_key") payload["custom_attributes"] = [ a for a in payload["custom_attributes"] if isinstance(a, dict) and str(a.get("attribute_code") or "").strip().lower() not in excluded_attrs ] if not payload["custom_attributes"]: del payload["custom_attributes"] if payload.get("extension_attributes"): payload["extension_attributes"].pop("stock_item", None) if not payload["extension_attributes"]: del payload["extension_attributes"] # Never set extension_attributes.stock_item when include_stock=False (default). # Plytix does not manage stock; Magento manages it locally. # Categories - need category_id; Magento expects IDs. For MVP we skip # category_links or pass empty; can enrich later with category lookup. cat_raw = row.get("categories") or row.get("categories_raw") if cat_raw and str(cat_raw).strip(): # Placeholder - real impl would resolve category paths to IDs # payload["extension_attributes"]["category_links"] = [] pass if selected_attribute_codes: payload = _filter_magento_product_payload(payload, selected_attribute_codes) return payload def _filter_magento_product_payload( payload: Dict[str, Any], selected_attribute_codes: Set[str], ) -> Dict[str, Any]: wanted = {str(code).strip().lower() for code in (selected_attribute_codes or set()) if str(code).strip()} if not wanted: return payload filtered: Dict[str, Any] = {} sku = payload.get("sku") if sku is not None: filtered["sku"] = sku for key in _TOP_LEVEL_MAGENTO_PRODUCT_FIELDS: if key in wanted and key in payload: filtered[key] = payload[key] custom_attrs = [ attr for attr in (payload.get("custom_attributes") or []) if isinstance(attr, dict) and str(attr.get("attribute_code") or "").strip().lower() in wanted ] if custom_attrs: filtered["custom_attributes"] = custom_attrs return filtered