"""Ensure Magento spec attributes are mapped and select options are synced from parsed master values. Examples: python -m app.jobs.ensure_magento_spec_attributes --connection-id 1 --dry-run python -m app.jobs.ensure_magento_spec_attributes --connection-id 1 --apply python -m app.jobs.ensure_magento_spec_attributes --connection-id 1 --apply --provision-missing """ from __future__ import annotations import argparse import json from typing import Any, Dict, List import sqlalchemy as sa from sqlalchemy import select from sqlalchemy.dialects.postgresql import insert from db.channel_remote_attributes import add_remote_attribute_mapping, promote_attribute_to_options, remove_remote_attribute_mapping from db.models import ChannelAttributeAlias, MagentoAttributeRegistry from db.session import get_session from db.source_imports import normalize_column_key REMOTE_MAPPER_SCOPE = "dynamic" REMOTE_MAPPER_NOTES = "spec_bundle:ensure_mapping" ATTRIBUTE_BUNDLE: List[Dict[str, Any]] = [ {"remote_code": "manufacturer", "canonical_code": "manufacturer", "data_type": "select", "sync_options": True}, {"remote_code": "family", "canonical_code": "collection", "data_type": "select", "sync_options": True}, {"remote_code": "collection_style", "canonical_code": "collection", "data_type": "select", "sync_options": True}, {"remote_code": "depth", "canonical_code": "depth", "data_type": "select", "sync_options": True}, {"remote_code": "door_style", "canonical_code": "door_style", "data_type": "select", "sync_options": True}, {"remote_code": "color_finish", "canonical_code": "color_finish", "data_type": "select", "sync_options": True}, {"remote_code": "overlay", "canonical_code": "overlay", "data_type": "select", "sync_options": True}, {"remote_code": "finish_type", "canonical_code": "finish_type", "data_type": "select", "sync_options": True}, {"remote_code": "drawer_slides", "canonical_code": "drawer_slides", "data_type": "select", "sync_options": True}, {"remote_code": "hinges", "canonical_code": "hinges", "data_type": "select", "sync_options": True}, {"remote_code": "box_construction", "canonical_code": "box_construction", "data_type": "select", "sync_options": True}, {"remote_code": "shelf_clips", "canonical_code": "shelf_clips", "data_type": "select", "sync_options": True}, {"remote_code": "materials_face_frame", "canonical_code": "materials_face_frame", "data_type": "select", "sync_options": True}, {"remote_code": "materials_box", "canonical_code": "materials_box", "data_type": "select", "sync_options": True}, {"remote_code": "soft_close", "canonical_code": "soft_close", "data_type": "boolean", "sync_options": False}, {"remote_code": "shelves_included", "canonical_code": "shelves_included", "data_type": "boolean", "sync_options": False}, {"remote_code": "pkg_height_in", "canonical_code": "packaged_dimension_height", "data_type": "decimal", "sync_options": False}, {"remote_code": "pkg_width_in", "canonical_code": "packaged_dimension_width", "data_type": "decimal", "sync_options": False}, {"remote_code": "pkg_length_in", "canonical_code": "packaged_dimension_depth", "data_type": "decimal", "sync_options": False}, {"remote_code": "pkg_weight_lb", "canonical_code": "packaged_gross_weight", "data_type": "decimal", "sync_options": False}, {"remote_code": "cabinet_back_material", "canonical_code": "cabinet_back_material", "data_type": "text", "sync_options": False}, {"remote_code": "cabinet_bottom_material", "canonical_code": "cabinet_bottom_material", "data_type": "text", "sync_options": False}, {"remote_code": "cabinet_top_material", "canonical_code": "cabinet_top_material", "data_type": "text", "sync_options": False}, {"remote_code": "cabinet_carcass_material", "canonical_code": "cabinet_carcass_material", "data_type": "text", "sync_options": False}, {"remote_code": "cabinet_toe_kick_construction", "canonical_code": "cabinet_toe_kick_construction", "data_type": "text", "sync_options": False}, {"remote_code": "cabinet_interior_finish", "canonical_code": "cabinet_interior_finish", "data_type": "text", "sync_options": False}, {"remote_code": "cabinet_corner_support", "canonical_code": "cabinet_corner_support", "data_type": "text", "sync_options": False}, {"remote_code": "cabinet_i_beam_support_braces", "canonical_code": "cabinet_i_beam_support_braces", "data_type": "text", "sync_options": False}, {"remote_code": "cabinet_shelf_material", "canonical_code": "cabinet_shelf_material", "data_type": "text", "sync_options": False}, {"remote_code": "cabinet_exterior_side_finish", "canonical_code": "cabinet_exterior_side_finish", "data_type": "text", "sync_options": False}, {"remote_code": "warranty", "canonical_code": "warranty", "data_type": "text", "sync_options": False}, {"remote_code": "care_instructions", "canonical_code": "care_instructions", "data_type": "text", "sync_options": False}, ] def ensure_magento_spec_attributes( *, connection_id: int, dry_run: bool = True, provision_missing: bool = False, ) -> Dict[str, Any]: from db.channel_attribute_provision import provision_pending_for_channels with get_session() as session: alias_cleanup_actions: List[Dict[str, Any]] = [] existing_codes = { normalize_column_key(code) for (code,) in session.execute( select(MagentoAttributeRegistry.attribute_code).where(MagentoAttributeRegistry.connection_id == int(connection_id)) ).all() if normalize_column_key(code) } results: List[Dict[str, Any]] = [] marked_create_new = 0 if not dry_run and "pkg_length_in" in existing_codes: cleanup = remove_remote_attribute_mapping( session, channel_type="magento", channel_attribute_code="packaged_dimension_depth", canonical_code="packaged_dimension_depth", ) if cleanup.get("removed"): alias_cleanup_actions.append(cleanup) for spec in ATTRIBUTE_BUNDLE: remote_code = normalize_column_key(spec["remote_code"]) canonical_code = normalize_column_key(spec["canonical_code"]) exists = remote_code in existing_codes row: Dict[str, Any] = { "remote_code": remote_code, "canonical_code": canonical_code, "data_type": spec["data_type"], "exists_in_registry": exists, "sync_options": bool(spec.get("sync_options")), } if exists: if not dry_run: add_remote_attribute_mapping( session, channel_type="magento", channel_attribute_code=remote_code, canonical_code=canonical_code, data_type=str(spec["data_type"]), sync_options_from_master=bool(spec.get("sync_options")), ) if spec.get("sync_options"): row["promotion"] = promote_attribute_to_options( session, channel_type="magento", connection_id=int(connection_id), channel_attribute_code=remote_code, canonical_code=canonical_code, dry_run=False, ) else: row["planned_action"] = "map_and_sync_options" if spec.get("sync_options") else "map_only" else: row["planned_action"] = "mark_create_new" if not dry_run: _upsert_create_new_alias( session, remote_code=remote_code, canonical_code=canonical_code, data_type=str(spec["data_type"]), ) marked_create_new += 1 results.append(row) provision_result = None if not dry_run: session.commit() if provision_missing: provision_result = provision_pending_for_channels( session, ["magento"], magento_connection_id=int(connection_id), dry_run=False, ) session.commit() for spec in ATTRIBUTE_BUNDLE: if not spec.get("sync_options"): continue if normalize_column_key(spec["remote_code"]) not in existing_codes: promote_attribute_to_options( session, channel_type="magento", connection_id=int(connection_id), channel_attribute_code=spec["remote_code"], canonical_code=spec["canonical_code"], dry_run=False, ) session.commit() return { "status": "ok", "connection_id": int(connection_id), "dry_run": dry_run, "attributes": results, "marked_create_new_count": marked_create_new, "provision_missing": provision_result, "cleanup_actions": alias_cleanup_actions, } def _upsert_create_new_alias( session, *, remote_code: str, canonical_code: str, data_type: str, ) -> None: stmt = insert(ChannelAttributeAlias).values( { "channel_code": "magento", "canonical_code": canonical_code, "channel_attribute_code": remote_code, "mapping_scope": REMOTE_MAPPER_SCOPE, "action": "create_new", "data_type": data_type, "transform_rule": None, "is_active": True, "notes": REMOTE_MAPPER_NOTES, } ) stmt = stmt.on_conflict_do_update( constraint="uq_channel_attribute_alias_channel_canonical_attr", set_={ "mapping_scope": REMOTE_MAPPER_SCOPE, "action": "create_new", "data_type": stmt.excluded.data_type, "is_active": True, "notes": REMOTE_MAPPER_NOTES, "updated_at": sa.func.now(), }, ) session.execute(stmt) def main() -> int: parser = argparse.ArgumentParser(description="Ensure Magento spec attributes are mapped and option-synced") parser.add_argument("--connection-id", type=int, required=True) parser.add_argument("--dry-run", dest="dry_run", action="store_true", default=True) parser.add_argument("--apply", dest="dry_run", action="store_false") parser.add_argument("--provision-missing", action="store_true", help="Create missing Magento attributes first") args = parser.parse_args() result = ensure_magento_spec_attributes( connection_id=args.connection_id, dry_run=args.dry_run, provision_missing=args.provision_missing, ) print(json.dumps(result, indent=2, default=str)) return 0 if __name__ == "__main__": raise SystemExit(main())