"""Push collection landing metadata to Magento category custom attributes.""" from __future__ import annotations import base64 import logging from datetime import datetime, timezone from typing import Any, Dict, List, Optional from sqlalchemy import select from sqlalchemy.orm import Session from channel.url_canonical import normalize_plp_path from db.collection_landing_pages import ( _relative_shopping_collection_value, _relative_shopping_l1_value, _relative_shopping_l2_value, ) from db.collection_landing_schema import ( COLLECTION_LANDING_FIELDS, MAGENTO_ATTRIBUTE_PREFIX, build_channel_field_payload, field_label, magento_attribute_code, ) from db.collection_assets import fetch_collection_asset from db.collection_landing_pages import DEFAULT_CATEGORY_L1 from db.models import MagentoAttributeRegistry from magento.collection_landing_setup import ( MAGENTO_MODULE_DEPLOY_ROOT, magento_eav_type, ) from magento.sku_policy import filter_magento_push_skus MAGENTO_BOOTSTRAP_MODE = "discover" MAGENTO_BOOTSTRAP_NOTE = ( "Magento REST API does not support creating category attributes. " "Install missing hs_* attributes via GET /api/channel/collections/magento-attribute-module " f"(deploy under {MAGENTO_MODULE_DEPLOY_ROOT}), run setup:upgrade, then re-run bootstrap." ) logger = logging.getLogger(__name__) def bootstrap_magento_collection_attributes( session: Session, api: Any, *, connection_id: int, dry_run: bool = False, ) -> Dict[str, Any]: existing = _existing_category_attribute_codes(session, connection_id) planned: List[str] = [] created: List[str] = [] skipped: List[str] = [] missing: List[str] = [] errors: List[str] = [] for field in COLLECTION_LANDING_FIELDS: code = magento_attribute_code(field["key"]) if code in existing: skipped.append(code) continue planned.append(code) if dry_run: continue if api is None: missing.append(code) continue status, body = api.get_category_attribute(code) if status == 200 and body: skipped.append(code) session.add( MagentoAttributeRegistry( connection_id=connection_id, attribute_code=code, attribute_id=int(body.get("attribute_id") or 0), frontend_label=body.get("default_frontend_label") or field_label(field["key"]), frontend_input=body.get("frontend_input") or field["magento_input"], backend_type=body.get("backend_type") or magento_eav_type(field["magento_input"]), is_user_defined=bool(body.get("is_user_defined", True)), fetched_at=datetime.now(timezone.utc), ) ) continue if status == 404: missing.append(code) continue errors.append(f"{code}: HTTP {status} {body}") if not dry_run and skipped: session.flush() result: Dict[str, Any] = { "channel": "magento", "connection_id": connection_id, "dry_run": dry_run, "mode": MAGENTO_BOOTSTRAP_MODE, "planned": planned, "created": created, "skipped": skipped, "missing": missing, "errors": errors, } if missing: result["note"] = MAGENTO_BOOTSTRAP_NOTE result["setup_module_url"] = "/api/channel/collections/magento-attribute-module" return result def build_magento_category_custom_attributes( collection_row: Dict[str, Any], *, field_keys: Optional[set[str]] = None, ) -> List[Dict[str, str]]: values = build_channel_field_payload(collection_row, field_keys=field_keys) attrs: List[Dict[str, str]] = [] for key, value in values.items(): if value is None: continue if key in {"is_collection", "has_collections"}: value = "1" if value == "true" else "0" attrs.append({"attribute_code": magento_attribute_code(key), "value": value}) return attrs def push_magento_collection_landing( api: Any, *, category_id: int, collection_row: Dict[str, Any], field_keys: Optional[set[str]] = None, dry_run: bool = False, ) -> Dict[str, Any]: asset_plan = _collection_asset_plan(collection_row, field_keys=field_keys) localized_row = collection_row uploaded_assets: List[Dict[str, str]] = [] if not dry_run and asset_plan: localized_row, uploaded_assets = localize_magento_collection_assets( api, collection_row, field_keys=field_keys, ) custom_attributes = build_magento_category_custom_attributes(localized_row, field_keys=field_keys) path_slug = str(collection_row.get("path_slug") or collection_row.get("taxonomy_path_slug") or "").strip() start_config_value = _custom_attribute_value(custom_attributes, "hs_start_shopping_config") legacy_intersections_value = _custom_attribute_value(custom_attributes, "hs_shopping_intersections") payload: Dict[str, Any] = {"id": int(category_id)} if custom_attributes: payload["custom_attributes"] = custom_attributes logger.info( "Magento collection landing payload path=%s category_id=%s attr_count=%s has_start_shopping_config=%s " "start_shopping_config_chars=%s has_shopping_intersections=%s shopping_intersections_chars=%s field_keys=%s", path_slug or "", int(category_id), len(custom_attributes), bool(start_config_value), len(start_config_value or ""), bool(legacy_intersections_value), len(legacy_intersections_value or ""), sorted(field_keys) if field_keys is not None else None, ) if dry_run: return { "channel": "magento", "dry_run": True, "category_id": int(category_id), "custom_attribute_count": len(custom_attributes), "custom_attributes": custom_attributes, "asset_plan": asset_plan, "has_start_shopping_config": bool(start_config_value), "start_shopping_config_chars": len(start_config_value or ""), "has_shopping_intersections": bool(legacy_intersections_value), "shopping_intersections_chars": len(legacy_intersections_value or ""), } status, body = api.update_category(int(category_id), payload) if status not in (200, 201): raise RuntimeError(f"update_category failed: HTTP {status} {body}") verify_status, verify_body = api.get_category(int(category_id)) verified_attrs = [] if verify_status == 200 and isinstance(verify_body, dict): verified_attrs = verify_body.get("custom_attributes") or [] verified_start_config = _custom_attribute_value(verified_attrs, "hs_start_shopping_config") verified_legacy_intersections = _custom_attribute_value(verified_attrs, "hs_shopping_intersections") logger.info( "Magento collection landing verify path=%s category_id=%s verify_status=%s has_start_shopping_config=%s " "start_shopping_config_chars=%s has_shopping_intersections=%s shopping_intersections_chars=%s", path_slug or "", int(category_id), verify_status, bool(verified_start_config), len(verified_start_config or ""), bool(verified_legacy_intersections), len(verified_legacy_intersections or ""), ) return { "channel": "magento", "category_id": int(category_id), "custom_attribute_count": len(custom_attributes), "pushed": len(custom_attributes), "uploaded_assets": uploaded_assets, "has_start_shopping_config": bool(start_config_value), "start_shopping_config_chars": len(start_config_value or ""), "verified_start_shopping_config": bool(verified_start_config), "verified_start_shopping_config_chars": len(verified_start_config or ""), } def _custom_attribute_value(custom_attributes: Any, attribute_code: str) -> Optional[str]: for item in custom_attributes or []: if not isinstance(item, dict): continue if str(item.get("attribute_code") or "").strip() != attribute_code: continue value = item.get("value") if value is None: return None return str(value) return None def _shopping_filter_error_text(body: Any) -> str: if isinstance(body, dict): message = str(body.get("message") or "") parameters = str(body.get("parameters") or "") return f"{message}{parameters}".lower() return str(body or "").lower() def _is_invalid_attribute_set_entity_type_error(status: int, body: Any) -> bool: if status != 400: return False return "invalid attribute set entity type" in _shopping_filter_error_text(body) def _merge_custom_attributes(existing: Any, updates: List[Dict[str, Any]]) -> List[Dict[str, Any]]: merged: Dict[str, Any] = {} for item in existing or []: if not isinstance(item, dict): continue code = str(item.get("attribute_code") or "").strip() if not code: continue merged[code] = item.get("value") for item in updates: if not isinstance(item, dict): continue code = str(item.get("attribute_code") or "").strip() if not code: continue merged[code] = item.get("value") return [ {"attribute_code": code, "value": value} for code, value in merged.items() ] def _update_magento_shopping_filters( api: Any, *, channel_sku: str, shopping_collection: str, shopping_l1: str, shopping_l2: str, ) -> tuple[int, Any]: custom_attributes = [ {"attribute_code": "shopping_collection", "value": shopping_collection}, {"attribute_code": "shopping_l1", "value": shopping_l1}, {"attribute_code": "shopping_l2", "value": shopping_l2}, ] payload = { "sku": channel_sku, "custom_attributes": custom_attributes, } status, body = api.put_product(channel_sku, payload) if status in (200, 201): return status, body if not _is_invalid_attribute_set_entity_type_error(status, body): return status, body product = api.get_product(channel_sku) or {} retry_payload: Dict[str, Any] = { "sku": channel_sku, "type_id": str(product.get("type_id") or "simple"), "custom_attributes": _merge_custom_attributes(product.get("custom_attributes"), custom_attributes), } attribute_set_id = product.get("attribute_set_id") try: if attribute_set_id is not None and str(attribute_set_id).strip() != "": retry_payload["attribute_set_id"] = int(attribute_set_id) except (TypeError, ValueError): pass return api.post_product(retry_payload) def sync_magento_collection_products( session: Session, api: Any, *, category_id: int, collection_row: Dict[str, Any], connection_id: Optional[int], dry_run: bool = False, on_progress: Optional[Any] = None, master_skus: Optional[List[str]] = None, ) -> Dict[str, Any]: """Materialize a collection's resolved master SKUs onto its Magento category and shopping filters.""" from db.channel_sku_mapping import mapping_by_master from db.channel_listing_path import resolve_listing_paths_for_sku from db.collection_sku_mapping import resolve_collection_skus from db.models import CollectionLandingPage from db.product_channel_taxonomy import upsert_sku_taxonomy_assignment if api is not None and not dry_run: _ensure_magento_shopping_filter_attributes(api) if master_skus is None: resolved = resolve_collection_skus( session, path_slug=collection_row.get("path_slug"), category_l1=collection_row.get("category_l1"), collection=collection_row.get("collection"), sku_prefixes=collection_row.get("sku_prefixes"), ) sku_items = resolved.get("skus") or [] master_skus = [str(item["master_sku"]) for item in sku_items] else: master_skus = [str(sku) for sku in master_skus if str(sku or "").strip()] master_skus = filter_magento_push_skus(master_skus) channel_skus = mapping_by_master( session, "magento", master_skus, connection_id=connection_id, ) active_collections = session.scalars( select(CollectionLandingPage).where(CollectionLandingPage.is_active.is_(True)) ).all() collection_paths = { normalize_plp_path(row.path_slug) for row in active_collections if normalize_plp_path(row.path_slug) } shopping_l1_paths: set[str] = set() shopping_l2_paths: set[str] = set() for landing in active_collections: config = landing.start_shopping_config if isinstance(landing.start_shopping_config, dict) else {} for item in config.get("items") or []: if not isinstance(item, dict): continue l1_path = normalize_plp_path(item.get("l1_path_slug") or "") if l1_path: shopping_l1_paths.add(l1_path) for raw_l2 in item.get("l2_path_slugs") or []: l2_path = normalize_plp_path(raw_l2 or "") if l2_path: shopping_l2_paths.add(l2_path) planned = [channel_skus.get(sku, sku) for sku in master_skus] if dry_run: return { "dry_run": True, "resolved": len(master_skus), "would_assign": len(planned), "sample_skus": planned[:25], "shopping_collection_sample": sorted(list(collection_paths))[:10], "shopping_l1_sample": sorted(list(shopping_l1_paths))[:10], "shopping_l2_sample": sorted(list(shopping_l2_paths))[:10], } status, existing_rows, error = api.get_category_products(int(category_id)) if status != 200: raise RuntimeError(f"get_category_products failed: HTTP {status} {error}") existing = { str(item.get("sku") or "").strip() for item in existing_rows if isinstance(item, dict) and str(item.get("sku") or "").strip() } assigned = 0 skipped_existing = 0 errors: List[Dict[str, Any]] = [] remote_path = str(collection_row.get("path_slug") or "") total = len(master_skus) processed = 0 for position, master_sku in enumerate(master_skus): try: channel_sku = channel_skus.get(master_sku, master_sku) upsert_sku_taxonomy_assignment( session, master_sku=master_sku, channel_code="magento", connection_id=connection_id, taxonomy_kind="category", remote_id=str(int(category_id)), remote_path=remote_path, match_source="collection_push", sort_order=position, ) if channel_sku in existing: skipped_existing += 1 else: link_status, body, link_error = api.assign_product_to_category( int(category_id), channel_sku, position=position, ) if link_status in (200, 201): assigned += 1 existing.add(channel_sku) else: errors.append( { "master_sku": master_sku, "channel_sku": channel_sku, "http_status": link_status, "error": link_error or body, } ) listing_paths = { normalize_plp_path(str(item.get("path_slug") or "")) for item in resolve_listing_paths_for_sku(session, master_sku) if normalize_plp_path(str(item.get("path_slug") or "")) } shopping_collection = _encode_magento_shopping_collection_value( [remote_path] if normalize_plp_path(remote_path) in collection_paths else [] ) shopping_l1 = _encode_magento_shopping_l1_value( _matching_shopping_branch_paths(listing_paths, shopping_l1_paths) ) shopping_l2 = _encode_magento_shopping_l2_value(sorted(listing_paths & shopping_l2_paths)) put_status, put_body = _update_magento_shopping_filters( api, channel_sku=channel_sku, shopping_collection=shopping_collection, shopping_l1=shopping_l1, shopping_l2=shopping_l2, ) if put_status not in (200, 201): errors.append( { "master_sku": master_sku, "channel_sku": channel_sku, "http_status": put_status, "error": put_body, "stage": "shopping_filters", } ) except Exception as exc: errors.append({"master_sku": master_sku, "error": str(exc)}) processed += 1 index = processed if on_progress and (index == 1 or index % 25 == 0 or index == total): on_progress( { "stage": "magento_sync", "completed": index, "total": total, "pushed": assigned, "failed": len(errors), "notes": f"Magento sync {index}/{total} ({assigned} assigned)", } ) if on_progress and total: on_progress( { "stage": "magento_sync", "completed": total, "total": total, "pushed": assigned, "failed": len(errors), "notes": f"Magento sync complete {total}/{total} ({assigned} assigned)", } ) return { "dry_run": False, "resolved": len(master_skus), "processed": processed, "assigned": assigned, "skipped_existing": skipped_existing, "failed": len(errors), "errors": errors[:25], } def _ensure_magento_shopping_filter_attributes(api: Any) -> None: """Ensure shopping_* exist and are indexed for OpenSearch field filters.""" searchable_payload = { "is_searchable": True, "is_filterable": 1, "is_filterable_in_search": True, "used_in_product_listing": True, } for attribute_code, frontend_label in [ ("shopping_collection", "Shopping Collection"), ("shopping_l1", "Shopping Level 1"), ("shopping_l2", "Shopping Level 2"), ]: status, body = api.get_attribute(attribute_code) if status == 200 and body: needs_index = not ( bool(body.get("is_searchable")) and int(body.get("is_filterable") or 0) > 0 and bool(body.get("is_filterable_in_search")) ) if needs_index: update_status, update_body = api.update_attribute(attribute_code, searchable_payload) if update_status not in (200, 201): raise RuntimeError( f"Could not make Magento attribute {attribute_code} searchable: " f"HTTP {update_status} {update_body}" ) continue create_status, create_body = api.create_attribute( attribute_code, frontend_label=frontend_label, frontend_input="text", backend_type="varchar", is_user_defined=True, is_required=False, used_in_product_listing=True, ) if create_status not in (200, 201): raise RuntimeError( f"Could not create Magento product attribute {attribute_code}: HTTP {create_status} {create_body}" ) update_status, update_body = api.update_attribute(attribute_code, searchable_payload) if update_status not in (200, 201): raise RuntimeError( f"Magento attribute {attribute_code} created but could not be made searchable: " f"HTTP {update_status} {update_body}" ) add_status, add_error = api.add_attribute_to_attribute_set(4, attribute_code) if add_status not in (200, 201) and add_error: raise RuntimeError( f"Magento attribute {attribute_code} created but could not be added to default set: {add_error}" ) def _first_normalized_path(values: List[str]) -> str: for value in values: cleaned = normalize_plp_path(value) if cleaned: return cleaned return "" def _encode_magento_shopping_collection_value(values: List[str]) -> str: return str(_relative_shopping_collection_value(_first_normalized_path(values)) or "") def _matching_shopping_branch_paths(assignments: set[str], branches: set[str]) -> List[str]: matched: List[str] = [] for branch in sorted(branches): normalized_branch = normalize_plp_path(branch) if not normalized_branch: continue if normalized_branch in assignments or any( assignment.startswith(f"{normalized_branch}/") for assignment in assignments ): matched.append(normalized_branch) return matched def _encode_magento_shopping_l1_value(values: List[str]) -> str: return str(_relative_shopping_l1_value(_first_normalized_path(values)) or "") def _encode_magento_shopping_l2_value(values: List[str]) -> str: return str(_relative_shopping_l2_value(_first_normalized_path(values)) or "") COLLECTION_ASSET_KEYS = {"thumbnail_image_url", "category_icon_url", "pdf_url", "hero_images"} def localize_magento_collection_assets( api: Any, collection_row: Dict[str, Any], *, field_keys: Optional[set[str]] = None, ) -> tuple[Dict[str, Any], List[Dict[str, str]]]: """Copy collection assets into Magento media and replace source URLs with local URLs.""" out = dict(collection_row) uploaded: List[Dict[str, str]] = [] cache: Dict[str, str] = {} def localize(source_url: str, field: str) -> str: source = str(source_url or "").strip() if not source: return "" if source in cache: return cache[source] asset = fetch_collection_asset(source) status, body = api.upload_collection_asset( filename=asset.filename, mime_type=asset.mime_type, base64_content=base64.b64encode(asset.content).decode("ascii"), file_size_bytes=len(asset.content), ) local_url = str((body or {}).get("url") or "").strip() if status not in (200, 201) or not local_url: raise RuntimeError(f"Magento collection asset upload failed: HTTP {status} {body}") cache[source] = local_url uploaded.append( { "field": field, "source_url": source, "url": local_url, "filename": asset.filename, } ) return local_url wanted = COLLECTION_ASSET_KEYS if field_keys is None else COLLECTION_ASSET_KEYS & field_keys if "thumbnail_image_url" in wanted and out.get("thumbnail_image_url"): out["thumbnail_image_url"] = localize(out["thumbnail_image_url"], "thumbnail_image_url") if "category_icon_url" in wanted and out.get("category_icon_url"): out["category_icon_url"] = localize(out["category_icon_url"], "category_icon_url") if "pdf_url" in wanted and out.get("pdf_url"): out["pdf_url"] = localize(out["pdf_url"], "pdf_url") if "hero_images" in wanted and out.get("hero_images"): hero_values = out["hero_images"] if isinstance(out["hero_images"], list) else [out["hero_images"]] out["hero_images"] = [ localize(value, "hero_images") for value in hero_values if str(value or "").strip() ] for node_key in ("child_taxonomy_nodes", "sibling_taxonomy_nodes"): nodes = out.get(node_key) if not isinstance(nodes, list): continue localized_nodes: List[Dict[str, Any]] = [] for node in nodes: if not isinstance(node, dict): localized_nodes.append(node) continue node_out = dict(node) icon_url = str(node_out.get("category_icon_url") or "").strip() if icon_url and ("category_icon_url" in wanted or field_keys is None): node_out["category_icon_url"] = localize(icon_url, "category_icon_url") localized_nodes.append(node_out) out[node_key] = localized_nodes return out, uploaded def _collection_asset_plan( collection_row: Dict[str, Any], *, field_keys: Optional[set[str]] = None, ) -> List[Dict[str, str]]: wanted = COLLECTION_ASSET_KEYS if field_keys is None else COLLECTION_ASSET_KEYS & field_keys plan: List[Dict[str, str]] = [] for key in ("thumbnail_image_url", "category_icon_url", "pdf_url"): value = collection_row.get(key) if key in wanted and value: plan.append({"field": key, "source_url": str(value)}) if "hero_images" in wanted: heroes = collection_row.get("hero_images") or [] if not isinstance(heroes, list): heroes = [heroes] plan.extend( {"field": "hero_images", "source_url": str(value)} for value in heroes if value ) for node_key in ("child_taxonomy_nodes", "sibling_taxonomy_nodes"): nodes = collection_row.get(node_key) or [] if not isinstance(nodes, list): continue for node in nodes: if isinstance(node, dict) and node.get("category_icon_url"): plan.append( {"field": "category_icon_url", "source_url": str(node["category_icon_url"])} ) return plan def push_magento_shopping_category_icons( api: Any, session: Session, *, collection_row: Dict[str, Any], connection_id: Optional[int], dry_run: bool = False, ) -> Dict[str, Any]: """Push hs_category_icon_url onto hub shopping categories (Base/Wall/etc.).""" from db.collection_landing_push import ( build_taxonomy_relationship_metadata, ensure_verified_magento_category_target, ) metadata = build_taxonomy_relationship_metadata( session, path_slug=str(collection_row.get("path_slug") or ""), channel_code="magento", connection_id=connection_id, ) nodes = _shopping_taxonomy_nodes(metadata) pushed = 0 skipped = 0 errors: List[Dict[str, Any]] = [] category_l1 = str(collection_row.get("category_l1") or DEFAULT_CATEGORY_L1) for node in nodes: path_slug = str(node.get("path_slug") or "").strip() icon_url = str(node.get("category_icon_url") or "").strip() if not path_slug or not icon_url: skipped += 1 continue try: from db.collection_landing_push import _authoritative_magento_collection_row node_row = _authoritative_magento_collection_row( session, { "path_slug": path_slug, "category_l1": category_l1, "collection": node.get("name"), "title": node.get("name"), }, ) target = ensure_verified_magento_category_target( session, node_row, connection_id=connection_id, api=api, provision_missing=False, ) remote_id = target.get("remote_id") if not remote_id or not str(remote_id).isdigit(): skipped += 1 continue push_magento_collection_landing( api, category_id=int(remote_id), collection_row={"category_icon_url": icon_url}, field_keys={"category_icon_url"}, dry_run=dry_run, ) pushed += 1 except Exception as exc: errors.append({"path_slug": path_slug, "error": str(exc)}) return { "dry_run": dry_run, "planned": len(nodes), "pushed": pushed, "skipped": skipped, "errors": errors[:25], } def _shopping_taxonomy_nodes(metadata: Dict[str, Any]) -> List[Dict[str, Any]]: if metadata.get("is_collection"): source = metadata.get("sibling_taxonomy_nodes") or [] else: source = metadata.get("child_taxonomy_nodes") or [] return [ node for node in source if isinstance(node, dict) and str(node.get("node_kind") or "") != "collection" ] def _existing_category_attribute_codes(session: Session, connection_id: int) -> set[str]: from sqlalchemy import select rows = session.scalars( select(MagentoAttributeRegistry.attribute_code).where( MagentoAttributeRegistry.connection_id == connection_id, ) ).all() prefix = f"{MAGENTO_ATTRIBUTE_PREFIX}" return {str(code) for code in rows if str(code).startswith(prefix)}