"""Hierarchical master taxonomy — categories + collections with channel remote IDs.""" from __future__ import annotations import re from typing import Any, Dict, 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 from db.channel_listing_path import upsert_listing_path, upsert_listing_path_target from db.collection_landing_pages import canonical_collection, collection_path_slug, DEFAULT_CATEGORY_L1 from db.master_taxonomy_registry import registry_code_from_name from db.models import ( ChannelListingPath, ChannelListingPathTarget, CollectionLandingPage, MagentoCategoryRegistry, MasterCollectionRegistry, MasterProduct, MasterProductAttributeValue, MasterTaxonomyChannelLink, MasterTaxonomyNode, ShopifyCollectionRegistry, ) from db.source_imports import normalize_column_key NODE_HUB = "hub" NODE_CATEGORY = "category" NODE_COLLECTION = "collection" MAGENTO_ROOT_PREFIXES = ("default category", "root catalog", "default") def build_hierarchy_from_master_catalog( session: Session, *, dry_run: bool = False, allow_invent: Optional[bool] = None, ) -> Dict[str, Any]: """Build master_taxonomy_node tree from master products + catalog intent paths.""" from db.collection_landing_pages import canonical_category from db.collection_path_guard import is_polluted_hub_collection_path, known_collection_names_for_hub from db.master_taxonomy_merge import resolve_canonical_path_slug from db.taxonomy_invent_policy import can_auto_invent invent_ok = can_auto_invent(allow_invent=allow_invent) products = list(session.scalars(select(MasterProduct).where(MasterProduct.is_active.is_(True))).all()) attrs_by_sku = _attrs_by_sku(session, [p.id for p in products]) stats = { "nodes_created": 0, "nodes_updated": 0, "paths_seen": 0, "collections_seen": 0, "polluted_skipped": 0, "invent_skipped": 0, "invent_frozen": not invent_ok, } known_by_hub: Dict[str, Set[str]] = {} node_cache = _load_node_cache(session) def _known_for(category_l1: Optional[str]) -> Set[str]: hub = canonical_category(category_l1) or DEFAULT_CATEGORY_L1 key = _norm_key(hub) cached = known_by_hub.get(key) if cached is not None: return cached names = known_collection_names_for_hub(session, category_l1=hub) known_by_hub[key] = names return names seen_paths: set[str] = set() collection_slugs = _collection_slugs_for_products(products) for product in products: attrs = attrs_by_sku.get(product.sku, {}) known = _known_for(product.category_l1) for path in category_targets_for_product_guarded(session, product, attrs, known_names=known): if _skip_path(path): continue slug = normalize_plp_path(resolve_canonical_path_slug(session, path) or path) if not slug or slug in seen_paths: continue if "/" not in slug and slug in collection_slugs: continue if is_polluted_hub_collection_path(session, slug, known_names=known): stats["polluted_skipped"] += 1 continue seen_paths.add(slug) stats["paths_seen"] += 1 kind = _kind_for_slug(slug, collection_slugs=collection_slugs) if not dry_run: created = _ensure_node_for_slug( session, slug, node_kind=kind, collection_slugs=collection_slugs, node_cache=node_cache, allow_invent=invent_ok, ) if created is None: stats["invent_skipped"] += 1 elif created: stats["nodes_created"] += 1 else: stats["nodes_updated"] += 1 coll = canonical_collection(product.collection) if coll: hub = product.category_l1 or DEFAULT_CATEGORY_L1 slug = normalize_plp_path( resolve_canonical_path_slug(session, collection_path_slug(hub, coll)) or collection_path_slug(hub, coll) ) if slug and slug not in seen_paths: if is_polluted_hub_collection_path(session, slug, known_names=known): stats["polluted_skipped"] += 1 continue seen_paths.add(slug) stats["collections_seen"] += 1 if not dry_run: registry_id = _collection_registry_id(session, coll, hub, allow_invent=invent_ok) created = _ensure_node_for_slug( session, slug, node_kind=NODE_COLLECTION, collection_registry_id=registry_id, collection_slugs=collection_slugs, node_cache=node_cache, allow_invent=invent_ok, ) if created is None: stats["invent_skipped"] += 1 elif created: stats["nodes_created"] += 1 else: stats["nodes_updated"] += 1 if not dry_run: session.flush() return {**stats, "dry_run": dry_run, "node_count": len(seen_paths)} def sync_channel_links_from_registries( session: Session, *, magento_connection_id: Optional[int] = None, shopify_connection_id: Optional[int] = None, dry_run: bool = False, write_listing_path_targets: bool = True, ) -> Dict[str, Any]: """Match master taxonomy nodes to pulled Magento/Shopify registries and store remote IDs.""" nodes = list(session.scalars(select(MasterTaxonomyNode).where(MasterTaxonomyNode.is_active.is_(True))).all()) magento_rows = _magento_registry_rows(session, magento_connection_id) shopify_rows = _shopify_registry_rows(session, shopify_connection_id) magento_index = _build_magento_match_index(magento_rows) shopify_index = _build_shopify_match_index(shopify_rows) parent_slug_by_id = {int(n.id): n.path_slug for n in nodes if n.id is not None} # Include inactive parents that may still be referenced. for node in session.scalars(select(MasterTaxonomyNode)).all(): if node.id is not None and node.id not in parent_slug_by_id: parent_slug_by_id[int(node.id)] = node.path_slug existing_links = _preload_channel_links( session, node_ids=[int(n.id) for n in nodes if n.id is not None], magento_connection_id=magento_connection_id, shopify_connection_id=shopify_connection_id, ) listing_paths_by_slug: Dict[str, ChannelListingPath] = {} listing_targets: Dict[Tuple[int, str, Optional[int], str, str], ChannelListingPathTarget] = {} if write_listing_path_targets and not dry_run: listing_paths_by_slug = { normalize_plp_path(row.path_slug): row for row in session.scalars(select(ChannelListingPath)).all() if normalize_plp_path(row.path_slug) } for target in session.scalars(select(ChannelListingPathTarget)).all(): listing_targets[ ( int(target.listing_path_id), str(target.channel_code), target.connection_id, str(target.taxonomy_kind), str(target.remote_id), ) ] = target stats = { "magento_linked": 0, "shopify_linked": 0, "magento_unmatched": 0, "shopify_unmatched": 0, "listing_path_targets": 0, } pending_listing_targets: List[Dict[str, Any]] = [] for node in nodes: magento_match = _match_magento_node(node, magento_rows, index=magento_index) if magento_match: if not dry_run: _upsert_channel_link( session, node_id=node.id, channel_code="magento", connection_id=magento_connection_id, taxonomy_kind="category" if node.node_kind != NODE_COLLECTION else "collection", remote_id=str(magento_match["category_id"]), remote_parent_id=str(magento_match["parent_id"]) if magento_match.get("parent_id") is not None else None, remote_path=magento_match.get("path_names"), existing_links=existing_links, flush=False, ) if write_listing_path_targets: parent_slug = parent_slug_by_id.get(int(node.parent_id)) if node.parent_id else None upsert_listing_path( session, path_slug=node.path_slug, title=node.name, path_kind=node.node_kind, parent_path_slug=parent_slug, notes="synced from master taxonomy hierarchy", flush=False, existing_by_slug=listing_paths_by_slug, ) pending_listing_targets.append( { "path_slug": node.path_slug, "channel_code": "magento", "remote_id": str(magento_match["category_id"]), "taxonomy_kind": "category" if node.node_kind != NODE_COLLECTION else "collection", "connection_id": magento_connection_id, "remote_path": magento_match.get("path_names"), } ) stats["listing_path_targets"] += 1 stats["magento_linked"] += 1 elif magento_connection_id is not None: stats["magento_unmatched"] += 1 shopify_match = _match_shopify_node(node, shopify_rows, index=shopify_index) if shopify_match: if not dry_run: _upsert_channel_link( session, node_id=node.id, channel_code="shopify", connection_id=shopify_connection_id, taxonomy_kind="collection", remote_id=str(shopify_match["collection_id"]), remote_parent_id="0", remote_path=shopify_match.get("title"), existing_links=existing_links, flush=False, ) if write_listing_path_targets: parent_slug = parent_slug_by_id.get(int(node.parent_id)) if node.parent_id else None upsert_listing_path( session, path_slug=node.path_slug, title=node.name, path_kind=node.node_kind, parent_path_slug=parent_slug, notes="synced from master taxonomy hierarchy", flush=False, existing_by_slug=listing_paths_by_slug, ) pending_listing_targets.append( { "path_slug": node.path_slug, "channel_code": "shopify", "remote_id": str(shopify_match["collection_id"]), "taxonomy_kind": "collection", "connection_id": shopify_connection_id, "remote_path": shopify_match.get("title"), } ) stats["listing_path_targets"] += 1 stats["shopify_linked"] += 1 elif shopify_connection_id is not None: stats["shopify_unmatched"] += 1 if not dry_run: # Assign IDs to newly created listing paths before writing targets. session.flush() for item in pending_listing_targets: upsert_listing_path_target( session, flush=False, existing_paths_by_slug=listing_paths_by_slug, existing_targets=listing_targets, **item, ) session.flush() return {**stats, "dry_run": dry_run} def list_taxonomy_tree( session: Session, *, node_kind: Optional[str] = None, include_channels: bool = True, magento_connection_id: Optional[int] = None, shopify_connection_id: Optional[int] = None, ) -> Dict[str, Any]: """Return flat nodes plus nested tree for dashboard/API.""" all_nodes = list( session.scalars( select(MasterTaxonomyNode).order_by(MasterTaxonomyNode.sort_order, MasterTaxonomyNode.name) ).all() ) active_node_ids = _active_lineage_node_ids(all_nodes) nodes = [ node for node in all_nodes if node.id in active_node_ids and (not node_kind or node.node_kind == node_kind.strip().lower()) ] links_by_node = ( _links_by_node( session, [n.id for n in nodes], magento_connection_id=magento_connection_id, shopify_connection_id=shopify_connection_id, ) if include_channels else {} ) icon_slugs = [node.path_slug for node in nodes if node.node_kind in {NODE_HUB, NODE_CATEGORY}] category_icons = _category_icons_by_slug(session, icon_slugs) magento_remote_ids = ( _active_magento_remote_ids(session, magento_connection_id) if include_channels else None ) flat = [ _node_dict( node, links_by_node.get(node.id, []), session=session, category_icon_url=category_icons.get(node.path_slug), magento_remote_ids=magento_remote_ids, ) for node in nodes ] categories = [row for row in flat if row["node_kind"] in {NODE_HUB, NODE_CATEGORY}] collections = [row for row in flat if row["node_kind"] == NODE_COLLECTION] return { "count": len(flat), "categories": categories, "collections": collections, "nodes": flat, "tree": _nest_tree(flat), } def active_lineage_nodes(session: Session) -> List[MasterTaxonomyNode]: all_nodes = list( session.scalars( select(MasterTaxonomyNode).order_by(MasterTaxonomyNode.sort_order, MasterTaxonomyNode.name) ).all() ) active_node_ids = _active_lineage_node_ids(all_nodes) return [node for node in all_nodes if node.id in active_node_ids] def active_lineage_path_set(session: Session) -> Set[str]: return { normalize_plp_path(node.path_slug) for node in active_lineage_nodes(session) if normalize_plp_path(node.path_slug) } def active_lineage_node_by_slug(session: Session, path_slug: str) -> Optional[MasterTaxonomyNode]: slug = normalize_plp_path(path_slug) if not slug: return None for node in active_lineage_nodes(session): if normalize_plp_path(node.path_slug) == slug: return node return None def _load_node_cache(session: Session) -> Dict[str, MasterTaxonomyNode]: return { normalize_plp_path(node.path_slug): node for node in session.scalars(select(MasterTaxonomyNode)).all() if normalize_plp_path(node.path_slug) } def _ensure_node_for_slug( session: Session, slug: str, *, node_kind: str, collection_registry_id: Optional[int] = None, collection_slugs: Optional[set[str]] = None, node_cache: Optional[Dict[str, MasterTaxonomyNode]] = None, allow_invent: bool = False, ) -> Optional[bool]: """Upsert node for path_slug. Returns True if created, False if updated/touched existing active, None if skipped because invent is frozen and the node (or a required parent) is missing. """ from db.master_taxonomy_merge import resolve_canonical_path_slug slug = normalize_plp_path(resolve_canonical_path_slug(session, slug) or slug) title = _title_from_slug(slug) code = registry_code_from_name(title) parent_slug = _parent_path_slug(slug) parent_id = None cache = node_cache if node_cache is not None else {} if parent_slug: parent_result = _ensure_node_for_slug( session, parent_slug, node_kind=_kind_for_slug(parent_slug, collection_slugs=collection_slugs), collection_slugs=collection_slugs, node_cache=node_cache, allow_invent=allow_invent, ) parent_row = cache.get(parent_slug) if parent_row is None: parent_row = session.scalar( select(MasterTaxonomyNode).where(MasterTaxonomyNode.path_slug == parent_slug).limit(1) ) if parent_row is not None: cache[parent_slug] = parent_row if parent_row is None: return None if parent_result is None else False parent_id = parent_row.id if parent_row else None row = cache.get(slug) if row is None: row = session.scalar(select(MasterTaxonomyNode).where(MasterTaxonomyNode.path_slug == slug).limit(1)) if row is not None: cache[slug] = row if row is not None and not row.is_active: return False if row is None and not allow_invent: return None if row is not None and not allow_invent: # Freeze: do not mutate existing active nodes from auto hierarchy rebuild. return False payload = { "parent_id": parent_id, "node_kind": node_kind, "code": code, "name": title, "path_slug": slug, "collection_registry_id": collection_registry_id, "is_active": True, } if row is None: row = MasterTaxonomyNode(**payload) session.add(row) session.flush() cache[slug] = row return True for key, value in payload.items(): if value is not None: setattr(row, key, value) cache[slug] = row return False def _collection_registry_id( session: Session, collection_name: str, category_l1: str, *, allow_invent: bool = False, ) -> Optional[int]: row = session.scalar( select(MasterCollectionRegistry) .where(MasterCollectionRegistry.name == collection_name) .where(MasterCollectionRegistry.is_active.is_(True)) .limit(1) ) if row: return row.id if not allow_invent: return None from db.master_taxonomy_registry import upsert_collection_registry created = upsert_collection_registry( session, {"name": collection_name, "code": registry_code_from_name(collection_name), "category_l1": category_l1}, allow_invent=True, ) return created.get("id") def _upsert_channel_link( session: Session, *, existing_links: Optional[Dict[Tuple[Any, ...], MasterTaxonomyChannelLink]] = None, flush: bool = True, **payload: Any, ) -> None: node_id = int(payload.get("taxonomy_node_id") or payload["node_id"]) channel = str(payload["channel_code"]).strip().lower() connection_id = payload.get("connection_id") kind = str(payload.get("taxonomy_kind") or "category").strip() link_key = (node_id, channel, connection_id, kind) existing = None if existing_links is not None: existing = existing_links.get(link_key) else: existing = session.scalar( select(MasterTaxonomyChannelLink) .where(MasterTaxonomyChannelLink.taxonomy_node_id == node_id) .where(MasterTaxonomyChannelLink.channel_code == channel) .where(MasterTaxonomyChannelLink.taxonomy_kind == kind) .where( MasterTaxonomyChannelLink.connection_id.is_(None) if connection_id is None else MasterTaxonomyChannelLink.connection_id == connection_id ) .limit(1) ) data = { "taxonomy_node_id": node_id, "channel_code": channel, "connection_id": connection_id, "taxonomy_kind": kind, "remote_id": str(payload["remote_id"]), "remote_parent_id": payload.get("remote_parent_id"), "remote_path": payload.get("remote_path"), "is_active": True, } if existing is None: row = MasterTaxonomyChannelLink(**data) session.add(row) if existing_links is not None: existing_links[link_key] = row else: for key, value in data.items(): setattr(existing, key, value) if flush: session.flush() def _preload_channel_links( session: Session, *, node_ids: List[int], magento_connection_id: Optional[int], shopify_connection_id: Optional[int], ) -> Dict[Tuple[Any, ...], MasterTaxonomyChannelLink]: if not node_ids: return {} stmt = select(MasterTaxonomyChannelLink).where(MasterTaxonomyChannelLink.taxonomy_node_id.in_(node_ids)) out: Dict[Tuple[Any, ...], MasterTaxonomyChannelLink] = {} for link in session.scalars(stmt).all(): if link.channel_code == "magento" and magento_connection_id is not None: if link.connection_id != magento_connection_id: continue if link.channel_code == "shopify" and shopify_connection_id is not None: if link.connection_id != shopify_connection_id: continue out[(int(link.taxonomy_node_id), str(link.channel_code), link.connection_id, str(link.taxonomy_kind))] = link return out def _taxonomy_leaf_title(name: str) -> str: text = str(name or "").strip() return text.split(" / ")[-1].strip() if " / " in text else text def magento_registry_candidates_for_node( node: MasterTaxonomyNode, registry_rows: List[Dict[str, Any]], *, index: Optional[Dict[str, Any]] = None, ) -> List[Dict[str, Any]]: """All Magento registry rows that may match a taxonomy node, best canonical path first.""" candidates_norm = {_normalize_path_names(c) for c in _magento_path_candidates(node.path_slug, node.name)} leaf_key = _norm_key(_taxonomy_leaf_title(node.name)) if index is not None: candidate_rows: List[Dict[str, Any]] = [] seen_ids: set[str] = set() for path_norm in candidates_norm: for row in index.get("by_path_norm", {}).get(path_norm, []): rid = str(row.get("category_id")) if rid not in seen_ids: seen_ids.add(rid) candidate_rows.append(row) for row in index.get("by_stripped_norm", {}).get(path_norm, []): rid = str(row.get("category_id")) if rid not in seen_ids: seen_ids.add(rid) candidate_rows.append(row) for row in index.get("by_leaf", {}).get(leaf_key, []): rid = str(row.get("category_id")) if rid not in seen_ids: seen_ids.add(rid) candidate_rows.append(row) registry_rows = candidate_rows results: List[Dict[str, Any]] = [] for row in registry_rows: path_names = str(row.get("path_names") or "") stripped = _strip_magento_roots(path_names) stripped_norm = _normalize_path_names(stripped) path_norm = _normalize_path_names(path_names) row_leaf = _norm_key(path_names.split("/")[-1] if path_names else str(row.get("name") or "")) if row_leaf != leaf_key and stripped_norm not in candidates_norm and path_norm not in candidates_norm: continue is_canonical = stripped_norm in candidates_norm or path_norm in candidates_norm results.append( { "remote_id": str(row.get("category_id")), "category_id": row.get("category_id"), "parent_id": row.get("parent_id"), "name": row.get("name"), "path_names": path_names, "stripped_path": stripped, "is_canonical_path": is_canonical, "match_reason": "path" if is_canonical else "leaf_name", "duplicate_group_size": 0, } ) results.sort(key=lambda item: (0 if item["is_canonical_path"] else 1, len(item.get("stripped_path") or ""))) group_size = len(results) for item in results: item["duplicate_group_size"] = group_size return results def _match_magento_node( node: MasterTaxonomyNode, rows: List[Dict[str, Any]], *, index: Optional[Dict[str, Any]] = None, ) -> Optional[Dict[str, Any]]: candidates = magento_registry_candidates_for_node(node, rows, index=index) if not candidates: return None best = candidates[0] if not best["is_canonical_path"] and len(candidates) != 1: return None by_id = (index or {}).get("by_id") if index is not None else None if isinstance(by_id, dict): hit = by_id.get(str(best["remote_id"])) if hit is not None: return hit return next((row for row in rows if str(row.get("category_id")) == best["remote_id"]), None) def _match_shopify_node( node: MasterTaxonomyNode, rows: List[Dict[str, Any]], *, index: Optional[Dict[str, Any]] = None, ) -> Optional[Dict[str, Any]]: title_key = _norm_key(node.name) handle_key = _norm_key(node.path_slug.split("/")[-1] if "/" in node.path_slug else node.path_slug) if index is not None: matches = [] seen: set[str] = set() for row in index.get("by_title", {}).get(title_key, []): rid = str(row.get("collection_id")) if rid not in seen: seen.add(rid) matches.append(row) for row in index.get("by_handle", {}).get(handle_key, []): rid = str(row.get("collection_id")) if rid not in seen: seen.add(rid) matches.append(row) if len(matches) == 1: return matches[0] return None matches = [] for row in rows: if _norm_key(row.get("title")) == title_key: matches.append(row) elif _norm_key(row.get("handle")) == handle_key: matches.append(row) if len(matches) == 1: return matches[0] return None def _build_magento_match_index(rows: List[Dict[str, Any]]) -> Dict[str, Any]: by_path_norm: Dict[str, List[Dict[str, Any]]] = {} by_stripped_norm: Dict[str, List[Dict[str, Any]]] = {} by_leaf: Dict[str, List[Dict[str, Any]]] = {} by_id: Dict[str, Dict[str, Any]] = {} for row in rows: path_names = str(row.get("path_names") or "") stripped = _strip_magento_roots(path_names) path_norm = _normalize_path_names(path_names) stripped_norm = _normalize_path_names(stripped) leaf = _norm_key(path_names.split("/")[-1] if path_names else str(row.get("name") or "")) by_id[str(row.get("category_id"))] = row if path_norm: by_path_norm.setdefault(path_norm, []).append(row) if stripped_norm: by_stripped_norm.setdefault(stripped_norm, []).append(row) if leaf: by_leaf.setdefault(leaf, []).append(row) return { "by_path_norm": by_path_norm, "by_stripped_norm": by_stripped_norm, "by_leaf": by_leaf, "by_id": by_id, } def _build_shopify_match_index(rows: List[Dict[str, Any]]) -> Dict[str, Any]: by_title: Dict[str, List[Dict[str, Any]]] = {} by_handle: Dict[str, List[Dict[str, Any]]] = {} for row in rows: title_key = _norm_key(row.get("title")) handle_key = _norm_key(row.get("handle")) if title_key: by_title.setdefault(title_key, []).append(row) if handle_key: by_handle.setdefault(handle_key, []).append(row) return {"by_title": by_title, "by_handle": by_handle} def _magento_registry_rows(session: Session, connection_id: Optional[int]) -> List[Dict[str, Any]]: if connection_id is None: return [] rows = session.scalars( select(MagentoCategoryRegistry) .where(MagentoCategoryRegistry.connection_id == connection_id) .where(MagentoCategoryRegistry.is_active.is_(True)) ).all() return [ { "category_id": row.category_id, "parent_id": row.parent_id, "path_names": row.path_names or row.name, "name": row.name, } for row in rows ] def _shopify_registry_rows(session: Session, connection_id: Optional[int]) -> List[Dict[str, Any]]: stmt = select(ShopifyCollectionRegistry) if connection_id is not None: stmt = stmt.where(ShopifyCollectionRegistry.connection_id == connection_id) return [ { "collection_id": row.collection_id, "title": row.title, "handle": row.handle, } for row in session.scalars(stmt).all() ] def _links_by_node( session: Session, node_ids: List[int], *, magento_connection_id: Optional[int] = None, shopify_connection_id: Optional[int] = None, ) -> Dict[int, List[MasterTaxonomyChannelLink]]: if not node_ids: return {} out: Dict[int, List[MasterTaxonomyChannelLink]] = {} stmt = ( select(MasterTaxonomyChannelLink) .where(MasterTaxonomyChannelLink.taxonomy_node_id.in_(node_ids)) .where(MasterTaxonomyChannelLink.is_active.is_(True)) ) if magento_connection_id is not None and shopify_connection_id is not None: # Strict dual-channel view: only the selected Magento + Shopify connections. stmt = stmt.where( ( (MasterTaxonomyChannelLink.channel_code == "magento") & (MasterTaxonomyChannelLink.connection_id == int(magento_connection_id)) ) | ( (MasterTaxonomyChannelLink.channel_code == "shopify") & (MasterTaxonomyChannelLink.connection_id == int(shopify_connection_id)) ) ) elif magento_connection_id is not None: # Magento-only selector: do not leak Shopify GIDs into the Magento tree view. stmt = stmt.where( (MasterTaxonomyChannelLink.channel_code == "magento") & (MasterTaxonomyChannelLink.connection_id == int(magento_connection_id)) ) elif shopify_connection_id is not None: stmt = stmt.where( (MasterTaxonomyChannelLink.channel_code == "shopify") & (MasterTaxonomyChannelLink.connection_id == int(shopify_connection_id)) ) for link in session.scalars(stmt).all(): out.setdefault(link.taxonomy_node_id, []).append(link) return out def reconcile_magento_channel_links_against_registry( session: Session, connection_id: int, ) -> Dict[str, int]: """Deactivate Magento taxonomy links / listing targets whose remote id is gone from registry.""" from db.models import ChannelListingPathTarget as ListingTarget active_ids = { str(int(row.category_id)) for row in session.scalars( select(MagentoCategoryRegistry) .where(MagentoCategoryRegistry.connection_id == int(connection_id)) .where(MagentoCategoryRegistry.is_active.is_(True)) ).all() } links_deactivated = 0 for link in session.scalars( select(MasterTaxonomyChannelLink) .where(MasterTaxonomyChannelLink.channel_code == "magento") .where(MasterTaxonomyChannelLink.connection_id == int(connection_id)) .where(MasterTaxonomyChannelLink.is_active.is_(True)) ).all(): remote_id = str(link.remote_id or "").strip() if remote_id and remote_id in active_ids: continue link.is_active = False links_deactivated += 1 targets_deactivated = 0 for target in session.scalars( select(ListingTarget) .where(ListingTarget.channel_code == "magento") .where(ListingTarget.connection_id == int(connection_id)) .where(ListingTarget.is_active.is_(True)) ).all(): remote_id = str(target.remote_id or "").strip() if remote_id and remote_id in active_ids: continue target.is_active = False targets_deactivated += 1 if links_deactivated or targets_deactivated: session.flush() return { "links_deactivated": links_deactivated, "listing_targets_deactivated": targets_deactivated, } def _active_magento_remote_ids(session: Session, connection_id: Optional[int]) -> Optional[set[str]]: if connection_id is None: return None return { str(int(row.category_id)) for row in session.scalars( select(MagentoCategoryRegistry) .where(MagentoCategoryRegistry.connection_id == int(connection_id)) .where(MagentoCategoryRegistry.is_active.is_(True)) ).all() } def _node_dict( node: MasterTaxonomyNode, links: List[MasterTaxonomyChannelLink], *, session: Optional[Session] = None, category_icon_url: Optional[str] = None, magento_remote_ids: Optional[set[str]] = None, ) -> Dict[str, Any]: channels: Dict[str, Any] = {} for link in links: remote_id = str(link.remote_id or "").strip() channel_payload: Dict[str, Any] = { "remote_id": remote_id, "remote_parent_id": link.remote_parent_id, "remote_path": link.remote_path, "taxonomy_kind": link.taxonomy_kind, "connection_id": link.connection_id, "linked": bool(remote_id), } if ( link.channel_code == "magento" and magento_remote_ids is not None and len(magento_remote_ids) > 0 and remote_id and remote_id not in magento_remote_ids ): # Stale DB link pointing at a Magento category deleted/hidden from registry. channel_payload["remote_missing"] = True channel_payload["linked"] = False channel_payload["stale_remote_id"] = remote_id channel_payload["remote_id"] = None if ( session is not None and link.channel_code == "shopify" and link.taxonomy_kind == "collection" and remote_id ): collection_row = session.scalar( select(ShopifyCollectionRegistry) .where(ShopifyCollectionRegistry.collection_id == remote_id) .limit(1) ) if collection_row is not None: channel_payload["collection_type"] = collection_row.collection_type channel_payload["collection_handle"] = collection_row.handle channel_payload["collection_title"] = collection_row.title if collection_row.collection_type == "smart": channel_payload["smart_collection_rules"] = collection_row.rules_json channels[link.channel_code] = channel_payload row = { "id": node.id, "parent_id": node.parent_id or 0, "node_kind": node.node_kind, "code": node.code, "name": node.name, "path_slug": node.path_slug, "collection_registry_id": node.collection_registry_id, "sort_order": node.sort_order, "channels": channels, } if session is not None and node.node_kind == NODE_COLLECTION: row["landing_status"] = _collection_landing_status(session, node) if node.node_kind in {NODE_HUB, NODE_CATEGORY}: row["category_icon_url"] = category_icon_url return row def _category_icons_by_slug(session: Session, path_slugs: List[str]) -> Dict[str, str]: 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: str(row.category_icon_url) for row in rows if row.category_icon_url } def _collection_landing_status(session: Session, node: MasterTaxonomyNode) -> str: landing = None if node.collection_registry_id: landing = session.scalar( select(CollectionLandingPage) .where(CollectionLandingPage.collection_registry_id == int(node.collection_registry_id)) .limit(1) ) if landing is None and node.path_slug: landing = session.scalar( select(CollectionLandingPage).where(CollectionLandingPage.path_slug == node.path_slug).limit(1) ) if landing is None: return "missing" if landing.path_slug != node.path_slug: return "mismatch" if node.collection_registry_id and landing.collection_registry_id != node.collection_registry_id: return "mismatch" return "linked" def _nest_tree(flat: List[Dict[str, Any]]) -> List[Dict[str, Any]]: by_id = {row["id"]: {**row, "children": []} for row in flat} roots: List[Dict[str, Any]] = [] for row in flat: node = by_id[row["id"]] parent_id = row.get("parent_id") or 0 if parent_id and parent_id in by_id: by_id[parent_id]["children"].append(node) else: roots.append(node) return roots def _active_lineage_node_ids(nodes: List[MasterTaxonomyNode]) -> Set[int]: by_id = {int(node.id): node for node in nodes if node.id is not None} memo: Dict[int, bool] = {} def _is_visible(node: MasterTaxonomyNode) -> bool: if node.id is None: return False node_id = int(node.id) cached = memo.get(node_id) if cached is not None: return cached if node.is_active is False: memo[node_id] = False return False parent_id = int(node.parent_id) if node.parent_id is not None else 0 if not parent_id: memo[node_id] = True return True parent = by_id.get(parent_id) if parent is None: memo[node_id] = False return False visible = _is_visible(parent) memo[node_id] = visible return visible return {int(node.id) for node in nodes if node.id is not None and _is_visible(node)} def _attrs_by_sku(session: Session, product_ids: List[int]) -> Dict[str, Dict[str, str]]: if not product_ids: return {} out: Dict[str, Dict[str, str]] = {} for value in session.scalars( select(MasterProductAttributeValue).where(MasterProductAttributeValue.product_id.in_(product_ids)) ).all(): if value.value is None or value.value == "": continue out.setdefault(value.sku, {})[normalize_column_key(value.attribute_code)] = str(value.value).strip() return out def _skip_path(path: str) -> bool: return _norm_key(path) == _norm_key(ALL_PRODUCTS) def _collection_slugs_for_products(products: List[MasterProduct]) -> set[str]: slugs: set[str] = set() for product in products: coll = canonical_collection(product.collection) if not coll: continue hub = product.category_l1 or DEFAULT_CATEGORY_L1 slugs.add(normalize_plp_path(collection_path_slug(hub, coll))) slugs.add(normalize_plp_path(coll)) return slugs def _kind_for_slug(slug: str, *, is_collection_path: bool = False, collection_slugs: Optional[set[str]] = None) -> str: slug = normalize_plp_path(slug) if is_collection_path or (collection_slugs and slug in collection_slugs): return NODE_COLLECTION parts = [p for p in slug.split("/") if p] if len(parts) <= 1: return NODE_HUB return NODE_CATEGORY def _parent_path_slug(slug: str) -> Optional[str]: parts = [p for p in normalize_plp_path(slug).split("/") if p] if len(parts) <= 1: return None return "/".join(parts[:-1]) def _parent_slug(session: Session, parent_id: Optional[int]) -> Optional[str]: if not parent_id: return None parent = session.get(MasterTaxonomyNode, int(parent_id)) return parent.path_slug if parent else None def _title_from_slug(slug: str) -> str: return " / ".join(part.replace("-", " ").title() for part in slug.split("/") if part) def _magento_path_candidates(path_slug: str, title: str) -> List[str]: from db.magento_path_names import plp_slug_to_magento_path_names, title_path_to_magento_path_names slug = normalize_plp_path(path_slug) human = plp_slug_to_magento_path_names(slug) paths = [human] titled = title_path_to_magento_path_names(title) if titled and _normalize_path_names(titled) != _normalize_path_names(human): paths.append(titled) # Keep a raw title-case slug variant only when it differs (legacy short "Tall"). raw_human = "/".join(part.replace("-", " ").title() for part in slug.split("/") if part) if raw_human and _normalize_path_names(raw_human) != _normalize_path_names(human): paths.append(raw_human) if "/" not in slug: leaf = title.split(" / ")[-1] if " / " in title else title if leaf and _normalize_path_names(leaf) != _normalize_path_names(human): paths.append(leaf) for prefix in ("Default Category", "Root Catalog"): paths.append(f"{prefix}/{human}") return list(dict.fromkeys(p for p in paths if p.strip())) def _strip_magento_roots(path_names: str) -> str: parts = [p.strip() for p in path_names.split("/") if p.strip()] while parts and parts[0].lower() in MAGENTO_ROOT_PREFIXES: parts.pop(0) return "/".join(parts) def _normalize_path_names(value: str) -> str: return re.sub(r"\s+", " ", str(value or "").strip().lower()) def _norm_key(value: Optional[str]) -> str: return re.sub(r"[^a-z0-9]+", "", str(value or "").lower())