from __future__ import annotations import hashlib import json import re from typing import Any, Dict, List, Optional, Sequence, Tuple import sqlalchemy as sa from sqlalchemy import select from sqlalchemy.dialects.postgresql import insert from sqlalchemy.orm import Session from db.tribeca_sku_parse import parse_master_raw_sku from db.channel_assignments import assign_skus from db.channel_sku_mapping import ACTIVE_STATUS as SKU_MAPPING_ACTIVE_STATUS from db.channel_sku_mapping import upsert_mappings from db.master_product_filters import apply_master_attribute_filters, parse_master_filters from db.master_relations import upsert_relations from db.models import MasterProduct, MasterProductAttributeValue, MasterProductRelation from db.source_imports import normalize_column_key DEFAULT_GROUP_ATTRS = ( "collection", "product_family", "cabinet_type_sub_category_1", "cabinet_type_sub_category_2", "variation_group_code", "style_family", "assembly_type", ) DEFAULT_OPTION_ATTRS = ( "variation_width_in", "variation_height_in", "variation_depth_in", "variation_angle", "variation_drawer_count", "variation_door_handing", "variation_panel_type", "variation_variant_code", "item_size", ) NON_AXIS_OPTION_ATTRS = frozenset({"item_size", "variation_option_label"}) PARENT_DESCRIPTOR_ATTRS = ( "cabinet_type_sub_category_2", "cabinet_type_sub_category_1", "cabinet_type", "type_code", "category_l3", "category_l2", "product_family", ) # Collection-level attrs that are uniform across siblings (workbook-validated). # Copied onto variation parents when every child shares the same value. # Kept for backwards-compatible callers/tests; copy logic now takes *all* # non-configuration attrs that are unanimous across children. COLLECTION_SHAREABLE_ATTRS = ( "door_style", "cabinet_door_style", "color", "finish", "cabinet_door_finish_type", "cabinet_construction", "cabinet_construction_type", "cabinet_door_overlay", "style_family", "collection_style", ) DIMENSION_OPTION_ATTRS = ( "variation_width_in", "variation_height_in", "variation_depth_in", "variation_angle", ) # Never copy these from children onto parent shells. PARENT_ATTR_COPY_SKIP_CODES = frozenset( { *DEFAULT_OPTION_ATTRS, *DIMENSION_OPTION_ATTRS, *NON_AXIS_OPTION_ATTRS, "width", "height", "depth", "assembled_width", "assembled_height", "assembled_depth", "sku", "name", "title", "display_name", "meta_title", "seo_title", "meta_description", "meta_keywords", "product_url_slug", "url_key", "handle", "description", "short_description", "price", "special_price", "tier_price", "qty", "quantity", "stock", "base_image", "additional_images", "image_urls", "thumbnail", "small_image", } ) def preview_variation_plan( session: Session, *, skus: Sequence[str], group_attrs: Optional[Sequence[str]] = None, option_attrs: Optional[Sequence[str]] = None, parent_suffix: str = "CONFIG", min_children: int = 2, option_axis_count: Optional[int] = None, exclude_linked_skus: bool = False, channel_code: str = "magento", channel_codes: Optional[Sequence[str]] = None, connection_id: Optional[int] = None, connection_ids: Optional[Dict[str, Optional[int]]] = None, ) -> Dict[str, Any]: products = _products(session, skus) skipped: List[Dict[str, Any]] = [] excluded_linked_count = 0 if exclude_linked_skus and products: linked = _relation_linked_skus(session, [product.sku for product in products]) if linked: excluded_linked_count = sum(1 for product in products if product.sku in linked) skipped.extend( {"sku": product.sku, "reason": "already_linked_relation"} for product in products if product.sku in linked ) products = [product for product in products if product.sku not in linked] attrs_by_sku = _attrs_by_sku(session, [p.id for p in products]) group_codes = _codes(group_attrs or DEFAULT_GROUP_ATTRS) option_codes = _codes(option_attrs or DEFAULT_OPTION_ATTRS) grouped: Dict[Tuple[Tuple[str, str], ...], List[MasterProduct]] = {} for product in products: values = _field_values(product, attrs_by_sku.get(product.sku, {}), group_codes) key = tuple((code, values.get(code) or "") for code in group_codes) grouped.setdefault(key, []).append(product) groups: List[Dict[str, Any]] = [] parent_skus: List[str] = [] group_index = 0 for key, members in sorted(grouped.items(), key=lambda item: _group_sort_key(item[1])): if len(members) < min_children: skipped.extend({"sku": p.sku, "reason": "group_below_min_children"} for p in members) continue attrs_for_members = {p.sku: attrs_by_sku.get(p.sku, {}) for p in members} family_subgroups = _subdivide_by_sku_family(members, attrs_for_members) for family_members in family_subgroups: if len(family_members) < min_children: skipped.extend({"sku": p.sku, "reason": "family_below_min_children"} for p in family_members) continue varying_option_codes = _varying_option_codes(family_members, attrs_for_members, option_codes) if not varying_option_codes: skipped.extend({"sku": p.sku, "reason": "no_varying_option_axis"} for p in family_members) continue dimension_axis_count = _varying_dimension_axis_count( family_members, attrs_for_members, DIMENSION_OPTION_ATTRS ) if option_axis_count is not None and dimension_axis_count != option_axis_count: skipped.extend( { "sku": p.sku, "reason": "option_axis_count_mismatch", "dimension_axis_count": dimension_axis_count, "requested_option_axis_count": option_axis_count, } for p in family_members ) continue group_index += 1 group_values = {code: value for code, value in key if value} sku_family = _sku_family_key(family_members[0], attrs_for_members.get(family_members[0].sku, {})) parent_sku = _parent_sku_for_group(family_members, group_values, parent_suffix=parent_suffix) parent_sku = _dedupe_parent_sku(parent_sku, parent_skus) parent_skus.append(parent_sku) groups.append( { "group_index": group_index, "parent_sku": parent_sku, "parent_name": _parent_name_for_group(family_members, group_values), "group_values": group_values, "sku_family": sku_family, "option_attrs": varying_option_codes, "dimension_axis_count": dimension_axis_count, "option_label": _option_axis_label(session, varying_option_codes, channel_code=channel_code, connection_id=connection_id), "children": [ { "sku": p.sku, "name": p.name, "option_mapping": _field_values(p, attrs_by_sku.get(p.sku, {}), varying_option_codes), } for p in sorted(family_members, key=lambda p: p.sku) ], } ) _attach_channel_variation_views( session, groups, channel_code=channel_code, channel_codes=channel_codes, connection_id=connection_id, connection_ids=connection_ids, ) return { "status": "ok", "input_skus": len(skus), "matched_skus": len(products), "group_attrs": group_codes, "option_attrs": option_codes, "option_axis_count": option_axis_count, "exclude_linked_skus": exclude_linked_skus, "excluded_linked_skus": excluded_linked_count, "group_count": len(groups), "relation_count": sum(len(group["children"]) for group in groups), "groups": groups, "skipped": skipped[:200], } def suggest_variation_plan( session: Session, *, category_l1: Optional[str] = "Kitchen Cabinets", product_family: Optional[str] = None, collection: Optional[str] = None, assembly_type: Optional[str] = "Assembled", search: Optional[str] = None, master_filters: Optional[Any] = None, group_attrs: Optional[Sequence[str]] = None, option_attrs: Optional[Sequence[str]] = None, parent_suffix: str = "CONFIG", min_children: int = 2, option_axis_count: Optional[int] = None, exclude_linked_skus: bool = False, limit: int = 10000, channel_code: str = "magento", channel_codes: Optional[Sequence[str]] = None, connection_id: Optional[int] = None, connection_ids: Optional[Dict[str, Optional[int]]] = None, ) -> Dict[str, Any]: """Find likely variation SKUs from the master catalog, then run the normal preview planner.""" stmt = select(MasterProduct.sku).where(MasterProduct.is_active.is_(True)) if category_l1: stmt = stmt.where(MasterProduct.category_l1.ilike(f"%{category_l1.strip()}%")) if product_family: stmt = stmt.where(MasterProduct.product_family.ilike(f"%{product_family.strip()}%")) if collection: stmt = stmt.where(MasterProduct.collection.ilike(f"%{collection.strip()}%")) if assembly_type: stmt = stmt.where(MasterProduct.assembly_type.ilike(f"%{assembly_type.strip()}%")) if search: pattern = f"%{search.strip()}%" stmt = stmt.where(sa.or_(MasterProduct.sku.ilike(pattern), MasterProduct.name.ilike(pattern))) stmt = apply_master_attribute_filters(stmt, filters=parse_master_filters(master_filters)) stmt = stmt.order_by(MasterProduct.sku).limit(max(1, min(int(limit or 10000), 25000))) skus = [sku for (sku,) in session.execute(stmt).all()] preview = preview_variation_plan( session, skus=skus, group_attrs=group_attrs, option_attrs=option_attrs, parent_suffix=parent_suffix, min_children=min_children, option_axis_count=option_axis_count, exclude_linked_skus=exclude_linked_skus, channel_code=channel_code, channel_codes=channel_codes, connection_id=connection_id, connection_ids=connection_ids, ) preview["candidate_filter"] = { "category_l1": category_l1, "product_family": product_family, "collection": collection, "assembly_type": assembly_type, "search": search, "option_axis_count": option_axis_count, "exclude_linked_skus": exclude_linked_skus, "limit": limit, } preview["candidate_sku_count"] = len(skus) return preview def commit_variation_plan( session: Session, *, groups: Sequence[Dict[str, Any]], source_label: str = "variation_builder", replace_existing: bool = False, assign_channels: Optional[Sequence[str]] = ("magento",), mapping_connection_ids: Optional[Dict[str, Optional[int]]] = None, create_identity_mappings: bool = True, ) -> Dict[str, Any]: parent_rows: List[Dict[str, Any]] = [] relation_rows: List[Dict[str, Any]] = [] for group in groups: parent_sku = str(group.get("parent_sku") or "").strip() children = group.get("children") or [] if not parent_sku or len(children) < 2: continue first_child = _first_child_product(session, children) parent_rows.append(_parent_row(parent_sku, group, first_child)) for order, child in enumerate(children): child_sku = str(child.get("sku") or "").strip() if not child_sku or child_sku == parent_sku: continue relation_rows.append( { "parent_sku": parent_sku, "child_sku": child_sku, "option_mapping": _axis_only_mapping(child.get("option_mapping") or {}), "sort_order": order, "source_label": source_label, } ) if not parent_rows: return { "status": "ok", "parents_upserted": 0, "relations_upserted": 0, "assignments_upserted": 0, "mappings_upserted": 0, } _upsert_parent_products(session, parent_rows) parent_skus = [row["sku"] for row in parent_rows] shareable_attrs_copied = _copy_collection_shareable_attrs_to_parents( session, parent_rows, source_label=f"{source_label}_shareable", ) assignments_upserted = _assign_parent_skus(session, parent_skus, assign_channels or []) mappings_upserted = ( _upsert_identity_mappings( session, parent_skus, assign_channels or [], mapping_connection_ids=mapping_connection_ids or {}, source_label=source_label, ) if create_identity_mappings else 0 ) if replace_existing: session.execute( sa.delete(MasterProductRelation).where(MasterProductRelation.parent_sku.in_(parent_skus)) ) relations_upserted = upsert_relations(session, relation_rows) return { "status": "ok", "parents_upserted": len(parent_rows), "relations_upserted": relations_upserted, "assignments_upserted": assignments_upserted, "mappings_upserted": mappings_upserted, "shareable_attrs_copied": shareable_attrs_copied, "parent_skus": parent_skus, } def _products(session: Session, skus: Sequence[str]) -> List[MasterProduct]: wanted = sorted({str(s).strip() for s in skus if str(s).strip()}) if not wanted: return [] return list( session.scalars( select(MasterProduct) .where(MasterProduct.is_active.is_(True)) .where(MasterProduct.sku.in_(wanted)) .order_by(MasterProduct.sku) ).all() ) def _attrs_by_sku(session: Session, product_ids: Sequence[int]) -> Dict[str, Dict[str, str]]: if not product_ids: return {} rows: Dict[str, Dict[str, str]] = {} for value in session.scalars( select(MasterProductAttributeValue).where(MasterProductAttributeValue.product_id.in_(list(product_ids))) ).all(): if value.value is None or value.value == "": continue rows.setdefault(value.sku, {})[normalize_column_key(value.attribute_code)] = str(value.value) return rows def _field_values(product: MasterProduct, attrs: Dict[str, str], codes: Sequence[str]) -> Dict[str, str]: values: Dict[str, str] = {} derived = _derived_option_values(product, attrs) for code in codes: value = _core_value(product, code) if value is None: value = attrs.get(code) if value is None: value = derived.get(normalize_column_key(code)) if value is not None and str(value).strip(): values[code] = str(value).strip() return values def _derived_option_values(product: MasterProduct, attrs: Dict[str, str]) -> Dict[str, str]: sku = str(getattr(product, "sku", "") or "").strip().upper() if not sku: return {} styled_sku = sku.split("-", 1)[1] if "-" in sku else sku derived: Dict[str, str] = {} raw_name = str(getattr(product, "name", "") or "") lower_name = raw_name.lower() drawer_match = re.search(r"-(\d+)$", sku) if drawer_match and "drawer" in lower_name: derived["variation_drawer_count"] = drawer_match.group(1) else: name_drawer_match = re.search(r"\b(\d+)\s*drawer\b", lower_name) if name_drawer_match: derived["variation_drawer_count"] = name_drawer_match.group(1) hand_match = re.search(r"(DLR|DL|DR|LR|RL)$", styled_sku) if hand_match: hand = hand_match.group(1) hand_labels = { "DL": "Left", "DR": "Right", "DLR": "Left/Right", "LR": "Left/Right", "RL": "Left/Right", } derived["variation_door_handing"] = hand_labels.get(hand, hand) if "GL" in styled_sku or " glass" in lower_name: derived["variation_panel_type"] = "Glass" elif re.match(r"^[A-Z]+-FPP", sku): derived["variation_panel_type"] = "Standard" parsed = parse_master_raw_sku(styled_sku) if parsed: variant_parts = [str(parsed.get("variant") or "").strip(), str(parsed.get("rest") or "").strip()] variant_parts = [part for part in variant_parts if part] if variant_parts: derived["variation_variant_code"] = "-".join(variant_parts) return derived def _relation_linked_skus(session: Session, skus: Sequence[str]) -> set[str]: wanted = sorted({str(sku).strip() for sku in skus if str(sku).strip()}) if not wanted: return set() child_rows = session.execute( select(MasterProductRelation.child_sku).where(MasterProductRelation.child_sku.in_(wanted)) ).all() parent_rows = session.execute( select(MasterProductRelation.parent_sku).where(MasterProductRelation.parent_sku.in_(wanted)) ).all() return {sku for (sku,) in child_rows} | {sku for (sku,) in parent_rows} def _subdivide_by_sku_family( members: Sequence[MasterProduct], attrs_by_sku: Dict[str, Dict[str, str]], ) -> List[List[MasterProduct]]: buckets: Dict[str, List[MasterProduct]] = {} for product in members: key = _sku_family_key(product, attrs_by_sku.get(product.sku, {})) buckets.setdefault(key, []).append(product) return list(buckets.values()) def _sku_family_key(product: MasterProduct, attrs: Dict[str, str]) -> str: group_code = (_field_values(product, attrs, ["variation_group_code"]).get("variation_group_code") or "").strip().upper() if group_code: return group_code return _sku_alpha_prefix(product.sku) def _varying_dimension_axis_count( members: Sequence[MasterProduct], attrs_by_sku: Dict[str, Dict[str, str]], dimension_codes: Sequence[str], ) -> int: return len(_varying_codes_for_members(members, attrs_by_sku, dimension_codes)) def _varying_codes_for_members( members: Sequence[MasterProduct], attrs_by_sku: Dict[str, Dict[str, str]], codes: Sequence[str], ) -> List[str]: varying: List[str] = [] for code in codes: values = { (_field_values(product, attrs_by_sku.get(product.sku, {}), [code]).get(code) or "").strip() for product in members } values.discard("") if len(values) > 1: varying.append(code) return varying def _varying_option_codes( members: Sequence[MasterProduct], attrs_by_sku: Dict[str, Dict[str, str]], option_codes: Sequence[str], ) -> List[str]: preferred = [code for code in option_codes if normalize_column_key(code) not in NON_AXIS_OPTION_ATTRS] if not preferred: return [] varying = _varying_codes_for_members(members, attrs_by_sku, preferred) if varying: return varying return [] def sanitize_option_attrs(option_attrs: Sequence[str]) -> List[str]: """Keep only Magento-usable configurable axes (drop variation_option_label / item_size).""" return [code for code in _codes(option_attrs) if code not in NON_AXIS_OPTION_ATTRS] def _option_axis_label( session: Session, option_codes: Sequence[str], *, channel_code: str = "magento", connection_id: Optional[int] = None, ) -> str: from db.channel_variation_axes import ChannelVariationAxisResolver return ChannelVariationAxisResolver(session, channel_code, connection_id=connection_id).option_axis_label( option_codes ) def _attach_channel_variation_views( session: Session, groups: List[Dict[str, Any]], *, channel_code: str, channel_codes: Optional[Sequence[str]], connection_id: Optional[int], connection_ids: Optional[Dict[str, Optional[int]]], ) -> None: from db.channel_variation_axes import ChannelVariationAxisResolver, variation_channel_views targets = list(channel_codes or [channel_code]) conn_ids = dict(connection_ids or {}) if connection_id is not None and channel_code not in conn_ids: conn_ids[channel_code] = connection_id for group in groups: if len(targets) == 1: resolver = ChannelVariationAxisResolver( session, targets[0], connection_id=conn_ids.get(targets[0]), ) group["channel"] = resolver.enrich_variation_group(group) else: group["channel_views"] = variation_channel_views( session, group, targets, connection_ids=conn_ids, ) def _core_value(product: MasterProduct, code: str) -> Optional[Any]: aliases = { "title": "name", "name": "name", "sku": "sku", "product_family": "product_family", "family": "product_family", "category_l1": "category_l1", "category_l2": "category_l2", "category_l3": "category_l3", "brand": "brand", "manufacturer": "manufacturer", "collection": "collection", "item_size": "item_size", "item_style": "item_style", "assembly_type": "assembly_type", "variation_group_code": "variation_group_code", "status": "status", } attr = aliases.get(normalize_column_key(code)) return getattr(product, attr, None) if attr else None def _parent_sku_for_group( members: Sequence[MasterProduct], group_values: Dict[str, str], *, parent_suffix: str, ) -> str: prefix = _common_catalog_prefix([p.sku for p in members]) if not prefix: tokens = [group_values.get(code) for code in ("collection", "cabinet_type", "type_code", "product_family")] prefix = "-".join(_sku_token(token) for token in tokens if token) suffix = _sku_token(parent_suffix) or "CONFIG" if prefix.upper().endswith(f"-{suffix}"): return prefix.upper() return f"{prefix}-{suffix}".upper() def _common_catalog_prefix(skus: Sequence[str]) -> str: parsed = [_sku_alpha_prefix(sku) for sku in skus if sku] parsed = [p for p in parsed if p] if not parsed: return "" prefix = parsed[0] for item in parsed[1:]: while prefix and not item.startswith(prefix): prefix = prefix[:-1] prefix = prefix.rstrip("-") return prefix.rstrip("-") def _sku_alpha_prefix(sku: str) -> str: text = str(sku or "").strip().upper() if not text: return "" match = re.match(r"^(.*?[A-Z])\d", text) if match: return match.group(1).rstrip("-") return text.rsplit("-", 1)[0] if "-" in text else text def _parent_name_for_group(members: Sequence[MasterProduct], group_values: Dict[str, str]) -> str: first = members[0] if members else None attrs_by_sku = {product.sku: group_values for product in members} return variation_parent_name_for_product(first, members, attrs_by_sku=attrs_by_sku, group_values=group_values) def variation_parent_name_for_product( parent: Optional[MasterProduct], children: Sequence[MasterProduct], *, attrs_by_sku: Optional[Dict[str, Dict[str, str]]] = None, group_values: Optional[Dict[str, str]] = None, ) -> str: values = dict(group_values or {}) first = children[0] if children else parent collection = _first_non_empty(values.get("collection"), getattr(parent, "collection", None), getattr(first, "collection", None)) descriptor = _first_non_empty( *[values.get(code) for code in PARENT_DESCRIPTOR_ATTRS], *[ _common_child_value(children, attrs_by_sku or {}, code) for code in PARENT_DESCRIPTOR_ATTRS ], ) descriptor = _clean_parent_descriptor(descriptor, collection) if collection and descriptor: return f"{collection} {descriptor}" if descriptor: manufacturer = _first_non_empty(getattr(parent, "manufacturer", None), getattr(first, "manufacturer", None)) return f"{descriptor} | {manufacturer}" if manufacturer else descriptor if collection: return collection return f"{getattr(first, 'name', None) or getattr(first, 'sku', None) or getattr(parent, 'sku', None) or 'Variation'} Options" def _common_child_value( children: Sequence[MasterProduct], attrs_by_sku: Dict[str, Dict[str, str]], code: str, ) -> Optional[str]: values = { str(_field_values(child, attrs_by_sku.get(child.sku, {}), [code]).get(code) or "").strip() for child in children } values.discard("") if len(values) == 1: return next(iter(values)) return None def _is_parent_attr_copy_skipped(code: str, *, extra_skip: Optional[Sequence[str]] = None) -> bool: normalized = normalize_column_key(code) if not normalized: return True if normalized in PARENT_ATTR_COPY_SKIP_CODES: return True if normalized.startswith("variation_"): return True if extra_skip and normalized in {normalize_column_key(c) for c in extra_skip}: return True return False def _copy_collection_shareable_attrs_to_parents( session: Session, parent_rows: Sequence[Dict[str, Any]], *, source_label: str = "variation_builder_shareable", ) -> int: """Fill missing parent attrs from values shared by all children in the group. Copies every non-configuration attribute that is unanimous across children (door/finish/construction/hardware/materials/etc.). Axis / size / SEO / media / price fields are skipped — parents keep their own titles/slugs. """ from db.canonical_attributes import canonicalize_attribute_fields copied = 0 for row in parent_rows: parent_sku = str(row.get("sku") or "").strip() payload = row.get("raw_payload") if isinstance(row.get("raw_payload"), dict) else {} child_skus = [str(sku).strip() for sku in (payload.get("children") or []) if str(sku).strip()] if not parent_sku or len(child_skus) < 1: continue parent = session.scalar(select(MasterProduct).where(MasterProduct.sku == parent_sku)) children = list( session.scalars(select(MasterProduct).where(MasterProduct.sku.in_(child_skus))).all() ) if parent is None or not children: continue attrs_by_sku = _attrs_by_sku(session, [child.id for child in children] + [parent.id]) axis_skip = [str(c).strip() for c in (payload.get("option_attrs") or []) if str(c).strip()] candidate_codes: set[str] = set() for child in children: candidate_codes.update(attrs_by_sku.get(child.sku, {}).keys()) # Always consider the known shareable set even if only present under aliases. candidate_codes.update(COLLECTION_SHAREABLE_ATTRS) common: Dict[str, str] = {} for code in sorted(candidate_codes): if _is_parent_attr_copy_skipped(code, extra_skip=axis_skip): continue value = _common_child_value(children, attrs_by_sku, code) if value: common[code] = value if not common: continue canonical = canonicalize_attribute_fields(common) # Preserve raw unanimous codes that canonicalize may not remap. for code, value in common.items(): canonical.setdefault(code, value) parent_attrs = attrs_by_sku.get(parent.sku, {}) for code in sorted({normalize_column_key(c) for c in canonical.keys() if c}): if _is_parent_attr_copy_skipped(code, extra_skip=axis_skip): continue value = str(canonical.get(code) or "").strip() if not value: continue if str(parent_attrs.get(code) or "").strip(): continue _upsert_attribute_value( session, product=parent, attribute_code=code, value=value, source_label=source_label, ) parent_attrs[code] = value copied += 1 return copied def copy_shared_child_attrs_to_parents_from_relations( session: Session, *, parent_skus: Optional[Sequence[str]] = None, source_label: str = "parent_attr_impute_from_children", dry_run: bool = True, ) -> Dict[str, Any]: """Backfill existing variation parents from MasterProductRelation children.""" wanted = {str(s).strip() for s in parent_skus or [] if str(s).strip()} stmt = select(MasterProductRelation).order_by( MasterProductRelation.parent_sku, MasterProductRelation.sort_order, MasterProductRelation.child_sku, ) if wanted: stmt = stmt.where(MasterProductRelation.parent_sku.in_(wanted)) relations = list(session.scalars(stmt).all()) children_by_parent: Dict[str, List[str]] = {} option_attrs_by_parent: Dict[str, List[str]] = {} for rel in relations: parent = str(rel.parent_sku or "").strip() child = str(rel.child_sku or "").strip() if not parent or not child: continue children_by_parent.setdefault(parent, []) if child not in children_by_parent[parent]: children_by_parent[parent].append(child) mapping = rel.option_mapping if isinstance(rel.option_mapping, dict) else {} for code in mapping.keys(): option_attrs_by_parent.setdefault(parent, []) norm = normalize_column_key(code) if norm and norm not in option_attrs_by_parent[parent]: option_attrs_by_parent[parent].append(norm) parent_rows = [ { "sku": parent_sku, "raw_payload": { "children": child_skus, "option_attrs": option_attrs_by_parent.get(parent_sku) or [], }, } for parent_sku, child_skus in sorted(children_by_parent.items()) if child_skus ] if dry_run: # Preview by running copy against a throwaway pattern: count would-be writes # without persisting — reuse the same selection logic. preview = _preview_parent_attr_copies(session, parent_rows) return { "status": "ok", "dry_run": True, "parent_count": len(parent_rows), "would_copy": preview["would_copy"], "sample": preview["sample"][:50], } copied = _copy_collection_shareable_attrs_to_parents( session, parent_rows, source_label=source_label, ) return { "status": "ok", "dry_run": False, "parent_count": len(parent_rows), "copied": copied, } def _preview_parent_attr_copies( session: Session, parent_rows: Sequence[Dict[str, Any]], ) -> Dict[str, Any]: from db.canonical_attributes import canonicalize_attribute_fields would_copy = 0 sample: List[Dict[str, Any]] = [] for row in parent_rows: parent_sku = str(row.get("sku") or "").strip() payload = row.get("raw_payload") if isinstance(row.get("raw_payload"), dict) else {} child_skus = [str(sku).strip() for sku in (payload.get("children") or []) if str(sku).strip()] parent = session.scalar(select(MasterProduct).where(MasterProduct.sku == parent_sku)) children = list( session.scalars(select(MasterProduct).where(MasterProduct.sku.in_(child_skus))).all() ) if parent is None or not children: continue attrs_by_sku = _attrs_by_sku(session, [child.id for child in children] + [parent.id]) axis_skip = [str(c).strip() for c in (payload.get("option_attrs") or []) if str(c).strip()] candidate_codes: set[str] = set() for child in children: candidate_codes.update(attrs_by_sku.get(child.sku, {}).keys()) candidate_codes.update(COLLECTION_SHAREABLE_ATTRS) common: Dict[str, str] = {} for code in sorted(candidate_codes): if _is_parent_attr_copy_skipped(code, extra_skip=axis_skip): continue value = _common_child_value(children, attrs_by_sku, code) if value: common[code] = value if not common: continue canonical = canonicalize_attribute_fields(common) for code, value in common.items(): canonical.setdefault(code, value) parent_attrs = attrs_by_sku.get(parent.sku, {}) for code in sorted({normalize_column_key(c) for c in canonical.keys() if c}): if _is_parent_attr_copy_skipped(code, extra_skip=axis_skip): continue value = str(canonical.get(code) or "").strip() if not value or str(parent_attrs.get(code) or "").strip(): continue would_copy += 1 if len(sample) < 50: sample.append( { "parent_sku": parent_sku, "attribute": code, "value": value, "child_count": len(children), } ) return {"would_copy": would_copy, "sample": sample} def _upsert_attribute_value( session: Session, *, product: MasterProduct, attribute_code: str, value: str, source_label: str, ) -> None: row = session.scalar( select(MasterProductAttributeValue) .where(MasterProductAttributeValue.product_id == product.id) .where(MasterProductAttributeValue.attribute_code == attribute_code) ) if row is None: row = MasterProductAttributeValue( product_id=product.id, sku=product.sku, attribute_code=attribute_code, ) session.add(row) row.value = value row.source_label = source_label def _clean_parent_descriptor(value: Optional[str], collection: Optional[str]) -> str: text = str(value or "").strip() if not text: return "" if collection and text.lower() == str(collection).strip().lower(): return "" if text.lower() in {"cabinet", "cabinets", "kitchen cabinets"}: return "" return text def _first_non_empty(*values: Any) -> str: for value in values: text = str(value or "").strip() if text: return text return "" def _dedupe_parent_sku(parent_sku: str, existing: List[str]) -> str: if parent_sku not in existing: return parent_sku index = 2 while f"{parent_sku}-{index}" in existing: index += 1 return f"{parent_sku}-{index}" def _first_child_product(session: Session, children: Sequence[Dict[str, Any]]) -> Optional[MasterProduct]: for child in children: sku = str(child.get("sku") or "").strip() if sku: return session.scalar(select(MasterProduct).where(MasterProduct.sku == sku)) return None def _parent_row(parent_sku: str, group: Dict[str, Any], first_child: Optional[MasterProduct]) -> Dict[str, Any]: group_values = dict(group.get("group_values") or {}) option_attrs = sanitize_option_attrs(group.get("option_attrs") or []) payload = { "generated_by": "variation_builder", "group_values": group_values, "option_attrs": option_attrs, "children": [child.get("sku") for child in group.get("children") or []], } return { "sku": parent_sku, "source_item": None, "name": group.get("parent_name") or parent_sku, "product_family": group_values.get("product_family") or (first_child.product_family if first_child else None), "category_l1": first_child.category_l1 if first_child else None, "category_l2": first_child.category_l2 if first_child else None, "category_l3": first_child.category_l3 if first_child else None, "brand": first_child.brand if first_child else None, "manufacturer": first_child.manufacturer if first_child else None, "collection": group_values.get("collection") or (first_child.collection if first_child else None), "item_size": None, "item_style": first_child.item_style if first_child else None, "assembly_type": first_child.assembly_type if first_child else None, "stocked": first_child.stocked if first_child else None, "status": "active", "raw_payload": payload, "row_hash": _stable_hash(payload), "is_active": True, } def _upsert_parent_products(session: Session, rows: List[Dict[str, Any]]) -> None: stmt = insert(MasterProduct).values(rows) stmt = stmt.on_conflict_do_update( constraint="uq_master_product_sku", set_={ "name": stmt.excluded.name, "product_family": stmt.excluded.product_family, "category_l1": stmt.excluded.category_l1, "category_l2": stmt.excluded.category_l2, "category_l3": stmt.excluded.category_l3, "brand": stmt.excluded.brand, "manufacturer": stmt.excluded.manufacturer, "collection": stmt.excluded.collection, "item_style": stmt.excluded.item_style, "assembly_type": stmt.excluded.assembly_type, "stocked": stmt.excluded.stocked, "status": stmt.excluded.status, "raw_payload": stmt.excluded.raw_payload, "row_hash": stmt.excluded.row_hash, "is_active": stmt.excluded.is_active, "updated_at": sa.func.now(), }, ) session.execute(stmt) def _assign_parent_skus(session: Session, parent_skus: Sequence[str], channels: Sequence[str]) -> int: total = 0 for channel in _codes(channels): result = assign_skus( session, skus=list(parent_skus), channel_code=channel, reason="variation_builder_parent_shell", ) total += int(result.get("upserted") or 0) return total def _upsert_identity_mappings( session: Session, parent_skus: Sequence[str], channels: Sequence[str], *, mapping_connection_ids: Dict[str, Optional[int]], source_label: str, ) -> int: rows: List[Dict[str, Any]] = [] for channel in _codes(channels): connection_id = mapping_connection_ids.get(channel) for sku in parent_skus: rows.append( { "master_sku": sku, "channel_code": channel, "connection_id": connection_id, "channel_sku": sku, "mapping_status": SKU_MAPPING_ACTIVE_STATUS, "match_source": source_label, "notes": "Generated parent shell identity mapping", } ) return upsert_mappings(session, rows) def export_variation_parent_child_rows( session: Session, *, parent_skus: Optional[Sequence[str]] = None, channel_code: str = "magento", connection_id: Optional[int] = None, ) -> Dict[str, Any]: """Flatten variation parents + children with option axis values for CSV export. Prefers `master_product_relation` rows; falls back to variation_builder `raw_payload.children` / `option_attrs` when relations are missing. """ from db.channel_variation_axes import ChannelVariationAxisResolver wanted_parents = {str(sku or "").strip() for sku in (parent_skus or []) if str(sku or "").strip()} payload_parents = _variation_builder_parent_products(session, wanted_parents or None) relations = list( session.scalars( select(MasterProductRelation).order_by( MasterProductRelation.parent_sku, MasterProductRelation.sort_order, MasterProductRelation.child_sku, ) ).all() ) by_parent_relations: Dict[str, List[MasterProductRelation]] = {} for relation in relations: if wanted_parents and relation.parent_sku not in wanted_parents: continue if relation.parent_sku not in payload_parents and relation.parent_sku not in wanted_parents: if not str(relation.source_label or "").startswith("variation_builder"): continue by_parent_relations.setdefault(relation.parent_sku, []).append(relation) parent_skus_ordered = sorted(set(by_parent_relations) | set(payload_parents)) if not parent_skus_ordered: return {"rows": [], "parent_count": 0, "child_count": 0, "option_attrs": []} all_child_skus: List[str] = [] for parent_sku in parent_skus_ordered: rel_children = [row.child_sku for row in by_parent_relations.get(parent_sku, [])] payload = payload_parents.get(parent_sku) payload_children = [] if payload and isinstance(payload.raw_payload, dict): payload_children = [ str(sku or "").strip() for sku in (payload.raw_payload.get("children") or []) if str(sku or "").strip() ] all_child_skus.extend(rel_children or payload_children) product_skus = sorted(set(parent_skus_ordered) | {sku for sku in all_child_skus if sku}) products = { product.sku: product for product in session.scalars(select(MasterProduct).where(MasterProduct.sku.in_(product_skus))).all() } attrs_by_sku = _attrs_by_sku(session, [product.id for product in products.values()]) resolver = ChannelVariationAxisResolver(session, channel_code, connection_id=connection_id) option_attr_set: List[str] = [] rows: List[Dict[str, Any]] = [] for parent_sku in parent_skus_ordered: parent = products.get(parent_sku) or payload_parents.get(parent_sku) if parent is None: continue raw_payload = parent.raw_payload if isinstance(parent.raw_payload, dict) else {} option_attrs = sanitize_option_attrs(raw_payload.get("option_attrs") or DEFAULT_OPTION_ATTRS) if not option_attrs: option_attrs = list(DIMENSION_OPTION_ATTRS) parent_relations = by_parent_relations.get(parent_sku) or [] if parent_relations: child_entries = [ { "sku": relation.child_sku, "option_mapping": _axis_only_mapping(relation.option_mapping or {}), "sort_order": relation.sort_order, "link_source": "relation", } for relation in parent_relations ] else: child_entries = [ { "sku": child_sku, "option_mapping": {}, "sort_order": index, "link_source": "raw_payload", } for index, child_sku in enumerate( [ str(sku or "").strip() for sku in (raw_payload.get("children") or []) if str(sku or "").strip() ] ) ] child_products = [products[entry["sku"]] for entry in child_entries if entry["sku"] in products] if child_products: inferred = _varying_option_codes( child_products, {product.sku: attrs_by_sku.get(product.sku, {}) for product in child_products}, option_attrs or list(DIMENSION_OPTION_ATTRS), ) if inferred: option_attrs = inferred for code in option_attrs: if code not in option_attr_set: option_attr_set.append(code) group_values = raw_payload.get("group_values") if isinstance(raw_payload.get("group_values"), dict) else {} child_skus = [entry["sku"] for entry in child_entries] option_mappings_for_fields: List[Dict[str, str]] = [] for entry in child_entries: child_sku = entry["sku"] child = products.get(child_sku) option_mapping = _axis_only_mapping(entry.get("option_mapping") or {}) if not option_mapping and option_attrs and child is not None: option_mapping = _field_values(child, attrs_by_sku.get(child_sku, {}), option_attrs) option_mappings_for_fields.append(option_mapping) relation_fields = resolver.build_relation_fields(child_skus, option_mappings_for_fields) option_label = resolver.option_axis_label(option_attrs) channel_attrs = resolver.channel_option_attrs(option_attrs) for index, entry in enumerate(child_entries): child_sku = entry["sku"] child = products.get(child_sku) option_mapping = option_mappings_for_fields[index] if index < len(option_mappings_for_fields) else {} row: Dict[str, Any] = { "parent_sku": parent_sku, "parent_name": parent.name or "", "parent_collection": parent.collection or group_values.get("collection") or "", "option_attrs": ",".join(option_attrs), "option_label": option_label, "magento_configurable_attributes": ",".join(channel_attrs), "variant_list": relation_fields.get("variant_list") or ",".join(child_skus), "variant_of": parent_sku, "configurable_variations": relation_fields.get("configurable_variations") or "", "configurable_variation_labels": relation_fields.get("configurable_variation_labels") or "", "child_sku": child_sku, "child_name": (child.name if child else "") or "", "sort_order": entry.get("sort_order", 0), "link_source": entry.get("link_source") or "", "explicit_seo_title": bool(raw_payload.get("explicit_seo_title")), } for code in option_attr_set: if code in NON_AXIS_OPTION_ATTRS: continue row[code] = option_mapping.get(code) or attrs_by_sku.get(child_sku, {}).get(code) or "" rows.append(row) return { "rows": rows, "parent_count": len(parent_skus_ordered), "child_count": len({row["child_sku"] for row in rows}), "option_attrs": [code for code in option_attr_set if code not in NON_AXIS_OPTION_ATTRS], } def repair_variation_relations_from_payload( session: Session, *, parent_skus: Optional[Sequence[str]] = None, source_label: str = "variation_builder_repair", dry_run: bool = False, ) -> Dict[str, Any]: """Backfill master_product_relation (+ clean option_attrs) from variation_builder raw_payload. Existing Magento parents often have children only in raw_payload, so channel exports never emit variant_list / variant_of / configurable_variations. """ wanted = {str(sku or "").strip() for sku in (parent_skus or []) if str(sku or "").strip()} parents = _variation_builder_parent_products(session, wanted or None) if not parents: return { "status": "ok", "dry_run": dry_run, "parent_count": 0, "relations_upserted": 0, "payloads_cleaned": 0, "skipped_no_children": 0, "skipped_no_axis": 0, "sample_parents": [], } existing = { (row.parent_sku, row.child_sku): row for row in session.scalars( select(MasterProductRelation).where(MasterProductRelation.parent_sku.in_(list(parents))) ).all() } all_child_skus: List[str] = [] for parent in parents.values(): payload = parent.raw_payload if isinstance(parent.raw_payload, dict) else {} all_child_skus.extend( str(sku or "").strip() for sku in (payload.get("children") or []) if str(sku or "").strip() ) child_products = { product.sku: product for product in session.scalars( select(MasterProduct).where(MasterProduct.sku.in_(sorted(set(all_child_skus)))) ).all() } if all_child_skus else {} attrs_by_sku = _attrs_by_sku(session, [product.id for product in child_products.values()]) relation_rows: List[Dict[str, Any]] = [] payloads_cleaned = 0 skipped_no_children = 0 skipped_no_axis = 0 sample_parents: List[Dict[str, Any]] = [] for parent_sku, parent in sorted(parents.items()): raw_payload = dict(parent.raw_payload or {}) if isinstance(parent.raw_payload, dict) else {} child_skus = [ str(sku or "").strip() for sku in (raw_payload.get("children") or []) if str(sku or "").strip() and str(sku or "").strip() != parent_sku ] if len(child_skus) < 1: skipped_no_children += 1 continue stored_attrs = sanitize_option_attrs(raw_payload.get("option_attrs") or []) candidate_option_attrs = list( dict.fromkeys( [ *stored_attrs, *list(DEFAULT_OPTION_ATTRS), ] ) ) members = [child_products[sku] for sku in child_skus if sku in child_products] option_attrs = _varying_option_codes( members, {product.sku: attrs_by_sku.get(product.sku, {}) for product in members}, candidate_option_attrs, ) if not option_attrs: for code in candidate_option_attrs: if any(attrs_by_sku.get(sku, {}).get(code) for sku in child_skus): option_attrs = [code] break if not option_attrs: skipped_no_axis += 1 continue cleaned_payload = dict(raw_payload) if cleaned_payload.get("option_attrs") != option_attrs: cleaned_payload["option_attrs"] = option_attrs payloads_cleaned += 1 if not dry_run: parent.raw_payload = cleaned_payload parent.row_hash = _stable_hash(cleaned_payload) parent_sample_children: List[Dict[str, str]] = [] for order, child_sku in enumerate(child_skus): child = child_products.get(child_sku) existing_row = existing.get((parent_sku, child_sku)) existing_mapping = _axis_only_mapping(existing_row.option_mapping if existing_row else None) derived_mapping = ( _field_values(child, attrs_by_sku.get(child_sku, {}), option_attrs) if child is not None else {} ) mapping: Dict[str, str] = {} for code in option_attrs: value = derived_mapping.get(code) or existing_mapping.get(code) if value: mapping[code] = value if not mapping: continue relation_rows.append( { "parent_sku": parent_sku, "child_sku": child_sku, "option_mapping": mapping, "sort_order": order, "source_label": source_label, } ) if len(parent_sample_children) < 3: parent_sample_children.append({"sku": child_sku, **mapping}) if parent_sample_children and len(sample_parents) < 25: sample_parents.append( { "parent_sku": parent_sku, "option_attrs": option_attrs, "child_count": len(child_skus), "sample_children": parent_sample_children, } ) relations_upserted = 0 if dry_run else upsert_relations(session, relation_rows) return { "status": "ok", "dry_run": dry_run, "parent_count": len(parents), "would_upsert_relations": len(relation_rows), "relations_upserted": relations_upserted, "payloads_cleaned": payloads_cleaned, "skipped_no_children": skipped_no_children, "skipped_no_axis": skipped_no_axis, "sample_parents": sample_parents, } def _axis_only_mapping(mapping: Optional[Dict[str, Any]]) -> Dict[str, str]: out: Dict[str, str] = {} for key, value in dict(mapping or {}).items(): code = normalize_column_key(key) if not code or code in NON_AXIS_OPTION_ATTRS: continue text = str(value or "").strip() if text and text.lower() not in {"nan", "none", "null"}: out[code] = text return out def _variation_builder_parent_products( session: Session, parent_skus: Optional[Sequence[str]] = None, ) -> Dict[str, MasterProduct]: stmt = ( select(MasterProduct) .where(MasterProduct.is_active.is_(True)) .where(MasterProduct.raw_payload.is_not(None)) .order_by(MasterProduct.sku) ) if parent_skus: stmt = stmt.where(MasterProduct.sku.in_(list(parent_skus))) out: Dict[str, MasterProduct] = {} for product in session.scalars(stmt).all(): payload = product.raw_payload if isinstance(product.raw_payload, dict) else {} if payload.get("generated_by") != "variation_builder": continue children = payload.get("children") or [] if not children and not parent_skus: continue out[product.sku] = product return out def _codes(values: Sequence[str]) -> List[str]: out: List[str] = [] for value in values: code = normalize_column_key(value) if code and code not in out: out.append(code) return out def _sku_token(value: Optional[str]) -> str: return re.sub(r"[^A-Z0-9]+", "-", str(value or "").upper()).strip("-") def _group_sort_key(members: Sequence[MasterProduct]) -> str: return members[0].sku if members else "" def _stable_hash(payload: Dict[str, Any]) -> str: blob = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str) return hashlib.sha256(blob.encode("utf-8")).hexdigest()