from __future__ import annotations import hashlib import json from typing import Any, Dict, List, Optional, Set from sqlalchemy import select from sqlalchemy.orm import Session from channel.url_canonical import normalize_plp_path from db.channel_aliases import channel_aliases_by_canonical from db.channel_assignments import active_skus_for_channel from db.channel_capabilities import resolve_product_structure, resolve_product_structure_for_connection from db.canonical_attributes import canonicalize_attribute_fields, matches_publish_constraints from db.location_experience import collection_location_context from db.master_overrides import overrides_for_channel from db.master_relations import ( build_parent_relation_fields, children_by_parent, expand_skus_with_children, expand_skus_with_parents, parent_by_child, parent_skus_with_children, ) from db.models import MasterLocationRegistry, MasterProduct, MasterProductAttributeValue, MasterProductLocationAvailability, MasterProductPrice from db.source_imports import normalize_column_key CHANNEL_CODES = {"magento", "shopify", "plytix"} def resolve_pipeline_channel_code( channel_code: Optional[str] = None, *, channel_type: Optional[str] = None, ) -> str: """Map a connection display code (e.g. magento-1, shop_code) to a pipeline channel code. Master-catalog export, aliases, assignments, and publish state all use the pipeline codes magento | shopify | plytix — not per-connection identifiers. """ if channel_type: normalized_type = normalize_column_key(channel_type) if normalized_type in CHANNEL_CODES: return normalized_type if channel_code: normalized_code = normalize_column_key(channel_code) if normalized_code in CHANNEL_CODES: return normalized_code raise ValueError( f"Unsupported channel_code: {channel_code!r} " f"(channel_type={channel_type!r}; expected one of {sorted(CHANNEL_CODES)})" ) # Per-channel sell price preference, first match wins. PRICE_TYPE_PRIORITY = { "shopify": ("shopify", "base", "rta", "assembled"), "magento": ("base", "rta", "assembled"), "plytix": ("base",), } CORE_CANONICAL_FIELDS = ( "title", "brand", "manufacturer", "product_family", "category_l1", "category_l2", "category_l3", "collection", "assembly_type", "item_size", "status", ) def _first_shopping_value(raw: Any) -> str: if isinstance(raw, list): for item in raw: text = str(item or "").strip() if text: return text return "" return str(raw or "").strip() def _shopping_collection_export_value(raw: Any) -> str: from db.collection_landing_pages import _relative_shopping_collection_value value = _first_shopping_value(raw) if not value: return "" return str(_relative_shopping_collection_value(value) or value) def _shopping_path_export_value(raw: Any, resolver) -> str: value = _first_shopping_value(raw) if not value: return "" return str(resolver(value) or value) def _relative_shopping_l1_export_value(path_slug: Any) -> str: from db.collection_landing_pages import _relative_shopping_l1_value return str(_relative_shopping_l1_value(path_slug) or "") def _relative_shopping_l2_export_value(path_slug: Any) -> str: from db.collection_landing_pages import _relative_shopping_l2_value return str(_relative_shopping_l2_value(path_slug) or "") def resolve_push_skus( session: Session, channel_code: str, *, skus: Optional[List[str]] = None, only_assigned: bool = True, product_structure: Optional[str] = None, capabilities: Optional[Dict[str, Any]] = None, expand_relations: Optional[bool] = None, ) -> Set[str]: """SKUs that should participate in a channel push/export after assignment + structure rules.""" channel = resolve_pipeline_channel_code(channel_code) structure = product_structure or resolve_product_structure_for_connection( session, channel, connection_id=None, capabilities=capabilities, ) assigned: Optional[Set[str]] = active_skus_for_channel(session, channel) if only_assigned else None if skus: wanted = {str(s).strip() for s in skus if str(s).strip()} if assigned is not None: wanted &= assigned elif assigned is not None: wanted = set(assigned) else: wanted = { sku for (sku,) in session.execute( select(MasterProduct.sku).where(MasterProduct.is_active.is_(True)) ).all() } should_expand = expand_relations if should_expand is None: # Curated SKU lists (dashboard selection / limit_skus) must not pull in siblings. should_expand = structure == "parent_children" and not skus if should_expand and structure == "parent_children": wanted = expand_skus_with_children(session, wanted) wanted = expand_skus_with_parents(session, wanted) elif structure != "parent_children": # Flat channels can know variation relationships, but they should not publish # generated/configurable parent shell SKUs as standalone products. wanted -= parent_skus_with_children(session, wanted) return wanted def build_channel_export_bundle( session: Session, channel_code: str, *, skus: Optional[List[str]] = None, limit: Optional[int] = None, only_assigned: bool = True, connection_id: Optional[int] = None, capabilities: Optional[Dict[str, Any]] = None, publish_constraints: Optional[Dict[str, List[Any]]] = None, prefer_projection_cache: bool = True, ) -> Dict[str, Any]: """Payloads plus relation metadata for channels that use parent/child structure.""" channel = resolve_pipeline_channel_code(channel_code) structure = resolve_product_structure_for_connection( session, channel, connection_id=connection_id, capabilities=capabilities, ) wanted = resolve_push_skus( session, channel, skus=skus, only_assigned=only_assigned, product_structure=structure, capabilities=capabilities, ) if limit and len(wanted) > limit: wanted = set(sorted(wanted)[:limit]) payloads = build_channel_product_payloads( session, channel, skus=sorted(wanted), only_assigned=False, connection_id=connection_id, product_structure=structure, capabilities=capabilities, publish_constraints=publish_constraints, prefer_projection_cache=prefer_projection_cache, ) relations = children_by_parent(session, parent_skus_with_children(session, wanted)) return { "channel_code": channel, "product_structure": structure, "sku_count": len(wanted), "payloads": payloads, "relations": relations, } def build_channel_product_payloads( session: Session, channel_code: str, *, skus: Optional[List[str]] = None, limit: Optional[int] = None, only_assigned: bool = True, connection_id: Optional[int] = None, product_structure: Optional[str] = None, capabilities: Optional[Dict[str, Any]] = None, skip_taxonomy: bool = False, publish_constraints: Optional[Dict[str, List[Any]]] = None, prefer_projection_cache: bool = True, ) -> List[Dict[str, Any]]: """Build one payload per SKU for a channel: master core fields + attribute values, with master_product_override applied on top, mapped through channel attribute aliases. """ channel = resolve_pipeline_channel_code(channel_code) structure = product_structure or resolve_product_structure_for_connection( session, channel, connection_id=connection_id, capabilities=capabilities, ) if only_assigned or product_structure is not None or capabilities is not None: wanted = resolve_push_skus( session, channel, skus=skus, only_assigned=only_assigned, product_structure=structure, capabilities=capabilities, ) if limit and len(wanted) > limit: wanted = set(sorted(wanted)[:limit]) skus = sorted(wanted) only_assigned = False requested_skus = [str(s).strip() for s in (skus or []) if str(s).strip()] cached_fresh: List[Dict[str, Any]] = [] if prefer_projection_cache and requested_skus: from db.channel_projection_cache import partition_projection_cache partitioned = partition_projection_cache( session, channel_code=channel, connection_id=connection_id, skus=requested_skus, product_structure=structure, ) cached_fresh = list(partitioned["fresh"]) rebuild_skus = list(partitioned["rebuild_skus"]) if not rebuild_skus: _apply_canonical_shopping_field_overrides(cached_fresh) return cached_fresh skus = rebuild_skus only_assigned = False assigned: Optional[Set[str]] = active_skus_for_channel(session, channel) if only_assigned else None products = _products(session, skus=skus, assigned=assigned, limit=limit) values = _attribute_values(session, [p.id for p in products]) prices = _prices(session, [p.id for p in products]) image_fields = _image_fields(session, [p.sku for p in products]) availability = _location_availability(session, [p.id for p in products]) overrides = overrides_for_channel(session, channel) alias_map = _build_alias_map(session, channel) taxonomy_index = None try: from db.master_taxonomy_product_paths import _NodeIndex taxonomy_index = _NodeIndex.load(session) except Exception: taxonomy_index = None product_skus = [p.sku for p in products] parent_set = parent_skus_with_children(session, product_skus) child_parent_map = parent_by_child(session, product_skus) payloads: List[Dict[str, Any]] = [] for product in products: attrs = enrich_catalog_attributes( {normalize_column_key(v.attribute_code): v.value for v in values.get(product.id, [])} ) if taxonomy_index is not None: try: from db.master_taxonomy_product_paths import canonical_taxonomy_leaf_label taxonomy_leaf = canonical_taxonomy_leaf_label( session, product, attrs, index=taxonomy_index, ) if taxonomy_leaf: attrs["taxonomy_leaf_label"] = taxonomy_leaf except Exception: pass attrs.update(image_fields.get(product.sku, {})) canonical_fields, override_codes = compose_canonical_fields( _core_fields(product), attrs, overrides.get(product.sku, {}), ) _attach_location_availability_fields(canonical_fields, availability.get(product.id) or {}) _attach_collection_location_fields( session, canonical_fields, category_l1=product.category_l1, collection=product.collection, ) canonical_fields = canonicalize_attribute_fields(canonical_fields) _apply_manual_taxonomy_field_override(session, product.sku, canonical_fields) if not matches_publish_constraints(canonical_fields, publish_constraints): continue channel_fields = map_fields_to_channel(canonical_fields, alias_map) price = select_price(prices.get(product.id, {}), channel) if product.sku in parent_set: role = "parent" elif product.sku in child_parent_map: role = "child" else: role = "standalone" payloads.append( { "sku": product.sku, "channel_code": channel, "product_role": role, "canonical_fields": canonical_fields, "fields": channel_fields, "override_codes": sorted(override_codes), "price": price, "prices": prices.get(product.id, {}), "payload_hash": payload_hash({"fields": channel_fields, "price": price}), "source_row_hash": product.row_hash, } ) if structure == "parent_children": parent_fields = build_parent_relation_fields( session, [p["sku"] for p in payloads if p["product_role"] == "parent"], channel_code=channel, connection_id=connection_id, ) child_parents = { child: parent for parent, children in children_by_parent(session, parent_fields.keys()).items() for child in (c["child_sku"] for c in children) } for payload in payloads: sku = payload["sku"] if sku in parent_fields: payload["relation_fields"] = parent_fields[sku] elif sku in child_parents: payload["product_role"] = "child" payload["relation_fields"] = {"variant_of": child_parents[sku], "product_type": "simple"} _attach_channel_skus(session, channel, payloads, connection_id=connection_id) _attach_association_fields(session, channel, payloads, connection_id=connection_id) _attach_url_canonicals(payloads) if skip_taxonomy: for payload in payloads: payload["payload_hash"] = payload_hash( { "fields": payload.get("fields") or {}, "price": payload.get("price"), "association_fields": payload.get("association_fields") or {}, } ) else: _attach_taxonomy_targets(session, channel, payloads, connection_id=connection_id) _apply_manual_taxonomy_projection_overrides(session, channel, payloads, connection_id=connection_id) _apply_canonical_shopping_field_overrides(payloads) if prefer_projection_cache and payloads: from db.channel_projection_cache import store_projection_payloads store_projection_payloads( session, channel_code=channel, connection_id=connection_id, payloads=payloads, product_structure=structure, scope_key="write_through", notes="Write-through after partial rebuild", replace_scope=False, ) if cached_fresh: _apply_canonical_shopping_field_overrides(cached_fresh) by_sku = {p["sku"]: p for p in payloads} for cached in cached_fresh: by_sku.setdefault(cached["sku"], cached) payloads = [by_sku[sku] for sku in requested_skus if sku in by_sku] or list(by_sku.values()) return payloads def _apply_manual_taxonomy_field_override( session: Session, master_sku: str, canonical_fields: Dict[str, Any], ) -> None: from db.manual_taxonomy_assignments import resolve_manual_taxonomy_overrides overrides = resolve_manual_taxonomy_overrides(session, master_skus=[master_sku]) item = overrides.get(str(master_sku or "").strip().upper()) if not item: return for field, value in ( ("category_l1", item.get("category_l1")), ("category_l2", item.get("category_l2")), ("category_l3", item.get("category_l3")), ("collection", item.get("collection")), ("shopping_l1", item.get("shopping_l1")), ("shopping_l2", item.get("shopping_l2")), ): if value: canonical_fields[field] = value def _apply_manual_taxonomy_projection_overrides( session: Session, channel: str, payloads: List[Dict[str, Any]], *, connection_id: Optional[int] = None, ) -> None: from db.collection_landing_pages import ( _relative_shopping_collection_value, _relative_shopping_l1_value, _relative_shopping_l2_value, ) from db.manual_taxonomy_assignments import resolve_manual_taxonomy_overrides from db.models import MasterTaxonomyChannelLink, MasterTaxonomyNode if not payloads: return overrides = resolve_manual_taxonomy_overrides(session, master_skus=[payload.get("sku") for payload in payloads]) if not overrides: return path_slugs = sorted( { str(path).strip() for item in overrides.values() for path in item.get("membership_path_slugs") or [] if str(path).strip() } ) link_rows: List[Any] = [] if path_slugs: stmt = ( select(MasterTaxonomyNode.path_slug, MasterTaxonomyChannelLink) .join( MasterTaxonomyChannelLink, MasterTaxonomyChannelLink.taxonomy_node_id == MasterTaxonomyNode.id, ) .where(MasterTaxonomyNode.path_slug.in_(path_slugs)) .where(MasterTaxonomyNode.is_active.is_(True)) .where(MasterTaxonomyChannelLink.channel_code == channel) .where(MasterTaxonomyChannelLink.is_active.is_(True)) ) if connection_id is None: stmt = stmt.where(MasterTaxonomyChannelLink.connection_id.is_(None)) else: stmt = stmt.where( (MasterTaxonomyChannelLink.connection_id == connection_id) | (MasterTaxonomyChannelLink.connection_id.is_(None)) ) link_rows = session.execute(stmt).all() links_by_path: Dict[str, Dict[str, List[Any]]] = {} for path_slug, link in link_rows: links_by_path.setdefault(str(path_slug), {}).setdefault(str(link.taxonomy_kind or ""), []).append(link) for payload in payloads: sku = str(payload.get("sku") or "").strip().upper() override = overrides.get(sku) if not override: continue canonical_fields = dict(payload.get("canonical_fields") or {}) channel_fields = dict(payload.get("fields") or {}) taxonomy_targets = dict(payload.get("taxonomy_targets") or {}) plp_path_slugs = [ normalized for normalized in ( normalize_plp_path(item) for item in list(taxonomy_targets.get("plp_path_slugs") or []) + list(override.get("plp_path_slugs") or []) ) if normalized ] seen_paths = set() plp_path_slugs = [item for item in plp_path_slugs if not (item in seen_paths or seen_paths.add(item))] taxonomy_targets["plp_path_slugs"] = plp_path_slugs for field, value in ( ("category_l1", override.get("category_l1")), ("category_l2", override.get("category_l2")), ("category_l3", override.get("category_l3")), ("collection", override.get("collection")), ): if value: canonical_fields[field] = value collection_value = _relative_shopping_collection_value(override.get("collection_path_slug") or "") shopping_values = { "shopping_collection": str( collection_value or _shopping_collection_export_value(override.get("shopping_collection")) or "" ).strip(), "shopping_l1": str( _shopping_path_export_value(override.get("shopping_l1"), _relative_shopping_l1_export_value) or _relative_shopping_l1_export_value("/".join(plp_path_slugs[:2]) if len(plp_path_slugs) >= 2 else "") or "" ).strip(), "shopping_l2": str( _shopping_path_export_value(override.get("shopping_l2"), _relative_shopping_l2_export_value) or _relative_shopping_l2_export_value(override.get("canonical_taxonomy_path_slug") or "") or "" ).strip(), } for code, value in shopping_values.items(): if value: canonical_fields[code] = value channel_fields[code] = value membership_paths = [normalize_plp_path(item) for item in (override.get("membership_path_slugs") or []) if normalize_plp_path(item)] if channel == "magento": existing_ids = [int(item) for item in (taxonomy_targets.get("magento_category_ids") or []) if str(item).isdigit()] existing_paths = [str(item).strip() for item in (taxonomy_targets.get("magento_category_paths") or []) if str(item).strip()] seen_ids = set(existing_ids) seen_category_paths = set(existing_paths) for path_slug in membership_paths: for kind, links in (links_by_path.get(path_slug) or {}).items(): if kind not in {"category", "collection"}: continue for link in links: if str(link.remote_id or "").isdigit(): category_id = int(str(link.remote_id)) if category_id not in seen_ids: seen_ids.add(category_id) existing_ids.append(category_id) remote_path = str(link.remote_path or "").strip() or path_slug if remote_path and remote_path not in seen_category_paths: seen_category_paths.add(remote_path) existing_paths.append(remote_path) taxonomy_targets["magento_category_ids"] = existing_ids taxonomy_targets["magento_category_paths"] = existing_paths elif channel == "shopify": existing_product_category = str(taxonomy_targets.get("shopify_product_category") or "").strip() existing_collection_ids = [str(item).strip() for item in (taxonomy_targets.get("shopify_collection_ids") or []) if str(item).strip()] seen_collection_ids = set(existing_collection_ids) for path_slug in membership_paths: for kind, links in (links_by_path.get(path_slug) or {}).items(): for link in links: remote_id = str(link.remote_id or "").strip() if not remote_id: continue if kind == "product_category" and not existing_product_category: existing_product_category = remote_id if kind == "collection" and remote_id not in seen_collection_ids: seen_collection_ids.add(remote_id) existing_collection_ids.append(remote_id) if existing_product_category: taxonomy_targets["shopify_product_category"] = existing_product_category taxonomy_targets["shopify_collection_ids"] = existing_collection_ids payload["canonical_fields"] = canonical_fields payload["fields"] = channel_fields payload["taxonomy_targets"] = taxonomy_targets payload["payload_hash"] = payload_hash( { "fields": channel_fields, "price": payload.get("price"), "taxonomy_targets": taxonomy_targets, "association_fields": payload.get("association_fields") or {}, } ) def _attach_association_fields( session: Session, channel: str, payloads: List[Dict[str, Any]], *, connection_id: Optional[int] = None, ) -> None: from db.association_rules import LINK_TYPES from db.channel_sku_mapping import mapping_by_master from db.master_associations import association_fields_by_sku if not payloads: return master_skus = [p["sku"] for p in payloads] associations = association_fields_by_sku(session, master_skus) sku_map = mapping_by_master(session, channel, master_skus, connection_id=connection_id) # Also map association targets that may not be in this push batch. target_skus = sorted( { target for by_type in associations.values() for targets in by_type.values() for target in targets } ) if target_skus: sku_map.update( mapping_by_master(session, channel, target_skus, connection_id=connection_id) ) for payload in payloads: master_assoc = associations.get(payload["sku"]) or {} payload["association_fields_master"] = { link_type: list(master_assoc.get(link_type) or []) for link_type in LINK_TYPES } payload["association_fields"] = { link_type: [sku_map.get(sku, sku) for sku in (master_assoc.get(link_type) or [])] for link_type in LINK_TYPES } def _attach_channel_skus( session: Session, channel: str, payloads: List[Dict[str, Any]], *, connection_id: Optional[int] = None, ) -> None: from db.channel_sku_mapping import mapping_by_master if not payloads: return sku_map = mapping_by_master(session, channel, [p["sku"] for p in payloads], connection_id=connection_id) for payload in payloads: master_sku = payload["sku"] channel_sku = sku_map.get(master_sku, master_sku) payload["channel_sku"] = channel_sku if payload.get("relation_fields"): from db.channel_sku_mapping import apply_sku_map_to_relation_fields payload["relation_fields_master"] = dict(payload["relation_fields"]) payload["relation_fields"] = apply_sku_map_to_relation_fields(payload["relation_fields"], sku_map) def _attach_url_canonicals(payloads: List[Dict[str, Any]]) -> None: from channel.url_canonical import build_pdp_slug from magento.payload_mapper import is_polluted_product_title for payload in payloads: canonical = dict(payload.get("canonical_fields") or {}) title = ( canonical.get("title") or canonical.get("name") or canonical.get("seo_title") or canonical.get("meta_title") ) if title and is_polluted_product_title(str(title)): title = canonical.get("title") or canonical.get("name") or title slug = build_pdp_slug(title, payload.get("channel_sku") or payload.get("sku") or "") canonical["product_url_slug"] = slug payload["canonical_fields"] = canonical fields = dict(payload.get("fields") or {}) fields["product_url_slug"] = slug if payload.get("channel_code") == "shopify": fields["handle"] = slug if payload.get("channel_code") == "magento": fields["url_key"] = slug payload["fields"] = fields def _attach_taxonomy_targets( session: Session, channel: str, payloads: List[Dict[str, Any]], *, connection_id: Optional[int] = None, ) -> None: from db.compat_connections import decode_compat_connection_id from db.product_channel_taxonomy import resolve_taxonomy_targets_for_push native_connection_id = connection_id if connection_id is not None: channel_type, native_id = decode_compat_connection_id(connection_id) if channel_type in {"magento", "shopify"}: native_connection_id = native_id for payload in payloads: taxonomy_targets = resolve_taxonomy_targets_for_push( session, master_sku=payload["sku"], channel_code=channel, canonical_fields=payload.get("canonical_fields") or {}, connection_id=native_connection_id, ) payload["taxonomy_targets"] = taxonomy_targets payload["payload_hash"] = payload_hash( { "fields": payload.get("fields") or {}, "price": payload.get("price"), "taxonomy_targets": taxonomy_targets, "association_fields": payload.get("association_fields") or {}, } ) def _build_alias_map(session: Session, channel: str) -> Dict[str, List[str]]: """Merge static wide-table mappings with channel_attribute_alias (supports 1→N fan-out). Static mappings seed the first target. Additional alias rows (typically mapping_scope=dynamic) append extra remote attribute codes without removing the static target — e.g. collection → collection_style + family. """ from db.master_static_field_mapping import static_alias_map_for_channel alias_map: Dict[str, List[str]] = { canonical: [channel_attr] for canonical, channel_attr in static_alias_map_for_channel(session, channel).items() } aliases = channel_aliases_by_canonical(session, channel) for canonical, alias_rows in aliases.items(): targets = alias_map.setdefault(canonical, []) for alias in alias_rows: code = normalize_column_key(alias.channel_attribute_code) if code and code not in targets: targets.append(code) return alias_map def enrich_catalog_attributes(attrs: Dict[str, Any]) -> Dict[str, Any]: """Derive shopping-facet attributes used for layered nav and category filters.""" from db.catalog_intent_planner import _normalize_customer_category_label from db.shopping_facet_catalog import magento_option_label enriched = dict(attrs) sub1 = str(enriched.get("cabinet_type_sub_category_1") or "").strip() cabinet_type = str(enriched.get("cabinet_type") or "").strip() if sub1: normalized_sub = _normalize_customer_category_label(sub1) if normalized_sub: enriched["cabinet_type_sub_category_1"] = normalized_sub if not cabinet_type: enriched["cabinet_type"] = magento_option_label(normalized_sub) if cabinet_type: normalized_type = _normalize_customer_category_label(cabinet_type) or cabinet_type enriched["cabinet_type"] = magento_option_label(normalized_type) if not sub1 and normalized_type: enriched["cabinet_type_sub_category_1"] = normalized_type return canonicalize_attribute_fields(enriched) def compose_canonical_fields( core_fields: Dict[str, Any], attribute_values: Dict[str, Any], overrides: Dict[str, Any], ) -> tuple[Dict[str, Any], Set[str]]: """Pure merge: core master fields, then discovered attribute values, then overrides win. Overrides may also introduce canonical codes that do not exist in master data (e.g. meta_title). Returns (fields, override_codes_applied). """ fields: Dict[str, Any] = {} for code, value in {**core_fields, **attribute_values}.items(): if value is None or value == "": continue fields[normalize_column_key(code)] = value applied: Set[str] = set() for code, value in overrides.items(): key = normalize_column_key(code) if value is None or value == "": continue fields[key] = value applied.add(key) return canonicalize_attribute_fields(fields), applied def map_fields_to_channel( canonical_fields: Dict[str, Any], alias_map: Dict[str, List[str]], ) -> Dict[str, Any]: """Pure projection: canonical code → channel attribute code(s) via aliases; unaliased codes pass through.""" mapped: Dict[str, Any] = {} for canonical_code, value in canonical_fields.items(): for channel_code in alias_map.get(canonical_code, [canonical_code]): mapped[channel_code] = value return mapped def _apply_canonical_shopping_field_overrides(payloads: List[Dict[str, Any]]) -> None: for payload in payloads: canonical_fields = dict(payload.get("canonical_fields") or {}) channel_fields = dict(payload.get("fields") or {}) shopping_values = _derive_shopping_fields_from_taxonomy_targets(payload) changed = False for code in ("shopping_collection", "shopping_l1", "shopping_l2"): value = shopping_values.get(code) if value: if canonical_fields.get(code) != value: canonical_fields[code] = value changed = True if channel_fields.get(code) != value: channel_fields[code] = value changed = True else: if code in canonical_fields: canonical_fields.pop(code, None) changed = True if code in channel_fields: channel_fields.pop(code, None) changed = True if not changed: continue payload["canonical_fields"] = canonical_fields payload["fields"] = channel_fields payload["payload_hash"] = payload_hash( { "fields": channel_fields, "price": payload.get("price"), "association_fields": payload.get("association_fields") or {}, } ) def _derive_shopping_fields_from_taxonomy_targets(payload: Dict[str, Any]) -> Dict[str, str]: from db.collection_landing_pages import ( DEFAULT_CATEGORY_L1, START_SHOPPING_L1_KEYS, _relative_shopping_collection_value, _relative_shopping_l1_value, _relative_shopping_l2_value, collection_path_slug, ) canonical_fields = dict(payload.get("canonical_fields") or {}) taxonomy_targets = payload.get("taxonomy_targets") or {} raw_path_slugs = taxonomy_targets.get("plp_path_slugs") or [] path_slugs = [ normalized for normalized in (normalize_plp_path(str(item or "")) for item in raw_path_slugs) if normalized ] path_set = set(path_slugs) if not path_set: return {} collection_slug = normalize_plp_path( collection_path_slug( canonical_fields.get("category_l1") or DEFAULT_CATEGORY_L1, canonical_fields.get("collection"), ) ) collection_path = collection_slug if collection_slug in path_set else "" l1_candidates = [ slug for slug in path_set if slug.split("/") and slug.split("/")[-1] in START_SHOPPING_L1_KEYS ] l1_path = min(l1_candidates, key=lambda slug: (len(slug.split("/")), slug)) if l1_candidates else "" l2_candidates = [ slug for slug in path_set if l1_path and slug.startswith(f"{l1_path}/") ] l2_path = min(l2_candidates, key=lambda slug: (len(slug.split("/")), slug)) if l2_candidates else "" out: Dict[str, str] = {} collection_value = _relative_shopping_collection_value(collection_path) l1_value = _relative_shopping_l1_value(l1_path) l2_value = _relative_shopping_l2_value(l2_path) if collection_value: out["shopping_collection"] = str(collection_value) if l1_value: out["shopping_l1"] = str(l1_value) if l2_value: out["shopping_l2"] = str(l2_value) return out def select_price(prices: Dict[str, Any], channel: str) -> Optional[float]: for price_type in PRICE_TYPE_PRIORITY.get(channel, ("base",)): amount = prices.get(price_type) if amount is not None: return float(amount) return None def payload_hash(payload: Dict[str, Any]) -> str: blob = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str) return hashlib.sha256(blob.encode("utf-8")).hexdigest() RELATION_FIELD_COLUMNS = ( "product_type", "variant_of", "variant_list", "configurable_attributes", "configurable_variation_labels", "configurable_variations", ) ASSOCIATION_FIELD_COLUMNS = ( "related", "upsell", "crosssell", ) def flatten_export_bundle(bundle: Dict[str, Any]) -> List[Dict[str, Any]]: """One CSV-ready row per SKU: identity + price + relation fields + every channel field. This is exactly what a push would send, so the dashboard can review it before going live. """ payloads = bundle.get("payloads") or [] field_columns: List[str] = [] seen: Set[str] = set() for payload in payloads: for code in payload.get("fields") or {}: if code not in seen: seen.add(code) field_columns.append(code) field_columns.sort() rows: List[Dict[str, Any]] = [] for payload in payloads: relation_fields = payload.get("relation_fields") or {} association_fields = payload.get("association_fields") or {} row: Dict[str, Any] = { "master_sku": payload.get("sku"), "channel_sku": payload.get("channel_sku") or payload.get("sku"), "channel_code": payload.get("channel_code"), "product_role": payload.get("product_role") or "standalone", "price": payload.get("price"), "override_codes": ",".join(payload.get("override_codes") or []), "payload_hash": payload.get("payload_hash"), } for column in RELATION_FIELD_COLUMNS: row[f"relation_{column}"] = relation_fields.get(column, "") for column in ASSOCIATION_FIELD_COLUMNS: targets = association_fields.get(column) or [] row[f"association_{column}"] = ",".join(targets) fields = payload.get("fields") or {} for column in field_columns: row[column] = fields.get(column, "") rows.append(row) return rows def build_channel_attribute_preview( session: Session, channel_code: str, *, limit: Optional[int] = None, only_assigned: bool = True, connection_id: Optional[int] = None, ) -> List[Dict[str, Any]]: """Flat per-attribute preview rows derived from the same payloads the push uses.""" payloads = build_channel_product_payloads( session, channel_code, limit=limit, only_assigned=only_assigned, connection_id=connection_id, ) channel = normalize_column_key(channel_code) alias_map = _build_alias_map(session, channel) rows: List[Dict[str, Any]] = [] for payload in payloads: override_codes = set(payload["override_codes"]) for canonical_code, value in payload["canonical_fields"].items(): for channel_attribute_code in alias_map.get(canonical_code, [canonical_code]): rows.append( { "channel_code": channel, "sku": payload["sku"], "canonical_code": canonical_code, "channel_attribute_code": channel_attribute_code, "value": value, "source": "override" if canonical_code in override_codes else "master_product", "action": "upsert_attribute", } ) if payload["price"] is not None: rows.append( { "channel_code": channel, "sku": payload["sku"], "canonical_code": "price", "channel_attribute_code": "price", "value": payload["price"], "source": "master_product_price", "action": "upsert_price", } ) return rows def _products( session: Session, *, skus: Optional[List[str]], assigned: Optional[Set[str]], limit: Optional[int], ) -> List[MasterProduct]: stmt = select(MasterProduct).where(MasterProduct.is_active.is_(True)).order_by(MasterProduct.sku) wanted = None if skus: wanted = {str(s).strip() for s in skus if str(s).strip()} if assigned is not None: wanted = wanted & assigned if wanted is not None else assigned if not wanted: return [] if wanted is not None: stmt = stmt.where(MasterProduct.sku.in_(sorted(wanted))) if limit: stmt = stmt.limit(limit) return list(session.scalars(stmt).all()) def _image_fields(session: Session, skus: List[str]) -> Dict[str, Dict[str, str]]: from db.master_product_images import image_fields_by_sku return image_fields_by_sku(session, skus) def _attribute_values(session: Session, product_ids: List[int]) -> Dict[int, List[MasterProductAttributeValue]]: if not product_ids: return {} rows: Dict[int, List[MasterProductAttributeValue]] = {} for value in session.scalars( select(MasterProductAttributeValue) .where(MasterProductAttributeValue.product_id.in_(product_ids)) .order_by(MasterProductAttributeValue.product_id, MasterProductAttributeValue.attribute_code) ).all(): rows.setdefault(value.product_id, []).append(value) return rows def _prices(session: Session, product_ids: List[int]) -> Dict[int, Dict[str, float]]: if not product_ids: return {} rows: Dict[int, Dict[str, float]] = {} for price in session.scalars( select(MasterProductPrice).where(MasterProductPrice.product_id.in_(product_ids)) ).all(): if price.amount is None: continue rows.setdefault(price.product_id, {})[price.price_type] = float(price.amount) return rows def _location_availability(session: Session, product_ids: List[int]) -> Dict[int, Dict[str, Any]]: if not product_ids: return {} location_rows = session.scalars( select(MasterProductLocationAvailability).where(MasterProductLocationAvailability.product_id.in_(product_ids)) ).all() location_ids = sorted({int(row.location_id) for row in location_rows if row.location_id is not None}) locations_by_id = { row.id: row for row in session.scalars(select(MasterLocationRegistry).where(MasterLocationRegistry.id.in_(location_ids))).all() } if location_ids else {} location_code_lookup = _location_code_lookup(session) grouped: Dict[int, Dict[str, Any]] = {} for row in location_rows: payload = grouped.setdefault( int(row.product_id), { "availability_by_location": {}, "available_location_codes": [], "available_location_labels": [], }, ) code = _canonical_location_code(str(row.location_code or "").strip(), location_code_lookup) if not code: continue location = locations_by_id.get(row.location_id) if row.location_id is not None else None if location is None: location = location_code_lookup.get(code) label = str(location.name).strip() if location is not None and str(location.name or "").strip() else code payload.setdefault("_source_values_by_location", {})[code] = row.source_value mode = _normalize_location_availability_mode(row.is_available, row.source_value) payload["availability_by_location"][code] = mode if row.is_available: payload["available_location_codes"].append(code) payload["available_location_labels"].append(label) if mode == "grab_and_go": payload.setdefault("grab_and_go_location_codes", []).append(code) payload.setdefault("grab_and_go_location_labels", []).append(label) elif mode == "available": payload.setdefault("available_non_grab_location_codes", []).append(code) payload.setdefault("available_non_grab_location_labels", []).append(label) else: payload.setdefault("special_order_location_codes", []).append(code) payload.setdefault("special_order_location_labels", []).append(label) for payload in grouped.values(): for key in list(payload.keys()): if key.endswith("_codes") or key.endswith("_labels"): payload[key] = sorted(dict.fromkeys(payload[key])) payload["is_grab_and_go_any"] = bool(payload.get("grab_and_go_location_codes")) payload["is_available_any"] = bool(payload.get("available_location_codes")) payload["is_special_order_only"] = not bool(payload.get("available_location_codes")) payload["default_location_availability_mode"] = _default_location_availability_mode(payload) return grouped def _attach_location_availability_fields(canonical_fields: Dict[str, Any], availability: Dict[str, Any]) -> None: if not availability: return if _prefers_grab_and_go(canonical_fields) and not availability.get("grab_and_go_location_codes"): promoted_codes = list(availability.get("available_location_codes") or []) promoted_labels = list(availability.get("available_location_labels") or []) if promoted_codes: availability["grab_and_go_location_codes"] = promoted_codes availability["grab_and_go_location_labels"] = promoted_labels availability["available_non_grab_location_codes"] = [] availability["available_non_grab_location_labels"] = [] availability["availability_by_location"] = { code: ("grab_and_go" if code in promoted_codes else value) for code, value in (availability.get("availability_by_location") or {}).items() } availability["is_grab_and_go_any"] = True availability["default_location_availability_mode"] = "grab_and_go" for key, value in availability.items(): if str(key).startswith("_"): continue canonical_fields[key] = value # Magento PLP Item Availability attrs (comma-separated location codes). grab_codes = [str(c).strip() for c in (availability.get("grab_and_go_location_codes") or []) if str(c).strip()] if availability.get("available_non_grab_location_codes") is not None: available_codes = [ str(c).strip() for c in (availability.get("available_non_grab_location_codes") or []) if str(c).strip() ] else: available_codes = [ str(c).strip() for c in (availability.get("available_location_codes") or []) if str(c).strip() and str(c).strip() not in set(grab_codes) ] special_codes = [str(c).strip() for c in (availability.get("special_order_location_codes") or []) if str(c).strip()] canonical_fields["pim_avail_grab_go"] = ",".join(grab_codes) canonical_fields["pim_avail_available"] = ",".join(available_codes) canonical_fields["pim_avail_special"] = ",".join(special_codes) if not canonical_fields.get("item_availability"): mode = availability.get("default_location_availability_mode") if mode == "grab_and_go": canonical_fields["item_availability"] = "Grab & Go" canonical_fields.setdefault("collection_availability", "Grab & Go") elif mode == "available": canonical_fields["item_availability"] = "Available" canonical_fields.setdefault("collection_availability", "Available") elif mode == "special_order": canonical_fields["item_availability"] = "Special Order" canonical_fields.setdefault("collection_availability", "Special Order") def _attach_collection_location_fields( session: Session, canonical_fields: Dict[str, Any], *, category_l1: Optional[str], collection: Optional[str], ) -> None: context = collection_location_context( session, category_l1=category_l1, collection_name=collection, ) settings = context.get("location_settings") if isinstance(context.get("location_settings"), dict) else {} resolved = context.get("resolved_location") if isinstance(context.get("resolved_location"), dict) else {} directory = context.get("location_directory") if isinstance(context.get("location_directory"), list) else [] if settings: canonical_fields["location_settings"] = settings canonical_fields["default_location_code"] = settings.get("default_location_code") canonical_fields["default_location_name"] = settings.get("default_location_name") canonical_fields["allow_geolocation"] = bool(settings.get("allow_geolocation")) canonical_fields["allow_manual_location_selection"] = bool(settings.get("allow_manual_selection")) canonical_fields["location_selection_mode"] = settings.get("selection_mode") canonical_fields["location_selector_label"] = settings.get("location_label") if resolved: canonical_fields["resolved_default_location_code"] = resolved.get("selected_location_code") canonical_fields["resolved_default_location_name"] = resolved.get("selected_location_name") canonical_fields["resolved_location_selection_strategy"] = resolved.get("selection_strategy") if directory: canonical_fields["location_directory"] = directory def _normalize_location_availability_mode(is_available: bool, source_value: Any) -> str: text = str(source_value or "").strip().lower() if "grab" in text: return "grab_and_go" return "available" if bool(is_available) else "special_order" def _prefers_grab_and_go(fields: Dict[str, Any]) -> bool: for key in ("grab_n_go", "grab_ngo", "grab_go"): if _truthy_token(fields.get(key)): return True text = str(fields.get("item_availability") or fields.get("collection_availability") or "").strip().lower() return "grab" in text def _default_location_availability_mode(payload: Dict[str, Any]) -> str: if payload.get("grab_and_go_location_codes"): return "grab_and_go" if payload.get("available_location_codes"): return "available" return "special_order" def _location_code_lookup(session: Session) -> Dict[str, MasterLocationRegistry]: rows = session.scalars(select(MasterLocationRegistry).where(MasterLocationRegistry.is_active.is_(True))).all() lookup: Dict[str, MasterLocationRegistry] = {} for row in rows: if row.code: lookup[str(row.code).strip()] = row labels = [] aliases = row.aliases if isinstance(row.aliases, dict) else {} labels.extend(aliases.get("labels", []) if isinstance(aliases.get("labels"), list) else []) labels.append(row.name) for label in labels: key = _normalize_location_token(label) if key and key not in lookup: lookup[key] = row return lookup def _canonical_location_code(raw_code: str, lookup: Dict[str, MasterLocationRegistry]) -> str: text = str(raw_code or "").strip() if not text: return "" if text in lookup: return str(lookup[text].code) normalized = _normalize_location_token(text) row = lookup.get(normalized) if row is not None: return str(row.code) return text def _normalize_location_token(value: Any) -> str: text = normalize_column_key(value) if text.startswith("location_"): text = text[len("location_") :] for suffix in ("_warehouse", "_store", "_pickup_location", "_location"): if text.endswith(suffix): text = text[: -len(suffix)] return text def _truthy_token(value: Any) -> bool: return str(value or "").strip().lower() in {"1", "true", "yes", "y", "available", "in stock", "grab go", "grab and go"} def _core_fields(product: MasterProduct) -> Dict[str, Any]: return { "title": product.name, "brand": product.brand, "manufacturer": product.manufacturer, "product_family": product.product_family, "category_l1": product.category_l1, "category_l2": product.category_l2, "category_l3": product.category_l3, "collection": product.collection, "assembly_type": product.assembly_type, "item_size": product.item_size, "status": product.status, }