from __future__ import annotations from collections import defaultdict import json from typing import Any, Dict, Iterable, List, Optional import pandas as pd import sqlalchemy as sa from sqlalchemy import func, select from sqlalchemy.dialects.postgresql import insert from sqlalchemy.orm import Session from db.attribute_audit import build_attribute_audit, semantic_family_key from db.channel_aliases import channel_aliases_by_attribute, channel_aliases_by_canonical from db.models import ( MagentoAttributeCache, MagentoAttributeRegistry, MasterAttributeDefinition, MasterProductAttributeValue, PlytixAttributeCatalog, ProductAttributeMapping, ShopifyAttributeRegistry, ) from db.source_imports import infer_destination_hint, normalize_column_key ADMIN_ONLY_KEYS = { "grab_n_go", "item_style", "item_style_1", "item_style_prefix", "notes", "plytix_notes", "source_image_file_s", "product_url_slug", "product_url_portandbell_com", "lookup_key_auto", } SLUG_FIELD_KEYS = {"product_url_slug", "url_key", "handle", "slug"} PRICE_FIELD_KEYS = { "price", "base_price", "rta_price", "assembled_price", "shopify_price", "magento_price", "special_price", } MAGENTO_CORE_CODES = { "id", "sku", "name", "title", "price", "special_price", "status", "visibility", "type_id", "attribute_set_id", "created_at", "updated_at", "weight", "description", "short_description", "meta_title", "meta_description", "url_key", "category_ids", } DYNAMIC_CHANNEL_FIELD_KEYS = { "configurable_attributes", "configurable_variations", "configurable_variation_labels", "variant_list", "variant_of", "child_skus", "parent_sku", "magento_configurable_attributes", "magento_configurable_variations", "magento_configurable_variation_labels", "plytix_variations", "plytix_variants", "plytix_variant_list", "shopify_option_1", "shopify_option_2", "shopify_option_3", "shopify_option1", "shopify_option2", "shopify_option3", "shopify_option_1_name", "shopify_option_2_name", "shopify_option_3_name", "shopify_option1_name", "shopify_option2_name", "shopify_option3_name", "shopify_option_1_value", "shopify_option_2_value", "shopify_option_3_value", "shopify_option1_value", "shopify_option2_value", "shopify_option3_value", } def build_attribute_mapping_review(session: Session, *, include_magento: bool = True) -> List[Dict[str, Any]]: """Build a reviewable source-to-canonical attribute mapping worksheet.""" candidates = _master_attribute_candidates(session) candidates.extend(_plytix_catalog_candidates(session)) candidates.extend(_shopify_registry_candidates(session)) candidates.extend(_source_attribute_candidates(session, include_magento=include_magento)) mappings = _current_mappings(session) family_counts: Dict[str, int] = defaultdict(int) for row in candidates: family_counts[row["family_key"]] += 1 rows: List[Dict[str, Any]] = [] for row in candidates: mapping_key = (row["source_label"], row["source_column_key"]) current = mappings.get(mapping_key, []) target_scope = _suggest_target_scope(row) purpose = _suggest_purpose(row, target_scope) canonical_code = _suggest_canonical_code(row) transform_rule = _suggest_transform_rule(row, canonical_code, target_scope) priority = _suggest_priority(row, target_scope, canonical_code) mapped_codes = sorted({m.canonical_code for m in current if m.is_active}) mapped_scopes = sorted({m.target_scope for m in current if m.is_active}) review_status = "mapped" if mapped_codes else "unmapped" recommendation = _recommend(row, family_counts[row["family_key"]], review_status, target_scope) rows.append( { "family_key": row["family_key"], "source_system": row["source_system"], "source_label": row["source_label"], "source_column_key": row["source_column_key"], "source_column_name": row["source_column_name"], "suggested_canonical_code": canonical_code, "suggested_target_scope": target_scope, "suggested_purpose": purpose, "suggested_priority": priority, "suggested_transform_rule": json.dumps(transform_rule, sort_keys=True) if transform_rule else "", "sku_count": row["sku_count"], "sample_values": row.get("sample_values") or "", "mapping_status": review_status, "current_canonical_codes": ", ".join(mapped_codes), "current_target_scopes": ", ".join(mapped_scopes), "value_strategy": _suggest_value_strategy(row, canonical_code, target_scope), "recommendation": recommendation, } ) rows.sort( key=lambda r: ( _recommend_sort(r["recommendation"]), r["family_key"], _source_sort(r["source_system"]), r["source_label"], r["source_column_key"], ) ) return rows def build_attribute_action_plan(session: Session, *, include_magento: bool = True) -> List[Dict[str, Any]]: """Classify attributes into cleanup/add/review actions per external system.""" review_rows = build_attribute_mapping_review(session, include_magento=include_magento) desired_magento = _desired_magento_codes(review_rows) existing_magento = _existing_magento_attributes(session) existing_shopify = _existing_shopify_attributes(session) magento_aliases_by_canonical = channel_aliases_by_canonical(session, "magento") magento_aliases_by_attribute = channel_aliases_by_attribute(session, "magento") shopify_aliases_by_canonical = channel_aliases_by_canonical(session, "shopify") shopify_aliases_by_attribute = channel_aliases_by_attribute(session, "shopify") rows: List[Dict[str, Any]] = [] catalog_rows = _plytix_catalog_action_rows(session, review_rows) rows.extend(catalog_rows or _plytix_action_rows(review_rows)) rows.extend( _magento_action_rows( review_rows, desired_magento, existing_magento, magento_aliases_by_canonical, magento_aliases_by_attribute, ) ) rows.extend( _shopify_action_rows( review_rows, existing_shopify, shopify_aliases_by_canonical, shopify_aliases_by_attribute, ) ) rows.sort(key=lambda r: (_system_sort(r["system"]), _action_sort(r["action"]), r["attribute_code"], r["source_column_key"])) return rows def import_attribute_mapping_dataframe( session: Session, df: pd.DataFrame, *, default_source_label: str = "mapping_csv", accept_suggestions: bool = False, ) -> Dict[str, int]: """Upsert product_attribute_mapping rows from a CSV/review worksheet.""" rows: List[Dict[str, Any]] = [] skipped = 0 for raw in df.to_dict(orient="records"): normalized = {normalize_column_key(k): v for k, v in raw.items()} source_label = _clean_text( _first(normalized, "source_label", "source_system", default=default_source_label) ) source_column_key = normalize_column_key(_first(normalized, "source_column_key", "column_key") or "") canonical_value = _first(normalized, "canonical_code", "current_canonical_codes") if accept_suggestions and not canonical_value: canonical_value = _first(normalized, "suggested_canonical_code") canonical_code = normalize_column_key(canonical_value or "") target_scope = normalize_column_key( _first(normalized, "target_scope", "current_target_scopes") or (_first(normalized, "suggested_target_scope") if accept_suggestions else None) or "canonical" ) if not source_label or not source_column_key or not canonical_code: skipped += 1 continue rows.append( { "source_label": source_label, "source_column_name": _clean_text(_first(normalized, "source_column_name", "column_name")), "source_column_key": source_column_key, "canonical_code": canonical_code, "canonical_label": _clean_text(_first(normalized, "canonical_label")), "target_scope": target_scope or "canonical", "data_type": normalize_column_key(_first(normalized, "data_type") or "text") or "text", "transform_rule": _parse_json_dict( _first(normalized, "transform_rule") or (_first(normalized, "suggested_transform_rule") if accept_suggestions else None) ), "priority": _to_int( _first(normalized, "priority") or (_first(normalized, "suggested_priority") if accept_suggestions else None), default=100, ), "is_active": _to_bool(_first(normalized, "is_active"), default=True), "notes": _clean_text(_first(normalized, "notes", "recommendation")), } ) if rows: stmt = insert(ProductAttributeMapping).values(rows) stmt = stmt.on_conflict_do_update( constraint="uq_product_attribute_mapping_source_target", set_={ "source_column_name": stmt.excluded.source_column_name, "canonical_code": stmt.excluded.canonical_code, "canonical_label": stmt.excluded.canonical_label, "data_type": stmt.excluded.data_type, "transform_rule": stmt.excluded.transform_rule, "priority": stmt.excluded.priority, "is_active": stmt.excluded.is_active, "notes": stmt.excluded.notes, "updated_at": sa.func.now(), }, ) session.execute(stmt) return {"input_rows": len(df), "upserted": len(rows), "skipped": skipped} def export_attribute_mappings(session: Session) -> List[Dict[str, Any]]: rows = [] for mapping in session.scalars( select(ProductAttributeMapping).order_by( ProductAttributeMapping.source_label, ProductAttributeMapping.target_scope, ProductAttributeMapping.source_column_key, ) ).all(): rows.append( { "source_label": mapping.source_label, "source_column_name": mapping.source_column_name or "", "source_column_key": mapping.source_column_key, "canonical_code": mapping.canonical_code, "canonical_label": mapping.canonical_label or "", "target_scope": mapping.target_scope, "data_type": mapping.data_type, "transform_rule": json.dumps(mapping.transform_rule or {}, sort_keys=True), "priority": mapping.priority, "is_active": mapping.is_active, "notes": mapping.notes or "", } ) return rows def _plytix_action_rows(review_rows: List[Dict[str, Any]]) -> List[Dict[str, Any]]: rows = [] seen_codes: set[str] = set() for row in review_rows: if row["source_system"] != "plytix": continue seen_codes.add(row["source_column_key"]) code = row["source_column_key"] scope = row["suggested_target_scope"] if scope == "dynamic": action = "delete_from_plytix_after_mapping" reason = "dynamic_channel_field_generated_from_canonical_variation_model" elif scope in {"shopify", "magento", "admin"} or code.startswith(("shopify_", "magento_", "mage_")): action = "delete_from_plytix_after_mapping" reason = f"{scope}_scoped_or_prefixed" elif row["recommendation"] == "merge_review": action = "merge_or_delete_duplicate_in_plytix" reason = "same canonical family exists in other sources" else: action = "keep_or_map_in_plytix" reason = "candidate canonical source" rows.append(_action_row("plytix", action, code, row, reason)) return rows def _plytix_catalog_action_rows(session: Session, review_rows: List[Dict[str, Any]]) -> List[Dict[str, Any]]: review_by_code = { row["source_column_key"]: row for row in review_rows if row.get("source_system") == "plytix" } rows = [] for attr in session.scalars( select(PlytixAttributeCatalog).where(PlytixAttributeCatalog.is_active.is_(True)) ).all(): code = normalize_column_key(attr.attribute_code) target_scope = _target_scope_for_catalog_code(code, attr.destination_hint) row = review_by_code.get(code) or { "source_system": "plytix_catalog", "source_label": attr.source_label, "source_column_key": code, "suggested_canonical_code": _canonical_code_for_key(code), "suggested_target_scope": target_scope, "suggested_purpose": "channel_dynamic" if target_scope == "dynamic" else "presentation", "sku_count": "", "sample_values": attr.label or "", "recommendation": "catalog_review", } scope = row["suggested_target_scope"] if scope == "dynamic": action = "delete_from_plytix_after_mapping" reason = "dynamic_channel_field_generated_from_canonical_variation_model" elif scope in {"shopify", "magento", "admin"} or code.startswith(("shopify_", "magento_", "mage_")): action = "delete_from_plytix_after_mapping" reason = f"{scope}_scoped_or_prefixed" elif code in review_by_code: action = "keep_or_map_in_plytix" reason = "present in current Plytix product source" else: action = "delete_from_plytix_review" reason = "catalog attribute not present in current Plytix product source" rows.append(_action_row("plytix", action, code, row, reason)) return rows def _magento_action_rows( review_rows: List[Dict[str, Any]], desired_magento: Dict[str, Dict[str, Any]], existing_magento: Dict[str, Dict[str, Any]], aliases_by_canonical: Optional[Dict[str, List[Any]]] = None, aliases_by_attribute: Optional[Dict[str, Any]] = None, ) -> List[Dict[str, Any]]: rows = [] aliases_by_canonical = aliases_by_canonical or {} aliases_by_attribute = aliases_by_attribute or {} emitted_alias_attributes: set[str] = set() for code, row in desired_magento.items(): if code in MAGENTO_CORE_CODES: rows.append(_action_row("magento", "use_core_field_or_existing_attribute", code, row, "Magento core field")) elif code in existing_magento: rows.append(_action_row("magento", "keep_existing_attribute", code, row, "exists in Magento registry/cache")) elif code in aliases_by_canonical: for alias in aliases_by_canonical[code]: emitted_alias_attributes.add(normalize_column_key(alias.channel_attribute_code)) alias_row = dict(row) alias_row["source_column_key"] = alias.channel_attribute_code rows.append( _action_row( "magento", "use_existing_channel_alias", alias.channel_attribute_code, alias_row, f"alias for canonical {code}", ) ) else: rows.append(_action_row("magento", "add_attribute_to_magento_review", code, row, "needed by mapped canonical/channel data")) for code, meta in existing_magento.items(): if code in desired_magento or code in MAGENTO_CORE_CODES: continue alias = aliases_by_attribute.get(code) if alias is not None: if normalize_column_key(code) in emitted_alias_attributes: continue rows.append( { "system": "magento", "action": "keep_existing_channel_alias", "attribute_code": code, "source_system": "magento_registry", "source_label": f"connection:{meta.get('connection_id', '')}", "source_column_key": code, "canonical_code": alias.canonical_code, "target_scope": "magento", "purpose": "channel_specific", "sku_count": "", "sample_values": meta.get("frontend_label") or "", "reason": f"alias for canonical {alias.canonical_code}", } ) continue action = "delete_from_magento_review" if meta.get("is_user_defined") else "keep_magento_system_attribute" reason = "existing Magento attribute not in desired mapping" rows.append( { "system": "magento", "action": action, "attribute_code": code, "source_system": "magento_registry", "source_label": f"connection:{meta.get('connection_id', '')}", "source_column_key": code, "canonical_code": "", "target_scope": "magento", "purpose": "channel_specific", "sku_count": "", "sample_values": meta.get("frontend_label") or "", "reason": reason, } ) return rows def _shopify_action_rows( review_rows: List[Dict[str, Any]], existing_shopify: Optional[Dict[str, Dict[str, Any]]] = None, aliases_by_canonical: Optional[Dict[str, List[Any]]] = None, aliases_by_attribute: Optional[Dict[str, Any]] = None, ) -> List[Dict[str, Any]]: rows = [] existing = existing_shopify or {} aliases_by_canonical = aliases_by_canonical or {} aliases_by_attribute = aliases_by_attribute or {} emitted_alias_attributes: set[str] = set() seen: set[str] = set() for row in review_rows: scope = row["suggested_target_scope"] code = row["suggested_canonical_code"] if scope in {"admin", "pricing", "magento", "dynamic"} or code in {"sku", "title", "description", "seo_title", "seo_description"}: continue if not _should_publish_to_shopify(row): continue if code not in seen: seen.add(code) if code in existing: action = "keep_existing_shopify_field" reason = "exists in Shopify registry" elif code in aliases_by_canonical: for alias in aliases_by_canonical[code]: emitted_alias_attributes.add(normalize_column_key(alias.channel_attribute_code)) alias_row = dict(row) alias_row["source_column_key"] = alias.channel_attribute_code rows.append( _action_row( "shopify", "use_existing_channel_alias", alias.channel_attribute_code, alias_row, f"alias for canonical {code}", ) ) continue else: action = "add_shopify_metafield_review" if scope != "shopify" else "keep_or_add_shopify_field_review" reason = "needed by mapped canonical/channel data" rows.append(_action_row("shopify", action, code, row, reason)) for code, meta in existing.items(): if code in seen or code in {"sku", "title", "description", "seo_title", "seo_description"}: continue real_code = meta.get("attribute_code") or code alias = aliases_by_attribute.get(normalize_column_key(real_code)) or aliases_by_attribute.get(code) if alias is not None: if normalize_column_key(real_code) in emitted_alias_attributes: continue rows.append( { "system": "shopify", "action": "keep_existing_channel_alias", "attribute_code": real_code, "source_system": "shopify_registry", "source_label": meta.get("shop_code") or "", "source_column_key": real_code, "canonical_code": alias.canonical_code, "target_scope": "shopify", "purpose": "channel_specific", "sku_count": "", "sample_values": meta.get("name") or "", "reason": f"alias for canonical {alias.canonical_code}", } ) continue action = "keep_shopify_standard_field" if meta.get("is_standard") else "delete_from_shopify_review" rows.append( { "system": "shopify", "action": action, "attribute_code": real_code, "source_system": "shopify_registry", "source_label": meta.get("shop_code") or "", "source_column_key": real_code, "canonical_code": code if real_code != code else "", "target_scope": "shopify", "purpose": "channel_specific", "sku_count": "", "sample_values": meta.get("name") or "", "reason": "existing Shopify field not in desired mapping", } ) return rows def _desired_magento_codes(review_rows: List[Dict[str, Any]]) -> Dict[str, Dict[str, Any]]: desired: Dict[str, Dict[str, Any]] = {} for row in review_rows: scope = row["suggested_target_scope"] purpose = row["suggested_purpose"] code = row["suggested_canonical_code"] if scope in {"admin", "pricing", "shopify", "dynamic"}: continue if not _should_publish_to_magento(row): continue if purpose not in {"presentation", "seo", "channel_specific"}: continue if code not in desired or int(row.get("suggested_priority") or 100) < int(desired[code].get("suggested_priority") or 100): desired[code] = row return desired def _should_publish_to_magento(row: Dict[str, Any]) -> bool: scope = row.get("suggested_target_scope") source = row.get("source_system") if scope == "magento": return True return source == "master" and scope in {"canonical", "seo"} def _should_publish_to_shopify(row: Dict[str, Any]) -> bool: scope = row.get("suggested_target_scope") source = row.get("source_system") if scope == "shopify": return source not in {"plytix", "plytix_catalog"} return source == "master" and scope in {"canonical", "seo"} def _existing_magento_attributes(session: Session) -> Dict[str, Dict[str, Any]]: rows: Dict[str, Dict[str, Any]] = {} for model in (MagentoAttributeRegistry, MagentoAttributeCache): for attr in session.scalars(select(model)).all(): code = normalize_column_key(attr.attribute_code) rows.setdefault( code, { "connection_id": getattr(attr, "connection_id", None), "frontend_label": getattr(attr, "frontend_label", None), "frontend_input": getattr(attr, "frontend_input", None), "backend_type": getattr(attr, "backend_type", None), "is_user_defined": bool(getattr(attr, "is_user_defined", False)), }, ) return rows def _existing_shopify_attributes(session: Session) -> Dict[str, Dict[str, Any]]: rows: Dict[str, Dict[str, Any]] = {} for attr in session.scalars(select(ShopifyAttributeRegistry)).all(): code = normalize_column_key(attr.attribute_code) match_code = _shopify_match_code(attr.namespace, attr.key, code) rows.setdefault( match_code, { "shop_code": attr.shop_code, "attribute_code": code, "owner_type": attr.owner_type, "name": attr.name, "namespace": attr.namespace, "key": attr.key, "data_type": attr.data_type, "is_standard": attr.is_standard, "fetched_at": attr.fetched_at, }, ) return rows def _action_row(system: str, action: str, attribute_code: str, row: Dict[str, Any], reason: str) -> Dict[str, Any]: return { "system": system, "action": action, "attribute_code": attribute_code, "source_system": row.get("source_system") or "", "source_label": row.get("source_label") or "", "source_column_key": row.get("source_column_key") or "", "canonical_code": row.get("suggested_canonical_code") or attribute_code, "target_scope": row.get("suggested_target_scope") or "", "purpose": row.get("suggested_purpose") or "", "sku_count": row.get("sku_count") or "", "sample_values": row.get("sample_values") or "", "reason": reason, } def _system_sort(system: str) -> int: return {"plytix": 0, "magento": 1, "shopify": 2}.get(system, 9) def _action_sort(action: str) -> int: order = { "delete_from_plytix_after_mapping": 0, "merge_or_delete_duplicate_in_plytix": 1, "add_attribute_to_magento_review": 2, "delete_from_magento_review": 3, "add_shopify_metafield_review": 4, "keep_existing_shopify_field": 5, "use_existing_channel_alias": 6, "delete_from_shopify_review": 7, } return order.get(action, 50) def _master_attribute_candidates(session: Session) -> List[Dict[str, Any]]: value_counts = { code: int(count) for code, count in session.execute( select(MasterProductAttributeValue.attribute_code, func.count(func.distinct(MasterProductAttributeValue.product_id))) .group_by(MasterProductAttributeValue.attribute_code) ).all() } samples = _attribute_samples(session) rows = [] for definition in session.scalars(select(MasterAttributeDefinition).where(MasterAttributeDefinition.is_active.is_(True))).all(): code = normalize_column_key(definition.attribute_code) rows.append( { "family_key": semantic_family_key(code), "source_system": "master", "source_label": "master_catalog", "source_column_key": code, "source_column_name": definition.label or code, "suggested_canonical_code": _canonical_code_for_key(code), "destination_hint": None, "sku_count": value_counts.get(code, 0), "sample_values": samples.get(code, ""), "purpose": definition.purpose, } ) return rows def _source_attribute_candidates(session: Session, *, include_magento: bool) -> List[Dict[str, Any]]: rows = [] for row in build_attribute_audit(session, include_magento=include_magento): column_key = normalize_column_key(row["column_key"]) rows.append( { "family_key": row["family_key"], "source_system": row["source_system"], "source_label": row["source_label"] or f"{row['source_system']}_source", "source_column_key": column_key, "source_column_name": row["column_name"], "suggested_canonical_code": _canonical_code_for_key(column_key, family_key=row["family_key"]), "destination_hint": row.get("destination_hint"), "sku_count": int(row.get("sku_count") or 0), "sample_values": row.get("sample_values") or "", "purpose": None, } ) return rows def _plytix_catalog_candidates(session: Session) -> List[Dict[str, Any]]: rows = [] for attr in session.scalars( select(PlytixAttributeCatalog).where(PlytixAttributeCatalog.is_active.is_(True)) ).all(): code = normalize_column_key(attr.attribute_code) scope = _target_scope_for_catalog_code(code, attr.destination_hint) rows.append( { "family_key": semantic_family_key(code), "source_system": "plytix_catalog", "source_label": attr.source_label, "source_column_key": code, "source_column_name": attr.label or code, "suggested_canonical_code": _canonical_code_for_key(code), "destination_hint": attr.destination_hint, "sku_count": "", "sample_values": attr.label or "", "purpose": "admin" if scope == "admin" else None, } ) return rows def _shopify_registry_candidates(session: Session) -> List[Dict[str, Any]]: rows = [] for attr in session.scalars(select(ShopifyAttributeRegistry)).all(): code = normalize_column_key(attr.attribute_code) match_code = _shopify_match_code(attr.namespace, attr.key, code) rows.append( { "family_key": semantic_family_key(match_code), "source_system": "shopify_registry", "source_label": attr.shop_code, "source_column_key": code, "source_column_name": attr.name or code, "suggested_canonical_code": _canonical_code_for_key(match_code), "destination_hint": "shopify", "sku_count": "", "sample_values": attr.name or "", "purpose": "channel_specific" if not attr.is_standard else None, } ) return rows def _shopify_match_code(namespace: Optional[str], key: Optional[str], fallback_code: str) -> str: ns = normalize_column_key(namespace or "") keyed = normalize_column_key(key or "") if ns == "custom" and keyed: return keyed return normalize_column_key(fallback_code) def _attribute_samples(session: Session) -> Dict[str, str]: samples: Dict[str, List[str]] = defaultdict(list) stmt = select(MasterProductAttributeValue.attribute_code, MasterProductAttributeValue.value).order_by( MasterProductAttributeValue.attribute_code, MasterProductAttributeValue.id, ) for code, value in session.execute(stmt).all(): key = normalize_column_key(code) text = _value_to_text(value) if text is None or text in samples[key]: continue if len(samples[key]) < 5: samples[key].append(text[:160]) return {key: " | ".join(values) for key, values in samples.items()} def _current_mappings(session: Session) -> Dict[tuple, List[ProductAttributeMapping]]: rows: Dict[tuple, List[ProductAttributeMapping]] = defaultdict(list) for mapping in session.scalars(select(ProductAttributeMapping)).all(): rows[(mapping.source_label, normalize_column_key(mapping.source_column_key))].append(mapping) return rows def _suggest_target_scope(row: Dict[str, Any]) -> str: hint = row.get("destination_hint") key = normalize_column_key(row["source_column_key"]) if _is_dynamic_channel_key(key): return "dynamic" if key in ADMIN_ONLY_KEYS: return "admin" if key in PRICE_FIELD_KEYS: return "pricing" if hint == "shopify" or key.startswith("shopify_"): return "shopify" if hint == "magento" or key.startswith(("magento_", "mage_")): return "magento" if hint == "seo" or key.startswith(("seo_", "meta_")): return "seo" if row.get("purpose") == "admin": return "admin" return "canonical" def _target_scope_for_catalog_code(code: str, destination_hint: Optional[str]) -> str: key = normalize_column_key(code) if _is_dynamic_channel_key(key): return "dynamic" if key in ADMIN_ONLY_KEYS: return "admin" if key in PRICE_FIELD_KEYS: return "pricing" if destination_hint: return normalize_column_key(destination_hint) if key.startswith("shopify_"): return "shopify" if key.startswith(("magento_", "mage_")): return "magento" if key.startswith(("seo_", "meta_")): return "seo" return "canonical" def _suggest_canonical_code(row: Dict[str, Any]) -> str: return _canonical_code_for_key(row["source_column_key"], family_key=row.get("family_key")) def _suggest_purpose(row: Dict[str, Any], target_scope: str) -> str: key = normalize_column_key(row["source_column_key"]) if target_scope == "dynamic" or _is_dynamic_channel_key(key): return "channel_dynamic" if target_scope == "admin" or key in ADMIN_ONLY_KEYS: return "admin" if target_scope == "seo": return "seo" if target_scope == "pricing" or key in PRICE_FIELD_KEYS: return "operational" if row.get("purpose"): return str(row["purpose"]) if target_scope in {"shopify", "magento"}: return "channel_specific" if key in {"price", "qty", "quantity", "is_in_stock", "stock_status"}: return "operational" return "presentation" def _suggest_priority(row: Dict[str, Any], target_scope: str, canonical_code: str) -> int: key = normalize_column_key(row["source_column_key"]) source = row.get("source_system") if target_scope == "dynamic": return 15 if target_scope == "pricing": return 20 if target_scope == "seo" or key.startswith(("seo_", "meta_")): return 10 if canonical_code in {"title", "description"} and key.startswith("shopify_"): return 20 if source == "master": return 30 if source == "supplemental": return 40 if source == "plytix": return 50 if source == "magento": return 90 return 100 def _suggest_transform_rule(row: Dict[str, Any], canonical_code: str, target_scope: str) -> Optional[Dict[str, Any]]: key = normalize_column_key(row["source_column_key"]) if key in SLUG_FIELD_KEYS: return { "type": "channel_slug_override", "default": "let_channel_generate", "when_explicit": "send_and_store_channel_response", "unique_retry": {"suffix_start": 2, "pattern": "{slug}-{n}"}, } if key == "rta_price": return { "type": "price_variant_split", "price_type": "rta", "base_sku_prefix": "RTA-", "if_prefixed_sku_exists": "assign_to_prefixed_sku", "if_missing": "create_prefixed_sku", } if key == "assembled_price": return { "type": "price_by_assembly_type", "price_type": "assembled", "use_when": {"assembly_type": "Assembled"}, } if key in {"price", "base_price"}: return { "type": "price_by_assembly_type", "assembled_source": "assembled_price", "rta_source": "rta_price", "fallback_source": key, } if key.endswith("_lb") or key in {"weight_lb", "packaged_gross_weight", "assembled_net_weight"}: return {"type": "unit", "quantity": "weight", "from": "lb", "to": "channel_weight_unit"} if key.endswith("_kg"): return {"type": "unit", "quantity": "weight", "from": "kg", "to": "channel_weight_unit"} if key.endswith("_in") or key in { "width_in", "height_in", "depth_in", "assembled_width", "assembled_height", "assembled_depth", "packaged_dimension_width", "packaged_dimension_height", "packaged_dimension_depth", }: return {"type": "unit", "quantity": "length", "from": "in", "to": "channel_length_unit"} if key.endswith("_cm"): return {"type": "unit", "quantity": "length", "from": "cm", "to": "channel_length_unit"} if canonical_code == "assembly_type": return {"type": "normalize_enum", "map": {"unassembled": "RTA", "rta": "RTA", "assembled": "Assembled"}} if canonical_code == "title" and target_scope == "canonical": return {"type": "clean_title", "fallbacks": ["seo_title", "label", "name", "display_name", "sku"]} if target_scope == "dynamic": if key.startswith("shopify_option"): return { "type": "channel_variation_projection", "channel": "shopify", "max_options": 3, "source": "canonical_variation_model", } if "configurable" in key: return { "type": "channel_variation_projection", "channel": "magento", "source": "canonical_variation_model", } return {"type": "channel_variation_projection", "source": "canonical_variation_model"} return None def _suggest_value_strategy(row: Dict[str, Any], canonical_code: str, target_scope: str) -> str: key = normalize_column_key(row["source_column_key"]) if target_scope == "dynamic": if key.startswith("shopify_option"): return "generated_for_shopify_from_canonical_variation_model_max_3_options" if "configurable" in key: return "generated_for_magento_from_canonical_variation_model" return "generated_channel_mechanic" if key in SLUG_FIELD_KEYS: return "channel_generates_slug_by_default; explicit_slug_override_requires_unique_retry_and_store_response" if target_scope == "pricing": if key == "rta_price": return "rta_price_goes_to_existing_or_new_rta_prefixed_sku" if key == "assembled_price": return "assembled_price_used_when_item_is_assembled" return "sell_price_source" if target_scope == "admin": return "admin_only" if canonical_code == "title": if target_scope == "seo" or key.startswith(("seo_", "meta_")): return "seo_title_override" if key in {"label", "name", "display_name", "product_title", "title"}: return "canonical_title_source; ai_title_can_override_later" if canonical_code == "description": if target_scope == "seo" or key.startswith(("seo_", "meta_")): return "seo_description_override" return "description_source" if target_scope in {"shopify", "magento"}: return f"{target_scope}_specific" return "canonical_source" def _canonical_code_for_key(column_key: str, *, family_key: Optional[str] = None) -> str: key = normalize_column_key(column_key) family = normalize_column_key(family_key or semantic_family_key(key)) for prefix in ("shopify_", "shopify_product_", "shopify_variant_", "magento_", "mage_", "plytix_"): if key.startswith(prefix): key = key[len(prefix):] break if key in {"label", "name", "display_name", "product_title", "title", "page_title", "title_tag"}: return "title" if key in {"meta_title", "metatitle", "seo_title"}: return "seo_title" if key in {"description", "body_html", "html_description", "product_description"}: return "description" if key in {"meta_description", "metadescription", "seo_description"}: return "seo_description" if key in {"assembled_or_rta"}: return "assembly_type" if key in PRICE_FIELD_KEYS: return key return family or key def _recommend(row: Dict[str, Any], family_count: int, mapping_status: str, target_scope: str) -> str: if mapping_status == "mapped": return "mapped_review" if target_scope == "dynamic": return "channel_dynamic_review" if target_scope in {"shopify", "magento"}: return "channel_specific_review" if target_scope == "seo": return "seo_review" if row["source_system"] == "master": return "canonical_candidate" if family_count > 1: return "merge_review" return "map_review" def _value_to_text(value: Any) -> Optional[str]: if value is None: return None text = str(value).strip() if not text or text.lower() in {"nan", "none", "null", "-", "--", "\u2014"}: return None return text def _source_sort(source: str) -> int: return {"master": 0, "plytix_catalog": 1, "plytix": 2, "supplemental": 3, "magento": 4}.get(source, 9) def _recommend_sort(recommendation: str) -> int: return { "merge_review": 0, "canonical_candidate": 1, "channel_dynamic_review": 2, "channel_specific_review": 3, "seo_review": 4, "map_review": 5, "mapped_review": 6, }.get(recommendation, 9) def _is_dynamic_channel_key(key: str) -> bool: normalized = normalize_column_key(key) if normalized in DYNAMIC_CHANNEL_FIELD_KEYS: return True if normalized.startswith(("shopify_option_", "shopify_option")): return True if normalized.startswith(("option_1", "option_2", "option_3", "option1", "option2", "option3")): return True return "configurable_variation" in normalized def _first(row: Dict[str, Any], *keys: str, default: Any = None) -> Any: for key in keys: wanted = normalize_column_key(key) value = row.get(wanted) if _clean_text(value) is not None: return value return default def _clean_text(value: Any) -> Optional[str]: if value is None: return None text = str(value).strip() if not text or text.lower() in {"nan", "none", "null"}: return None return text def _parse_json_dict(value: Any) -> Optional[Dict[str, Any]]: text = _clean_text(value) if not text: return None try: parsed = json.loads(text) except json.JSONDecodeError: return {"rule": text} return parsed if isinstance(parsed, dict) else {"value": parsed} def _to_int(value: Any, *, default: int) -> int: text = _clean_text(value) if not text: return default try: return int(float(text)) except ValueError: return default def _to_bool(value: Any, *, default: bool) -> bool: text = _clean_text(value) if not text: return default return text.lower() in {"1", "true", "yes", "y", "on", "active"}