from __future__ import annotations from typing import Any, Dict, Iterable, List, Optional, Set import pandas as pd import sqlalchemy as sa from sqlalchemy import select from sqlalchemy.dialects.postgresql import insert from sqlalchemy.orm import Session from db.channel_exports import CHANNEL_CODES, CORE_CANONICAL_FIELDS from db.models import ChannelAttributeAlias, MasterAttributeDefinition, MasterStaticFieldMapping from db.source_imports import normalize_column_key STATIC_FIELD_GROUPS = ("core", "seo", "commerce", "media", "status") CHANNEL_COLUMNS = ("magento_attribute_code", "shopify_attribute_code", "plytix_attribute_code") CREATE_NEW_MARKER = "__create_new__" CREATE_NEW_ACTION = "create_new" CHANNEL_COLUMN_TO_CODE = { "magento_attribute_code": "magento", "shopify_attribute_code": "shopify", "plytix_attribute_code": "plytix", } TAXONOMY_FIELDS = {"category_l1", "category_l2", "category_l3", "collection"} MAGENTO_OPERATIONAL_FIELDS = { "store_code", "store_codes", "website_id", "website_ids", "source_code", "source_codes", "msi_source", "msi_sources", "inventory_source", "inventory_sources", } IMAGE_FIELDS = {"base_image", "additional_images"} DYNAMIC_TRANSFORM_FIELDS = TAXONOMY_FIELDS | MAGENTO_OPERATIONAL_FIELDS | IMAGE_FIELDS DEFAULT_CHANNEL_FIELD_ALIASES: Dict[str, Dict[str, str]] = { "shopify": { "title": "title", "brand": "vendor", "manufacturer": "vendor", "product_family": "product_type", "status": "status", "description": "description", "short_description": "custom_short_description", "meta_title": "seo_title", "meta_description": "seo_description", "meta_keywords": "meta_keywords", "product_url_slug": "handle", }, "magento": { "title": "name", "brand": "family", "manufacturer": "manufacturer", "status": "status", "description": "description", "additional_information": "additional_information", "materials": "materials", "short_description": "short_description", "meta_title": "meta_title", "meta_description": "meta_description", "meta_keywords": "meta_keyword", "product_url_slug": "url_key", }, "plytix": { "title": "label", "brand": "brand", "manufacturer": "manufacturer", "status": "status", "description": "description", "short_description": "short_description", }, } # Built-in static master fields (left column seeds). Dynamic/presentation attrs are out of scope here. DEFAULT_STATIC_MASTER_FIELDS: List[Dict[str, Any]] = [ {"master_attribute_code": "title", "master_label": "Product Title", "field_group": "core", "data_type": "text", "sort_order": 10}, {"master_attribute_code": "brand", "master_label": "Brand", "field_group": "core", "data_type": "text", "sort_order": 20}, {"master_attribute_code": "manufacturer", "master_label": "Manufacturer", "field_group": "core", "data_type": "text", "sort_order": 30}, {"master_attribute_code": "product_family", "master_label": "Product Family", "field_group": "core", "data_type": "text", "sort_order": 40}, {"master_attribute_code": "category_l1", "master_label": "Category L1", "field_group": "taxonomy", "data_type": "text", "sort_order": 50}, {"master_attribute_code": "category_l2", "master_label": "Category L2", "field_group": "taxonomy", "data_type": "text", "sort_order": 60}, {"master_attribute_code": "category_l3", "master_label": "Category L3", "field_group": "taxonomy", "data_type": "text", "sort_order": 70}, {"master_attribute_code": "collection", "master_label": "Collection", "field_group": "taxonomy", "data_type": "text", "sort_order": 80}, {"master_attribute_code": "assembly_type", "master_label": "Assembly Type", "field_group": "core", "data_type": "text", "sort_order": 90}, {"master_attribute_code": "item_size", "master_label": "Item Size", "field_group": "core", "data_type": "text", "sort_order": 100}, {"master_attribute_code": "status", "master_label": "Status", "field_group": "status", "data_type": "text", "sort_order": 110}, {"master_attribute_code": "description", "master_label": "Description", "field_group": "core", "data_type": "html", "sort_order": 120}, {"master_attribute_code": "short_description", "master_label": "Short Description", "field_group": "core", "data_type": "text", "sort_order": 130}, {"master_attribute_code": "additional_information", "master_label": "Additional Information", "field_group": "core", "data_type": "html", "sort_order": 140}, {"master_attribute_code": "materials", "master_label": "Materials", "field_group": "core", "data_type": "html", "sort_order": 150}, {"master_attribute_code": "meta_title", "master_label": "SEO Title", "field_group": "seo", "data_type": "text", "sort_order": 200}, {"master_attribute_code": "meta_description", "master_label": "SEO Description", "field_group": "seo", "data_type": "text", "sort_order": 210}, {"master_attribute_code": "meta_keywords", "master_label": "SEO Keywords", "field_group": "seo", "data_type": "text", "sort_order": 220}, {"master_attribute_code": "product_url_slug", "master_label": "Product URL Slug", "field_group": "seo", "data_type": "text", "sort_order": 230}, {"master_attribute_code": "price", "master_label": "Sell Price", "field_group": "commerce", "data_type": "number", "sort_order": 300}, {"master_attribute_code": "base_image", "master_label": "Base Image", "field_group": "media_transform", "data_type": "image", "sort_order": 400}, {"master_attribute_code": "additional_images", "master_label": "Additional Images", "field_group": "media_transform", "data_type": "image", "sort_order": 410}, ] def build_static_field_mapping_table( session: Session, *, include_inactive: bool = False, magento_connection_id: Optional[int] = None, shopify_connection_id: Optional[int] = None, include_channel_options: bool = True, ) -> Dict[str, Any]: """Pivot table for UI/CSV: all master attributes on the left, one channel column each.""" from db.channel_attribute_options import channel_attribute_options_bundle from db.master_attributes import list_master_attributes ensure_all_master_attribute_rows(session) master_attrs = list_master_attributes(session) stmt = select(MasterStaticFieldMapping) if not include_inactive: stmt = stmt.where(MasterStaticFieldMapping.is_active.is_(True)) existing = { row.master_attribute_code: row for row in session.scalars(stmt).all() } pending_create = _pending_create_by_channel(session) rows: List[Dict[str, Any]] = [] for attr in master_attrs: code = attr["attribute_code"] mapped = existing.get(code) row_snapshot = _static_mapping_snapshot(mapped) mapped_channels = [ channel for col, channel in CHANNEL_COLUMN_TO_CODE.items() if row_snapshot.get(col) ] rows.append( { "master_attribute_code": code, "master_label": row_snapshot.get("master_label") or attr.get("label") or code, "field_group": row_snapshot.get("field_group") or _default_group(code), "data_type": row_snapshot.get("data_type") or attr.get("data_type") or "text", "purpose": attr.get("purpose") or "", "product_count": int(attr.get("product_count") or 0), "attribute_source": attr.get("source") or "", "magento": _ui_channel_value_from_snapshot(row_snapshot, "magento", code, pending_create), "shopify": _ui_channel_value_from_snapshot(row_snapshot, "shopify", code, pending_create), "plytix": _ui_channel_value_from_snapshot(row_snapshot, "plytix", code, pending_create), "mapped_channel_count": len(mapped_channels), "mapped_channels": mapped_channels, "has_mapping_row": mapped is not None, "is_active": row_snapshot.get("is_active", True), "notes": row_snapshot.get("notes") or "", } ) rows.sort(key=lambda row: (row["field_group"], row["master_attribute_code"])) result: Dict[str, Any] = { "mapping_kind": "static", "channels": sorted(CHANNEL_CODES), "columns": [ "master_attribute_code", "master_label", "field_group", "data_type", "magento", "shopify", "plytix", "mapped_channel_count", "is_active", "notes", ], "row_count": len(rows), "master_attribute_count": len(master_attrs), "mapping_row_count": len(existing), "rows": rows, } if include_channel_options: result["channel_options"] = channel_attribute_options_bundle( session, magento_connection_id=magento_connection_id, shopify_connection_id=shopify_connection_id, ) return result def ensure_all_master_attribute_rows(session: Session) -> int: """Ensure every master catalog attribute has a row in the wide mapping table.""" from db.master_attributes import list_master_attributes _ensure_seeded_rows(session) existing = { code for (code,) in session.execute(select(MasterStaticFieldMapping.master_attribute_code)).all() } seeds: List[Dict[str, Any]] = [] sort_order = 5000 for attr in list_master_attributes(session): code = normalize_column_key(attr["attribute_code"]) if not code or code in existing: continue seeds.append( { "master_attribute_code": code, "master_label": attr.get("label") or code, "field_group": _default_group(code), "data_type": attr.get("data_type") or "text", "sort_order": sort_order, "is_active": True, "notes": f"auto-seeded from {attr.get('source', 'master')}", } ) sort_order += 1 return upsert_static_field_mappings(session, seeds) def save_static_field_mappings_batch( session: Session, rows: List[Dict[str, Any]], *, sync_aliases: bool = True, ) -> Dict[str, int]: """Persist UI dropdown selections (batch).""" normalized: List[Dict[str, Any]] = [] pending_actions: Dict[tuple[str, str], str] = {} for raw in rows: master_code = _clean( raw.get("master_attribute_code") or raw.get("master_attribute") or raw.get("attribute_code") ) if not master_code: continue row_payload: Dict[str, Any] = { "master_attribute_code": master_code, "master_label": _clean(raw.get("master_label") or raw.get("label")), "field_group": _clean(raw.get("field_group")) or _default_group(master_code), "data_type": _clean(raw.get("data_type")) or "text", "sort_order": _to_int(raw.get("sort_order"), default=0), "is_active": _to_bool(raw.get("is_active"), default=True), "notes": _clean(raw.get("notes")), } for channel in CHANNEL_COLUMN_TO_CODE.values(): channel_attr, action = _resolve_channel_mapping(raw, channel, master_code) row_payload[f"{channel}_attribute_code"] = channel_attr if action == CREATE_NEW_ACTION: pending_actions[(channel, master_code)] = CREATE_NEW_ACTION normalized.append(row_payload) upserted = upsert_static_field_mappings(session, normalized) aliases_synced = ( sync_static_mappings_to_channel_aliases(session, pending_actions=pending_actions) if sync_aliases and normalized else 0 ) pending_create_channels = sorted({channel for channel, _master in pending_actions.keys()}) return { "upserted": upserted, "aliases_synced": aliases_synced, "pending_create_channels": pending_create_channels, } def export_static_field_mapping_csv_rows( session: Session, *, magento_connection_id: Optional[int] = None, shopify_connection_id: Optional[int] = None, ) -> List[Dict[str, Any]]: table = build_static_field_mapping_table( session, include_inactive=True, magento_connection_id=magento_connection_id, shopify_connection_id=shopify_connection_id, include_channel_options=True, ) rows: List[Dict[str, Any]] = [] for item in table.get("rows") or []: rows.append( { "row_type": "mapping", "master_attribute_code": item.get("master_attribute_code") or "", "master_label": item.get("master_label") or "", "field_group": item.get("field_group") or "", "data_type": item.get("data_type") or "", "magento": item.get("magento") or "", "shopify": item.get("shopify") or "", "plytix": item.get("plytix") or "", "mapped_channel_count": item.get("mapped_channel_count") or 0, "is_active": item.get("is_active", True), "notes": item.get("notes") or "", "channel_code": "", "channel_attribute_code": "", "channel_label": "", "channel_data_type": "", "channel_owner_type": "", "channel_source": "", "connection_id": "", } ) channel_options = table.get("channel_options") or {} connection_ids = { "magento": magento_connection_id, "shopify": shopify_connection_id, "plytix": None, } for channel in CHANNEL_CODES: options = sorted( channel_options.get(channel, []), key=lambda opt: ( str(opt.get("label") or opt.get("code") or "").lower(), str(opt.get("code") or "").lower(), ), ) for opt in options: rows.append( { "row_type": "channel_attribute", "master_attribute_code": "", "master_label": "", "field_group": "", "data_type": "", "magento": "", "shopify": "", "plytix": "", "mapped_channel_count": "", "is_active": "", "notes": "", "channel_code": channel, "channel_attribute_code": opt.get("code") or "", "channel_label": opt.get("label") or "", "channel_data_type": opt.get("data_type") or "", "channel_owner_type": opt.get("owner_type") or "", "channel_source": opt.get("source") or "", "connection_id": connection_ids.get(channel) or "", } ) return rows def import_static_field_mapping_dataframe( session: Session, df: pd.DataFrame, *, sync_aliases: bool = True, ) -> Dict[str, int]: 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()} row_type = _clean(_first(normalized, "row_type")) or "mapping" if row_type == "channel_attribute": skipped += 1 continue master_code = _clean( _first( normalized, "master_attribute_code", "master_attribute", "canonical_code", "canonical_attribute_code", "attribute_code", ) ) if not master_code: skipped += 1 continue rows.append( { "master_attribute_code": master_code, "master_label": _clean(_first(normalized, "master_label", "label", "canonical_label")), "field_group": _clean(_first(normalized, "field_group", "group")) or _default_group(master_code), "data_type": _clean(_first(normalized, "data_type")) or "text", "magento_attribute_code": _channel_value(normalized, "magento"), "shopify_attribute_code": _channel_value(normalized, "shopify"), "plytix_attribute_code": _channel_value(normalized, "plytix"), "sort_order": _to_int(_first(normalized, "sort_order"), default=0), "is_active": _to_bool(_first(normalized, "is_active"), default=True), "notes": _clean(_first(normalized, "notes")), } ) upserted = upsert_static_field_mappings(session, rows) aliases_synced = sync_static_mappings_to_channel_aliases(session) if sync_aliases and rows else 0 return {"input_rows": len(df), "upserted": upserted, "skipped": skipped, "aliases_synced": aliases_synced} def upsert_static_field_mappings(session: Session, rows: List[Dict[str, Any]]) -> int: if not rows: return 0 stmt = insert(MasterStaticFieldMapping).values(rows) stmt = stmt.on_conflict_do_update( constraint="uq_master_static_field_mapping_master_code", set_={ "master_label": stmt.excluded.master_label, "field_group": stmt.excluded.field_group, "data_type": stmt.excluded.data_type, "magento_attribute_code": stmt.excluded.magento_attribute_code, "shopify_attribute_code": stmt.excluded.shopify_attribute_code, "plytix_attribute_code": stmt.excluded.plytix_attribute_code, "sort_order": stmt.excluded.sort_order, "is_active": stmt.excluded.is_active, "notes": stmt.excluded.notes, "updated_at": sa.func.now(), }, ) session.execute(stmt) return len(rows) def bootstrap_static_field_rows(session: Session, *, sync_aliases: bool = False) -> Dict[str, int]: """Seed left-column master static attributes (no channel mappings unless already present).""" existing = { code for (code,) in session.execute(select(MasterStaticFieldMapping.master_attribute_code)).all() } seeds = [] sort = 1000 for item in DEFAULT_STATIC_MASTER_FIELDS: code = item["master_attribute_code"] if code in existing: continue seeds.append({**item, "is_active": True, "notes": "bootstrap static field"}) for definition in session.scalars( select(MasterAttributeDefinition) .where(MasterAttributeDefinition.is_active.is_(True)) .where(MasterAttributeDefinition.purpose.in_(("seo", "operational"))) .order_by(MasterAttributeDefinition.attribute_code) ).all(): code = normalize_column_key(definition.attribute_code) if not code or code in existing or code in {s["master_attribute_code"] for s in seeds}: continue seeds.append( { "master_attribute_code": code, "master_label": definition.label or code, "field_group": "seo" if definition.purpose == "seo" else "commerce", "data_type": definition.data_type or "text", "sort_order": sort, "is_active": True, "notes": "bootstrap from master_attribute_definition", } ) sort += 1 upserted = upsert_static_field_mappings(session, seeds) ensured = ensure_all_master_attribute_rows(session) aliases_synced = sync_static_mappings_to_channel_aliases(session) if sync_aliases else 0 return {"seeded": upserted, "ensured_master_attributes": ensured, "aliases_synced": aliases_synced} def auto_match_static_field_mappings( session: Session, *, magento_connection_id: Optional[int] = None, shopify_connection_id: Optional[int] = None, channels: Optional[Iterable[str]] = None, overwrite: bool = False, sync_aliases: bool = True, ) -> Dict[str, int]: """Persist conservative 1:1 mappings where the master code exists in a channel registry. This intentionally does not infer fuzzy matches or mark missing attributes as create-new. Those choices can change channel data shape, so the UI/user should make them explicitly. """ from db.channel_attribute_options import channel_attribute_options_bundle ensure_all_master_attribute_rows(session) wanted_channels = { normalize_column_key(channel) for channel in (channels or CHANNEL_CODES) if normalize_column_key(channel) in CHANNEL_CODES } if not wanted_channels: wanted_channels = set(CHANNEL_CODES) options = channel_attribute_options_bundle( session, magento_connection_id=magento_connection_id, shopify_connection_id=shopify_connection_id, ) option_codes = { channel: { normalize_column_key(opt.get("code")): str(opt.get("code") or "").strip() for opt in channel_options if str(opt.get("code") or "").strip() } for channel, channel_options in options.items() } rows = session.scalars( select(MasterStaticFieldMapping) .where(MasterStaticFieldMapping.is_active.is_(True)) .order_by(MasterStaticFieldMapping.master_attribute_code) ).all() matched_by_channel = {channel: 0 for channel in CHANNEL_CODES} changed_rows = 0 dynamic_transform_rows = 0 for row in rows: master_key = normalize_column_key(row.master_attribute_code) changed = False if master_key in DYNAMIC_TRANSFORM_FIELDS or _default_group(master_key).endswith("_transform"): dynamic_transform_rows += 1 for channel in wanted_channels: column = f"{channel}_attribute_code" if not overwrite and _clean(getattr(row, column, None)): continue default_channel_code = DEFAULT_CHANNEL_FIELD_ALIASES.get(channel, {}).get(master_key) channel_code = None if default_channel_code and _default_alias_allowed(channel, master_key, default_channel_code, option_codes): channel_code = default_channel_code if not channel_code and master_key not in DYNAMIC_TRANSFORM_FIELDS: channel_code = option_codes.get(channel, {}).get(master_key) if not channel_code: continue setattr(row, column, channel_code) matched_by_channel[channel] += 1 changed = True if changed: changed_rows += 1 aliases_synced = sync_static_mappings_to_channel_aliases(session) if sync_aliases and changed_rows else 0 return { "changed_rows": changed_rows, "magento_matched": matched_by_channel.get("magento", 0), "shopify_matched": matched_by_channel.get("shopify", 0), "plytix_matched": matched_by_channel.get("plytix", 0), "dynamic_transform_rows": dynamic_transform_rows, "aliases_synced": aliases_synced, } def _default_alias_allowed( channel: str, master_key: str, channel_code: str, option_codes: Dict[str, Dict[str, str]], ) -> bool: """Allow built-in aliases for known top-level channel fields even if no registry row exists.""" if channel == "shopify" and channel_code in { "title", "vendor", "product_type", "status", "description", "seo_title", "seo_description", "meta_keywords", "handle", "custom_short_description", }: return True if channel == "plytix" and channel_code in {"label", "status"}: return True if channel_code in option_codes.get(channel, {}): return True # Magento payload mapper knows these top-level fields separately from custom attrs. if channel == "magento" and ( (master_key in {"title"} and channel_code == "name") or channel_code in {"description", "short_description", "meta_title", "meta_description", "meta_keyword", "url_key"} ): return True return False def import_wide_mappings_from_channel_aliases(session: Session) -> Dict[str, int]: """Fold existing static-like channel_attribute_alias rows into the wide table (1:1 per channel).""" pivot: Dict[str, Dict[str, Any]] = {} for alias in session.scalars( select(ChannelAttributeAlias) .where(ChannelAttributeAlias.is_active.is_(True)) .where(ChannelAttributeAlias.mapping_scope != "dynamic") .order_by(ChannelAttributeAlias.canonical_code, ChannelAttributeAlias.channel_code) ).all(): canonical = normalize_column_key(alias.canonical_code) channel = normalize_column_key(alias.channel_code) if channel not in CHANNEL_CODES: continue row = pivot.setdefault( canonical, { "master_attribute_code": canonical, "master_label": canonical, "field_group": _default_group(canonical), "data_type": alias.data_type or "text", "sort_order": 0, "is_active": True, "notes": "imported from channel_attribute_alias", }, ) col = f"{channel}_attribute_code" if col in CHANNEL_COLUMN_TO_CODE and not row.get(col): row[col] = normalize_column_key(alias.channel_attribute_code) for alias in session.scalars( select(ChannelAttributeAlias) .where(ChannelAttributeAlias.is_active.is_(True)) .where(ChannelAttributeAlias.mapping_scope == "dynamic") .order_by(ChannelAttributeAlias.canonical_code, ChannelAttributeAlias.channel_code) ).all(): canonical = normalize_column_key(alias.canonical_code) if canonical in CORE_CANONICAL_FIELDS or canonical.startswith(("meta_", "seo_")): channel = normalize_column_key(alias.channel_code) if channel not in CHANNEL_CODES: continue row = pivot.setdefault( canonical, { "master_attribute_code": canonical, "master_label": canonical, "field_group": _default_group(canonical), "data_type": alias.data_type or "text", "sort_order": 0, "is_active": True, "notes": "imported legacy alias row", }, ) col = f"{channel}_attribute_code" if col in CHANNEL_COLUMN_TO_CODE and not row.get(col): row[col] = normalize_column_key(alias.channel_attribute_code) upserted = upsert_static_field_mappings(session, list(pivot.values())) aliases_synced = sync_static_mappings_to_channel_aliases(session) return {"imported_master_fields": upserted, "aliases_synced": aliases_synced} def sync_static_mappings_to_channel_aliases( session: Session, *, pending_actions: Optional[Dict[tuple[str, str], str]] = None, ) -> int: """Push wide static 1:1 mappings into channel_attribute_alias for the export pipeline.""" pending_actions = pending_actions or {} rows = session.scalars( select(MasterStaticFieldMapping) .where(MasterStaticFieldMapping.is_active.is_(True)) .order_by(MasterStaticFieldMapping.sort_order, MasterStaticFieldMapping.master_attribute_code) ).all() master_codes = [row.master_attribute_code for row in rows] if master_codes: session.execute( sa.delete(ChannelAttributeAlias).where( ChannelAttributeAlias.mapping_scope == "static", ChannelAttributeAlias.canonical_code.in_(master_codes), ) ) alias_rows: List[Dict[str, Any]] = [] for row in rows: for col, channel in CHANNEL_COLUMN_TO_CODE.items(): channel_attr = _clean(getattr(row, col)) if not channel_attr: continue action = pending_actions.get((channel, row.master_attribute_code), USE_EXISTING_ACTION) alias_rows.append( { "channel_code": channel, "canonical_code": row.master_attribute_code, "channel_attribute_code": channel_attr, "mapping_scope": "static", "action": action, "data_type": row.data_type, "transform_rule": None, "is_active": True, "notes": f"static_field_map:{row.master_attribute_code}", } ) if not alias_rows: return 0 stmt = insert(ChannelAttributeAlias).values(alias_rows) stmt = stmt.on_conflict_do_update( constraint="uq_channel_attribute_alias_channel_canonical_attr", set_={ "mapping_scope": stmt.excluded.mapping_scope, "action": stmt.excluded.action, "data_type": stmt.excluded.data_type, "is_active": stmt.excluded.is_active, "notes": stmt.excluded.notes, "updated_at": sa.func.now(), }, ) session.execute(stmt) return len(alias_rows) def static_alias_map_for_channel( session: Session, channel_code: str, *, master_codes: Optional[Iterable[str]] = None, ) -> Dict[str, str]: """canonical master code → channel attribute code for static fields only.""" channel = normalize_column_key(channel_code) mapping: Dict[str, str] = {} stmt = ( select(MasterStaticFieldMapping) .where(MasterStaticFieldMapping.is_active.is_(True)) .order_by(MasterStaticFieldMapping.sort_order) ) if master_codes is not None: wanted = {normalize_column_key(code) for code in master_codes if normalize_column_key(code)} if not wanted: return mapping stmt = stmt.where(MasterStaticFieldMapping.master_attribute_code.in_(sorted(wanted))) for row in session.scalars(stmt).all(): col = f"{channel}_attribute_code" value = _clean(getattr(row, col, None)) if value: mapping[row.master_attribute_code] = value return mapping def _ensure_seeded_rows(session: Session) -> None: count = session.scalar(select(sa.func.count()).select_from(MasterStaticFieldMapping)) or 0 if count == 0: bootstrap_static_field_rows(session, sync_aliases=False) def _default_group(master_code: str) -> str: code = normalize_column_key(master_code) if code in TAXONOMY_FIELDS: return "taxonomy" if code in MAGENTO_OPERATIONAL_FIELDS or code.startswith(("location_", "msi_", "inventory_")): return "operations_transform" if code in IMAGE_FIELDS or "image" in code: return "media_transform" if code in CORE_CANONICAL_FIELDS: return "core" if code.startswith(("meta_", "seo_")): return "seo" if code in {"price", "msrp", "shopify_price"} or "price" in code: return "commerce" if code == "status": return "status" return "core" USE_EXISTING_ACTION = "use_existing" def _pending_create_by_channel(session: Session) -> Dict[tuple[str, str], bool]: from db.channel_attribute_provision import CREATE_NEW_ACTION pending: Dict[tuple[str, str], bool] = {} for alias in session.scalars( select(ChannelAttributeAlias) .where(ChannelAttributeAlias.mapping_scope == "static") .where(ChannelAttributeAlias.is_active.is_(True)) .where(ChannelAttributeAlias.action == CREATE_NEW_ACTION) ).all(): pending[(alias.channel_code, alias.canonical_code)] = True return pending def _static_mapping_snapshot(mapped: Optional[MasterStaticFieldMapping]) -> Dict[str, Any]: if mapped is None: return {} return { "master_label": mapped.master_label, "field_group": mapped.field_group, "data_type": mapped.data_type, "magento_attribute_code": mapped.magento_attribute_code, "shopify_attribute_code": mapped.shopify_attribute_code, "plytix_attribute_code": mapped.plytix_attribute_code, "is_active": mapped.is_active, "notes": mapped.notes, } def _ui_channel_value_from_snapshot( snapshot: Dict[str, Any], channel: str, master_code: str, pending_create: Dict[tuple[str, str], bool], ) -> str: if pending_create.get((channel, master_code)): return CREATE_NEW_MARKER return _clean(snapshot.get(f"{channel}_attribute_code")) or "" def _resolve_channel_mapping( raw: Dict[str, Any], channel: str, master_code: str, ) -> tuple[Optional[str], str]: value = _channel_value(raw, channel) if value == CREATE_NEW_MARKER: return master_code, CREATE_NEW_ACTION return value, USE_EXISTING_ACTION def _channel_value(normalized: Dict[str, Any], channel: str) -> Optional[str]: return _clean( _first( normalized, f"{channel}_attribute_code", f"{channel}_attribute", f"{channel}_code", channel, ) ) def _first(row: Dict[str, Any], *keys: str) -> Any: for key in keys: value = row.get(normalize_column_key(key)) if _clean(value) is not None: return value return None def _clean(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 _to_bool(value: Any, *, default: bool) -> bool: text = _clean(value) if text is None: return default return text.lower() in {"1", "true", "yes", "y", "on", "active"} def _to_int(value: Any, *, default: int) -> int: text = _clean(value) if text is None: return default try: return int(float(text)) except ValueError: return default