""" Resolve master-catalog variation axes to downstream channel attribute codes and labels. Master relations store canonical option_mapping keys (e.g. variation_width_in). Channel push and the Variation Builder UI resolve those through channel_attribute_alias and attribute registries so Magento width/height and Shopify options stay in sync with mappings. """ from __future__ import annotations from typing import Any, Dict, Iterable, List, Optional, Sequence from sqlalchemy import select from sqlalchemy.orm import Session from db.channel_exports import resolve_pipeline_channel_code from db.compat_connections import decode_compat_connection_id from db.models import MagentoAttributeRegistry, ShopifyAttributeRegistry, ShopifyConnection from db.source_imports import normalize_column_key # Fallback when no alias row exists (Magento historical defaults). _CANONICAL_AXIS_DEFAULTS: Dict[str, Dict[str, str]] = { "magento": { "variation_width_in": "width", "variation_height_in": "height", "variation_depth_in": "depth", "variation_angle": "angle", "width_in": "width", "height_in": "height", "depth_in": "depth", "angle": "angle", }, } # Short Magento-facing labels for single-axis configurables. _CANONICAL_AXIS_LABELS: Dict[str, str] = { "variation_width_in": "Width", "variation_height_in": "Height", "variation_depth_in": "Depth", "variation_angle": "Angle", "variation_drawer_count": "Drawer Count", "variation_door_handing": "Door Handing", "variation_panel_type": "Panel Type", "variation_variant_code": "Variant", "width": "Width", "height": "Height", "depth": "Depth", "angle": "Angle", "width_in": "Width", "height_in": "Height", "depth_in": "Depth", } _DIMENSION_CANONICAL_KEYS = ( "variation_width_in", "variation_height_in", "variation_depth_in", "variation_angle", "width_in", "height_in", "depth_in", "width", "height", "depth", "angle", ) _NON_AXIS_OPTION_KEYS = frozenset( { "variation_option_label", "item_size", } ) class ChannelVariationAxisResolver: """Maps canonical variation axes → channel attribute codes + display labels.""" def __init__( self, session: Session, channel_code: str, *, connection_id: Optional[int] = None, ) -> None: self._session = session self._channel = resolve_pipeline_channel_code(channel_code) self._connection_id = connection_id self._alias_cache: Optional[Dict[str, str]] = None self._channel_attr_labels: Optional[Dict[str, str]] = None @property def channel_code(self) -> str: return self._channel def resolve_canonical_axis(self, canonical_code: str) -> str: """Canonical master attribute code → channel attribute code.""" code = normalize_column_key(canonical_code) if not code: return code aliases = self._canonical_to_channel_map() if code in aliases: return aliases[code] defaults = _CANONICAL_AXIS_DEFAULTS.get(self._channel, {}) if code in defaults: return defaults[code] return code def axis_label(self, channel_axis_code: str, *, canonical_code: Optional[str] = None) -> str: """Display label for a channel axis (short dimension labels win for Magento).""" channel_axis = normalize_column_key(channel_axis_code) canon = normalize_column_key(canonical_code or "") # Prefer short Width/Height/Depth/Angle for known dimension axes. if canon and canon in _CANONICAL_AXIS_LABELS: return _CANONICAL_AXIS_LABELS[canon] if channel_axis in _CANONICAL_AXIS_LABELS: return _CANONICAL_AXIS_LABELS[channel_axis] registry_labels = self._load_channel_attr_labels() label = registry_labels.get(channel_axis) if label: return label return channel_axis.replace("_", " ").title() def map_option_mapping(self, canonical_mapping: Dict[str, str]) -> Dict[str, str]: """Rewrite option_mapping keys to channel attribute codes; values unchanged.""" mapped: Dict[str, str] = {} for key, value in self._canonical_option_mapping(canonical_mapping).items(): channel_key = self.resolve_canonical_axis(str(key)) if channel_key and channel_key not in _NON_AXIS_OPTION_KEYS: mapped[channel_key] = str(value).strip() return mapped def channel_option_attrs(self, canonical_attrs: Sequence[str]) -> List[str]: """Ordered unique channel axis codes for canonical option attrs.""" seen: set[str] = set() out: List[str] = [] for canonical in canonical_attrs: code = normalize_column_key(canonical) if not code or code in _NON_AXIS_OPTION_KEYS: continue channel_axis = self.resolve_canonical_axis(code) if channel_axis and channel_axis not in _NON_AXIS_OPTION_KEYS and channel_axis not in seen: seen.add(channel_axis) out.append(channel_axis) return out def variation_labels_string( self, channel_axes: Sequence[str], *, canonical_axes: Optional[Sequence[str]] = None, ) -> str: canon_list = list(canonical_axes or []) parts: List[str] = [] for index, axis in enumerate(channel_axes): canon = canon_list[index] if index < len(canon_list) else None parts.append(f"{axis}={self.axis_label(axis, canonical_code=canon)}") return ",".join(parts) def option_axis_label(self, canonical_attrs: Sequence[str]) -> str: """Short combined label for UI (e.g. 'Width / Height').""" names = [ self.axis_label( self.resolve_canonical_axis(code), canonical_code=code, ) for code in canonical_attrs ] return " / ".join(names) def build_relation_fields( self, child_skus: Sequence[str], option_mappings: Sequence[Dict[str, str]], ) -> Dict[str, str]: """Magento-style relation fields using channel attribute codes.""" child_list = [str(s).strip() for s in child_skus if str(s).strip()] if not child_list: return {} normalized_mappings = [self._canonical_option_mapping(m) for m in option_mappings] master_attrs: List[str] = [] seen_master_attrs: set[str] = set() for mapping in normalized_mappings: for key in (mapping or {}).keys(): canon_key = normalize_column_key(key) if ( not canon_key or canon_key in seen_master_attrs or canon_key in {"nan", "none", "null"} ): continue seen_master_attrs.add(canon_key) master_attrs.append(canon_key) channel_attrs = self.channel_option_attrs(master_attrs) cv_parts: List[str] = [] for child_sku, mapping in zip(child_list, normalized_mappings): channel_map = self.map_option_mapping(mapping or {}) if not channel_map: continue pairs = [f"{k}={v}" for k, v in sorted(channel_map.items())] cv_parts.append(f"sku={child_sku}," + ",".join(pairs)) fields: Dict[str, str] = { "variant_list": ",".join(child_list), "product_type": "configurable", } if channel_attrs: joined = ",".join(channel_attrs) fields["configurable_attributes"] = joined fields["configurable_variation_axis"] = joined fields["configurable_variation_labels"] = self.variation_labels_string( channel_attrs, canonical_axes=master_attrs, ) if cv_parts: fields["configurable_variations"] = "|".join(cv_parts) return fields def enrich_variation_group( self, group: Dict[str, Any], *, include_relation_preview: bool = True, ) -> Dict[str, Any]: """Add channel-specific axis/mapping view to a variation builder group dict.""" canonical_attrs = list(group.get("option_attrs") or []) channel_attrs = self.channel_option_attrs(canonical_attrs) children = group.get("children") or [] child_skus = [str(c.get("sku") or "").strip() for c in children] option_mappings = [dict(c.get("option_mapping") or {}) for c in children] channel_children = [] for child in children: canonical_map = self._canonical_option_mapping(dict(child.get("option_mapping") or {})) channel_children.append( { **child, "channel_option_mapping": self.map_option_mapping(canonical_map), } ) view: Dict[str, Any] = { "channel_code": self._channel, "connection_id": self._connection_id, "option_attrs": channel_attrs, "canonical_option_attrs": canonical_attrs, "option_label": self.option_axis_label(canonical_attrs), "variation_labels": self.variation_labels_string(channel_attrs, canonical_axes=canonical_attrs), "children": channel_children, } if include_relation_preview and child_skus: view["relation_fields"] = self.build_relation_fields(child_skus, option_mappings) return view def _canonical_option_mapping(self, canonical_mapping: Optional[Dict[str, str]]) -> Dict[str, str]: """Keep all real axes, preferring dimension-style keys first for stable output.""" mapping = canonical_mapping or {} normalized: Dict[str, str] = {} for key in _DIMENSION_CANONICAL_KEYS: canon_key = normalize_column_key(key) value = mapping.get(canon_key) if value in (None, ""): continue text = str(value).strip() if text and text.lower() not in {"nan", "none", "null"}: normalized[canon_key] = text for key, value in mapping.items(): canon_key = normalize_column_key(str(key)) if canon_key in _NON_AXIS_OPTION_KEYS or canon_key in normalized or value in (None, ""): continue text = str(value).strip() if text and text.lower() not in {"nan", "none", "null"}: normalized[canon_key] = text return normalized def _canonical_to_channel_map(self) -> Dict[str, str]: if self._alias_cache is not None: return self._alias_cache mapping: Dict[str, str] = {} try: from db.channel_aliases import channel_aliases_by_canonical alias_rows = channel_aliases_by_canonical(self._session, self._channel) for canonical, aliases in alias_rows.items(): for alias in aliases: channel_attr = normalize_column_key( str(getattr(alias, "channel_attribute_code", "") or "") ) if channel_attr: mapping[normalize_column_key(canonical)] = channel_attr break except Exception: mapping = dict(mapping) self._alias_cache = mapping return mapping def _load_channel_attr_labels(self) -> Dict[str, str]: if self._channel_attr_labels is not None: return self._channel_attr_labels labels: Dict[str, str] = {} try: if self._channel == "magento": labels = self._magento_attr_labels() elif self._channel == "shopify": labels = self._shopify_attr_labels() except Exception: labels = {} self._channel_attr_labels = labels return labels def _magento_attr_labels(self) -> Dict[str, str]: native_id = self._native_magento_connection_id() stmt = select( MagentoAttributeRegistry.attribute_code, MagentoAttributeRegistry.frontend_label, ) if native_id is not None: stmt = stmt.where(MagentoAttributeRegistry.connection_id == native_id) labels: Dict[str, str] = {} for code, frontend_label in self._session.execute(stmt).all(): attr_code = normalize_column_key(str(code or "")) label = str(frontend_label or "").strip() if attr_code and label: labels[attr_code] = label return labels def _shopify_attr_labels(self) -> Dict[str, str]: shop_code = self._shopify_shop_code() stmt = select(ShopifyAttributeRegistry.attribute_code, ShopifyAttributeRegistry.name) if shop_code: stmt = stmt.where(ShopifyAttributeRegistry.shop_code == shop_code) labels: Dict[str, str] = {} for code, name in self._session.execute(stmt).all(): attr_code = normalize_column_key(str(code or "")) label = str(name or "").strip() if attr_code and label: labels[attr_code] = label return labels def _native_magento_connection_id(self) -> Optional[int]: if self._connection_id is None: return None channel_type, native_id = decode_compat_connection_id(self._connection_id) return native_id if channel_type == "magento" else None def _shopify_shop_code(self) -> Optional[str]: if self._connection_id is None: return None channel_type, native_id = decode_compat_connection_id(self._connection_id) if channel_type != "shopify": return None return self._session.scalar( select(ShopifyConnection.shop_code).where(ShopifyConnection.id == native_id) ) def build_parent_relation_fields_for_channel( session: Session, parent_skus: Iterable[str], *, channel_code: str, connection_id: Optional[int] = None, ) -> Dict[str, Dict[str, str]]: """Relation fields keyed by parent SKU for a specific downstream channel.""" from db.master_relations import children_by_parent resolver = ChannelVariationAxisResolver(session, channel_code, connection_id=connection_id) grouped = children_by_parent(session, parent_skus) fields_by_parent: Dict[str, Dict[str, str]] = {} for parent_sku, children in grouped.items(): if not children: continue child_skus = [c["child_sku"] for c in children] option_mappings = [c.get("option_mapping") or {} for c in children] fields = resolver.build_relation_fields(child_skus, option_mappings) if fields: fields_by_parent[parent_sku] = fields return fields_by_parent def variation_channel_views( session: Session, group: Dict[str, Any], channel_codes: Sequence[str], *, connection_ids: Optional[Dict[str, Optional[int]]] = None, ) -> Dict[str, Dict[str, Any]]: """Build per-channel variation preview slices for the Variation Builder UI.""" connection_ids = connection_ids or {} views: Dict[str, Dict[str, Any]] = {} for channel in channel_codes: code = resolve_pipeline_channel_code(channel) if not code: continue resolver = ChannelVariationAxisResolver( session, code, connection_id=connection_ids.get(code), ) views[code] = resolver.enrich_variation_group(group) return views