from __future__ import annotations import re from dataclasses import dataclass from datetime import datetime, timedelta, timezone from typing import Any, Dict, Iterable, List, Optional, Set, Tuple from sqlalchemy import select from sqlalchemy.orm import Session from channel.url_canonical import normalize_plp_path from db.catalog_intent_planner import ( ALL_PRODUCTS, category_targets_for_product_guarded, hub_path_slug, listing_path_kind, listing_path_target_from_category_path, listing_path_title, parent_path_slug, shopify_collection_targets_for_product, shopify_path_slug_for_collection_title, slugify_segment, ) from db.channel_assignments import active_skus_for_channel from db.channel_listing_path import upsert_listing_path, upsert_listing_path_target, upsert_product_listing_path_assignment from db.magento_repositories import SqlAlchemyCategoryRegistryRepository from db.models import ( ChannelListingPath, MagentoCategoryRegistry, MasterProduct, MasterProductAttributeValue, ProductChannelTaxonomyAssignment, ProductListingPathAssignment, ShopifyCollectionRegistry, ) from db.product_channel_taxonomy import ( ACTIVE_STATUS, MAGENTO_KIND, SHOPIFY_COLLECTION_KIND, SHOPIFY_PRODUCT_CATEGORY_KIND, upsert_sku_taxonomy_assignment, ) from db.source_imports import normalize_column_key MATCH_SOURCE = "catalog_provision" @dataclass(frozen=True) class ProvisionOptions: magento_connection_id: Optional[int] = None shopify_connection_id: Optional[int] = None only_assigned: bool = True dry_run: bool = True include_magento: bool = True include_shopify: bool = True create_missing: bool = True assign_skus: bool = True delta_only: bool = True seed_master_listing_paths: bool = True provision_attributes: bool = True include_brand_collections: bool = True shopify_brand_filter: Optional[List[str]] = None limit: Optional[int] = None def provision_channel_catalogs( session: Session, *, options: ProvisionOptions, magento_api: Optional[Any] = None, shopify_client: Optional[Any] = None, shopify_shop_code: Optional[str] = None, ) -> Dict[str, Any]: products = _load_products(session, options=options) plan = build_channel_catalog_plan(session, products, options=options) master_listing_plan = ( build_master_listing_path_plan(session, _load_master_placement_products(session, options=options), options=options) if options.seed_master_listing_paths else {"path_targets": [], "assignments": [], "product_count": 0} ) result: Dict[str, Any] = { "status": "ok", "dry_run": options.dry_run, "assignment_mode": "delta" if options.delta_only else "full", "product_count": len(products), "channel_product_counts": plan["product_counts"], "master_catalog": { "product_count": master_listing_plan["product_count"], "listing_paths": len(master_listing_plan["path_targets"]), "listing_path_assignments": len(master_listing_plan["assignments"]), }, "attributes": {}, "targets": { "magento_categories": len(plan["magento"]["category_paths"]), "shopify_collections": len(plan["shopify"]["collections"]), "shopify_product_categories": len(plan["shopify"]["product_categories"]), }, "assignments": { "magento_category": len(plan["magento"]["assignments"]), "shopify_collection": len(plan["shopify"]["collection_assignments"]), "shopify_product_category": len(plan["shopify"]["product_category_assignments"]), }, "samples": { "magento_categories": plan["magento"]["category_paths"][:25], "shopify_collections": plan["shopify"]["collections"][:25], "magento_assignments": plan["magento"]["assignments"][:25], "shopify_assignments": ( plan["shopify"]["collection_assignments"][:15] + plan["shopify"]["product_category_assignments"][:10] ), "master_listing_paths": master_listing_plan["path_targets"][:25], "master_listing_assignments": master_listing_plan["assignments"][:25], }, } if options.dry_run: if options.provision_attributes: result["attributes"] = _provision_attributes(session, options=options, dry_run=True) return result if options.seed_master_listing_paths: result["master_catalog"].update( _apply_master_listing_paths(session, master_listing_plan, options=options) ) if options.include_magento: result["magento"] = _apply_magento( session, plan, options=options, api=magento_api, ) if options.include_shopify: result["shopify"] = _apply_shopify( session, plan, options=options, client=shopify_client, shop_code=shopify_shop_code, ) if options.provision_attributes: result["attributes"] = _provision_attributes(session, options=options, dry_run=False) return result def _provision_attributes(session: Session, *, options: ProvisionOptions, dry_run: bool) -> Dict[str, Any]: from db.channel_attribute_provision import provision_pending_for_channels channels = [] if options.include_magento: channels.append("magento") if options.include_shopify: channels.append("shopify") if not channels: return {} return provision_pending_for_channels( session, channels, magento_connection_id=options.magento_connection_id, shopify_connection_id=options.shopify_connection_id, dry_run=dry_run, ) def build_channel_catalog_plan( session: Session, products: Iterable[MasterProduct], *, options: ProvisionOptions, ) -> Dict[str, Any]: magento_paths: Dict[str, Dict[str, Any]] = {} shopify_collections: Dict[str, Dict[str, Any]] = {} magento_assignments: List[Dict[str, Any]] = [] shopify_collection_assignments: List[Dict[str, Any]] = [] shopify_product_category_assignments: List[Dict[str, Any]] = [] shopify_product_categories: Dict[str, Dict[str, Any]] = {} products = list(products) attrs_by_sku = _attrs_by_sku(session, [product.id for product in products]) magento_allowed_skus, shopify_allowed_skus = _channel_allowed_skus(session, products, options=options) existing_assignments = _existing_assignment_keys(session, products, options=options) existing_magento = _existing_magento_categories(session, options.magento_connection_id) existing_shopify = _existing_shopify_collections(session, options.shopify_connection_id) from db.product_channel_taxonomy import sku_has_master_taxonomy_placements canonical_taxonomy_skus = { product.sku for product in products if hasattr(session, "scalars") and sku_has_master_taxonomy_placements(session, product.sku) } for product in products: attrs = attrs_by_sku.get(product.sku, {}) canonical_taxonomy = product.sku in canonical_taxonomy_skus # Once a product has canonical master taxonomy placements, its channel # assignments are maintained from linked taxonomy leaves. Do not let the # legacy broad-path provisioner append categories beside them. if product.sku in magento_allowed_skus and not canonical_taxonomy: category_targets = category_targets_for_product_guarded( session, product, attrs, magento_connection_id=options.magento_connection_id, ) for sort_order, path in enumerate(category_targets, start=10): category_id = existing_magento.get(path) assignment_exists = ( category_id is not None and _assignment_key(product.sku, "magento", MAGENTO_KIND, category_id) in existing_assignments ) if not options.delta_only or not assignment_exists: magento_paths.setdefault( path, { "path": path, "exists": path in existing_magento, "category_id": category_id, }, ) magento_assignments.append( { "master_sku": product.sku, "path": path, "sort_order": sort_order, } ) if product.sku in shopify_allowed_skus and _matches_shopify_brand(product, options): category_targets = category_targets_for_product_guarded( session, product, attrs, magento_connection_id=options.magento_connection_id, ) collection_targets = [] if canonical_taxonomy else shopify_collection_targets_for_product( product, category_targets, attrs, include_brand=options.include_brand_collections, ) for sort_order, title in enumerate(collection_targets, start=10): handle = slugify_segment(title) existing = existing_shopify.get(handle) or existing_shopify.get(_norm(title)) assignment_exists = ( existing is not None and _assignment_key(product.sku, "shopify", SHOPIFY_COLLECTION_KIND, existing) in existing_assignments ) if not options.delta_only or not assignment_exists: shopify_collections.setdefault( title, { "title": title, "handle": handle, "exists": bool(existing), "collection_id": existing, }, ) shopify_collection_assignments.append( { "master_sku": product.sku, "title": title, "handle": handle, "sort_order": sort_order, } ) suggestion = _suggest_shopify_category(session, product) if suggestion: taxonomy_id = suggestion["taxonomy_id"] assignment_exists = ( _assignment_key(product.sku, "shopify", SHOPIFY_PRODUCT_CATEGORY_KIND, taxonomy_id) in existing_assignments ) if not options.delta_only or not assignment_exists: shopify_product_categories.setdefault(taxonomy_id, suggestion) shopify_product_category_assignments.append( { "master_sku": product.sku, "taxonomy_id": taxonomy_id, "full_name": suggestion.get("full_name"), "sort_order": 10, } ) return { "product_counts": { "magento": len(magento_allowed_skus), "shopify": sum( 1 for product in products if product.sku in shopify_allowed_skus and _matches_shopify_brand(product, options) ), }, "magento": { "category_paths": list(magento_paths.values()), "assignments": magento_assignments, }, "shopify": { "collections": list(shopify_collections.values()), "product_categories": list(shopify_product_categories.values()), "collection_assignments": shopify_collection_assignments, "product_category_assignments": shopify_product_category_assignments, }, } def build_master_listing_path_plan( session: Session, products: Iterable[MasterProduct], *, options: ProvisionOptions, ) -> Dict[str, Any]: products = list(products) attrs_by_sku = _attrs_by_sku(session, [product.id for product in products]) existing_paths = _existing_listing_paths(session) existing_assignments = _existing_listing_assignment_keys(session, products, existing_paths=existing_paths, options=options) path_targets: Dict[str, Dict[str, Any]] = {} assignments: List[Dict[str, Any]] = [] for product in products: attrs = attrs_by_sku.get(product.sku, {}) category_targets = category_targets_for_product_guarded( session, product, attrs, magento_connection_id=options.magento_connection_id, ) for sort_order, path in enumerate(category_targets, start=10): slug = normalize_plp_path(path) if not slug: continue existing_id = existing_paths.get(slug) assignment_exists = ( existing_id is not None and (product.sku, int(existing_id)) in existing_assignments ) if options.delta_only and assignment_exists: continue path_targets.setdefault( slug, { **listing_path_target_from_category_path(path), "exists": existing_id is not None, "listing_path_id": existing_id, }, ) assignments.append( { "master_sku": product.sku, "path_slug": slug, "sort_order": sort_order, } ) return { "product_count": len(products), "path_targets": list(path_targets.values()), "assignments": assignments, } def _apply_master_listing_paths( session: Session, plan: Dict[str, Any], *, options: ProvisionOptions, ) -> Dict[str, Any]: path_ids_by_slug: Dict[str, int] = {} created_or_updated_paths = 0 assigned = 0 for target in plan["path_targets"]: row = upsert_listing_path( session, path_slug=target["path_slug"], title=target["title"], path_kind=target.get("path_kind"), parent_path_slug=target.get("parent_path_slug"), hub_path_slug=target.get("hub_path_slug"), notes="seeded by catalog provision", ) if row.get("id"): path_ids_by_slug[target["path_slug"]] = int(row["id"]) created_or_updated_paths += 1 if options.assign_skus: for item in plan["assignments"]: path_id = path_ids_by_slug.get(item["path_slug"]) if path_id is None: continue upsert_product_listing_path_assignment( session, master_sku=item["master_sku"], listing_path_id=path_id, match_source=MATCH_SOURCE, sort_order=int(item.get("sort_order") or 0), ) assigned += 1 return { "upserted_listing_paths": created_or_updated_paths, "upserted_listing_path_assignments": assigned, } def _apply_magento( session: Session, plan: Dict[str, Any], *, options: ProvisionOptions, api: Optional[Any], ) -> Dict[str, Any]: category_ids_by_path: Dict[str, int] = {} created = 0 missing = 0 errors: List[Dict[str, Any]] = [] repo = SqlAlchemyCategoryRegistryRepository(session) ttl_cutoff = datetime.now(timezone.utc) - timedelta(days=3650) for target in plan["magento"]["category_paths"]: path = target["path"] category_id = target.get("category_id") if category_id: category_ids_by_path[path] = int(category_id) continue if _norm(path) == _norm(ALL_PRODUCTS): root_id = None if options.magento_connection_id: root_id = repo.resolve_path(options.magento_connection_id, ALL_PRODUCTS, ttl_cutoff) if not root_id: root_id = repo.resolve_path(options.magento_connection_id, "Default Category", ttl_cutoff) if root_id: category_ids_by_path[path] = int(root_id) else: missing += 1 continue category_id = repo.resolve_path(options.magento_connection_id or 0, path, ttl_cutoff) if options.magento_connection_id else None if category_id: category_ids_by_path[path] = int(category_id) continue if not options.create_missing or api is None or options.magento_connection_id is None: missing += 1 continue try: category_id = _ensure_magento_category_path( session, api, connection_id=options.magento_connection_id, path=path, ) if category_id: category_ids_by_path[path] = int(category_id) created += 1 else: missing += 1 except Exception as exc: errors.append({"path": path, "error": str(exc)}) upserted = 0 if options.assign_skus: for item in plan["magento"]["assignments"]: category_id = category_ids_by_path.get(item["path"]) if not category_id: continue upsert_sku_taxonomy_assignment( session, master_sku=item["master_sku"], channel_code="magento", connection_id=options.magento_connection_id, taxonomy_kind=MAGENTO_KIND, remote_id=str(category_id), remote_path=item["path"], match_source=MATCH_SOURCE, sort_order=int(item.get("sort_order") or 0), ) upserted += 1 linked_targets = _link_provision_listing_path_targets_magento( session, category_ids_by_path=category_ids_by_path, connection_id=options.magento_connection_id, ) return { "created_categories": created, "missing_categories": missing, "assigned": upserted, "linked_listing_path_targets": linked_targets, "errors": errors[:25], "error_count": len(errors), } def _apply_shopify( session: Session, plan: Dict[str, Any], *, options: ProvisionOptions, client: Optional[Any], shop_code: Optional[str], ) -> Dict[str, Any]: collection_ids_by_title: Dict[str, str] = {} created = 0 missing = 0 errors: List[Dict[str, Any]] = [] for target in plan["shopify"]["collections"]: title = target["title"] collection_id = target.get("collection_id") if collection_id: collection_ids_by_title[title] = str(collection_id) continue if not options.create_missing or client is None or not shop_code: missing += 1 continue try: from shopify.collection_sync import create_shopify_collection created_row = create_shopify_collection( session, client, shop_code=shop_code, connection_id=options.shopify_connection_id, title=title, handle=target.get("handle"), ) collection_ids_by_title[title] = created_row["collection_id"] created += 1 except Exception as exc: errors.append({"title": title, "error": str(exc)}) collection_upserted = 0 product_category_upserted = 0 if options.assign_skus: for item in plan["shopify"]["collection_assignments"]: collection_id = collection_ids_by_title.get(item["title"]) if not collection_id: continue upsert_sku_taxonomy_assignment( session, master_sku=item["master_sku"], channel_code="shopify", connection_id=options.shopify_connection_id, taxonomy_kind=SHOPIFY_COLLECTION_KIND, remote_id=collection_id, remote_path=item["title"], match_source=MATCH_SOURCE, sort_order=int(item.get("sort_order") or 0), ) collection_upserted += 1 for item in plan["shopify"]["product_category_assignments"]: upsert_sku_taxonomy_assignment( session, master_sku=item["master_sku"], channel_code="shopify", connection_id=options.shopify_connection_id, taxonomy_kind=SHOPIFY_PRODUCT_CATEGORY_KIND, remote_id=item["taxonomy_id"], remote_path=item.get("full_name"), match_source=MATCH_SOURCE, sort_order=int(item.get("sort_order") or 0), ) product_category_upserted += 1 linked_targets = _link_provision_listing_path_targets_shopify( session, plan=plan, collection_ids_by_title=collection_ids_by_title, connection_id=options.shopify_connection_id, ) return { "created_collections": created, "missing_collections": missing, "assigned_collections": collection_upserted, "assigned_product_categories": product_category_upserted, "linked_listing_path_targets": linked_targets, "errors": errors[:25], "error_count": len(errors), } def _link_provision_listing_path_targets_magento( session: Session, *, category_ids_by_path: Dict[str, int], connection_id: Optional[int], ) -> int: if connection_id is None: return 0 linked = 0 for path, category_id in category_ids_by_path.items(): if _norm(path) == _norm(ALL_PRODUCTS): continue target = listing_path_target_from_category_path(path) slug = target.get("path_slug") if not slug: continue upsert_listing_path( session, path_slug=slug, title=target["title"], path_kind=target.get("path_kind"), parent_path_slug=target.get("parent_path_slug"), hub_path_slug=target.get("hub_path_slug"), notes="linked by catalog provision", ) upsert_listing_path_target( session, path_slug=slug, channel_code="magento", remote_id=str(int(category_id)), taxonomy_kind="category", connection_id=connection_id, remote_path=path, ) linked += 1 return linked def _link_provision_listing_path_targets_shopify( session: Session, *, plan: Dict[str, Any], collection_ids_by_title: Dict[str, str], connection_id: Optional[int], ) -> int: linked = 0 for target in plan.get("shopify", {}).get("collections") or []: title = str(target.get("title") or "").strip() collection_id = collection_ids_by_title.get(title) or target.get("collection_id") if not title or not collection_id: continue slug = shopify_path_slug_for_collection_title(title) if not slug: continue upsert_listing_path( session, path_slug=slug, title=title, path_kind="collection", hub_path_slug=hub_path_slug(slug), notes="linked by catalog provision", ) upsert_listing_path_target( session, path_slug=slug, channel_code="shopify", remote_id=str(collection_id), taxonomy_kind="collection", connection_id=connection_id, remote_path=target.get("handle") or slugify_segment(title), ) linked += 1 return linked def _load_products(session: Session, *, options: ProvisionOptions) -> List[MasterProduct]: stmt = select(MasterProduct).where(MasterProduct.is_active.is_(True)) assigned_skus = set() if options.only_assigned: if options.include_magento: assigned_skus |= active_skus_for_channel(session, "magento") if options.include_shopify: assigned_skus |= active_skus_for_channel(session, "shopify") if not assigned_skus: return [] stmt = stmt.where(MasterProduct.sku.in_(sorted(assigned_skus))) stmt = stmt.order_by(MasterProduct.sku) if options.limit: stmt = stmt.limit(options.limit) return list(session.scalars(stmt).all()) def _load_master_placement_products(session: Session, *, options: ProvisionOptions) -> List[MasterProduct]: stmt = select(MasterProduct).where(MasterProduct.is_active.is_(True)).order_by(MasterProduct.sku) if options.limit: stmt = stmt.limit(options.limit) return list(session.scalars(stmt).all()) def _existing_listing_paths(session: Session) -> Dict[str, int]: return { row.path_slug: int(row.id) for row in session.scalars(select(ChannelListingPath).where(ChannelListingPath.is_active.is_(True))).all() if row.path_slug and row.id } def _existing_listing_assignment_keys( session: Session, products: Iterable[MasterProduct], *, existing_paths: Dict[str, int], options: ProvisionOptions, ) -> Set[Tuple[str, int]]: if not options.delta_only: return set() product_skus = sorted({product.sku for product in products if product.sku}) path_ids = sorted(set(existing_paths.values())) if not product_skus or not path_ids: return set() keys: Set[Tuple[str, int]] = set() for row in session.scalars( select(ProductListingPathAssignment) .where(ProductListingPathAssignment.master_sku.in_(product_skus)) .where(ProductListingPathAssignment.listing_path_id.in_(path_ids)) .where(ProductListingPathAssignment.assignment_status == ACTIVE_STATUS) ).all(): keys.add((row.master_sku, int(row.listing_path_id))) return keys def _channel_allowed_skus( session: Session, products: Iterable[MasterProduct], *, options: ProvisionOptions, ) -> Tuple[Set[str], Set[str]]: product_skus = {product.sku for product in products if product.sku} if not options.only_assigned: return ( set(product_skus) if options.include_magento else set(), set(product_skus) if options.include_shopify else set(), ) magento_skus = ( active_skus_for_channel(session, "magento") & product_skus if options.include_magento else set() ) shopify_skus = ( active_skus_for_channel(session, "shopify") & product_skus if options.include_shopify else set() ) return magento_skus, shopify_skus def _existing_assignment_keys( session: Session, products: Iterable[MasterProduct], *, options: ProvisionOptions, ) -> Set[Tuple[str, str, str, str]]: if not options.delta_only: return set() product_skus = sorted({product.sku for product in products if product.sku}) if not product_skus: return set() channels = [] if options.include_magento: channels.append(("magento", options.magento_connection_id)) if options.include_shopify: channels.append(("shopify", options.shopify_connection_id)) if not channels: return set() keys: Set[Tuple[str, str, str, str]] = set() for channel, connection_id in channels: stmt = ( select(ProductChannelTaxonomyAssignment) .where(ProductChannelTaxonomyAssignment.master_sku.in_(product_skus)) .where(ProductChannelTaxonomyAssignment.channel_code == channel) .where(ProductChannelTaxonomyAssignment.assignment_status == ACTIVE_STATUS) ) stmt = stmt.where( ProductChannelTaxonomyAssignment.connection_id.is_(None) if connection_id is None else ProductChannelTaxonomyAssignment.connection_id == connection_id ) for row in session.scalars(stmt).all(): keys.add(_assignment_key(row.master_sku, row.channel_code, row.taxonomy_kind, row.remote_id)) return keys def _assignment_key(master_sku: str, channel_code: str, taxonomy_kind: str, remote_id: Any) -> Tuple[str, str, str, str]: return ( str(master_sku or "").strip(), str(channel_code or "").strip().lower(), str(taxonomy_kind or "").strip(), str(remote_id or "").strip(), ) def _matches_shopify_brand(product: MasterProduct, options: ProvisionOptions) -> bool: wanted = {_norm(item) for item in (options.shopify_brand_filter or []) if _clean(item)} if not wanted: return True values = { _norm(product.brand), _norm(product.category_l2), _norm(product.manufacturer), } return bool(values & wanted) def _attrs_by_sku(session: Session, product_ids: Iterable[int]) -> Dict[str, Dict[str, str]]: ids = [product_id for product_id in product_ids if product_id] if not ids: return {} rows: Dict[str, Dict[str, str]] = {} for value in session.scalars( select(MasterProductAttributeValue).where(MasterProductAttributeValue.product_id.in_(ids)) ).all(): if value.value is None or value.value == "": continue rows.setdefault(value.sku, {})[normalize_column_key(value.attribute_code)] = str(value.value).strip() return rows def _first_attr(attrs: Dict[str, str], *codes: str) -> Optional[str]: for code in codes: value = attrs.get(normalize_column_key(code)) if _clean(value): return str(value).strip() return None def _suggest_shopify_category(session: Session, product: MasterProduct) -> Optional[Dict[str, Any]]: from shopify.taxonomy_sync import suggest_shopify_product_category return suggest_shopify_product_category( session, category_l1=product.category_l1, category_l2=product.category_l2, category_l3=product.category_l3, collection=product.collection, product_name=product.name, ) def _existing_magento_categories(session: Session, connection_id: Optional[int]) -> Dict[str, int]: if connection_id is None: return {} rows = session.scalars( select(MagentoCategoryRegistry).where(MagentoCategoryRegistry.connection_id == connection_id) ).all() out: Dict[str, int] = {} for row in rows: path = _clean(row.path_names) if path: out[path] = row.category_id simplified = _strip_root_segments(path) if simplified: out.setdefault(simplified, row.category_id) return out def _existing_shopify_collections(session: Session, connection_id: Optional[int]) -> Dict[str, str]: stmt = select(ShopifyCollectionRegistry) if connection_id is not None: stmt = stmt.where(ShopifyCollectionRegistry.connection_id == connection_id) rows = session.scalars(stmt).all() out: Dict[str, str] = {} for row in rows: if row.handle: out[_norm(row.handle)] = row.collection_id if row.title: out[_norm(row.title)] = row.collection_id return out def _ensure_magento_category_path( session: Session, api: Any, *, connection_id: int, path: str, root_category_id: int = 2, ) -> Optional[int]: repo = SqlAlchemyCategoryRegistryRepository(session) ttl_cutoff = datetime.now(timezone.utc) - timedelta(days=3650) parent_id = root_category_id built_path = "" for index, segment in enumerate([part.strip() for part in path.split("/") if part.strip()], start=1): built_path = f"{built_path}/{segment}" if built_path else segment existing_id = repo.resolve_path(connection_id, built_path, ttl_cutoff) if existing_id: parent_id = int(existing_id) _sync_magento_category_segment_name(api, category_id=parent_id, expected_name=segment) continue search_status, matches = api.search_categories(name=segment, parent_id=parent_id) if search_status == 200 and matches: match = matches[0] category_id = int(match.get("id") or match.get("entity_id")) _sync_magento_category_segment_name(api, category_id=category_id, expected_name=segment) repo.upsert_category( connection_id, { "category_id": category_id, "parent_id": parent_id, "name": segment, "path": str(match.get("path") or ""), "path_names": built_path, "level": index, "is_active": match.get("is_active", True), }, datetime.now(timezone.utc), ) parent_id = category_id continue from db.taxonomy_invent_policy import can_provision_remotes if not can_provision_remotes(): raise ValueError( f"Remote taxonomy provision is frozen; Magento category missing for '{built_path}'. " "Unlock MASTER_TAXONOMY_AUTO_INVENT (or taxonomy_remote_provision_enabled) only when " "intentionally creating remotes from locked master nodes." ) status, body = api.create_category( { "parent_id": parent_id, "name": segment, "is_active": True, "include_in_menu": True, } ) if status not in (200, 201) or not isinstance(body, dict) or not body.get("id"): raise RuntimeError(f"Could not create Magento category '{built_path}': HTTP {status} {body}") category_id = int(body["id"]) repo.upsert_category( connection_id, { "category_id": category_id, "parent_id": parent_id, "name": segment, "path": str(body.get("path") or ""), "path_names": built_path, "level": index, "is_active": True, }, datetime.now(timezone.utc), ) parent_id = category_id return parent_id def _sync_magento_category_segment_name(api: Any, *, category_id: int, expected_name: str) -> None: """Rename an existing Magento category when the display name drifted from the path segment.""" expected = str(expected_name or "").strip() if not expected: return status, body = api.get_category(int(category_id)) if status != 200 or not isinstance(body, dict): return current_name = str(body.get("name") or "").strip() if not current_name or current_name == expected: return parent_id = body.get("parent_id") payload: Dict[str, Any] = { "id": int(category_id), "name": expected, "is_active": body.get("is_active", True), } if parent_id is not None: payload["parent_id"] = int(parent_id) update_status, update_body = api.update_category(int(category_id), payload) if update_status not in (200, 201): raise RuntimeError( f"Could not rename Magento category {category_id} from {current_name!r} to {expected!r}: " f"HTTP {update_status} {update_body}" ) def _strip_root_segments(path: str) -> str: parts = [part.strip() for part in path.split("/") if part.strip()] while parts and parts[0].lower() in {"root catalog", "default category", "default"}: parts.pop(0) return "/".join(parts) def _slugify(value: str) -> str: text = _norm(value) text = re.sub(r"[^a-z0-9]+", "-", text).strip("-") return text[:250] or "collection" def _norm(value: Optional[str]) -> str: return " ".join(str(value or "").strip().lower().replace("_", " ").replace("-", " ").split()) def _clean(value: Any) -> Optional[str]: text = str(value or "").strip() if not text or text.lower() in {"nan", "none", "null"}: return None return text