"""Orchestrate collection landing bootstrap + push to Magento and Shopify.""" from __future__ import annotations from copy import deepcopy import logging from datetime import datetime, timedelta, 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, path_segments from db.channel_listing_path import upsert_listing_path, upsert_listing_path_target from db.collection_channel_export import export_collections_for_channel from db.collection_channel_placement import build_channel_placement, shopify_collection_title from db.collection_landing_pages import get_collection_landing_page, DEFAULT_CATEGORY_L1 from db.collection_landing_schema import ( COLLECTION_LANDING_FIELDS, COLLECTION_LANDING_NON_ASSET_FIELD_KEYS, MAGENTO_ATTRIBUTE_PREFIX, RELATIONSHIP_FIELD_KEYS, SHOPIFY_NAMESPACE, field_label, magento_attribute_code, shopify_metafield_key, ) from db.models import ( ChannelListingPath, ChannelListingPathTarget, ChannelTaxonomyMapping, MagentoCategoryRegistry, MagentoAttributeRegistry, CollectionLandingPage, MasterTaxonomyChannelLink, MasterTaxonomyNode, ShopifyAttributeRegistry, ShopifyCollectionRegistry, ) TARGET_SOURCE_LISTING_PATH = "listing_path_target" TARGET_SOURCE_CHANNEL_TAXONOMY = "channel_taxonomy" TARGET_SOURCE_TAXONOMY_MAPPING = "taxonomy_mapping" TARGET_SOURCE_REGISTRY = "registry" TARGET_SOURCE_PROVISIONED = "provisioned" logger = logging.getLogger(__name__) def _localize_collection_row_skus( session: Session, row: Dict[str, Any], *, channel_code: str, connection_id: Optional[int], ) -> Dict[str, Any]: from db.channel_sku_mapping import mapping_by_master localized = deepcopy(row) sample = localized.get("sample_product") master_skus: set[str] = set() if isinstance(sample, dict): sample_master = str(sample.get("master_sku") or sample.get("sku") or "").strip() if sample_master: master_skus.add(sample_master) metadata = localized.get("landing_metadata") _collect_master_skus_from_metadata(metadata, master_skus) sku_map = mapping_by_master(session, channel_code, sorted(master_skus), connection_id=connection_id) if master_skus else {} if isinstance(sample, dict): sample_master = str(sample.get("master_sku") or sample.get("sku") or "").strip() if sample_master: channel_sku = sku_map.get(sample_master, sample_master) sample["master_sku"] = sample_master sample["channel_sku"] = channel_sku sample["sku"] = channel_sku _localize_cta_sample_skus(localized, sample_master=sample_master, channel_sku=channel_sku) if isinstance(metadata, dict) and sku_map: localized["landing_metadata"] = _apply_sku_map_to_metadata(metadata, sku_map) return localized def _localize_cta_sample_skus(row: Dict[str, Any], *, sample_master: str, channel_sku: str) -> None: cta_rows = row.get("cta_rows") if not isinstance(cta_rows, list): return for item in cta_rows: if not isinstance(item, dict): continue action = str(item.get("action") or "").strip().lower() sku = str(item.get("sku") or "").strip() if action == "order_sample" or sku == sample_master: item["sku"] = channel_sku def _collect_master_skus_from_metadata(value: Any, master_skus: set[str]) -> None: if isinstance(value, dict): explicit_master = str(value.get("master_sku") or value.get("expected_master_sku") or "").strip() if explicit_master: master_skus.add(explicit_master) elif "sku" in value: sku_value = str(value.get("sku") or "").strip() if _looks_like_master_sku(sku_value): master_skus.add(sku_value) for nested in value.values(): _collect_master_skus_from_metadata(nested, master_skus) return if isinstance(value, list): for nested in value: _collect_master_skus_from_metadata(nested, master_skus) def _apply_sku_map_to_metadata(value: Any, sku_map: Dict[str, str]) -> Any: if isinstance(value, dict): out = {key: _apply_sku_map_to_metadata(nested, sku_map) for key, nested in value.items()} master_sku = str(out.get("master_sku") or out.get("expected_master_sku") or "").strip() if master_sku: out["channel_sku"] = sku_map.get(master_sku, master_sku) elif "sku" in out: sku_value = str(out.get("sku") or "").strip() if sku_value in sku_map: out["channel_sku"] = sku_map[sku_value] return out if isinstance(value, list): return [_apply_sku_map_to_metadata(item, sku_map) for item in value] return value def _looks_like_master_sku(value: str) -> bool: text = str(value or "").strip().upper() if not text or " " in text: return False return "-" in text def bootstrap_collection_landing_channel( session: Session, *, channel_code: str, connection_id: Optional[int] = None, dry_run: bool = False, ) -> Dict[str, Any]: channel = str(channel_code or "").strip().lower() if channel == "shopify": return _bootstrap_shopify(session, connection_id=connection_id, dry_run=dry_run) if channel == "magento": return _bootstrap_magento(session, connection_id=connection_id, dry_run=dry_run) raise ValueError("channel must be magento or shopify") def push_collection_landing_channel( session: Session, *, channel_code: str, connection_id: Optional[int] = None, path_slug: Optional[str] = None, category_l1: Optional[str] = None, only_with_landing_content: bool = True, dry_run: bool = False, provision_missing_remote: bool = False, apply_listing_paths: bool = True, sync_products: bool = True, sync_shopping_icons: bool = True, skip_collection_assets: bool = False, limit: int = 200, ) -> Dict[str, Any]: channel = str(channel_code or "").strip().lower() if channel not in {"magento", "shopify"}: raise ValueError("channel must be magento or shopify") if path_slug: row = get_collection_landing_page(session, path_slug) if not row: raise ValueError(f"Collection not found: {path_slug}") collections = [row] else: payload = export_collections_for_channel( session, channel_code=channel, category_l1=category_l1, connection_id=connection_id, include_inferred=not only_with_landing_content, include_skus=False, limit=limit, ) collections = payload.get("collections") or [] if only_with_landing_content: collections = [row for row in collections if row.get("has_landing_content", True) and not row.get("inferred")] results: List[Dict[str, Any]] = [] errors: List[Dict[str, Any]] = [] pushed = 0 skipped = 0 for row in collections: slug = normalize_plp_path(row.get("path_slug") or "") if not slug: skipped += 1 continue try: logger.info( "Collection landing push start channel=%s connection_id=%s path_slug=%s dry_run=%s " "sync_products=%s sync_shopping_icons=%s apply_listing_paths=%s", channel, connection_id, slug, dry_run, sync_products, sync_shopping_icons, apply_listing_paths, ) if channel == "shopify": item = _push_one_shopify( session, row, connection_id=connection_id, dry_run=dry_run, provision_missing_remote=provision_missing_remote, skip_collection_assets=skip_collection_assets, ) else: item = _push_one_magento( session, row, connection_id=connection_id, dry_run=dry_run, provision_missing_remote=provision_missing_remote, apply_listing_paths=apply_listing_paths, sync_products=sync_products, sync_shopping_icons=sync_shopping_icons, skip_collection_assets=skip_collection_assets, ) results.append(item) if item.get("status") == "pushed": pushed += 1 else: skipped += 1 logger.info( "Collection landing push done channel=%s connection_id=%s path_slug=%s status=%s", channel, connection_id, slug, item.get("status"), ) except Exception as exc: logger.exception( "Collection landing push failed channel=%s connection_id=%s path_slug=%s", channel, connection_id, slug, ) errors.append({"path_slug": slug, "error": str(exc)}) return { "channel_code": channel, "connection_id": connection_id, "dry_run": dry_run, "provision_missing_remote": provision_missing_remote, "apply_listing_paths": apply_listing_paths, "sync_products": sync_products, "sync_shopping_icons": sync_shopping_icons, "skip_collection_assets": skip_collection_assets, "count": len(collections), "pushed": pushed, "skipped": skipped, "errors": errors, "results": results[:50], } def sync_magento_collection_products_channel( session: Session, *, connection_id: Optional[int], path_slug: Optional[str] = None, category_l1: Optional[str] = None, apply_listing_paths: bool = True, dry_run: bool = False, limit: int = 200, on_progress: Optional[Any] = None, master_skus: Optional[List[str]] = None, ) -> Dict[str, Any]: """Assign resolved collection SKUs to linked Magento categories without re-pushing landing fields.""" from magento.collection_landing_push import ( push_magento_shopping_category_icons, sync_magento_collection_products, ) from db.collection_landing_pages import apply_collection_to_skus if path_slug: row = get_collection_landing_page(session, path_slug) if not row: raise ValueError(f"Collection not found: {path_slug}") collections = [row] else: payload = export_collections_for_channel( session, channel_code="magento", category_l1=category_l1, connection_id=connection_id, include_inferred=False, include_skus=False, limit=limit, ) collections = payload.get("collections") or [] results: List[Dict[str, Any]] = [] errors: List[Dict[str, Any]] = [] synced = 0 skipped = 0 for row in collections: slug = normalize_plp_path(row.get("path_slug") or "") if not slug: skipped += 1 continue try: target = resolve_magento_category_target(session, row, connection_id=connection_id) if not target.get("remote_id") or not str(target["remote_id"]).isdigit(): results.append( { "path_slug": slug, "status": "skipped", "reason": SKIP_REASON_MAGENTO_NOT_FOUND, "remote_path": target.get("remote_path"), } ) skipped += 1 continue listing_apply = None if apply_listing_paths and not dry_run: listing_apply = apply_collection_to_skus( session, path_slug=slug, category_l1=row.get("category_l1"), collection=row.get("collection"), dry_run=False, ) category_id = int(target["remote_id"]) channel_row = { **row, **build_taxonomy_relationship_metadata( session, path_slug=slug, channel_code="magento", connection_id=connection_id, ), } api, _ = _build_magento_api(session, connection_id) if api is None and not dry_run: raise ValueError("Magento connection not found") product_sync = sync_magento_collection_products( session, api, category_id=category_id, collection_row=channel_row, connection_id=connection_id, dry_run=dry_run, on_progress=on_progress, master_skus=master_skus, ) shopping_icons = push_magento_shopping_category_icons( api, session, collection_row=channel_row, connection_id=connection_id, dry_run=dry_run, ) results.append( { "path_slug": slug, "status": "preview" if dry_run else "synced", "category_id": category_id, "listing_apply": listing_apply, "product_sync": product_sync, "shopping_icons": shopping_icons, } ) synced += 1 except Exception as exc: errors.append({"path_slug": slug, "error": str(exc), "aborted": True}) return { "channel_code": "magento", "connection_id": connection_id, "dry_run": dry_run, "apply_listing_paths": apply_listing_paths, "count": len(collections), "synced": synced, "skipped": skipped, "errors": errors, "results": results[:50], } def resolve_shopify_collection_gid( session: Session, collection_row: Dict[str, Any], *, connection_id: Optional[int], ) -> Optional[str]: target = resolve_shopify_collection_target(session, collection_row, connection_id=connection_id) remote_id = target.get("remote_id") return str(remote_id) if remote_id else None def resolve_shopify_collection_target( session: Session, collection_row: Dict[str, Any], *, connection_id: Optional[int], ) -> Dict[str, Any]: slug = normalize_plp_path(collection_row.get("path_slug") or "") placement = _placement_for_row(collection_row, channel_code="shopify", path_slug=slug) title = placement.get("remote_title") or shopify_collection_title( collection_row.get("collection"), slug, ) handle = placement.get("remote_handle") remote = _listing_path_target(session, slug, channel_code="shopify", connection_id=connection_id) if remote: return { "remote_id": str(remote), "source": TARGET_SOURCE_LISTING_PATH, "remote_path": handle, } taxonomy = collection_row.get("channel", {}).get("taxonomy") if isinstance(collection_row.get("channel"), dict) else None if isinstance(taxonomy, dict) and taxonomy.get("channel_remote_id"): return { "remote_id": str(taxonomy["channel_remote_id"]), "source": TARGET_SOURCE_CHANNEL_TAXONOMY, "remote_path": taxonomy.get("channel_remote_path") or handle, } mapping = session.scalar( select(ChannelTaxonomyMapping.channel_remote_id) .where(ChannelTaxonomyMapping.channel_code == "shopify") .where(ChannelTaxonomyMapping.master_path_key.in_([slug, title])) .where(ChannelTaxonomyMapping.is_active.is_(True)) .limit(1) ) if mapping: return { "remote_id": str(mapping), "source": TARGET_SOURCE_TAXONOMY_MAPPING, "remote_path": handle, } native_id = _native_shopify_id(connection_id) stmt = select(ShopifyCollectionRegistry.collection_id) if native_id is not None: stmt = stmt.where(ShopifyCollectionRegistry.connection_id == native_id) if handle: row = session.scalar(stmt.where(ShopifyCollectionRegistry.handle == handle).limit(1)) if row: return { "remote_id": str(row), "source": TARGET_SOURCE_REGISTRY, "remote_path": handle, } if title: row = session.scalar(stmt.where(ShopifyCollectionRegistry.title.ilike(title)).limit(1)) if row: return { "remote_id": str(row), "source": TARGET_SOURCE_REGISTRY, "remote_path": handle, } return {"remote_id": None, "source": None, "remote_path": handle} def resolve_magento_category_id( session: Session, collection_row: Dict[str, Any], *, connection_id: Optional[int], ) -> Optional[int]: target = resolve_magento_category_target(session, collection_row, connection_id=connection_id) remote_id = target.get("remote_id") if remote_id is None: return None return int(remote_id) if str(remote_id).isdigit() else None def _authoritative_magento_collection_row( session: Session, collection_row: Dict[str, Any], ) -> Dict[str, Any]: """Resolve taxonomy aliases and prefer collection landing page fields for Magento placement.""" from db.master_taxonomy_merge import resolve_canonical_path_slug, resolve_canonical_node slug = normalize_plp_path(collection_row.get("path_slug") or "") canonical = resolve_canonical_path_slug(session, slug) or slug out = dict(collection_row) out["path_slug"] = canonical landing_row = session.scalar( select(CollectionLandingPage).where(CollectionLandingPage.path_slug == canonical).limit(1) ) if landing_row is not None: out["collection"] = landing_row.collection out["title"] = landing_row.title or landing_row.collection out["category_l1"] = landing_row.category_l1 if landing_row.parent_path_slug: out["parent_path_slug"] = landing_row.parent_path_slug return out node = resolve_canonical_node(session, path_slug=canonical) if node is not None: out.setdefault("collection", node.name) out.setdefault("title", node.name) return out def _can_provision_magento_category_path(session: Session, path_slug: str) -> bool: """Only provision Magento categories backed by a landing page or active taxonomy node.""" from db.master_taxonomy_hierarchy import NODE_CATEGORY, NODE_COLLECTION, NODE_HUB from db.master_taxonomy_merge import resolve_canonical_node, resolve_canonical_path_slug canonical = resolve_canonical_path_slug(session, path_slug) or normalize_plp_path(path_slug) if not canonical: return False landing_row = session.scalar( select(CollectionLandingPage).where(CollectionLandingPage.path_slug == canonical).limit(1) ) if landing_row is not None: return True node = resolve_canonical_node(session, path_slug=canonical) return node is not None and node.node_kind in {NODE_HUB, NODE_CATEGORY, NODE_COLLECTION} def resolve_magento_category_target( session: Session, collection_row: Dict[str, Any], *, connection_id: Optional[int], ) -> Dict[str, Any]: row = _authoritative_magento_collection_row(session, collection_row) slug = normalize_plp_path(row.get("path_slug") or "") placement = _placement_for_row(row, channel_code="magento", path_slug=slug) breadcrumb = placement.get("breadcrumb") or [] path_names = "/".join(str(part) for part in breadcrumb if part) if connection_id is None: return {"remote_id": None, "source": None, "remote_path": path_names or None} remote = _listing_path_target(session, slug, channel_code="magento", connection_id=connection_id) if remote and str(remote).isdigit(): return { "remote_id": str(remote), "source": TARGET_SOURCE_LISTING_PATH, "remote_path": path_names or None, } taxonomy = row.get("channel", {}).get("taxonomy") if isinstance(row.get("channel"), dict) else None if isinstance(taxonomy, dict) and taxonomy.get("channel_remote_id"): if str(taxonomy["channel_remote_id"]).isdigit(): return { "remote_id": str(taxonomy["channel_remote_id"]), "source": TARGET_SOURCE_CHANNEL_TAXONOMY, "remote_path": taxonomy.get("channel_remote_path") or path_names or None, } _, native_id = _decode_magento_connection_id(connection_id) if native_id is None: return {"remote_id": None, "source": None, "remote_path": path_names or None} ttl = datetime.now(timezone.utc) - timedelta(days=3650) registry_row = session.scalar( select(MagentoCategoryRegistry.category_id) .where(MagentoCategoryRegistry.connection_id == native_id) .where(MagentoCategoryRegistry.path_names == path_names) .where(MagentoCategoryRegistry.fetched_at >= ttl) .limit(1) ) if registry_row: return { "remote_id": str(int(registry_row)), "source": TARGET_SOURCE_REGISTRY, "remote_path": path_names or None, } collection_name = str(row.get("collection") or row.get("title") or "").strip() if collection_name: registry_row = session.scalar( select(MagentoCategoryRegistry.category_id) .where(MagentoCategoryRegistry.connection_id == native_id) .where(MagentoCategoryRegistry.name.ilike(collection_name)) .where(MagentoCategoryRegistry.fetched_at >= ttl) .limit(1) ) if registry_row: return { "remote_id": str(int(registry_row)), "source": TARGET_SOURCE_REGISTRY, "remote_path": path_names or None, } return {"remote_id": None, "source": None, "remote_path": path_names or None} def provision_shopify_collection_target( session: Session, collection_row: Dict[str, Any], *, connection_id: Optional[int], ) -> Dict[str, Any]: from shopify.collection_sync import create_shopify_collection from shopify.connections import build_client slug = normalize_plp_path(collection_row.get("path_slug") or "") placement = _placement_for_row(collection_row, channel_code="shopify", path_slug=slug) title = str(placement.get("remote_title") or "").strip() handle = placement.get("remote_handle") if not title: raise ValueError("Cannot provision Shopify collection without a title") _ensure_listing_path_for_collection(session, collection_row, placement=placement, path_slug=slug) connection = _get_shopify_connection(session, connection_id) if not connection: raise ValueError("No active Shopify connection") client = build_client(connection) native_id = _native_shopify_id(connection_id) created = create_shopify_collection( session, client, shop_code=connection.shop_code, connection_id=native_id, title=title, handle=handle, description_html=collection_row.get("short_description"), ) remote_id = str(created["collection_id"]) upsert_listing_path_target( session, path_slug=slug, channel_code="shopify", remote_id=remote_id, taxonomy_kind="collection", connection_id=connection_id, remote_path=created.get("handle") or handle, ) return { "remote_id": remote_id, "source": TARGET_SOURCE_PROVISIONED, "remote_path": created.get("handle") or handle, "provisioned": True, } def provision_magento_category_target( session: Session, collection_row: Dict[str, Any], *, connection_id: Optional[int], ) -> Dict[str, Any]: from db.channel_catalog_provision import _ensure_magento_category_path row = _authoritative_magento_collection_row(session, collection_row) slug = normalize_plp_path(row.get("path_slug") or "") if not _can_provision_magento_category_path(session, slug): raise ValueError(f"Cannot provision Magento category for unmapped path slug: {slug}") placement = _placement_for_row(row, channel_code="magento", path_slug=slug) breadcrumb = placement.get("breadcrumb") or [] path_names = "/".join(str(part) for part in breadcrumb if part) if not path_names: raise ValueError("Cannot provision Magento category without a breadcrumb path") api, native_id = _build_magento_api(session, connection_id) if api is None or native_id is None: raise ValueError("Magento connection not found") category_id = _ensure_magento_category_path( session, api, connection_id=native_id, path=path_names, ) if not category_id: raise ValueError(f"Could not provision Magento category for {path_names}") _ensure_listing_path_for_collection(session, row, placement=placement, path_slug=slug) 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_names, ) return { "remote_id": str(int(category_id)), "source": TARGET_SOURCE_PROVISIONED, "remote_path": path_names, "provisioned": True, } def ensure_magento_hub_shopping_categories( session: Session, *, connection_id: Optional[int], hub_path_slug: str = "kitchen-cabinets", dry_run: bool = False, ) -> Dict[str, Any]: """Ensure hub shopping facet categories exist and carry PieHS taxonomy metadata.""" from db.channel_catalog_provision import _ensure_magento_category_path from db.listing_intersections import shopping_facet_options_for_hub from magento.collection_landing_push import push_magento_collection_landing hub = normalize_plp_path(hub_path_slug) if not hub: raise ValueError("hub_path_slug is required") options = shopping_facet_options_for_hub(session, hub) title = _title_from_path_slug(hub) planned = [ { "path_slug": hub, "remote_path": title, "title": title, "path_kind": "hub", "is_collection": False, "has_collections": True, } ] for option in options: facet_slug = normalize_plp_path(str(option.get("slug") or "")) if not facet_slug: continue label = str(option.get("label") or option.get("title") or _title_from_path_slug(facet_slug)).strip() path_slug = normalize_plp_path(f"{hub}/{facet_slug}") planned.append( { "path_slug": path_slug, "remote_path": f"{title}/{label}", "title": label, "path_kind": "facet", "parent_path_slug": hub, "hub_path_slug": hub, "is_collection": False, "has_collections": False, } ) if dry_run: return { "dry_run": True, "connection_id": connection_id, "hub_path_slug": hub, "planned": planned, "created_or_updated": 0, "errors": [], } if connection_id is None: raise ValueError("connection_id is required to ensure Magento shopping categories") api, native_id = _build_magento_api(session, connection_id) if api is None or native_id is None: raise ValueError("Magento connection not found") field_keys = { "page_type", "is_collection", "has_collections", "taxonomy_path_slug", "parent_path_slug", } created_or_updated = 0 errors: List[Dict[str, Any]] = [] results: List[Dict[str, Any]] = [] for item in planned: try: category_id = _ensure_magento_category_path( session, api, connection_id=native_id, path=str(item["remote_path"]), ) if not category_id: raise RuntimeError(f"Could not resolve Magento category for {item['remote_path']}") upsert_listing_path( session, path_slug=str(item["path_slug"]), title=str(item["title"]), path_kind=str(item["path_kind"]), parent_path_slug=item.get("parent_path_slug"), hub_path_slug=item.get("hub_path_slug") or (hub if item["path_slug"] != hub else None), notes="Magento shopping category repair", ) upsert_listing_path_target( session, path_slug=str(item["path_slug"]), channel_code="magento", remote_id=str(int(category_id)), taxonomy_kind="category", connection_id=connection_id, remote_path=str(item["remote_path"]), ) _upsert_taxonomy_channel_link_for_path( session, path_slug=str(item["path_slug"]), channel_code="magento", connection_id=native_id, remote_id=str(int(category_id)), remote_path=str(item["remote_path"]), taxonomy_kind="category", ) push_magento_collection_landing( api, category_id=int(category_id), collection_row={ "page_type": item["path_kind"], "title": item["title"], "is_collection": item["is_collection"], "has_collections": item["has_collections"], "taxonomy_path_slug": item["path_slug"], "parent_path_slug": item.get("parent_path_slug"), }, field_keys=field_keys, dry_run=False, ) created_or_updated += 1 results.append( { "path_slug": item["path_slug"], "category_id": int(category_id), "remote_path": item["remote_path"], "status": "updated", } ) except Exception as exc: errors.append({"path_slug": item.get("path_slug"), "error": str(exc)}) return { "dry_run": False, "connection_id": connection_id, "hub_path_slug": hub, "planned": planned, "created_or_updated": created_or_updated, "errors": errors, "results": results, } def _upsert_taxonomy_channel_link_for_path( session: Session, *, path_slug: str, channel_code: str, connection_id: Optional[int], remote_id: str, remote_path: Optional[str], taxonomy_kind: str, ) -> None: node = session.scalar( select(MasterTaxonomyNode) .where( MasterTaxonomyNode.path_slug == normalize_plp_path(path_slug), MasterTaxonomyNode.is_active.is_(True), ) .limit(1) ) if node is None: return row = session.scalar( select(MasterTaxonomyChannelLink) .where( MasterTaxonomyChannelLink.taxonomy_node_id == node.id, MasterTaxonomyChannelLink.channel_code == channel_code, MasterTaxonomyChannelLink.connection_id == connection_id, MasterTaxonomyChannelLink.taxonomy_kind == taxonomy_kind, ) .limit(1) ) if row is None: row = MasterTaxonomyChannelLink( taxonomy_node_id=node.id, channel_code=channel_code, connection_id=connection_id, taxonomy_kind=taxonomy_kind, remote_id=remote_id, remote_path=remote_path, is_active=True, ) session.add(row) else: row.remote_id = remote_id row.remote_path = remote_path row.is_active = True session.flush() def _title_from_path_slug(path_slug: str) -> str: return " / ".join(part.replace("-", " ").title() for part in path_segments(path_slug)) SKIP_REASON_SHOPIFY_NOT_FOUND = "shopify_collection_not_found" SKIP_REASON_MAGENTO_NOT_FOUND = "magento_category_not_found" SKIP_REASON_CONNECTION_REQUIRED = "connection_required" SKIP_REASON_NO_ACTIVE_CONNECTION = "no_active_connection" READINESS_LINKED = "linked" READINESS_PROVISIONABLE = "provisionable" READINESS_BLOCKED = "blocked" def describe_collection_channel_target( session: Session, collection_row: Dict[str, Any], *, channel_code: str, connection_id: Optional[int] = None, ) -> Dict[str, Any]: """Resolved remote target + operator-facing readiness for one collection/channel.""" channel = str(channel_code or "").strip().lower() if channel not in {"magento", "shopify"}: raise ValueError("channel must be magento or shopify") slug = normalize_plp_path(collection_row.get("path_slug") or "") placement = _placement_for_row(collection_row, channel_code=channel, path_slug=slug) if channel == "shopify": target = resolve_shopify_collection_target(session, collection_row, connection_id=connection_id) missing_reason = SKIP_REASON_SHOPIFY_NOT_FOUND connection = _get_shopify_connection(session, connection_id) connection_available = connection is not None requires_connection_id = False else: target = resolve_magento_category_target(session, collection_row, connection_id=connection_id) missing_reason = SKIP_REASON_MAGENTO_NOT_FOUND api, native_id = _build_magento_api(session, connection_id) connection_available = api is not None and native_id is not None requires_connection_id = True linked = bool(target.get("remote_id")) skip_reason = None readiness = READINESS_LINKED if not linked: if requires_connection_id and connection_id is None: skip_reason = SKIP_REASON_CONNECTION_REQUIRED readiness = READINESS_BLOCKED elif not connection_available: skip_reason = SKIP_REASON_NO_ACTIVE_CONNECTION readiness = READINESS_BLOCKED else: skip_reason = missing_reason readiness = READINESS_PROVISIONABLE return { "channel_code": channel, "connection_id": connection_id, "path_slug": slug, "remote_id": target.get("remote_id"), "target_source": target.get("source"), "remote_path": target.get("remote_path"), "linked": linked, "can_provision": readiness == READINESS_PROVISIONABLE, "readiness": readiness, "skip_reason": skip_reason, "placement": { "remote_title": placement.get("remote_title"), "remote_handle": placement.get("remote_handle"), "breadcrumb": placement.get("breadcrumb"), "remote_taxonomy_kind": placement.get("remote_taxonomy_kind"), }, } def describe_collection_channel_targets( session: Session, collection_row: Dict[str, Any], *, channel_code: Optional[str] = None, connection_id: Optional[int] = None, ) -> Dict[str, Any]: channels = ( [str(channel_code).strip().lower()] if channel_code else ["magento", "shopify"] ) targets = [ describe_collection_channel_target( session, collection_row, channel_code=channel, connection_id=connection_id, ) for channel in channels ] from db.collection_taxonomy_bridge import describe_registry_consistency registry = describe_registry_consistency(session, path_slug=normalize_plp_path(collection_row.get("path_slug") or "")) return { "path_slug": normalize_plp_path(collection_row.get("path_slug") or ""), "connection_id": connection_id, "targets": targets, "registry": registry, "field_contract": describe_collection_field_contract( session, channel_code=channels[0] if len(channels) == 1 else None, connection_id=connection_id, ), "relationship_metadata": { channel: build_taxonomy_relationship_metadata( session, path_slug=normalize_plp_path(collection_row.get("path_slug") or ""), channel_code=channel, connection_id=connection_id, ) for channel in channels }, } def describe_collection_field_contract( session: Session, *, channel_code: Optional[str], connection_id: Optional[int], ) -> Dict[str, Any]: """Template-facing Magento attribute / Shopify metafield identifiers.""" channels = [channel_code] if channel_code else ["magento", "shopify"] out: Dict[str, Any] = {} if "magento" in channels: _, native_id = _decode_magento_connection_id(connection_id) rows = [] if native_id is not None: rows = list( session.scalars( select(MagentoAttributeRegistry).where( MagentoAttributeRegistry.connection_id == native_id, MagentoAttributeRegistry.attribute_code.in_( [magento_attribute_code(field["key"]) for field in COLLECTION_LANDING_FIELDS] ), ) ).all() ) registry = {row.attribute_code: row for row in rows} out["magento"] = { "attribute_prefix": MAGENTO_ATTRIBUTE_PREFIX, "fields": [ { "name": field_label(field["key"]), "canonical_key": field["key"], "attribute_code": magento_attribute_code(field["key"]), "attribute_id": getattr(registry.get(magento_attribute_code(field["key"])), "attribute_id", None), "frontend_input": field["magento_input"], "created": magento_attribute_code(field["key"]) in registry, } for field in COLLECTION_LANDING_FIELDS ], } if "shopify" in channels: connection = _get_shopify_connection(session, connection_id) rows = [] if connection is not None: rows = list( session.scalars( select(ShopifyAttributeRegistry).where( ShopifyAttributeRegistry.shop_code == connection.shop_code, ShopifyAttributeRegistry.owner_type == "collection_metafield", ) ).all() ) registry = {str(row.key or ""): row for row in rows} def definition_gid(row: Optional[ShopifyAttributeRegistry]) -> Optional[str]: raw = row.raw_json if row and isinstance(row.raw_json, dict) else {} return raw.get("definition_id") or raw.get("id") or raw.get("admin_graphql_api_id") out["shopify"] = { "namespace": SHOPIFY_NAMESPACE, "owner_type": "COLLECTION", "fields": [ { "name": field_label(field["key"]), "canonical_key": field["key"], "namespace": SHOPIFY_NAMESPACE, "key": shopify_metafield_key(field["key"]), "type": field["shopify_type"], "definition_gid": definition_gid(registry.get(shopify_metafield_key(field["key"]))), "created": shopify_metafield_key(field["key"]) in registry, } for field in COLLECTION_LANDING_FIELDS ], } return out def build_taxonomy_relationship_metadata( session: Session, *, path_slug: str, channel_code: str, connection_id: Optional[int], ) -> Dict[str, Any]: """Flatten master hierarchy into channel-neutral relationship metafields.""" slug = normalize_plp_path(path_slug) node = session.scalar( select(MasterTaxonomyNode) .where(MasterTaxonomyNode.path_slug == slug, MasterTaxonomyNode.is_active.is_(True)) .limit(1) ) if node is None: return { "is_collection": True, "has_collections": False, "taxonomy_path_slug": slug, "parent_path_slug": slug.rsplit("/", 1)[0] if "/" in slug else None, "parent_remote_id": None, "child_remote_ids": [], "sibling_remote_ids": [], "child_taxonomy_nodes": [], "sibling_taxonomy_nodes": [], } nodes = list( session.scalars(select(MasterTaxonomyNode).where(MasterTaxonomyNode.is_active.is_(True))).all() ) by_id = {item.id: item for item in nodes} children = [item for item in nodes if item.parent_id == node.id] siblings = [item for item in nodes if item.parent_id == node.parent_id and item.id != node.id] links = _taxonomy_links_by_node( session, node_ids=[item.id for item in nodes], channel_code=channel_code, connection_id=connection_id, ) landing_by_slug = _collection_landing_by_slug( session, [item.path_slug for item in children + siblings], ) def item_payload(item: MasterTaxonomyNode) -> Dict[str, Any]: link = links.get(item.id) landing = landing_by_slug.get(item.path_slug) payload: Dict[str, Any] = { "name": item.name, "path_slug": item.path_slug, "node_kind": item.node_kind, "remote_id": link.remote_id if link else None, } if landing and landing.category_icon_url: payload["category_icon_url"] = landing.category_icon_url return payload parent = by_id.get(node.parent_id) if node.parent_id else None parent_link = links.get(parent.id) if parent else None child_payloads = [item_payload(item) for item in children] sibling_payloads = [item_payload(item) for item in siblings] return { "is_collection": node.node_kind == "collection", "has_collections": any(item.node_kind == "collection" for item in children), "taxonomy_path_slug": node.path_slug, "parent_path_slug": parent.path_slug if parent else None, "parent_remote_id": parent_link.remote_id if parent_link else None, "child_remote_ids": [item["remote_id"] for item in child_payloads if item.get("remote_id")], "sibling_remote_ids": [item["remote_id"] for item in sibling_payloads if item.get("remote_id")], "child_taxonomy_nodes": child_payloads, "sibling_taxonomy_nodes": sibling_payloads, } def _collection_landing_by_slug(session: Session, path_slugs: List[str]) -> Dict[str, CollectionLandingPage]: slugs = [normalize_plp_path(slug) for slug in path_slugs if str(slug or "").strip()] if not slugs: return {} rows = session.scalars( select(CollectionLandingPage).where(CollectionLandingPage.path_slug.in_(slugs)) ).all() return {row.path_slug: row for row in rows} def _taxonomy_links_by_node( session: Session, *, node_ids: List[int], channel_code: str, connection_id: Optional[int], ) -> Dict[int, MasterTaxonomyChannelLink]: if not node_ids: return {} stmt = select(MasterTaxonomyChannelLink).where( MasterTaxonomyChannelLink.taxonomy_node_id.in_(node_ids), MasterTaxonomyChannelLink.channel_code == channel_code, MasterTaxonomyChannelLink.is_active.is_(True), ) links = list(session.scalars(stmt).all()) connection_candidates = {connection_id} if channel_code == "shopify": connection_candidates.add(_native_shopify_id(connection_id)) else: connection_candidates.add(_decode_magento_connection_id(connection_id)[1]) out: Dict[int, MasterTaxonomyChannelLink] = {} for link in links: if connection_id is not None and link.connection_id not in connection_candidates and link.connection_id is not None: continue current = out.get(link.taxonomy_node_id) if current is None or (current.connection_id is None and link.connection_id is not None): out[link.taxonomy_node_id] = link return out def _bootstrap_shopify(session: Session, *, connection_id: Optional[int], dry_run: bool) -> Dict[str, Any]: from shopify.collection_landing_push import bootstrap_shopify_collection_metafields from shopify.connections import build_client connection = _get_shopify_connection(session, connection_id) if not connection and not dry_run: raise ValueError("No active Shopify connection") shop_code = connection.shop_code if connection else "default" client = build_client(connection) if connection and not dry_run else None if dry_run or client is None: return bootstrap_shopify_collection_metafields( session, client, shop_code=shop_code, dry_run=True, ) return bootstrap_shopify_collection_metafields(session, client, shop_code=shop_code, dry_run=False) def _bootstrap_magento(session: Session, *, connection_id: Optional[int], dry_run: bool) -> Dict[str, Any]: from magento.collection_landing_push import bootstrap_magento_collection_attributes api, native_id = _build_magento_api(session, connection_id) if api is None and not dry_run: raise ValueError("Magento connection not found") if dry_run or api is None: return bootstrap_magento_collection_attributes(session, api, connection_id=native_id or 0, dry_run=True) return bootstrap_magento_collection_attributes(session, api, connection_id=native_id, dry_run=False) def _push_one_shopify( session: Session, row: Dict[str, Any], *, connection_id: Optional[int], dry_run: bool, provision_missing_remote: bool, skip_collection_assets: bool = False, ) -> Dict[str, Any]: from shopify.collection_landing_push import push_shopify_collection_landing from shopify.connections import build_client slug = row.get("path_slug") target, skip = _resolve_or_provision_target( session, row, channel_code="shopify", connection_id=connection_id, dry_run=dry_run, provision_missing_remote=provision_missing_remote, provision_fn=provision_shopify_collection_target, missing_reason="shopify_collection_not_found", ) if skip: return skip gid = str(target["remote_id"]) _refresh_shopping_intersections( session, row, connection_id=connection_id, shopify_collection_id=gid, shopify_handle=str(target.get("remote_path") or ""), ) channel_row = { **row, **build_taxonomy_relationship_metadata( session, path_slug=str(slug or ""), channel_code="shopify", connection_id=connection_id, ), } channel_row = _localize_collection_row_skus( session, channel_row, channel_code="shopify", connection_id=connection_id, ) field_keys = COLLECTION_LANDING_NON_ASSET_FIELD_KEYS if skip_collection_assets else None if dry_run: return push_shopify_collection_landing( client=None, collection_gid=gid, collection_row=channel_row, field_keys=field_keys, dry_run=True, ) | _target_result_fields(target, status="preview", path_slug=slug) connection = _get_shopify_connection(session, connection_id) if not connection: raise ValueError("No active Shopify connection") client = build_client(connection) result = push_shopify_collection_landing( client, collection_gid=gid, collection_row=channel_row, field_keys=field_keys, dry_run=False, ) relationship_sync = _push_related_taxonomy_metadata( session, source_path_slug=str(slug or ""), channel_code="shopify", connection_id=connection_id, client_or_api=client, ) return result | {"relationship_sync": relationship_sync} | _target_result_fields( target, status="pushed", path_slug=slug ) def _push_one_magento( session: Session, row: Dict[str, Any], *, connection_id: Optional[int], dry_run: bool, provision_missing_remote: bool, apply_listing_paths: bool = True, sync_products: bool = True, sync_shopping_icons: bool = True, skip_collection_assets: bool = False, ) -> Dict[str, Any]: from magento.collection_landing_push import ( push_magento_collection_landing, push_magento_shopping_category_icons, sync_magento_collection_products, ) from db.collection_landing_pages import apply_collection_to_skus slug = row.get("path_slug") target, skip = _resolve_or_provision_target( session, row, channel_code="magento", connection_id=connection_id, dry_run=dry_run, provision_missing_remote=provision_missing_remote, provision_fn=provision_magento_category_target, missing_reason="magento_category_not_found", ) if skip: return skip category_id = int(target["remote_id"]) logger.info( "Collection landing Magento stage=refresh_intersections connection_id=%s path_slug=%s category_id=%s", connection_id, slug, category_id, ) _refresh_shopping_intersections( session, row, connection_id=connection_id, magento_category_id=category_id, ) logger.info( "Collection landing Magento stage=build_relationship_metadata connection_id=%s path_slug=%s category_id=%s", connection_id, slug, category_id, ) channel_row = { **row, **build_taxonomy_relationship_metadata( session, path_slug=str(slug or ""), channel_code="magento", connection_id=connection_id, ), } channel_row = _localize_collection_row_skus( session, channel_row, channel_code="magento", connection_id=connection_id, ) field_keys = COLLECTION_LANDING_NON_ASSET_FIELD_KEYS if skip_collection_assets else None api, _ = _build_magento_api(session, connection_id) if api is None and not dry_run: raise ValueError("Magento connection not found") listing_apply = None if apply_listing_paths and not dry_run: logger.info( "Collection landing Magento stage=apply_listing_paths connection_id=%s path_slug=%s category_id=%s", connection_id, slug, category_id, ) listing_apply = apply_collection_to_skus( session, path_slug=str(slug or ""), category_l1=row.get("category_l1"), collection=row.get("collection"), dry_run=False, ) if dry_run: logger.info( "Collection landing Magento stage=dry_run_preview connection_id=%s path_slug=%s category_id=%s", connection_id, slug, category_id, ) preview = push_magento_collection_landing( api, category_id=category_id, collection_row=channel_row, field_keys=field_keys, dry_run=True, ) product_sync = ( sync_magento_collection_products( session, api, category_id=category_id, collection_row=channel_row, connection_id=connection_id, dry_run=True, ) if sync_products else {"dry_run": True, "skipped": True} ) shopping_icons = ( push_magento_shopping_category_icons( api, session, collection_row=channel_row, connection_id=connection_id, dry_run=True, ) if sync_shopping_icons and not skip_collection_assets else {"dry_run": True, "skipped": True} ) return preview | { "listing_apply": listing_apply, "product_sync": product_sync, "shopping_icons": shopping_icons, } | _target_result_fields( target, status="preview", path_slug=slug ) result = push_magento_collection_landing( api, category_id=category_id, collection_row=channel_row, field_keys=field_keys, dry_run=False, ) logger.info( "Collection landing Magento stage=push_category_metadata_done connection_id=%s path_slug=%s category_id=%s", connection_id, slug, category_id, ) product_sync = ( sync_magento_collection_products( session, api, category_id=category_id, collection_row=channel_row, connection_id=connection_id, dry_run=False, ) if sync_products else {"dry_run": False, "skipped": True} ) logger.info( "Collection landing Magento stage=product_sync_done connection_id=%s path_slug=%s category_id=%s skipped=%s", connection_id, slug, category_id, bool(product_sync.get("skipped")), ) shopping_icons = ( push_magento_shopping_category_icons( api, session, collection_row=channel_row, connection_id=connection_id, dry_run=False, ) if sync_shopping_icons and not skip_collection_assets else {"dry_run": False, "skipped": True} ) logger.info( "Collection landing Magento stage=shopping_icons_done connection_id=%s path_slug=%s category_id=%s skipped=%s", connection_id, slug, category_id, bool(shopping_icons.get("skipped")), ) relationship_sync = _push_related_taxonomy_metadata( session, source_path_slug=str(slug or ""), channel_code="magento", connection_id=connection_id, client_or_api=api, ) logger.info( "Collection landing Magento stage=relationship_sync_done connection_id=%s path_slug=%s category_id=%s updated=%s errors=%s", connection_id, slug, category_id, relationship_sync.get("updated"), len(relationship_sync.get("errors") or []), ) return result | { "listing_apply": listing_apply, "product_sync": product_sync, "shopping_icons": shopping_icons, "relationship_sync": relationship_sync, } | _target_result_fields( target, status="pushed", path_slug=slug ) def _push_related_taxonomy_metadata( session: Session, *, source_path_slug: str, channel_code: str, connection_id: Optional[int], client_or_api: Any, ) -> Dict[str, Any]: """Refresh parent and sibling relationship fields when one collection is pushed.""" source = session.scalar( select(MasterTaxonomyNode) .where( MasterTaxonomyNode.path_slug == normalize_plp_path(source_path_slug), MasterTaxonomyNode.is_active.is_(True), ) .limit(1) ) if source is None: return {"updated": 0, "skipped": 0, "errors": []} stmt = select(MasterTaxonomyNode).where(MasterTaxonomyNode.is_active.is_(True)) if source.parent_id is None: stmt = stmt.where(MasterTaxonomyNode.parent_id == source.id) else: stmt = stmt.where( (MasterTaxonomyNode.id == source.parent_id) | (MasterTaxonomyNode.parent_id == source.parent_id) ) related = [node for node in session.scalars(stmt).all() if node.id != source.id] links = _taxonomy_links_by_node( session, node_ids=[node.id for node in related], channel_code=channel_code, connection_id=connection_id, ) updated = 0 skipped = 0 errors: List[Dict[str, str]] = [] for node in related: link = links.get(node.id) if link is None: skipped += 1 continue metadata = build_taxonomy_relationship_metadata( session, path_slug=node.path_slug, channel_code=channel_code, connection_id=connection_id, ) try: if channel_code == "shopify": from shopify.collection_landing_push import push_shopify_collection_landing push_shopify_collection_landing( client_or_api, collection_gid=str(link.remote_id), collection_row=metadata, field_keys=RELATIONSHIP_FIELD_KEYS, dry_run=False, ) elif str(link.remote_id).isdigit(): from magento.collection_landing_push import push_magento_collection_landing target = ensure_verified_magento_category_target( session, { "path_slug": node.path_slug, "category_l1": metadata.get("category_l1") or DEFAULT_CATEGORY_L1, "collection": node.name, "title": node.name, }, connection_id=connection_id, api=client_or_api, provision_missing=False, ) category_id = target.get("remote_id") if not category_id or not str(category_id).isdigit(): skipped += 1 continue push_magento_collection_landing( client_or_api, category_id=int(category_id), collection_row=metadata, field_keys=RELATIONSHIP_FIELD_KEYS, dry_run=False, ) else: skipped += 1 continue updated += 1 except Exception as exc: errors.append({"path_slug": node.path_slug, "error": str(exc)}) return {"updated": updated, "skipped": skipped, "errors": errors} def _confirm_magento_category_target( session: Session, target: Dict[str, Any], *, path_slug: str, connection_id: Optional[int], api: Any = None, ) -> Dict[str, Any]: """Drop cached Magento category IDs that no longer exist on the remote store.""" from db.master_taxonomy_merge import resolve_canonical_path_slug remote_id = target.get("remote_id") if not remote_id or not str(remote_id).isdigit(): return target if api is None: api, _ = _build_magento_api(session, connection_id) if api is None: return target status, _ = api.get_category(int(remote_id)) if status == 200: return target if status != 404: return target canonical_slug = resolve_canonical_path_slug(session, path_slug) or normalize_plp_path(path_slug) _invalidate_stale_magento_category_mappings( session, path_slug=canonical_slug, connection_id=connection_id, remote_id=str(remote_id), ) return { "remote_id": None, "source": None, "remote_path": target.get("remote_path"), } def _invalidate_stale_magento_category_mappings( session: Session, *, path_slug: str, connection_id: Optional[int], remote_id: str, ) -> None: slug = normalize_plp_path(path_slug) path = session.scalar( select(ChannelListingPath).where(ChannelListingPath.path_slug == slug).limit(1) ) if path is not None: targets = session.scalars( select(ChannelListingPathTarget) .where(ChannelListingPathTarget.listing_path_id == path.id) .where(ChannelListingPathTarget.channel_code == "magento") .where(ChannelListingPathTarget.remote_id == str(remote_id)) .where(ChannelListingPathTarget.is_active.is_(True)) ).all() for target in targets: if connection_id is not None and target.connection_id not in {connection_id, None}: continue target.is_active = False node = session.scalar( select(MasterTaxonomyNode) .where(MasterTaxonomyNode.path_slug == slug, MasterTaxonomyNode.is_active.is_(True)) .limit(1) ) if node is not None: links = session.scalars( select(MasterTaxonomyChannelLink) .where(MasterTaxonomyChannelLink.taxonomy_node_id == node.id) .where(MasterTaxonomyChannelLink.channel_code == "magento") .where(MasterTaxonomyChannelLink.remote_id == str(remote_id)) .where(MasterTaxonomyChannelLink.is_active.is_(True)) ).all() for link in links: if connection_id is not None and link.connection_id not in {connection_id, None}: continue link.is_active = False session.flush() def ensure_verified_magento_category_target( session: Session, collection_row: Dict[str, Any], *, connection_id: Optional[int], api: Any = None, provision_missing: bool = True, ) -> Dict[str, Any]: """Resolve a Magento category target and verify or repair stale remote IDs.""" row = _authoritative_magento_collection_row(session, collection_row) slug = normalize_plp_path(row.get("path_slug") or "") target = resolve_magento_category_target(session, row, connection_id=connection_id) target = _confirm_magento_category_target( session, target, path_slug=slug, connection_id=connection_id, api=api, ) if target.get("remote_id") or not provision_missing: return target if not _can_provision_magento_category_path(session, slug): return target if api is None: api, native_id = _build_magento_api(session, connection_id) else: _, native_id = _decode_magento_connection_id(connection_id) if api is None or native_id is None: return target placement = _placement_for_row(row, channel_code="magento", path_slug=slug) path_names = "/".join(str(part) for part in (placement.get("breadcrumb") or []) if part) if not path_names: return target from db.channel_catalog_provision import _ensure_magento_category_path category_id = _ensure_magento_category_path( session, api, connection_id=native_id, path=path_names, ) if not category_id: return target _ensure_listing_path_for_collection(session, row, placement=placement, path_slug=slug) 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_names, ) _upsert_taxonomy_channel_link_for_path( session, path_slug=slug, channel_code="magento", connection_id=native_id, remote_id=str(int(category_id)), remote_path=path_names, taxonomy_kind="category", ) return { "remote_id": str(int(category_id)), "source": TARGET_SOURCE_PROVISIONED, "remote_path": path_names, "provisioned": True, } def _resolve_or_provision_target( session: Session, row: Dict[str, Any], *, channel_code: str, connection_id: Optional[int], dry_run: bool, provision_missing_remote: bool, provision_fn: Any, missing_reason: str, ) -> tuple[Dict[str, Any], Optional[Dict[str, Any]]]: channel = str(channel_code or "").strip().lower() if channel == "shopify": target = resolve_shopify_collection_target(session, row, connection_id=connection_id) else: target = resolve_magento_category_target(session, row, connection_id=connection_id) if target.get("remote_id") and not dry_run: authoritative = _authoritative_magento_collection_row(session, row) target = _confirm_magento_category_target( session, target, path_slug=normalize_plp_path(authoritative.get("path_slug") or ""), connection_id=connection_id, ) if target.get("remote_id"): return target, None if not provision_missing_remote: return target, { "status": "skipped", "path_slug": row.get("path_slug"), "reason": missing_reason, "target_source": target.get("source"), "remote_path": target.get("remote_path"), } if dry_run: placement = _placement_for_row( row, channel_code=channel, path_slug=normalize_plp_path(row.get("path_slug") or ""), ) return target, { "status": "preview", "path_slug": row.get("path_slug"), "would_provision": True, "proposed_remote_title": placement.get("remote_title"), "proposed_remote_handle": placement.get("remote_handle"), "proposed_remote_path": target.get("remote_path"), "target_source": None, } if channel == "magento" and connection_id is None: raise ValueError("connection_id is required to provision Magento categories") provisioned = provision_fn(session, row, connection_id=connection_id) return provisioned, None def _target_result_fields( target: Dict[str, Any], *, status: str, path_slug: Optional[str], ) -> Dict[str, Any]: return { "status": status, "path_slug": path_slug, "remote_id": target.get("remote_id"), "target_source": target.get("source"), "remote_path": target.get("remote_path"), "provisioned": bool(target.get("provisioned")), } def _placement_for_row( collection_row: Dict[str, Any], *, channel_code: str, path_slug: str, ) -> Dict[str, Any]: channel_placement = None channel = collection_row.get("channel") if isinstance(channel, dict): channel_placement = channel.get("placement") if isinstance(channel_placement, dict) and channel_placement.get("channel_code") == channel_code: return channel_placement return build_channel_placement( channel_code=channel_code, path_slug=path_slug, category_l1=str(collection_row.get("category_l1") or "Kitchen Cabinets"), collection=str(collection_row.get("collection") or collection_row.get("title") or ""), parent_path_slug=collection_row.get("parent_path_slug"), ) def _ensure_listing_path_for_collection( session: Session, collection_row: Dict[str, Any], *, placement: Dict[str, Any], path_slug: str, ) -> None: upsert_listing_path( session, path_slug=path_slug, title=str(collection_row.get("title") or collection_row.get("collection") or path_slug), path_kind="collection", parent_path_slug=placement.get("master_parent_path_slug") or collection_row.get("parent_path_slug"), hub_path_slug=placement.get("hub_path_slug") or normalize_plp_path(collection_row.get("category_l1") or ""), notes="collection landing page", ) def _decode_magento_connection_id(connection_id: Optional[int]) -> tuple[Optional[Any], Optional[int]]: if connection_id is None: return None, None from db.compat_connections import decode_compat_connection_id, SHOPIFY_COMPAT_ID_OFFSET native_id = connection_id if connection_id >= SHOPIFY_COMPAT_ID_OFFSET: channel_type, native_id = decode_compat_connection_id(connection_id) if channel_type != "magento": return None, None return None, native_id def _listing_path_target( session: Session, path_slug: str, *, channel_code: str, connection_id: Optional[int], ) -> Optional[str]: path = session.scalar( select(ChannelListingPath).where(ChannelListingPath.path_slug == normalize_plp_path(path_slug)).limit(1) ) if path is None: return None stmt = ( select(ChannelListingPathTarget.remote_id) .where(ChannelListingPathTarget.listing_path_id == path.id) .where(ChannelListingPathTarget.channel_code == channel_code) .where(ChannelListingPathTarget.is_active.is_(True)) ) if connection_id is not None: stmt = stmt.where( (ChannelListingPathTarget.connection_id == connection_id) | (ChannelListingPathTarget.connection_id.is_(None)) ) return session.scalar(stmt.limit(1)) def _build_magento_api(session: Session, connection_id: Optional[int]): from db.compat_connections import decode_compat_connection_id, SHOPIFY_COMPAT_ID_OFFSET from db.magento_repositories import SqlAlchemyMagentoConnectionRepository from magento.magento_api import MagentoOAuthClient, MagentoRestClient from magento.oauth_client import build_magento_oauth_kwargs native_id = connection_id if connection_id is not None and connection_id >= SHOPIFY_COMPAT_ID_OFFSET: channel_type, native_id = decode_compat_connection_id(connection_id) if channel_type != "magento": return None, None if native_id is None: return None, None conn_repo = SqlAlchemyMagentoConnectionRepository(session) conn = conn_repo.get_for_sync(native_id) if not conn: return None, native_id oauth_kwargs = build_magento_oauth_kwargs(conn, timeout=30.0) oauth_kwargs["max_recovery_rounds"] = 1 api = MagentoRestClient(MagentoOAuthClient(**oauth_kwargs)) return api, native_id def _native_shopify_id(connection_id: Optional[int]) -> Optional[int]: if connection_id is None: return None from db.compat_connections import decode_compat_connection_id, SHOPIFY_COMPAT_ID_OFFSET if connection_id >= SHOPIFY_COMPAT_ID_OFFSET: channel_type, native_id = decode_compat_connection_id(connection_id) if channel_type != "shopify": return None return native_id return connection_id def _get_shopify_connection(session: Session, connection_id: Optional[int]): from shopify.connections import get_active_connection from db.models import ShopifyConnection native_id = _native_shopify_id(connection_id) if native_id is not None: row = session.get(ShopifyConnection, native_id) if row and row.status != "disabled": return row return get_active_connection(session) def _refresh_shopping_intersections( session: Session, row: Dict[str, Any], *, connection_id: Optional[int] = None, magento_category_id: Optional[int | str] = None, shopify_collection_id: Optional[str] = None, shopify_handle: Optional[str] = None, ) -> None: from db.collection_landing_pages import _attach_shopping_intersections from db.listing_intersections import sync_shopping_intersections_for_collection sync_shopping_intersections_for_collection( session, row, connection_id=connection_id, magento_category_id=magento_category_id, shopify_collection_id=shopify_collection_id, shopify_handle=shopify_handle, ) _attach_shopping_intersections(session, row)