"""One-time seed of v1 master attributes into the v2 attribute registry. Copies definitions, locked vocabularies, observed select values, and channel aliases. Existing v2 rows are left alone so operator edits survive a re-run. """ from __future__ import annotations from dataclasses import dataclass, field from datetime import datetime, timezone import re from typing import Any, Iterable from sqlalchemy import delete, inspect, select from sqlalchemy.exc import OperationalError, ProgrammingError from sqlalchemy.orm import Session from db.channel_variation_axes import _CANONICAL_AXIS_LABELS from db.models import ( AttributeDef, AttributeOption, ChannelAttributeAlias, MagentoAttributeOptionRegistry, MagentoAttributeRegistry, ManualAttributeVocabulary, ManualAttributeVocabularyValue, MasterAttributeDefinition, MasterCollectionRegistry, MasterProductAttributeValue, MasterProductFamily, MasterProductRelation, MasterTaxonomyNode, ShopifyAttributeRegistry, ) from db.models_v2 import ( V2Attribute, V2AttributeSeedCandidate, V2AttributeSeedMappingCandidate, V2AttributeValue, V2AttributeValueSeedCandidate, V2ChannelAttributeMapping, V2ChannelAttributeValueMapping, V2ConfigurableLink, V2Product, ) from db.source_imports import normalize_column_key from db.v2_attribute_types import ( APPLY_READY_STATUSES, DEFAULT_OWNER_SCOPE, OWNER_SCOPES, SELECTABLE_TYPES, channel_compatibility, default_ui_control, magento_frontend_input, map_canonical_data_type, shopify_metafield_type, ) from db.v2_import.matching import attribute_code_from_header, normalize_lookup_value from db.collection_landing_schema import COLLECTION_LANDING_FIELDS, MAGENTO_ATTRIBUTE_PREFIX IDENTITY_CODES = frozenset( { "sku", "product_type", "lifecycle_status", "brand", "manufacturer", } ) CONTENT_SEED_CODES = ( "title", "description", "short_description", "features", "specifications", "additional_information", "materials", "meta_title", "meta_description", ) CONTENT_LABELS = { "title": "Product Title", "description": "Description", "short_description": "Short Description", "features": "Features", "specifications": "Specifications", "additional_information": "Additional Information", "materials": "Materials", "meta_title": "SEO Title", "meta_description": "SEO Description", "name": "Product Title", } CONTENT_DATA_TYPES = { "title": "text", "description": "html", "short_description": "html", "features": "html", "specifications": "html", "additional_information": "html", "materials": "html", "meta_title": "text", "meta_description": "textarea", } CONTENT_PRODUCT_FIELDS = frozenset( { "title", "name", "description", "meta_title", "seo_title", "meta_description", "seo_description", "meta_keywords", "url_key", "url_slug", "product_url_slug", "slug", "handle", } ) CONTENT_CODES = frozenset(CONTENT_SEED_CODES) | CONTENT_PRODUCT_FIELDS | { "short_description", "additional_information", "features", "specifications", "specs", "materials", "cabinet_description", "cabinet_style_short_description", "display_name", "product_title", "product_features", } CONTENT_ALIAS_SKIP = frozenset( { "name", "display_name", "product_title", "specs", "product_features", "seo_title", "seo_description", } ) CONTENT_COMPOSITION = { "title": ("name", "title", "display_name"), "description": ( "description", "cabinet_description", "cabinet_style_short_description", "short_description", "overlay", "soft_close", "drawer_slides", "hinges", "shelves_included", ), "short_description": ("short_description", "cabinet_style_short_description", "cabinet_description"), "features": ("overlay", "soft_close", "drawer_slides", "hinges", "shelves_included"), "specifications": ( "additional_information", "materials", "manufacturer", "family", "collection_style", "warranty", "care_instructions", "packaged_dimension_height", "packaged_dimension_width", "packaged_dimension_depth", "packaged_gross_weight", "soft_close", "shelves_included", "assembly_type", "cabinet_back_material", "cabinet_bottom_material", "cabinet_top_material", "cabinet_carcass_material", "cabinet_shelf_material", "drawer_box", "box_construction", ), "additional_information": ( "additional_information", "manufacturer", "family", "collection_style", "warranty", "care_instructions", "packaged_dimension_height", "packaged_dimension_width", "packaged_dimension_depth", "packaged_gross_weight", "soft_close", "shelves_included", "assembly_type", ), "materials": ( "materials", "cabinet_exterior_side_finish", "cabinet_back_material", "cabinet_bottom_material", "cabinet_top_material", "cabinet_carcass_material", "cabinet_toe_kick_construction", "cabinet_interior_finish", "materials_box", "materials_face_frame", "shelf_clips", "box_construction", "drawer_box", ), "meta_title": ("meta_title", "seo_title", "title", "name"), "meta_description": ("meta_description", "seo_description", "short_description", "description"), } CONTENT_CHANNEL_ALIASES = { "title": {"magento": "name", "shopify": "title"}, "description": {"magento": "description", "shopify": "description"}, "short_description": {"magento": "short_description", "shopify": "custom_short_description"}, "additional_information": {"magento": "additional_information"}, "materials": {"magento": "materials"}, "meta_title": {"magento": "meta_title", "shopify": "seo_title"}, "meta_description": {"magento": "meta_description", "shopify": "seo_description"}, } CONTENT_NOTE = ( "Product content field. Title/description/SEO live on the product record, not as select options. " "Features and specs are often composed from several attributes at publish time. " "Do not treat unique titles or HTML blobs as option values." ) CONTENT_PRODUCT_NOTE = ( "Maps to a V2 product column (title, description, or SEO), not a v2_attribute row. " "Magento uses name/description/short_description; Shopify uses title/descriptionHtml. " "Apply skips creating an EAV attribute." ) TAXONOMY_CODES = frozenset( { "hub", "category", "categories", "categories_raw", "category_ids", "category_l1", "category_l2", "category_l3", "collection", "shopping_l1", "shopping_l2", "shopping_collection", } ) PIPELINE_CHANNELS = frozenset({"magento", "shopify", "plytix"}) SYSTEM_DYNAMIC_CODES = frozenset( { "item_style_1", "grab_n_go", "grab_go", "grab_ngo", "price", "rta_price", "assembled_price", "shopify_price", "price_starting_at", "image", "thumbnail", "small_image", "swatch_image", "children", "components", "configurable_attributes", "configurable_variations", "configurable_variation_labels", "configurable_variation_axis", "variant_list", "variants_list", "variant_of", "variants", "variant", "associated_skus", "associated_products", } ) CHANNEL_POLICY_CODES = frozenset( { "visibility", "status", "product_online", "shopify_status", "published", "published_scope", "published_at", } ) SKIP_ATTRIBUTE_CODES = IDENTITY_CODES | SYSTEM_DYNAMIC_CODES SKIP_APPLY_CODES = SKIP_ATTRIBUTE_CODES | TAXONOMY_CODES | CHANNEL_POLICY_CODES IDENTITY_AND_TAXONOMY_CODES = SKIP_APPLY_CODES SYSTEM_DYNAMIC_NOTE = ( "System/dynamic field (variants, relations, Magento blobs). " "Owned by the variation builder / channel APIs, not a V2 product attribute." ) CHANNEL_POLICY_NOTE = ( "Channel listing policy, not a shared product attribute. " "Magento uses visibility (Not Visible Individually, Catalog, Search, Catalog/Search) and status. " "Shopify uses product status (ACTIVE/DRAFT) plus sales-channel publications " "(Online Store, POS, Shop, and installed apps such as Google or Temu). " "Do not map Magento visibility options onto Shopify. Apply skips this row." ) TAXONOMY_NOTE = ( "Taxonomy-owned. Values are active V1 master_taxonomy_node names. " "V2 stores these on v2_category, not as a product attribute. Apply will skip this row." ) COLLECTION_FAMILY_CODES = frozenset({"family", "collection_style", "style_family"}) COLLECTION_FAMILY_NOTE = ( "Magento Family is the storefront name for Collection, not product family " "(Cabinets, Vanities, Doors). Keep collection names and shared accessory groups " "(Snow White Accessories). Drop category prefixes (Kitchen Cabinets / …) and " "category-tree paths (Base Cabinets, Wall Cabinets, mouldings)." ) PRODUCT_FAMILY_NOTE = ( "High-level product family: Cabinets, Vanities, Doors, and similar registry names. " "Do not copy Magento Family option labels here — those are collections and accessories." ) _COLLECTION_FAMILY_PREFIXES = ( "kitchen cabinets /", "bathroom vanities /", "kitchen cabinets", "bathroom vanities", ) _FALLBACK_PRODUCT_FAMILY_LABELS = ( "Cabinets", "Vanities", "Doors", "Windows", "Showers", "Locks & Hardware", "Kitchen Package", "Sample Door", ) _ACCESSORY_HYPHEN_RE = re.compile(r"^(?P.+?)\s*[-–]\s*accessories$", re.I) _ACCESSORY_SPACE_RE = re.compile(r"^(?P.+?)\s+accessories$", re.I) _VALUE_LOCKED_STATUSES = frozenset({"approved", "modified", "merged", "excluded"}) TAXONOMY_LABELS = frozenset( { "category", "categories", "category ids", "hub", "collection", "shopping l1", "shopping l2", "shopping collection", } ) SYSTEM_DYNAMIC_LABELS = frozenset( { "variants", "variant", "variants list", "variant list", "configurable variations", "configurable attributes", "children", "associated products", "associated skus", "components", } ) CHANNEL_POLICY_LABELS = frozenset( { "visibility", "product visibility", "online visibility", "product online", "shopify status", "published scope", } ) COLLECTION_LANDING_KEYS = frozenset(str(field["key"]) for field in COLLECTION_LANDING_FIELDS) VARIANT_AXIS_CODES = frozenset( key for key in _CANONICAL_AXIS_LABELS if str(key).startswith("variation_") ) | { "configurable_variation_axis", "variation_axis", "width", "height", "depth", "assembled_width", "assembled_height", "assembled_depth", } @dataclass class AttributeDraft: code: str label: str data_type: str is_variant_axis: bool = False is_channel_required: bool = False source: str = "master_attribute_definition" sources: list[str] = field(default_factory=list) magento_frontend_input: str | None = None magento_attribute_code: str | None = None shopify_representation: dict = field(default_factory=dict) conflict_reasons: list[str] = field(default_factory=list) values: dict[str, str] = field(default_factory=dict) value_aliases: dict[str, list[str]] = field(default_factory=dict) owner_scope: str = DEFAULT_OWNER_SCOPE @dataclass class ChannelMappingDraft: channel: str canonical_code: str channel_attribute_code: str channel_attribute_id: str | None = None sync_mode: str | None = None is_optional: bool = False representation: dict = field(default_factory=dict) @dataclass class ChannelValueMappingDraft: channel: str canonical_code: str value_code: str channel_option_id: str @dataclass class V2AttributeSeedStats: candidates_created: int = 0 candidates_updated: int = 0 candidates_preserved: int = 0 values_created: int = 0 values_updated: int = 0 mappings_created: int = 0 mappings_updated: int = 0 attributes_created: int = 0 attributes_skipped_existing: int = 0 values_applied: int = 0 channel_mappings_created: int = 0 channel_mappings_skipped_existing: int = 0 channel_value_mappings_created: int = 0 channel_value_mappings_skipped_existing: int = 0 skipped_identity_codes: int = 0 source_counts: dict[str, int] = field(default_factory=dict) review_status_counts: dict[str, int] = field(default_factory=dict) def as_dict(self) -> dict[str, Any]: return { "candidates_created": self.candidates_created, "candidates_updated": self.candidates_updated, "candidates_preserved": self.candidates_preserved, "values_created": self.values_created, "values_updated": self.values_updated, "mappings_created": self.mappings_created, "mappings_updated": self.mappings_updated, "attributes_created": self.attributes_created, "attributes_skipped_existing": self.attributes_skipped_existing, "values_applied": self.values_applied, "channel_mappings_created": self.channel_mappings_created, "channel_mappings_skipped_existing": self.channel_mappings_skipped_existing, "channel_value_mappings_created": self.channel_value_mappings_created, "channel_value_mappings_skipped_existing": self.channel_value_mappings_skipped_existing, "skipped_identity_codes": self.skipped_identity_codes, "source_counts": dict(self.source_counts), "review_status_counts": dict(self.review_status_counts), } def seed_v2_attributes( session: Session, *, apply: bool = False, reset: bool = False, overrides: dict[str, Any] | None = None, ) -> dict[str, Any]: """Discover V1 candidates for review. Apply only approved/modified records when asked.""" stats = discover_v2_attribute_candidates(session, reset=reset) if overrides: from db.v2_attribute_review import apply_override_payload apply_override_payload(session, overrides) payload = stats.as_dict() payload["report"] = build_seed_review_report(session) if apply: payload["apply"] = apply_approved_attribute_seed(session) return payload def discover_v2_attribute_candidates(session: Session, *, reset: bool = False) -> V2AttributeSeedStats: stats = V2AttributeSeedStats() drafts, mappings, _value_mappings = _collect_v1_catalog(session, stats) if reset: _clear_candidate_tables(session) usage, empty_usage = _product_usage_split(session) value_usage = _product_value_usage(session) duplicate_groups = _duplicate_groups(drafts) _persist_candidates(session, drafts, mappings, usage, empty_usage, value_usage, duplicate_groups, stats) stats.review_status_counts = _review_status_counts(session) return stats def refresh_observed_values(session: Session, candidate_ids: list[int] | None = None) -> dict[str, int]: """Re-read MasterProductAttributeValue into pending value candidates.""" stats = V2AttributeSeedStats() drafts, mappings, _value_mappings = _collect_v1_catalog(session, stats) usage, empty_usage = _product_usage_split(session) value_usage = _product_value_usage(session) wanted = {int(item) for item in (candidate_ids or [])} rows = list(session.scalars(select(V2AttributeSeedCandidate)).all()) if wanted: rows = [row for row in rows if int(row.id) in wanted] now = datetime.now(timezone.utc) created = 0 updated = 0 existing_values = { (str(row.attribute_source_code), str(row.source_value_code)): row for row in session.scalars(select(V2AttributeValueSeedCandidate)).all() } for row in rows: draft = drafts.get(row.source_code) if draft is None: continue if row.review_status == "pending_review": row.product_usage_count = usage.get(row.source_code, 0) row.value_count = len(draft.values) if empty_usage.get(row.source_code) and not usage.get(row.source_code): row.notes = row.notes or ( f"{empty_usage[row.source_code]} products have this field assigned with a blank value." ) row.updated_at = now updated += 1 for index, (value_code, label) in enumerate(draft.values.items()): key = (row.source_code, value_code) current = existing_values.get(key) usage_count = value_usage.get((row.source_code, normalize_lookup_value(label)), 0) if current is None: session.add( V2AttributeValueSeedCandidate( attribute_source_code=row.source_code, source_value_code=value_code, proposed_value_code=value_code, proposed_label=label, sort_order=index * 10, product_usage_count=usage_count, aliases=_merge_text_aliases([], draft.value_aliases.get(value_code) or []), created_at=now, updated_at=now, ) ) created += 1 elif current.review_status == "pending_review": current.proposed_label = label current.product_usage_count = usage_count _apply_draft_value_aliases(current, draft, value_code) current.updated_at = now session.flush() _fold_collection_family_seed_values(session) from db.v2_attribute_review import recount_candidate_usage, refold_merged_candidates refold_merged_candidates(session) for row in rows: recount_candidate_usage(session, row) row.updated_at = now return {"candidates_updated": updated, "values_created": created} def _attribute_index(session: Session) -> dict[str, V2Attribute]: with session.no_autoflush: rows = session.scalars(select(V2Attribute)).all() return {str(row.code): row for row in rows if str(row.code or "").strip()} def _clear_v2_attribute_tables(session: Session) -> None: session.execute(delete(V2ChannelAttributeValueMapping)) session.execute(delete(V2ChannelAttributeMapping)) session.execute(delete(V2AttributeValue)) session.execute(delete(V2Attribute)) session.flush() def _collect_v1_catalog( session: Session, stats: V2AttributeSeedStats, ) -> tuple[dict[str, AttributeDraft], list[ChannelMappingDraft], list[ChannelValueMappingDraft]]: definitions = { _code(row.attribute_code): row for row in session.scalars( select(MasterAttributeDefinition).where(MasterAttributeDefinition.is_active.is_(True)) ).all() if _code(row.attribute_code) } plytix_defs = { _code(row.attribute_code): row for row in (session.scalars(select(AttributeDef)).all() if _has_table(session, AttributeDef) else []) if _code(row.attribute_code) } vocabs = [ row for row in session.scalars( select(ManualAttributeVocabulary).where(ManualAttributeVocabulary.is_active.is_(True)) ).all() if _code(row.attribute_code) ] vocab_values = _group_vocab_values(session, vocabs) aliases = [ row for row in session.scalars( select(ChannelAttributeAlias).where(ChannelAttributeAlias.is_active.is_(True)) ).all() if _code(row.canonical_code) and str(row.channel_attribute_code or "").strip() ] magento_by_code = _magento_registry_by_code(session) magento_options_by_code = _magento_options_by_code(session) shopify_by_code = _shopify_registry_by_code(session) plytix_options = _plytix_options_by_code(session) observed_values = _observed_values_by_code(session) stats.source_counts = { "master_attribute_definition": len(definitions), "attribute_def": len(plytix_defs), "manual_attribute_vocabulary": len(vocabs), "channel_attribute_alias": len(aliases), "master_product_attribute_value_codes": len(observed_values), "magento_attribute_registry": len(magento_by_code), } candidate_codes: set[str] = set() candidate_codes.update(CONTENT_SEED_CODES) candidate_codes.update(definitions) candidate_codes.update(plytix_defs) candidate_codes.update(_code(row.attribute_code) for row in vocabs) for vocab in vocabs: candidate_codes.update(_vocab_covered_codes(vocab)) candidate_codes.update(_code(row.canonical_code) for row in aliases) candidate_codes.update(observed_values) for row in magento_by_code.values(): code = _code(row.attribute_code) if not code: continue if ( _is_taxonomy_code(code) or _is_taxonomy_label(row.frontend_label) or _is_system_dynamic_code(code) or _is_system_dynamic_label(row.frontend_label) or _is_channel_policy_code(code) or _is_channel_policy_label(row.frontend_label) ): candidate_codes.add(code) elif code.startswith(MAGENTO_ATTRIBUTE_PREFIX): candidate_codes.add(code) for row in shopify_by_code.values(): code = _code(row.attribute_code) owner = str(row.owner_type or "").strip().lower() if "collection" in owner or _is_channel_policy_code(code): candidate_codes.add(code) drafts: dict[str, AttributeDraft] = {} skipped_identity = 0 for code in sorted(candidate_codes): if not code or _should_skip_draft(code): if code: skipped_identity += 1 continue definition = definitions.get(code) plytix = plytix_defs.get(code) vocab = next((row for row in vocabs if code in _vocab_covered_codes(row)), None) alias_rows = [row for row in aliases if _code(row.canonical_code) == code] magento_meta = _magento_meta_for_canonical(code, alias_rows, magento_by_code) type_sources = [ magento_meta.frontend_input if magento_meta is not None else None, next((row.data_type for row in alias_rows if row.data_type), None), definition.data_type if definition is not None else None, plytix.type if plytix is not None else None, ] data_type = _resolve_data_type( code=code, definition=definition, plytix=plytix, alias_rows=alias_rows, magento_meta=magento_meta, has_vocab=bool(vocab_values.get(code) or (vocab and vocab_values.get(_code(vocab.attribute_code)))), has_options=bool(plytix_options.get(code) or magento_options_by_code.get(_magento_code_for(code, alias_rows))), ) if _is_taxonomy_code(code) or _is_taxonomy_label( magento_meta.frontend_label if magento_meta is not None else None ): data_type = "select" if _is_content_code(code): data_type = content_data_type(code) label = _resolve_label(code, definition=definition, plytix=plytix, vocab=vocab, magento_meta=magento_meta) if _is_content_code(code): label = CONTENT_LABELS.get(code) or label is_taxonomy = _is_taxonomy_code(code) or _is_taxonomy_label(label) required = False if definition is not None: required = bool(definition.is_required) elif plytix is not None: required = bool(plytix.is_required) if definition is not None and str(definition.purpose or "").strip().lower() == "admin": required = False source_names = _source_names(definition, plytix, vocab, observed_values.get(code), alias_rows) magento_code = _magento_code_for(code, alias_rows) if magento_meta is not None or alias_rows else None shopify_meta = _shopify_meta_for_canonical(code, alias_rows, shopify_by_code) draft = AttributeDraft( code=code, label=label, data_type=data_type, is_variant_axis=_is_variant_axis(code), is_channel_required=required, source=source_names[0] if source_names else "unknown", sources=source_names, magento_frontend_input=str(magento_meta.frontend_input).strip() if magento_meta and magento_meta.frontend_input else None, magento_attribute_code=magento_code if magento_meta is not None else None, shopify_representation=_shopify_representation(shopify_meta, data_type), conflict_reasons=[] if _is_content_code(code) else _type_conflicts(data_type, type_sources), owner_scope=infer_owner_scope( code=code, magento_code=magento_code, shopify_owner_type=str(shopify_meta.owner_type or "") if shopify_meta is not None else "", label=label, ), ) if is_taxonomy: draft.conflict_reasons = [] for label_text in _taxonomy_labels(session, code): _add_value(draft, label_text) elif not _is_content_code(code): for label_text in _value_labels_for( code=code, data_type=data_type, vocab=vocab, vocab_values=vocab_values, plytix_options=plytix_options, magento_options=magento_options_by_code.get(_magento_code_for(code, alias_rows), []), observed=observed_values.get(code, []), ): _add_value(draft, label_text) drafts[code] = draft _apply_family_value_policy(session, drafts) stats.skipped_identity_codes = skipped_identity mappings = _channel_mapping_drafts(aliases, magento_by_code, drafts) value_mappings = _channel_value_mapping_drafts(mappings, drafts, magento_options_by_code) return drafts, mappings, value_mappings def _seed_values( session: Session, attribute: V2Attribute, values: dict[str, str], stats: V2AttributeSeedStats, *, created_at: datetime, ) -> None: if not values: return with session.no_autoflush: existing_rows = list( session.scalars(select(V2AttributeValue).where(V2AttributeValue.attribute_id == int(attribute.id))).all() ) existing_codes = {str(row.value_code) for row in existing_rows} existing_labels = {normalize_lookup_value(row.display_label) for row in existing_rows} for value_code, display_label in values.items(): if value_code in existing_codes or normalize_lookup_value(display_label) in existing_labels: stats.values_skipped_existing += 1 continue session.add( V2AttributeValue( attribute_id=int(attribute.id), value_code=value_code, display_label=display_label, created_at=created_at, ) ) existing_codes.add(value_code) existing_labels.add(normalize_lookup_value(display_label)) stats.values_created += 1 def _seed_channel_mappings( session: Session, attributes: dict[str, V2Attribute], mappings: Iterable[ChannelMappingDraft], stats: V2AttributeSeedStats, *, created_at: datetime, ) -> None: with session.no_autoflush: existing = { (str(row.channel), int(row.canonical_attribute_id)): row for row in session.scalars(select(V2ChannelAttributeMapping)).all() } for draft in mappings: attribute = attributes.get(draft.canonical_code) if attribute is None: continue key = (draft.channel, int(attribute.id)) current = existing.get(key) if current is None: current = V2ChannelAttributeMapping( channel=draft.channel, canonical_attribute_id=int(attribute.id), channel_attribute_code=draft.channel_attribute_code, channel_attribute_id=draft.channel_attribute_id, sync_mode=draft.sync_mode, is_optional=draft.is_optional, created_at=created_at, ) session.add(current) existing[key] = current stats.channel_mappings_created += 1 continue stats.channel_mappings_skipped_existing += 1 def _seed_channel_value_mappings( session: Session, attributes: dict[str, V2Attribute], drafts: Iterable[ChannelValueMappingDraft], stats: V2AttributeSeedStats, *, created_at: datetime, ) -> None: values_by_attr: dict[int, dict[str, V2AttributeValue]] = {} with session.no_autoflush: existing = { (str(row.channel), int(row.canonical_attribute_value_id)): row for row in session.scalars(select(V2ChannelAttributeValueMapping)).all() } for draft in drafts: attribute = attributes.get(draft.canonical_code) if attribute is None: continue by_code = values_by_attr.get(int(attribute.id)) if by_code is None: with session.no_autoflush: by_code = { str(row.value_code): row for row in session.scalars( select(V2AttributeValue).where(V2AttributeValue.attribute_id == int(attribute.id)) ).all() } values_by_attr[int(attribute.id)] = by_code value = by_code.get(draft.value_code) if value is None: continue key = (draft.channel, int(value.id)) if key in existing: stats.channel_value_mappings_skipped_existing += 1 continue row = V2ChannelAttributeValueMapping( channel=draft.channel, canonical_attribute_value_id=int(value.id), channel_option_id=draft.channel_option_id, created_at=created_at, ) session.add(row) existing[key] = row stats.channel_value_mappings_created += 1 def _channel_mapping_drafts( aliases: list[ChannelAttributeAlias], magento_by_code: dict[str, MagentoAttributeRegistry], drafts: dict[str, AttributeDraft], ) -> list[ChannelMappingDraft]: chosen: dict[tuple[str, str], ChannelMappingDraft] = {} for alias in aliases: canonical = _code(alias.canonical_code) channel = _pipeline_channel(alias.channel_code) if not canonical or not channel or canonical not in drafts: continue channel_code = str(alias.channel_attribute_code or "").strip() if not channel_code: continue magento = magento_by_code.get(normalize_column_key(channel_code)) representation = {} if channel == "magento": representation = { "frontend_input": str(magento.frontend_input or "").strip() if magento is not None else magento_frontend_input(drafts[canonical].data_type), "backend_type": str(magento.backend_type or "").strip() if magento is not None else None, "attribute_id": str(magento.attribute_id) if magento is not None else None, "attribute_code": channel_code, } elif channel == "shopify": representation = dict(drafts[canonical].shopify_representation or {}) representation.setdefault("attribute_code", channel_code) candidate = ChannelMappingDraft( channel=channel, canonical_code=canonical, channel_attribute_code=channel_code, channel_attribute_id=str(magento.attribute_id) if magento is not None and channel == "magento" else None, sync_mode=str(alias.action or "").strip() or None, is_optional=str(alias.action or "").strip().lower() not in {"use_existing", "create_new"}, representation=representation, ) key = (channel, canonical) current = chosen.get(key) if current is None or _mapping_rank(candidate) < _mapping_rank(current): chosen[key] = candidate _ensure_content_channel_mappings(chosen, drafts, magento_by_code) return list(chosen.values()) def _ensure_content_channel_mappings( chosen: dict[tuple[str, str], ChannelMappingDraft], drafts: dict[str, AttributeDraft], magento_by_code: dict[str, MagentoAttributeRegistry], ) -> None: for canonical, channels in CONTENT_CHANNEL_ALIASES.items(): if canonical not in drafts: continue for channel, channel_code in channels.items(): key = (channel, canonical) if key in chosen or not channel_code: continue magento = magento_by_code.get(normalize_column_key(channel_code)) if channel == "magento" else None representation: dict[str, Any] = {"attribute_code": channel_code} if channel == "magento": representation["frontend_input"] = ( str(magento.frontend_input or "").strip() if magento is not None else magento_frontend_input(drafts[canonical].data_type) ) if magento is not None: representation["backend_type"] = str(magento.backend_type or "").strip() representation["attribute_id"] = str(magento.attribute_id) chosen[key] = ChannelMappingDraft( channel=channel, canonical_code=canonical, channel_attribute_code=channel_code, channel_attribute_id=str(magento.attribute_id) if magento is not None else None, sync_mode="use_existing", is_optional=False, representation=representation, ) def _channel_value_mapping_drafts( mappings: Iterable[ChannelMappingDraft], drafts: dict[str, AttributeDraft], magento_options_by_code: dict[str, list[MagentoAttributeOptionRegistry]], ) -> list[ChannelValueMappingDraft]: out: list[ChannelValueMappingDraft] = [] for mapping in mappings: if mapping.channel != "magento": continue draft = drafts.get(mapping.canonical_code) if draft is None: continue options = magento_options_by_code.get(normalize_column_key(mapping.channel_attribute_code), []) by_norm = {normalize_lookup_value(label): code for code, label in draft.values.items()} by_norm.update({normalize_lookup_value(code): code for code in draft.values}) for value_code, aliases in (draft.value_aliases or {}).items(): for alias in aliases: key = normalize_lookup_value(alias) if key: by_norm.setdefault(key, value_code) chosen: dict[str, ChannelValueMappingDraft] = {} preferred: dict[str, bool] = {} for option in options: value_code = by_norm.get(normalize_lookup_value(option.option_label_norm or option.option_label)) if not value_code: continue exact = normalize_lookup_value(option.option_label) == normalize_lookup_value(draft.values.get(value_code, "")) previous = chosen.get(value_code) if previous is None or (exact and not preferred.get(value_code)): chosen[value_code] = ChannelValueMappingDraft( channel="magento", canonical_code=mapping.canonical_code, value_code=value_code, channel_option_id=str(option.value_index), ) preferred[value_code] = exact out.extend(chosen.values()) return out def _value_labels_for( *, code: str, data_type: str, vocab: ManualAttributeVocabulary | None, vocab_values: dict[str, list[str]], plytix_options: dict[str, list[str]], magento_options: list[MagentoAttributeOptionRegistry], observed: list[str], ) -> list[str]: labels: list[str] = [] seen: set[str] = set() def add(raw: str) -> None: text = str(raw or "").strip() key = normalize_lookup_value(text) if not text or not key or key in seen: return seen.add(key) labels.append(text) for item in vocab_values.get(code, []): add(item) if vocab is not None: for item in vocab_values.get(_code(vocab.attribute_code), []): add(item) for item in plytix_options.get(code, []): add(item) for option in magento_options: add(option.option_label) include_observed = data_type in SELECTABLE_TYPES or data_type == "boolean" or bool(labels) or len(observed) <= 200 if include_observed: for item in observed[:200]: add(item) return labels def _add_value(draft: AttributeDraft, label: str) -> None: code = attribute_code_from_header(label) if not code: code = f"value_{len(draft.values) + 1}" if code in draft.values and normalize_lookup_value(draft.values[code]) != normalize_lookup_value(label): suffix = 2 candidate = f"{code}_{suffix}" while candidate in draft.values: suffix += 1 candidate = f"{code}_{suffix}" code = candidate draft.values.setdefault(code, label) def _is_collection_family_code(code: str) -> bool: return _code(code) in COLLECTION_FAMILY_CODES def _merge_text_aliases(existing: list[str] | None, extras: Iterable[object]) -> list[str]: merged = [str(item).strip() for item in (existing or []) if str(item).strip()] seen = {normalize_lookup_value(item) for item in merged} for extra in extras: text = str(extra or "").strip() key = normalize_lookup_value(text) if not text or not key or key in seen: continue merged.append(text) seen.add(key) return merged def _registry_name_lookup(session: Session, model: type, fallback: Iterable[str] = ()) -> dict[str, str]: lookup: dict[str, str] = {} for raw in fallback: text = str(raw or "").strip() key = normalize_lookup_value(text) if text and key: lookup[key] = text if not _has_table(session, model): return lookup rows = session.scalars(select(model).where(model.is_active.is_(True))).all() for row in rows: text = str(getattr(row, "name", "") or "").strip() key = normalize_lookup_value(text) if text and key: lookup[key] = text return lookup def _collection_name_lookup(session: Session) -> dict[str, str]: return _registry_name_lookup(session, MasterCollectionRegistry) def _product_family_name_lookup(session: Session) -> dict[str, str]: return _registry_name_lookup(session, MasterProductFamily, _FALLBACK_PRODUCT_FAMILY_LABELS) def _strip_collection_family_prefix(label: str) -> str: text = re.sub(r"\s+", " ", str(label or "").strip()) lowered = text.casefold() for prefix in _COLLECTION_FAMILY_PREFIXES: if lowered.startswith(prefix): return text[len(prefix) :].strip(" /-") return text def _accessory_group_label(label: str, family_names: dict[str, str]) -> str | None: text = re.sub(r"\s+", " ", str(label or "").strip()) if not text or "/" in text: return None match = _ACCESSORY_HYPHEN_RE.match(text) or _ACCESSORY_SPACE_RE.match(text) if match is None: return None theme = re.sub(r"\s+", " ", str(match.group("theme") or "").strip(" -")) if not theme or normalize_lookup_value(theme) in family_names: return None if normalize_lookup_value(theme) in {"kitchen cabinets", "bathroom vanities", "accessories"}: return None return f"{theme} Accessories" def canonicalize_collection_family_label( label: str, collections: dict[str, str], family_names: dict[str, str], ) -> str | None: """Keep collection names and shared accessory groups; drop category paths and product families.""" text = re.sub(r"\s+", " ", str(label or "").strip()) if not text: return None stripped = _strip_collection_family_prefix(text) if not stripped: return None if "/" in stripped: return None key = normalize_lookup_value(stripped) if not key or key in {"unassigned", "n/a", "na", "none", "null"}: return None if key in family_names or key == "accessories": return None accessory = _accessory_group_label(stripped, family_names) if accessory: return accessory return collections.get(key) or stripped def _collapse_collection_family_draft( draft: AttributeDraft, collections: dict[str, str], family_names: dict[str, str], ) -> None: grouped: dict[str, list[str]] = {} order: list[str] = [] for label in draft.values.values(): canonical = canonicalize_collection_family_label(label, collections, family_names) if canonical is None: continue key = normalize_lookup_value(canonical) if key not in grouped: grouped[key] = [] order.append(canonical) if label not in grouped[key]: grouped[key].append(label) draft.values = {} draft.value_aliases = {} for canonical in order: key = normalize_lookup_value(canonical) _add_value(draft, canonical) value_code = next( (code for code, label in draft.values.items() if normalize_lookup_value(label) == key), attribute_code_from_header(canonical), ) aliases = [ item for item in grouped[key] if normalize_lookup_value(item) != key ] if value_code and aliases: draft.value_aliases[value_code] = aliases def _apply_family_value_policy(session: Session, drafts: dict[str, AttributeDraft]) -> None: collections = _collection_name_lookup(session) family_names = _product_family_name_lookup(session) registry_families = _registry_name_lookup(session, MasterProductFamily) for code, draft in drafts.items(): if _is_collection_family_code(code): _collapse_collection_family_draft(draft, collections, family_names) elif _code(code) == "product_family": names = registry_families or family_names if not names: continue draft.values = {} draft.value_aliases = {} for name in sorted(set(names.values()), key=str.casefold): _add_value(draft, name) def _alias_seed_value_into( survivor: V2AttributeValueSeedCandidate, item: V2AttributeValueSeedCandidate, now: datetime, ) -> None: survivor.aliases = _merge_text_aliases( survivor.aliases, ( item.source_value_code, item.proposed_label, item.proposed_value_code, *(item.aliases or []), ), ) item.review_status = "merged" item.merge_into_value_code = survivor.proposed_value_code or survivor.source_value_code item.updated_at = now def _fold_collection_family_seed_values(session: Session) -> None: collections = _collection_name_lookup(session) family_names = _product_family_name_lookup(session) now = datetime.now(timezone.utc) for attr_code in COLLECTION_FAMILY_CODES: rows = list( session.scalars( select(V2AttributeValueSeedCandidate).where( V2AttributeValueSeedCandidate.attribute_source_code == attr_code ) ).all() ) if not rows: continue groups: dict[str, list[tuple[str, V2AttributeValueSeedCandidate]]] = {} for row in rows: if row.review_status == "merged": continue canonical = canonicalize_collection_family_label(row.proposed_label, collections, family_names) if canonical is None: if row.review_status == "pending_review": row.review_status = "excluded" row.updated_at = now continue groups.setdefault(normalize_lookup_value(canonical), []).append((canonical, row)) for items in groups.values(): canonical_label = items[0][0] candidates = [row for _, row in items] survivor = next( (row for row in candidates if row.review_status in {"approved", "modified"}), None, ) if survivor is None: survivor = next( ( row for row in candidates if normalize_lookup_value(row.proposed_label) == normalize_lookup_value(canonical_label) ), candidates[0], ) if survivor.review_status == "pending_review": survivor.proposed_label = canonical_label code = attribute_code_from_header(canonical_label) if code: survivor.proposed_value_code = code survivor.updated_at = now for row in candidates: if row is survivor: continue if row.review_status in {"approved", "modified", "excluded"}: continue _alias_seed_value_into(survivor, row, now) parent = session.scalar( select(V2AttributeSeedCandidate).where(V2AttributeSeedCandidate.source_code == attr_code) ) if parent is not None: parent.value_count = sum( 1 for row in rows if row.review_status not in {"merged", "excluded"} ) parent.updated_at = now def _apply_draft_value_aliases(row: V2AttributeValueSeedCandidate, draft: AttributeDraft, value_code: str) -> None: extras = list(draft.value_aliases.get(value_code) or []) extras.extend((value_code, row.proposed_label, row.source_value_code)) row.aliases = _merge_text_aliases(row.aliases, extras) def _resolve_data_type( *, code: str, definition: MasterAttributeDefinition | None, plytix: AttributeDef | None, alias_rows: list[ChannelAttributeAlias], magento_meta: MagentoAttributeRegistry | None, has_vocab: bool, has_options: bool, ) -> str: candidates = [ magento_meta.frontend_input if magento_meta is not None else None, next((row.data_type for row in alias_rows if row.data_type), None), definition.data_type if definition is not None else None, plytix.type if plytix is not None else None, ] mapped = "text" for raw in candidates: if not raw: continue mapped = map_canonical_data_type(raw) break else: mapped = _heuristic_data_type(code) if has_vocab or has_options: if mapped not in SELECTABLE_TYPES and mapped not in {"boolean", "decimal", "integer", "image", "html"}: mapped = "select" elif mapped == "text": mapped = "select" return mapped def _resolve_label( code: str, *, definition: MasterAttributeDefinition | None, plytix: AttributeDef | None, vocab: ManualAttributeVocabulary | None, magento_meta: MagentoAttributeRegistry | None, ) -> str: for raw in ( definition.label if definition is not None else None, vocab.label if vocab is not None else None, plytix.frontend_label if plytix is not None else None, magento_meta.frontend_label if magento_meta is not None else None, ): text = str(raw or "").strip() if text: return text return CONTENT_LABELS.get(_code(code)) or code.replace("_", " ").title() def _map_data_type(raw: object) -> str: return map_canonical_data_type(raw) def _heuristic_data_type(code: str) -> str: if _is_content_code(code): return content_data_type(code) if any(token in code for token in ("price", "width", "height", "depth", "weight", "count", "length")): return "decimal" if code.endswith("_included") or code.endswith("_compliant") or code in {"easy_clean", "scratch_resistant"}: return "boolean" if "image" in code or "thumbnail" in code: return "image" return "text" def _is_variant_axis(code: str) -> bool: return code.startswith("variation_") or code in VARIANT_AXIS_CODES def _should_skip_draft(code: str) -> bool: key = _code(code) if key in CONTENT_ALIAS_SKIP: return True return _is_identity_code(key) def _should_skip_apply(candidate: V2AttributeSeedCandidate) -> bool: return ( _is_identity_code(candidate.source_code) or _is_system_dynamic_code(candidate.source_code) or _is_system_dynamic_code(candidate.proposed_code) or _is_channel_policy_code(candidate.source_code) or _is_channel_policy_code(candidate.proposed_code) or _is_content_product_field(candidate.source_code) or _is_content_product_field(candidate.proposed_code) or _is_taxonomy_code(candidate.source_code) or _is_taxonomy_code(candidate.proposed_code) or _is_taxonomy_label(candidate.proposed_label) or _is_system_dynamic_label(candidate.proposed_label) or _is_channel_policy_label(candidate.proposed_label) ) def candidate_ownership(code: str, label: str | None = None, proposed_code: str | None = None) -> str: if _is_system_dynamic_code(code) or _is_system_dynamic_code(proposed_code or "") or _is_system_dynamic_label(label): return "system" if _is_channel_policy_code(code) or _is_channel_policy_code(proposed_code or "") or _is_channel_policy_label(label): return "channel_policy" if _is_content_code(code) or _is_content_code(proposed_code or ""): return "content" if _is_taxonomy_code(code) or _is_taxonomy_code(proposed_code or "") or _is_taxonomy_label(label): return "taxonomy" if _is_identity_code(code): return "identity" return "attribute" def _is_identity_code(code: str) -> bool: return _code(code) in IDENTITY_CODES def _is_taxonomy_code(code: str) -> bool: return _code(code) in TAXONOMY_CODES def _normalized_phrase(raw: object) -> str: return " ".join(str(raw or "").strip().lower().replace("_", " ").replace("-", " ").split()) def _is_taxonomy_label(raw: object) -> bool: return _normalized_phrase(raw) in TAXONOMY_LABELS def _is_system_dynamic_label(raw: object) -> bool: return _normalized_phrase(raw) in SYSTEM_DYNAMIC_LABELS def _is_system_dynamic_code(code: str) -> bool: key = _code(code) if key in SYSTEM_DYNAMIC_CODES: return True if key.endswith("_variant_list") or key.endswith("_variants_list") or key.endswith("_variants"): return True if key.startswith("configurable_variation"): return True return False def _is_channel_policy_code(code: str) -> bool: return _code(code) in CHANNEL_POLICY_CODES def _is_channel_policy_label(raw: object) -> bool: return _normalized_phrase(raw) in CHANNEL_POLICY_LABELS def _is_content_code(code: str) -> bool: return _code(code) in CONTENT_CODES def _is_content_product_field(code: str) -> bool: return _code(code) in CONTENT_PRODUCT_FIELDS def content_data_type(code: str) -> str: return CONTENT_DATA_TYPES.get(_code(code), "html" if _is_content_code(code) else "text") def content_composition_sources(code: str) -> list[str]: return [item for item in CONTENT_COMPOSITION.get(_code(code), ()) if item] def content_note_for(code: str) -> str: sources = content_composition_sources(code) note = CONTENT_PRODUCT_NOTE if _is_content_product_field(code) else CONTENT_NOTE if sources: return f"{note} Composed from: {', '.join(sources)}." return note def infer_owner_scope( *, code: str, magento_code: str | None = None, shopify_owner_type: str | None = None, label: str | None = None, ) -> str: key = _code(code) magento = _code(magento_code or "") owner = str(shopify_owner_type or "").strip().lower() if ( _is_taxonomy_code(key) or _is_taxonomy_label(label) or key in COLLECTION_LANDING_KEYS or key.startswith(MAGENTO_ATTRIBUTE_PREFIX) or magento.startswith(MAGENTO_ATTRIBUTE_PREFIX) or "collection" in owner or "category" in owner ): return "category" return DEFAULT_OWNER_SCOPE def _taxonomy_node_kinds(code: str) -> set[str]: key = _code(code) if key in {"hub", "shopping_l1"}: return {"hub"} if key in {"collection", "shopping_collection"}: return {"collection"} if key in {"category_l2", "category_l3", "shopping_l2"}: return {"category"} return {"hub", "category", "collection"} def _taxonomy_labels(session: Session, code: str) -> list[str]: labels: list[str] = [] seen: set[str] = set() def add(raw: object) -> None: text = str(raw or "").strip() key = normalize_lookup_value(text) if not text or not key or key in seen: return seen.add(key) labels.append(text) kinds = _taxonomy_node_kinds(code) if not _has_table(session, MasterTaxonomyNode): return labels rows = session.scalars( select(MasterTaxonomyNode) .where(MasterTaxonomyNode.is_active.is_(True)) .order_by(MasterTaxonomyNode.sort_order.asc(), MasterTaxonomyNode.name.asc()) ).all() for row in rows: if kinds and str(row.node_kind or "").strip() not in kinds: continue add(row.name) return labels def _source_names( definition: MasterAttributeDefinition | None, plytix: AttributeDef | None, vocab: ManualAttributeVocabulary | None, observed: list[str] | None, alias_rows: list[ChannelAttributeAlias] | None = None, ) -> list[str]: names: list[str] = [] if definition is not None: names.append("master_attribute_definition") if vocab is not None: names.append("manual_attribute_vocabulary") if plytix is not None: names.append("attribute_def") if observed: names.append("master_product_attribute_value") if alias_rows: names.append("channel_attribute_alias") return names or ["unknown"] def _source_name( definition: MasterAttributeDefinition | None, plytix: AttributeDef | None, vocab: ManualAttributeVocabulary | None, observed: list[str] | None, ) -> str: return _source_names(definition, plytix, vocab, observed)[0] def _code(value: object) -> str: return normalize_column_key(str(value or "")) def _pipeline_channel(channel_code: object) -> str | None: text = normalize_column_key(str(channel_code or "")) if text in PIPELINE_CHANNELS: return text for known in PIPELINE_CHANNELS: if text.startswith(f"{known}_") or text.startswith(f"{known}-"): return known return None def _mapping_rank(draft: ChannelMappingDraft) -> int: mode = str(draft.sync_mode or "").strip().lower() if mode == "use_existing": return 0 if mode == "create_new": return 1 return 2 def _vocab_covered_codes(vocab: ManualAttributeVocabulary) -> list[str]: raw = vocab.covered_codes_json items: list[Any] if isinstance(raw, list): items = raw elif isinstance(raw, dict): items = list(raw.get("items") or raw.get("codes") or []) else: items = [] items.append(vocab.attribute_code) out: list[str] = [] seen: set[str] = set() for item in items: code = _code(item) if code and code not in seen: seen.add(code) out.append(code) return out def _group_vocab_values( session: Session, vocabs: list[ManualAttributeVocabulary], ) -> dict[str, list[str]]: if not vocabs: return {} by_id = {int(row.id): row for row in vocabs} grouped: dict[str, list[str]] = {} rows = session.scalars( select(ManualAttributeVocabularyValue).where( ManualAttributeVocabularyValue.vocabulary_id.in_(list(by_id)), ManualAttributeVocabularyValue.is_active.is_(True), ) ).all() for row in rows: vocab = by_id.get(int(row.vocabulary_id)) if vocab is None: continue label = str(row.value or "").strip() if not label: continue for code in _vocab_covered_codes(vocab): grouped.setdefault(code, []).append(label) return grouped def _plytix_options_by_code(session: Session) -> dict[str, list[str]]: grouped: dict[str, list[str]] = {} if _has_table(session, AttributeOption): for row in session.scalars(select(AttributeOption)).all(): code = _code(row.attribute_code) label = str(row.option_value or "").strip() if code and label: grouped.setdefault(code, []).append(label) if _has_table(session, AttributeDef): for row in session.scalars(select(AttributeDef)).all(): code = _code(row.attribute_code) options = row.options_json if not code or not isinstance(options, list): continue for item in options: label = str(item or "").strip() if label: grouped.setdefault(code, []).append(label) return grouped def _observed_values_by_code(session: Session) -> dict[str, list[str]]: grouped: dict[str, list[str]] = {} seen: dict[str, set[str]] = {} for code, value in session.execute( select(MasterProductAttributeValue.attribute_code, MasterProductAttributeValue.value).distinct() ).all(): attr_code = _code(code) label = str(value or "").strip() if not attr_code or not label: continue key = normalize_lookup_value(label) bucket = seen.setdefault(attr_code, set()) if key in bucket: continue bucket.add(key) grouped.setdefault(attr_code, []).append(label) return grouped def _magento_registry_by_code(session: Session) -> dict[str, MagentoAttributeRegistry]: by_code: dict[str, MagentoAttributeRegistry] = {} if not _has_table(session, MagentoAttributeRegistry): return by_code rows = session.scalars( select(MagentoAttributeRegistry).order_by(MagentoAttributeRegistry.connection_id.asc()) ).all() for row in rows: code = normalize_column_key(row.attribute_code) if code and code not in by_code: by_code[code] = row return by_code def _magento_options_by_code(session: Session) -> dict[str, list[MagentoAttributeOptionRegistry]]: grouped: dict[str, list[MagentoAttributeOptionRegistry]] = {} if not _has_table(session, MagentoAttributeOptionRegistry): return grouped rows = session.scalars( select(MagentoAttributeOptionRegistry).order_by( MagentoAttributeOptionRegistry.connection_id.asc(), MagentoAttributeOptionRegistry.id.asc(), ) ).all() seen: dict[str, set[str]] = {} for row in rows: code = normalize_column_key(row.attribute_code) key = normalize_lookup_value(row.option_label_norm or row.option_label) if not code or not key: continue bucket = seen.setdefault(code, set()) if key in bucket: continue bucket.add(key) grouped.setdefault(code, []).append(row) return grouped def _magento_code_for(canonical: str, alias_rows: list[ChannelAttributeAlias]) -> str: for row in alias_rows: if _pipeline_channel(row.channel_code) == "magento" and str(row.channel_attribute_code or "").strip(): return normalize_column_key(row.channel_attribute_code) return canonical def _magento_meta_for_canonical( canonical: str, alias_rows: list[ChannelAttributeAlias], magento_by_code: dict[str, MagentoAttributeRegistry], ) -> MagentoAttributeRegistry | None: magento_code = _magento_code_for(canonical, alias_rows) return magento_by_code.get(magento_code) or magento_by_code.get(canonical) def _has_table(session: Session, model: type) -> bool: try: return bool(inspect(session.connection()).has_table(model.__tablename__)) except (OperationalError, ProgrammingError): return False def _shopify_registry_by_code(session: Session) -> dict[str, ShopifyAttributeRegistry]: by_code: dict[str, ShopifyAttributeRegistry] = {} if not _has_table(session, ShopifyAttributeRegistry): return by_code rows = session.scalars(select(ShopifyAttributeRegistry).order_by(ShopifyAttributeRegistry.id.asc())).all() for row in rows: code = normalize_column_key(row.attribute_code) if code and code not in by_code: by_code[code] = row return by_code def _shopify_meta_for_canonical( canonical: str, alias_rows: list[ChannelAttributeAlias], shopify_by_code: dict[str, ShopifyAttributeRegistry], ) -> ShopifyAttributeRegistry | None: for row in alias_rows: if _pipeline_channel(row.channel_code) == "shopify" and str(row.channel_attribute_code or "").strip(): found = shopify_by_code.get(normalize_column_key(row.channel_attribute_code)) if found is not None: return found return shopify_by_code.get(canonical) def _shopify_representation(row: ShopifyAttributeRegistry | None, data_type: str) -> dict[str, Any]: if row is None: return { "kind": "metafield", "metafield_type": shopify_metafield_type(data_type), } return { "kind": "product_option" if str(row.owner_type or "").lower() in {"product", "variant"} else "metafield", "attribute_code": row.attribute_code, "namespace": row.namespace, "key": row.key, "metafield_type": row.data_type or shopify_metafield_type(data_type), "name": row.name, } def _type_conflicts(resolved: str, raw_sources: list[object]) -> list[str]: mapped = [map_canonical_data_type(item) for item in raw_sources if item] unique = {item for item in mapped if item} if len(unique) > 1: return [f"source types disagree: {', '.join(sorted(unique))} (proposed {resolved})"] return [] def _clear_candidate_tables(session: Session) -> None: session.execute(delete(V2AttributeSeedMappingCandidate)) session.execute(delete(V2AttributeValueSeedCandidate)) session.execute(delete(V2AttributeSeedCandidate)) session.flush() def _product_usage_split(session: Session) -> tuple[dict[str, int], dict[str, int]]: populated: dict[str, int] = {} empty: dict[str, int] = {} for code, value, sku in session.execute( select( MasterProductAttributeValue.attribute_code, MasterProductAttributeValue.value, MasterProductAttributeValue.sku, ).distinct() ).all(): key = _code(code) if not key: continue if str(value or "").strip(): populated[key] = populated.get(key, 0) + 1 else: empty[key] = empty.get(key, 0) + 1 return populated, empty def _product_usage_by_code(session: Session) -> dict[str, int]: populated, empty = _product_usage_split(session) totals = dict(populated) for key, count in empty.items(): totals[key] = totals.get(key, 0) + count return totals def _product_value_usage(session: Session) -> dict[tuple[str, str], int]: counts: dict[tuple[str, str], int] = {} for code, value, sku in session.execute( select( MasterProductAttributeValue.attribute_code, MasterProductAttributeValue.value, MasterProductAttributeValue.sku, ).distinct() ).all(): attr = _code(code) label = normalize_lookup_value(value) if attr and label: counts[(attr, label)] = counts.get((attr, label), 0) + 1 return counts def _duplicate_groups(drafts: dict[str, AttributeDraft]) -> dict[str, str]: grouped: dict[str, list[str]] = {} for draft in drafts.values(): key = _fingerprint(draft.label) or _fingerprint(draft.code) if not key: continue grouped.setdefault(key, []).append(draft.code) mapping: dict[str, str] = {} for key, codes in grouped.items(): if len(codes) < 2: continue for code in codes: mapping[code] = key return mapping def _fingerprint(value: object) -> str: return "".join(ch for ch in str(value or "").casefold() if ch.isalnum()) def _recount_merged_usage(session: Session, existing: dict[str, V2AttributeSeedCandidate]) -> None: from db.v2_attribute_review import recount_candidate_usage for row in existing.values(): if row.aliases: recount_candidate_usage(session, row) def _persist_candidates( session: Session, drafts: dict[str, AttributeDraft], mappings: list[ChannelMappingDraft], usage: dict[str, int], empty_usage: dict[str, int], value_usage: dict[tuple[str, str], int], duplicate_groups: dict[str, str], stats: V2AttributeSeedStats, ) -> None: now = datetime.now(timezone.utc) existing = { str(row.source_code): row for row in session.scalars(select(V2AttributeSeedCandidate)).all() } locked = {"approved", "modified", "merged", "excluded"} for draft in drafts.values(): current = existing.get(draft.code) ui_control = default_ui_control(draft.data_type) review_status = "conflict" if draft.conflict_reasons else "pending_review" proposed_action = "merge" if draft.code in duplicate_groups else "keep" if current is None: current = V2AttributeSeedCandidate( source_code=draft.code, proposed_code=draft.code, proposed_label=draft.label, discovered_data_type=draft.data_type, proposed_data_type=draft.data_type, proposed_ui_control=ui_control, is_variant_axis=draft.is_variant_axis, is_channel_required=draft.is_channel_required, owner_scope=draft.owner_scope if draft.owner_scope in OWNER_SCOPES else DEFAULT_OWNER_SCOPE, review_status=review_status, sources=draft.sources, product_usage_count=usage.get(draft.code, 0), value_count=len(draft.values), channel_map_count=0, magento_frontend_input=draft.magento_frontend_input, magento_attribute_code=draft.magento_attribute_code, shopify_representation=draft.shopify_representation, conflict_reasons=draft.conflict_reasons, proposed_action=proposed_action, duplicate_group=duplicate_groups.get(draft.code), created_at=now, updated_at=now, ) session.add(current) existing[draft.code] = current stats.candidates_created += 1 if _is_taxonomy_code(draft.code): current.notes = current.notes or TAXONOMY_NOTE elif current.review_status in locked: current.discovered_data_type = draft.data_type current.sources = draft.sources current.product_usage_count = usage.get(draft.code, 0) current.value_count = len(draft.values) current.magento_frontend_input = draft.magento_frontend_input current.magento_attribute_code = draft.magento_attribute_code current.shopify_representation = draft.shopify_representation current.duplicate_group = duplicate_groups.get(draft.code) if any(_is_variant_axis(code) for code in [current.source_code, *(current.aliases or [])]): current.is_variant_axis = True current.updated_at = now stats.candidates_preserved += 1 else: current.proposed_code = draft.code current.proposed_label = draft.label current.discovered_data_type = draft.data_type current.proposed_data_type = draft.data_type current.proposed_ui_control = ui_control current.is_variant_axis = draft.is_variant_axis current.is_channel_required = draft.is_channel_required current.owner_scope = draft.owner_scope if draft.owner_scope in OWNER_SCOPES else DEFAULT_OWNER_SCOPE current.review_status = review_status current.sources = draft.sources current.product_usage_count = usage.get(draft.code, 0) current.value_count = len(draft.values) current.magento_frontend_input = draft.magento_frontend_input current.magento_attribute_code = draft.magento_attribute_code current.shopify_representation = draft.shopify_representation current.conflict_reasons = draft.conflict_reasons current.proposed_action = proposed_action current.duplicate_group = duplicate_groups.get(draft.code) current.updated_at = now stats.candidates_updated += 1 session.flush() existing_values = { (str(row.attribute_source_code), str(row.source_value_code)): row for row in session.scalars(select(V2AttributeValueSeedCandidate)).all() } for draft in drafts.values(): for index, (value_code, label) in enumerate(draft.values.items()): key = (draft.code, value_code) current = existing_values.get(key) usage_count = value_usage.get((draft.code, normalize_lookup_value(label)), 0) if current is None: current = V2AttributeValueSeedCandidate( attribute_source_code=draft.code, source_value_code=value_code, proposed_value_code=value_code, proposed_label=label, sort_order=index * 10, product_usage_count=usage_count, aliases=_merge_text_aliases([], draft.value_aliases.get(value_code) or []), created_at=now, updated_at=now, ) session.add(current) stats.values_created += 1 elif current.review_status in {"approved", "modified", "merged", "excluded"}: current.product_usage_count = usage_count current.updated_at = now else: current.proposed_label = label current.proposed_value_code = value_code current.product_usage_count = usage_count current.sort_order = index * 10 _apply_draft_value_aliases(current, draft, value_code) current.updated_at = now stats.values_updated += 1 session.flush() _fold_collection_family_seed_values(session) from db.v2_attribute_review import refold_merged_candidates refold_merged_candidates(session) _recount_merged_usage(session, existing) existing_maps = { (str(row.channel), str(row.attribute_source_code)): row for row in session.scalars(select(V2AttributeSeedMappingCandidate)).all() } map_counts: dict[str, int] = {} for mapping in mappings: representation = dict(mapping.representation or {}) representation["channel_attribute_code"] = mapping.channel_attribute_code representation["sync_mode"] = mapping.sync_mode representation["is_optional"] = mapping.is_optional draft = drafts.get(mapping.canonical_code) candidate_row = existing.get(mapping.canonical_code) data_type = ( candidate_row.proposed_data_type if candidate_row is not None and candidate_row.review_status in locked else (draft.data_type if draft else "text") ) compat = channel_compatibility(data_type, mapping.channel, representation) key = (mapping.channel, mapping.canonical_code) current = existing_maps.get(key) if current is None: session.add( V2AttributeSeedMappingCandidate( attribute_source_code=mapping.canonical_code, channel=mapping.channel, channel_attribute_code=mapping.channel_attribute_code, channel_attribute_id=mapping.channel_attribute_id, representation=representation, compatibility_status=compat, is_manual=False, created_at=now, updated_at=now, ) ) stats.mappings_created += 1 else: if getattr(current, "is_manual", False): current.compatibility_status = channel_compatibility( data_type, mapping.channel, dict(current.representation or {}) ) current.updated_at = now else: current.channel_attribute_code = mapping.channel_attribute_code current.channel_attribute_id = mapping.channel_attribute_id current.representation = representation current.compatibility_status = compat current.updated_at = now stats.mappings_updated += 1 map_counts[mapping.canonical_code] = map_counts.get(mapping.canonical_code, 0) + 1 session.flush() from sqlalchemy import func as sqlfunc for row in existing.values(): row.channel_map_count = int( session.scalar( select(sqlfunc.count()) .select_from(V2AttributeSeedMappingCandidate) .where(V2AttributeSeedMappingCandidate.attribute_source_code == row.source_code) ) or 0 ) for code, row in existing.items(): ownership = candidate_ownership(code, row.proposed_label, row.proposed_code) unlocked = row.review_status in {"pending_review", "conflict"} if ownership == "system" and unlocked: row.review_status = "excluded" row.proposed_action = "exclude" row.notes = row.notes or SYSTEM_DYNAMIC_NOTE row.conflict_reasons = [] row.updated_at = now elif ownership == "channel_policy" and unlocked: row.review_status = "excluded" row.proposed_action = "exclude" row.notes = row.notes or CHANNEL_POLICY_NOTE row.conflict_reasons = [] row.updated_at = now elif ownership == "content": row.notes = row.notes or content_note_for(code) row.conflict_reasons = [] row.updated_at = now elif ownership == "identity" and unlocked: row.review_status = "excluded" row.proposed_action = "exclude" row.notes = row.notes or "Structural/identity field — not a V2 catalog attribute." row.updated_at = now elif ownership == "taxonomy": was_identity_skip = "Structural/identity" in str(row.notes or "") if was_identity_skip or not row.notes: row.notes = TAXONOMY_NOTE if row.review_status == "excluded" and was_identity_skip: row.review_status = "pending_review" row.proposed_action = "keep" row.conflict_reasons = [] row.updated_at = now if _is_collection_family_code(code): row.notes = row.notes or COLLECTION_FAMILY_NOTE elif _code(code) == "product_family": row.notes = row.notes or PRODUCT_FAMILY_NOTE blank = empty_usage.get(code, 0) populated = usage.get(code, 0) if row.review_status == "pending_review" and blank and not populated: row.notes = row.notes or f"{blank} products have this field assigned with a blank value." def _review_status_counts(session: Session) -> dict[str, int]: return {str(status): int(count) for status, count in _status_counts(session)} def _status_counts(session: Session) -> list[tuple[str, int]]: from sqlalchemy import func return list( session.execute( select(V2AttributeSeedCandidate.review_status, func.count()) .group_by(V2AttributeSeedCandidate.review_status) ).all() ) def build_seed_review_report(session: Session) -> dict[str, Any]: candidates = list(session.scalars(select(V2AttributeSeedCandidate).order_by(V2AttributeSeedCandidate.source_code)).all()) status_counts: dict[str, int] = {} likely_duplicates: list[dict[str, Any]] = [] conflicts: list[dict[str, Any]] = [] items: list[dict[str, Any]] = [] grouped: dict[str, list[str]] = {} for row in candidates: status_counts[row.review_status] = status_counts.get(row.review_status, 0) + 1 if row.duplicate_group: grouped.setdefault(row.duplicate_group, []).append(row.source_code) if row.conflict_reasons: conflicts.append( { "source_code": row.source_code, "reasons": row.conflict_reasons, "discovered_data_type": row.discovered_data_type, "proposed_data_type": row.proposed_data_type, } ) items.append(_candidate_report_row(row)) for key, codes in grouped.items(): if len(codes) > 1: likely_duplicates.append({"group": key, "codes": codes}) return { "candidate_count": len(candidates), "review_status_counts": status_counts, "likely_duplicates": likely_duplicates, "conflicts": conflicts, "candidates": items, } def _candidate_report_row(row: V2AttributeSeedCandidate) -> dict[str, Any]: return { "id": int(row.id), "source_code": row.source_code, "proposed_code": row.proposed_code, "proposed_label": row.proposed_label, "sources": row.sources or [], "discovered_data_type": row.discovered_data_type, "proposed_data_type": row.proposed_data_type, "proposed_ui_control": row.proposed_ui_control, "value_count": int(row.value_count or 0), "product_usage_count": int(row.product_usage_count or 0), "channel_map_count": int(row.channel_map_count or 0), "magento_frontend_input": row.magento_frontend_input, "magento_attribute_code": row.magento_attribute_code, "shopify_representation": row.shopify_representation or {}, "review_status": row.review_status, "proposed_action": row.proposed_action, "duplicate_group": row.duplicate_group, "conflict_reasons": row.conflict_reasons or [], "merge_into_code": row.merge_into_code, "lifecycle_status": row.lifecycle_status, } def apply_approved_attribute_seed(session: Session) -> dict[str, Any]: from db.v2_attribute_review import refold_merged_candidates now = datetime.now(timezone.utc) stats = V2AttributeSeedStats() refold_merged_candidates(session) session.flush() ready = [ row for row in session.scalars(select(V2AttributeSeedCandidate)).all() if row.review_status in APPLY_READY_STATUSES ] existing = _attribute_index(session) created: dict[str, V2Attribute] = dict(existing) for candidate in ready: code = str(candidate.proposed_code or candidate.source_code).strip() if _should_skip_apply(candidate): if candidate_ownership(candidate.source_code, candidate.proposed_label, candidate.proposed_code) == "taxonomy": candidate.notes = candidate.notes or TAXONOMY_NOTE elif candidate_ownership(candidate.source_code, candidate.proposed_label, candidate.proposed_code) == "system": candidate.notes = candidate.notes or SYSTEM_DYNAMIC_NOTE elif candidate_ownership(candidate.source_code, candidate.proposed_label, candidate.proposed_code) == "channel_policy": candidate.notes = candidate.notes or CHANNEL_POLICY_NOTE elif candidate_ownership(candidate.source_code, candidate.proposed_label, candidate.proposed_code) == "content": candidate.notes = candidate.notes or content_note_for(candidate.source_code) continue current = created.get(code) aliases = list(candidate.aliases or []) if candidate.source_code not in aliases: aliases.append(candidate.source_code) if current is None: current = V2Attribute( code=code, label=candidate.proposed_label, data_type=candidate.proposed_data_type, ui_control=candidate.proposed_ui_control or default_ui_control(candidate.proposed_data_type), is_variant_axis=bool(candidate.is_variant_axis), is_channel_required=bool(candidate.is_channel_required), lifecycle_status=candidate.lifecycle_status or "active", owner_scope=candidate.owner_scope if getattr(candidate, "owner_scope", None) in OWNER_SCOPES else DEFAULT_OWNER_SCOPE, aliases=aliases, applies_to_product_types=[], created_at=now, ) session.add(current) session.flush() created[code] = current stats.attributes_created += 1 else: current.is_variant_axis = bool(current.is_variant_axis) or bool(candidate.is_variant_axis) merged_aliases = list(current.aliases or []) for alias in aliases: if alias and alias not in merged_aliases: merged_aliases.append(alias) current.aliases = merged_aliases if getattr(candidate, "owner_scope", None) in OWNER_SCOPES: current.owner_scope = candidate.owner_scope stats.attributes_skipped_existing += 1 candidate.applied_at = now _apply_candidate_values(session, candidate, current, stats, created_at=now) _apply_candidate_mappings(session, candidate, current, stats, created_at=now) _rewrite_merged_product_keys(session, candidate.source_code, code) session.flush() _apply_merged_aliases(session, created) _apply_magento_option_mappings(session, created, stats, created_at=now) return stats.as_dict() def _apply_candidate_values( session: Session, candidate: V2AttributeSeedCandidate, attribute: V2Attribute, stats: V2AttributeSeedStats, *, created_at: datetime, ) -> None: from db.v2_attribute_review import _values_for_candidate values = _values_for_candidate(session, candidate) existing = { str(row.value_code): row for row in session.scalars(select(V2AttributeValue).where(V2AttributeValue.attribute_id == int(attribute.id))).all() } for item in values: if item.review_status not in APPLY_READY_STATUSES: continue code = str(item.proposed_value_code or item.source_value_code).strip() if code in existing: row = existing[code] row.display_label = item.proposed_label or row.display_label merged = list(row.aliases or []) for alias in (item.source_value_code, *(item.aliases or [])): if alias and alias not in merged: merged.append(alias) row.aliases = merged continue aliases = list(item.aliases or []) if item.source_value_code not in aliases: aliases.append(item.source_value_code) row = V2AttributeValue( attribute_id=int(attribute.id), value_code=code, display_label=item.proposed_label, sort_order=int(item.sort_order or 0), lifecycle_status=item.lifecycle_status or "active", aliases=aliases, created_at=created_at, ) session.add(row) existing[code] = row stats.values_applied += 1 _apply_merged_value_aliases(values, existing) def _apply_merged_value_aliases( values: list[V2AttributeValueSeedCandidate], existing: dict[str, V2AttributeValue], ) -> None: by_any: dict[str, V2AttributeValue] = dict(existing) for row in existing.values(): by_any[normalize_lookup_value(row.display_label)] = row for alias in row.aliases or []: key = normalize_lookup_value(alias) if key: by_any[key] = row for item in values: if item.review_status != "merged" or not item.merge_into_value_code: continue target = ( existing.get(str(item.merge_into_value_code).strip()) or by_any.get(normalize_lookup_value(item.merge_into_value_code)) ) if target is None: continue aliases = list(target.aliases or []) for alias in (item.source_value_code, item.proposed_label, item.proposed_value_code, *(item.aliases or [])): if alias and alias not in aliases: aliases.append(alias) target.aliases = aliases def _apply_candidate_mappings( session: Session, candidate: V2AttributeSeedCandidate, attribute: V2Attribute, stats: V2AttributeSeedStats, *, created_at: datetime, ) -> None: mappings = list( session.scalars( select(V2AttributeSeedMappingCandidate).where( V2AttributeSeedMappingCandidate.attribute_source_code == candidate.source_code ) ).all() ) existing = { (str(row.channel), int(row.canonical_attribute_id)): row for row in session.scalars(select(V2ChannelAttributeMapping)).all() } for mapping in mappings: representation = dict(mapping.representation or {}) representation["channel_attribute_code"] = mapping.channel_attribute_code key = (str(mapping.channel), int(attribute.id)) if key in existing: row = existing[key] row.channel_attribute_code = mapping.channel_attribute_code row.channel_attribute_id = mapping.channel_attribute_id row.sync_mode = representation.get("sync_mode") row.is_optional = bool(representation.get("is_optional")) row.representation = representation row.compatibility_status = mapping.compatibility_status or channel_compatibility( candidate.proposed_data_type, mapping.channel, representation ) stats.channel_mappings_skipped_existing += 1 continue row = V2ChannelAttributeMapping( channel=mapping.channel, canonical_attribute_id=int(attribute.id), channel_attribute_code=mapping.channel_attribute_code, channel_attribute_id=mapping.channel_attribute_id, sync_mode=representation.get("sync_mode"), is_optional=bool(representation.get("is_optional")), representation=representation, compatibility_status=mapping.compatibility_status or channel_compatibility( candidate.proposed_data_type, mapping.channel, representation ), created_at=created_at, ) session.add(row) existing[key] = row stats.channel_mappings_created += 1 def _apply_merged_aliases(session: Session, created: dict[str, V2Attribute]) -> None: merged = [ row for row in session.scalars(select(V2AttributeSeedCandidate)).all() if row.review_status == "merged" and row.merge_into_code ] for row in merged: survivor = created.get(str(row.merge_into_code).strip()) or created.get(_code(row.merge_into_code)) if survivor is None: continue aliases = list(survivor.aliases or []) for alias in (row.source_code, row.proposed_code): if alias and alias not in aliases: aliases.append(alias) survivor.aliases = aliases _rewrite_merged_product_keys(session, row.source_code, survivor.code) def _rewrite_merged_product_keys(session: Session, source_code: str, target_code: str) -> None: if not source_code or source_code == target_code: return if _has_table(session, V2Product): products = session.scalars(select(V2Product)).all() for product in products: rewritten = _rewrite_mapping_dict(product.attributes or {}, source_code, target_code) if rewritten is not None: product.attributes = rewritten if _has_table(session, V2ConfigurableLink): for link in session.scalars(select(V2ConfigurableLink)).all(): rewritten = _rewrite_mapping_dict(link.axis_attributes or {}, source_code, target_code) if rewritten is not None: link.axis_attributes = rewritten if _has_table(session, MasterProductRelation): for relation in session.scalars(select(MasterProductRelation)).all(): rewritten = _rewrite_mapping_dict(relation.option_mapping or {}, source_code, target_code) if rewritten is not None: relation.option_mapping = rewritten def _rewrite_mapping_dict(payload: dict[str, Any] | None, source_code: str, target_code: str) -> dict[str, Any] | None: if not isinstance(payload, dict) or source_code not in payload: return None rewritten = dict(payload) if target_code not in rewritten or rewritten.get(target_code) in (None, ""): rewritten[target_code] = rewritten[source_code] rewritten.pop(source_code, None) return rewritten def _apply_magento_option_mappings( session: Session, attributes: dict[str, V2Attribute], stats: V2AttributeSeedStats, *, created_at: datetime, ) -> None: options_by_code = _magento_options_by_code(session) if not options_by_code: return mappings = session.scalars(select(V2ChannelAttributeMapping).where(V2ChannelAttributeMapping.channel == "magento")).all() existing = { (str(row.channel), int(row.canonical_attribute_value_id)): row for row in session.scalars(select(V2ChannelAttributeValueMapping)).all() } values_by_attr: dict[int, list[V2AttributeValue]] = {} for mapping in mappings: values = values_by_attr.get(int(mapping.canonical_attribute_id)) if values is None: values = list( session.scalars( select(V2AttributeValue).where(V2AttributeValue.attribute_id == int(mapping.canonical_attribute_id)) ).all() ) values_by_attr[int(mapping.canonical_attribute_id)] = values by_norm = {normalize_lookup_value(row.display_label): row for row in values} by_norm.update({normalize_lookup_value(row.value_code): row for row in values}) for row in values: for alias in row.aliases or []: key = normalize_lookup_value(alias) if key: by_norm.setdefault(key, row) chosen: dict[int, V2AttributeValue] = {} preferred: dict[int, bool] = {} option_ids: dict[int, str] = {} for option in options_by_code.get(normalize_column_key(mapping.channel_attribute_code or ""), []): value = by_norm.get(normalize_lookup_value(option.option_label_norm or option.option_label)) if value is None: continue value_id = int(value.id) exact = normalize_lookup_value(option.option_label) == normalize_lookup_value(value.display_label) if value_id not in chosen or (exact and not preferred.get(value_id)): chosen[value_id] = value preferred[value_id] = exact option_ids[value_id] = str(option.value_index) for value_id, value in chosen.items(): key = ("magento", value_id) if key in existing: stats.channel_value_mappings_skipped_existing += 1 continue row = V2ChannelAttributeValueMapping( channel="magento", canonical_attribute_value_id=value_id, channel_option_id=option_ids[value_id], created_at=created_at, ) session.add(row) existing[key] = row stats.channel_value_mappings_created += 1