"""Generic SEO intersection paths for collection/listing + facet combinations.""" from __future__ import annotations from typing import Any, Dict, Iterable, List, Optional from sqlalchemy import select from sqlalchemy.orm import Session from channel.url_canonical import build_magento_plp_url, build_plp_url, normalize_plp_path, path_segments, slugify_segment from db.channel_listing_path import list_listing_paths, upsert_listing_path, upsert_listing_path_target from db.collection_landing_pages import collection_start_shopping_options from db.models import ChannelListingPath, ChannelListingPathTarget, ListingIntersectionRule from db.shopping_facet_catalog import ( HUB_SHOPPING_OPTION_DEFAULTS, KITCHEN_CABINET_DETAIL_OPTIONS, KITCHEN_CABINET_SHOPPING_OPTIONS, canonical_cabinet_type_label, compact_filter_value, ) # Hub-level defaults until facet_terms.shopping_facet_attribute is set on channel_listing_path. HUB_SHOPPING_FACET_DEFAULTS: Dict[str, str] = { "kitchen-cabinets": "cabinet_type", } SHOPPING_FACET_ATTRIBUTE_CANDIDATES: List[Dict[str, str]] = [ {"code": "cabinet_type", "label": "Cabinet Type"}, {"code": "door_core", "label": "Door Core"}, {"code": "door_type", "label": "Door Type"}, {"code": "vanity_width", "label": "Vanity Width"}, {"code": "vanity_type", "label": "Vanity Type"}, {"code": "material", "label": "Material"}, {"code": "width", "label": "Width"}, {"code": "height", "label": "Height"}, {"code": "finish", "label": "Finish"}, ] # Re-export for backward compatibility. __all__ = [ "KITCHEN_CABINET_SHOPPING_OPTIONS", "HUB_SHOPPING_OPTION_DEFAULTS", "compact_filter_value", ] def intersection_path_slug(group_path_slug: str, facet_slug: str) -> str: return normalize_plp_path(f"{group_path_slug}/{facet_slug}") def upsert_listing_intersection_rule( session: Session, *, base_path_slug: str, group_path_slug: str, group_kind: str = "collection", facet_attribute: str, facet_label: str, facet_value: Optional[str] = None, facet_slug: Optional[str] = None, title: Optional[str] = None, target_path_slug: Optional[str] = None, magento_category_id: Optional[int | str] = None, category_filter_path_slug: Optional[str] = None, magento_filter_category_id: Optional[int | str] = None, shopify_collection_id: Optional[str] = None, shopify_handle: Optional[str] = None, connection_id: Optional[int] = None, is_indexable: bool = True, notes: Optional[str] = None, ) -> Dict[str, Any]: base = normalize_plp_path(base_path_slug) group = normalize_plp_path(group_path_slug) if not base or not group: raise ValueError("base_path_slug and group_path_slug are required") attribute = str(facet_attribute or "").strip() label = str(facet_label or "").strip() if not attribute or not label: raise ValueError("facet_attribute and facet_label are required") f_slug = slugify_segment(facet_slug or label) f_value = str(facet_value or compact_filter_value(label)).strip() seo_path = intersection_path_slug(group, f_slug) target_slug = normalize_plp_path(target_path_slug or group) page_title = str(title or f"{label} - {group.split('/')[-1].replace('-', ' ').title()}").strip() target_path = _ensure_path(session, target_slug, title=target_slug.replace("-", " ").title()) listing_path = _ensure_path( session, seo_path, title=page_title, path_kind="intersection", parent_path_slug=group, hub_path_slug=base, facet_terms=[group, f_slug], notes=notes or "SEO intersection path", ) category_filter_path = normalize_plp_path(category_filter_path_slug or "") resolved_filter_category_id = str(magento_filter_category_id or "").strip() if category_filter_path and not resolved_filter_category_id: resolved_filter_category_id = _channel_target_remote_id( session, path_slug=category_filter_path, channel_code="magento", connection_id=connection_id, ) payload = { "attribute": attribute, "label": label, "value": f_value, "slug": f_slug, } if category_filter_path: payload.update( { "type": "category", "attribute": "category_ids", "value": resolved_filter_category_id or category_filter_path, "category_path": category_filter_path, "category_id": resolved_filter_category_id or None, } ) channel_payload: Dict[str, Any] = {} if magento_category_id: channel_payload["magento"] = { "target_category_id": str(magento_category_id), "target_path_slug": target_slug, "filter": payload, } if category_filter_path: channel_payload["magento"]["filter_category_path"] = category_filter_path if resolved_filter_category_id: channel_payload["magento"]["filter_category_id"] = resolved_filter_category_id upsert_listing_path_target( session, path_slug=seo_path, channel_code="magento", remote_id=str(magento_category_id), taxonomy_kind="category_filter", connection_id=connection_id, remote_path=target_slug, ) if shopify_collection_id or shopify_handle: channel_payload["shopify"] = { "collection_id": shopify_collection_id, "handle": shopify_handle or seo_path.replace("/", "-"), "target_path_slug": target_slug, "filter": payload, } if shopify_collection_id: upsert_listing_path_target( session, path_slug=seo_path, channel_code="shopify", remote_id=str(shopify_collection_id), taxonomy_kind="collection_filter", connection_id=connection_id, remote_path=shopify_handle or seo_path.replace("/", "-"), ) row = session.scalar(select(ListingIntersectionRule).where(ListingIntersectionRule.seo_path_slug == seo_path)) data = { "base_path_slug": base, "group_path_slug": group, "group_kind": group_kind, "facet_attribute": attribute, "facet_label": label, "facet_value": f_value, "facet_slug": f_slug, "seo_path_slug": seo_path, "title": page_title, "listing_path_id": listing_path["id"], "target_listing_path_id": target_path["id"], "filter_payload": payload, "channel_payload": channel_payload or None, "is_indexable": is_indexable, "is_active": True, "notes": notes, } if row is None: row = ListingIntersectionRule(**data) session.add(row) else: for key, value in data.items(): setattr(row, key, value) session.flush() return listing_intersection_dict(row) def materialize_listing_intersections( session: Session, *, base_path_slug: str, group_path_slug: str, facet_attribute: str, options: Iterable[Dict[str, Any] | str], group_kind: str = "collection", target_path_slug: Optional[str] = None, magento_category_id: Optional[int | str] = None, shopify_collection_id: Optional[str] = None, shopify_handle_prefix: Optional[str] = None, connection_id: Optional[int] = None, use_category_filters: bool = False, ) -> List[Dict[str, Any]]: """Materialize SEO intersection rules for a collection/group + facet options. When ``use_category_filters`` is True (Start Shopping / taxonomy hubs), each option asserts ``filter_type=category`` with ``category_path`` under the hub (e.g. kitchen-cabinets/wall-cabinets) instead of attribute params. """ out: List[Dict[str, Any]] = [] hub = normalize_plp_path(base_path_slug) for option in options: if isinstance(option, str): item = {"label": option} else: item = dict(option) label = str(item.get("label") or item.get("name") or "").strip() if not label: continue facet_slug = item.get("slug") or slugify_segment(label) seo_path = intersection_path_slug(group_path_slug, facet_slug) category_filter_path = None if use_category_filters or item.get("category_path") or item.get("category_filter_path_slug"): category_filter_path = normalize_plp_path( item.get("category_path") or item.get("category_filter_path_slug") or f"{hub}/{facet_slug}" ) out.append( upsert_listing_intersection_rule( session, base_path_slug=base_path_slug, group_path_slug=group_path_slug, group_kind=group_kind, facet_attribute=facet_attribute, facet_label=label, facet_value=item.get("value"), facet_slug=facet_slug, title=item.get("title"), target_path_slug=target_path_slug, magento_category_id=magento_category_id, category_filter_path_slug=category_filter_path, magento_filter_category_id=item.get("magento_filter_category_id") or item.get("filter_category_id"), shopify_collection_id=shopify_collection_id, shopify_handle=( f"{normalize_plp_path(shopify_handle_prefix).replace('/', '-')}-{facet_slug}" if shopify_handle_prefix else item.get("shopify_handle") ), connection_id=connection_id, is_indexable=bool(item.get("is_indexable", True)), notes=item.get("notes"), ) ) out[-1]["seo_path_slug"] = seo_path return out def list_listing_intersections( session: Session, *, base_path_slug: Optional[str] = None, group_path_slug: Optional[str] = None, is_active: bool = True, include_nested: bool = True, ) -> List[Dict[str, Any]]: stmt = select(ListingIntersectionRule).order_by( ListingIntersectionRule.group_path_slug, ListingIntersectionRule.facet_slug, ) if is_active: stmt = stmt.where(ListingIntersectionRule.is_active.is_(True)) if base_path_slug: stmt = stmt.where(ListingIntersectionRule.base_path_slug == normalize_plp_path(base_path_slug)) rows = [listing_intersection_dict(row) for row in session.scalars(stmt).all()] if not group_path_slug: return rows group = normalize_plp_path(group_path_slug) if not include_nested: return [row for row in rows if normalize_plp_path(row.get("group_path_slug")) == group] # Include L1 (group == collection) and L2 (group/seo under collection/...). return [ row for row in rows if normalize_plp_path(row.get("group_path_slug")) == group or str(row.get("group_path_slug") or "").startswith(f"{group}/") or str(row.get("seo_path_slug") or "").startswith(f"{group}/") ] def intersection_form_options( session: Session, *, hub_path_slug: Optional[str] = None, facet_attribute: Optional[str] = None, ) -> Dict[str, Any]: """Dashboard form choices: hubs, facet attributes, facet option labels, collection groups.""" hubs: List[Dict[str, Any]] = [] seen_hubs: set[str] = set() for item in list_listing_paths(session, path_kind="hub", limit=200): slug = normalize_plp_path(item.get("path_slug") or "") if not slug or slug in seen_hubs: continue seen_hubs.add(slug) hubs.append( { "path_slug": slug, "title": item.get("title") or slug.replace("-", " ").title(), "shopping_facet_attribute": item.get("shopping_facet_attribute") or HUB_SHOPPING_FACET_DEFAULTS.get(slug), } ) for item in list_listing_paths(session, limit=500): slug = normalize_plp_path(item.get("path_slug") or "") if not slug or "/" in slug or slug in seen_hubs: continue seen_hubs.add(slug) hubs.append( { "path_slug": slug, "title": item.get("title") or slug.replace("-", " ").title(), "shopping_facet_attribute": item.get("shopping_facet_attribute") or HUB_SHOPPING_FACET_DEFAULTS.get(slug), } ) hubs.sort(key=lambda row: str(row.get("title") or row.get("path_slug") or "")) facet_codes = {row["code"] for row in SHOPPING_FACET_ATTRIBUTE_CANDIDATES} for hub in hubs: attribute = str(hub.get("shopping_facet_attribute") or "").strip() if attribute: facet_codes.add(attribute) facet_attributes = list(SHOPPING_FACET_ATTRIBUTE_CANDIDATES) for code in sorted(facet_codes - {row["code"] for row in facet_attributes}): facet_attributes.append({"code": code, "label": code.replace("_", " ").title()}) hub = normalize_plp_path(hub_path_slug or "") selected_attribute = str(facet_attribute or "").strip() or ( shopping_facet_attribute_for_hub(session, hub) if hub else None ) facet_options = shopping_facet_options_for_hub(session, hub, facet_attribute=selected_attribute) if hub else [] if hub and selected_attribute and not facet_options: from db.master_product_filters import list_master_attribute_values facet_options = [ {"label": value, "slug": slugify_segment(value), "value": compact_filter_value(value)} for value in list_master_attribute_values(session, selected_attribute, limit=100) if str(value or "").strip() ] groups: List[Dict[str, str]] = [] if hub: groups = intersection_group_paths_for_hub(session, hub) return { "hubs": hubs, "facet_attributes": facet_attributes, "facet_options": facet_options, "groups": groups, "shopping_facet_attribute": selected_attribute, } def intersection_group_paths_for_hub(session: Session, hub_path_slug: str) -> List[Dict[str, str]]: """Collection/group paths for intersection materialize — taxonomy collections only, deduped.""" hub = normalize_plp_path(hub_path_slug) if not hub: return [] facet_slugs = _hub_facet_path_slugs(session, hub) from db.collection_landing_pages import canonical_collection, list_collection_landing_pages from db.master_taxonomy_merge import resolve_canonical_path_slug from db.models import MasterTaxonomyNode candidates: Dict[str, Dict[str, str]] = {} def _add_candidate(raw_slug: str, title: str) -> None: from db.collection_path_guard import is_polluted_hub_collection_path from db.master_taxonomy_product_paths import is_shopping_facet_rooted_path canonical = resolve_canonical_path_slug(session, raw_slug) or normalize_plp_path(raw_slug) segments = path_segments(canonical) if len(segments) < 2 or segments[0] != hub or segments[-1] in facet_slugs: return # Product taxonomy under shopping facets (accessories/clearance-kit) is not a # finish collection — never materialize Start Shopping SEO intersections there. if is_shopping_facet_rooted_path(canonical): return if is_polluted_hub_collection_path(session, canonical): return label = str(title or canonical_collection(segments[1].replace("-", " ").title()) or segments[1]).strip() if not label: return existing = candidates.get(canonical) if existing is None or len(label) < len(existing["title"]): candidates[canonical] = {"path_slug": canonical, "title": label} elif existing["title"].lower().startswith("kitchen cabinets /"): candidates[canonical] = {"path_slug": canonical, "title": label} for node in session.scalars( select(MasterTaxonomyNode) .where(MasterTaxonomyNode.is_active.is_(True)) .where(MasterTaxonomyNode.node_kind == "collection") .where(MasterTaxonomyNode.path_slug.like(f"{hub}/%")) .order_by(MasterTaxonomyNode.sort_order, MasterTaxonomyNode.name) ): _add_candidate(str(node.path_slug), str(node.name)) hub_title = hub.replace("-", " ").title() for page in list_collection_landing_pages(session, include_inferred=False, limit=500): slug = str(page.get("path_slug") or "") if not slug.startswith(f"{hub}/"): continue if page.get("inferred"): continue title = str(page.get("title") or page.get("collection") or "").strip() if title.lower().startswith(f"{hub_title.lower()} /"): title = title.split("/", 1)[-1].strip() _add_candidate(slug, title) return sorted(candidates.values(), key=lambda row: str(row.get("title") or row.get("path_slug") or "").lower()) def _hub_facet_path_slugs(session: Session, hub: str) -> set[str]: slugs = {str(item.get("slug") or "").strip() for item in shopping_facet_options_for_hub(session, hub)} slugs.discard("") for item in list_listing_paths(session, hub_path_slug=hub, path_kind="facet", limit=200): segments = path_segments(normalize_plp_path(item.get("path_slug") or "")) if len(segments) == 2 and segments[0] == hub: slugs.add(segments[1]) for item in list_listing_paths(session, hub_path_slug=hub, limit=500): if str(item.get("path_kind") or "") != "facet": continue segments = path_segments(normalize_plp_path(item.get("path_slug") or "")) if len(segments) == 2 and segments[0] == hub: slugs.add(segments[1]) return slugs def shopping_facet_attribute_for_hub(session: Session, hub_path_slug: str) -> Optional[str]: hub = normalize_plp_path(hub_path_slug) if not hub: return None path = session.scalar(select(ChannelListingPath).where(ChannelListingPath.path_slug == hub)) if path and isinstance(path.facet_terms, dict): attribute = str(path.facet_terms.get("shopping_facet_attribute") or "").strip() if attribute: return attribute return HUB_SHOPPING_FACET_DEFAULTS.get(hub) def shopping_facet_options_for_hub( session: Session, hub_path_slug: str, *, facet_attribute: Optional[str] = None, ) -> List[Dict[str, str]]: """Facet tile options for a hub: listing-path children first, then hub defaults.""" hub = normalize_plp_path(hub_path_slug) if not hub: return [] selected_attribute = str(facet_attribute or shopping_facet_attribute_for_hub(session, hub) or "").strip() default_attribute = HUB_SHOPPING_FACET_DEFAULTS.get(hub) defaults = HUB_SHOPPING_OPTION_DEFAULTS.get(hub, []) if defaults and (not selected_attribute or selected_attribute == default_attribute): return [{**dict(item), "source": "default"} for item in defaults] options: List[Dict[str, str]] = [] seen: set[str] = set() for item in list_listing_paths(session, hub_path_slug=hub, path_kind="facet", limit=100): slug = normalize_plp_path(item.get("path_slug") or "") segments = path_segments(slug) if len(segments) != 2 or segments[0] != hub: continue facet_slug = segments[1] label = str(item.get("title") or facet_slug.replace("-", " ").title()).strip() key = facet_slug.lower() if not label or key in seen: continue seen.add(key) options.append({"label": label, "slug": facet_slug, "title": label, "source": "listing_path"}) if options: return options return [{**dict(item), "source": "default"} for item in defaults] def resolve_channel_filter_value( session: Session, *, attribute_code: str, label: str, connection_id: Optional[int] = None, ) -> str: """Compact URL slug for layered-nav query params (e.g. basecabinets), not value_index.""" _ = session, connection_id # reserved for future channel-specific slug tables display = canonical_cabinet_type_label(label) if attribute_code == "cabinet_type" else label return compact_filter_value(display or label) def sync_shopping_intersections_for_collection( session: Session, collection_row: Dict[str, Any], *, connection_id: Optional[int] = None, facet_attribute: Optional[str] = None, facet_options: Optional[Iterable[Dict[str, Any] | str]] = None, magento_category_id: Optional[int | str] = None, shopify_collection_id: Optional[str] = None, shopify_handle: Optional[str] = None, ) -> List[Dict[str, Any]]: """Materialize intersection rules for one collection from hub shopping facet config.""" from db.master_taxonomy_product_paths import is_shopping_facet_rooted_path group_path = normalize_plp_path(str(collection_row.get("path_slug") or "")) if not group_path: return [] # Never invent hub/accessories/clearance-kit/base-cabinets/... SEO trees. if is_shopping_facet_rooted_path(group_path): return [] segments = path_segments(group_path) if len(segments) < 2: return [] hub = normalize_plp_path( collection_row.get("parent_path_slug") or collection_row.get("category_l1") or segments[0] ) if "/" not in hub: hub = segments[0] attribute = facet_attribute or shopping_facet_attribute_for_hub(session, hub) if not attribute: return [] explicit_config = _explicit_start_shopping_items(session, collection_row, hub_path_slug=hub) options = list(facet_options) if facet_options is not None else shopping_facet_options_for_hub(session, hub) if not options and not explicit_config: return [] materialized: List[Dict[str, Any]] = [] detail_jobs: List[tuple[Dict[str, Any], str, List[Dict[str, str]], str]] = [] keep_paths: set[str] = set() for option in explicit_config or options: if isinstance(option, str): item: Dict[str, Any] = {"label": option} else: item = dict(option) label = str(item.get("label") or item.get("name") or "").strip() if not label: continue if not item.get("value"): item["value"] = resolve_channel_filter_value( session, attribute_code=attribute, label=label, connection_id=connection_id, ) row = upsert_listing_intersection_rule( session, base_path_slug=hub, group_path_slug=group_path, group_kind="collection", facet_attribute=attribute, facet_label=label, facet_value=item.get("value"), facet_slug=item.get("slug"), title=item.get("title"), target_path_slug=group_path, category_filter_path_slug=normalize_plp_path(f"{hub}/{item.get('slug') or slugify_segment(label)}"), magento_category_id=magento_category_id or _magento_target_category_id(collection_row), shopify_collection_id=shopify_collection_id or _shopify_target_collection_id(collection_row), shopify_handle=shopify_handle or _shopify_target_handle(collection_row, hub, group_path), connection_id=connection_id, is_indexable=bool(item.get("is_indexable", True)), notes=item.get("notes"), ) materialized.append(row) keep_paths.add(row["seo_path_slug"]) detail_options = _detail_options_for_shopping_option( session, hub, label, selected_paths=item.get("selected_detail_paths"), ) detail_attribute = _detail_facet_attribute(label) if detail_options and detail_attribute: detail_jobs.append((row, label, detail_options, detail_attribute)) for parent_row, parent_label, detail_options, detail_attribute in detail_jobs: for detail in detail_options: detail_item = dict(detail) detail_label = str(detail_item.get("label") or "").strip() if not detail_label: continue if not detail_item.get("value"): detail_item["value"] = resolve_channel_filter_value( session, attribute_code=detail_attribute, label=detail_label, connection_id=connection_id, ) detail_row = upsert_listing_intersection_rule( session, base_path_slug=hub, group_path_slug=parent_row["seo_path_slug"], group_kind="collection_filter", facet_attribute=detail_attribute, facet_label=detail_label, facet_value=detail_item.get("value"), facet_slug=detail_item.get("slug"), title=detail_item.get("title"), target_path_slug=group_path, category_filter_path_slug=detail_item.get("category_path") or normalize_plp_path( f"{hub}/{slugify_segment(parent_label)}/{detail_item.get('slug') or slugify_segment(detail_label)}" ), magento_category_id=magento_category_id or _magento_target_category_id(collection_row), shopify_collection_id=shopify_collection_id or _shopify_target_collection_id(collection_row), shopify_handle=shopify_handle or _shopify_target_handle(collection_row, hub, group_path), connection_id=connection_id, is_indexable=bool(detail_item.get("is_indexable", True)), notes=detail_item.get("notes") or f"SEO sub-intersection path for {parent_label}", ) materialized.append(detail_row) keep_paths.add(detail_row["seo_path_slug"]) _deactivate_stale_intersections(session, group_path_slug=group_path, keep_seo_paths=keep_paths) return materialized def _explicit_start_shopping_items( session: Session, collection_row: Dict[str, Any], *, hub_path_slug: str, ) -> List[Dict[str, Any]]: config = collection_row.get("start_shopping_config") if not isinstance(config, dict) or not isinstance(config.get("items"), list): return [] available = collection_start_shopping_options( session, path_slug=collection_row.get("path_slug"), category_l1=collection_row.get("category_l1"), parent_path_slug=config.get("hub_path_slug") or hub_path_slug, ) by_path = { str(item.get("path_slug")): item for item in available.get("options") or [] if isinstance(item, dict) and str(item.get("path_slug") or "").strip() } explicit: List[Dict[str, Any]] = [] for item in config.get("items") or []: if not isinstance(item, dict): continue l1_path = normalize_plp_path(item.get("l1_path_slug")) option = by_path.get(l1_path) if not option: continue explicit.append( { "label": option.get("label"), "slug": option.get("slug"), "category_path": option.get("path_slug"), "selected_detail_paths": [ normalize_plp_path(path) for path in item.get("l2_path_slugs") or [] if str(path or "").strip() ], } ) return explicit def _detail_options_for_shopping_option( session: Session, hub_path_slug: str, label: str, *, selected_paths: Optional[Iterable[str]] = None, ) -> List[Dict[str, str]]: if normalize_plp_path(hub_path_slug) != "kitchen-cabinets": return [] taxonomy_options = _taxonomy_child_options_for_shopping_option( session, hub_path_slug, label, selected_paths=selected_paths, ) if taxonomy_options: return taxonomy_options if selected_paths: # Explicit Start Shopping L2 selections must come from canonical taxonomy paths, # not from legacy shorthand fallback buckets. return [] return [dict(item) for item in KITCHEN_CABINET_DETAIL_OPTIONS.get(label, [])] def _detail_facet_attribute(label: str) -> Optional[str]: normalized = slugify_segment(label).replace("-", "_") if normalized in {"base_cabinets", "wall_cabinets", "panels_and_fillers", "mouldings", "accessories"}: return normalized return None def _channel_target_remote_id( session: Session, *, path_slug: str, channel_code: str, connection_id: Optional[int] = None, ) -> str: slug = normalize_plp_path(path_slug) if not slug: return "" stmt = ( select(ChannelListingPathTarget.remote_id) .join(ChannelListingPath, ChannelListingPath.id == ChannelListingPathTarget.listing_path_id) .where(ChannelListingPath.path_slug == slug) .where(ChannelListingPathTarget.channel_code == channel_code) .where(ChannelListingPathTarget.is_active.is_(True)) .order_by(ChannelListingPathTarget.id.desc()) ) if connection_id is not None: stmt = stmt.where(ChannelListingPathTarget.connection_id == int(connection_id)) remote_id = session.scalar(stmt.limit(1)) return str(remote_id or "").strip() def _taxonomy_child_options_for_shopping_option( session: Session, hub_path_slug: str, label: str, *, selected_paths: Optional[Iterable[str]] = None, ) -> List[Dict[str, str]]: from db.models import MasterTaxonomyNode parent_path = normalize_plp_path(f"{hub_path_slug}/{slugify_segment(label)}") parent = session.scalar( select(MasterTaxonomyNode) .where(MasterTaxonomyNode.path_slug == parent_path) .where(MasterTaxonomyNode.is_active.is_(True)) .limit(1) ) if parent is None: return [] rows = session.scalars( select(MasterTaxonomyNode) .where(MasterTaxonomyNode.parent_id == int(parent.id)) .where(MasterTaxonomyNode.is_active.is_(True)) .where(MasterTaxonomyNode.node_kind == "category") .order_by(MasterTaxonomyNode.sort_order, MasterTaxonomyNode.name) ).all() if not rows: rows = session.scalars( select(MasterTaxonomyNode) .where(MasterTaxonomyNode.path_slug.like(f"{parent_path}/%")) .where(MasterTaxonomyNode.is_active.is_(True)) .where(MasterTaxonomyNode.node_kind == "category") .order_by(MasterTaxonomyNode.sort_order, MasterTaxonomyNode.name) ).all() by_path: Dict[str, Dict[str, str]] = {} taxonomy_order: List[Dict[str, str]] = [] seen: set[str] = set() parent_depth = len(path_segments(parent_path)) for row in rows: row_path = normalize_plp_path(row.path_slug) segments = path_segments(row_path) if len(segments) != parent_depth + 1: continue slug = segments[-1] if not slug or slug in seen: continue seen.add(slug) item = { "label": _taxonomy_leaf_label(row), "slug": slug, "magento_label": slug.replace("-", " "), "category_path": row_path, } by_path[row_path] = item taxonomy_order.append(item) selected_list = [ normalize_plp_path(path) for path in selected_paths or [] if str(path or "").strip() ] if selected_list: # Preserve explicit Start Shopping L2 array order from start_shopping_config. out: List[Dict[str, str]] = [] seen_selected: set[str] = set() for path in selected_list: item = by_path.get(path) if not item: continue slug = item["slug"] if slug in seen_selected: continue seen_selected.add(slug) out.append(item) return out return taxonomy_order def _taxonomy_leaf_label(row: Any) -> str: name = str(getattr(row, "name", "") or "").strip() if " / " in name: leaf = name.split(" / ")[-1].strip() if leaf: return leaf slug = normalize_plp_path(str(getattr(row, "path_slug", "") or "")) if slug: return slug.rsplit("/", 1)[-1].replace("-", " ").title() return name def _deactivate_stale_intersections( session: Session, *, group_path_slug: str, keep_seo_paths: set[str], ) -> int: group = normalize_plp_path(group_path_slug) rows = session.scalars( select(ListingIntersectionRule).where( ListingIntersectionRule.group_path_slug == group, ListingIntersectionRule.is_active.is_(True), ) ).all() deactivated = 0 for row in rows: if row.seo_path_slug in keep_seo_paths: continue row.is_active = False deactivated += 1 if deactivated: session.flush() return deactivated def deactivate_shopping_facet_rooted_intersections( session: Session, *, dry_run: bool = True, hub_path_slug: Optional[str] = None, note: Optional[str] = None, ) -> Dict[str, Any]: """Deactivate polluted SEO / collection rows under hub/shopping-facet taxonomy. Keeps real product leaves such as ``kitchen-cabinets/accessories/clearance-kit``. Removes invalid Collections-dashboard rows and SEO trees like ``.../clearance-kit/base-cabinets`` and ``.../base-cabinets/double-door-...``. """ from db.master_taxonomy_product_paths import is_shopping_facet_rooted_path from db.models import CollectionLandingPage, MasterTaxonomyNode cleanup_note = note or "deactivate_shopping_facet_rooted_intersections" hub = normalize_plp_path(hub_path_slug) if hub_path_slug else "" samples: List[Dict[str, Any]] = [] would = 0 deactivated_paths = 0 deactivated_rules = 0 deactivated_targets = 0 deactivated_landings = 0 deactivated_taxonomy_nodes = 0 def _sample(row: Dict[str, Any]) -> None: if len(samples) < 40: samples.append(row) def _is_polluted_seo_slug(slug: str) -> bool: parts = [p for p in normalize_plp_path(slug).split("/") if p] # Depth 4+ under accessories/... is never a valid finish-collection / SEO page. return is_shopping_facet_rooted_path(slug) and len(parts) >= 4 path_stmt = ( select(ChannelListingPath) .where(ChannelListingPath.is_active.is_(True)) .order_by(ChannelListingPath.path_slug) ) if hub: path_stmt = path_stmt.where(ChannelListingPath.path_slug.like(f"{hub}/%")) for path in session.scalars(path_stmt).all(): slug = normalize_plp_path(path.path_slug) kind = str(path.path_kind or "").strip().lower() polluted_seo = _is_polluted_seo_slug(slug) polluted_collection = is_shopping_facet_rooted_path(slug) and kind in { "intersection", "collection", } if not (polluted_seo or polluted_collection): continue # Keep real product L3 leaves (hub/facet/detail) even if path_kind drifted. parts = [p for p in slug.split("/") if p] if len(parts) == 3 and kind not in {"intersection", "collection"}: continue _sample( { "path_slug": slug, "path_kind": kind, "action": "deactivate_facet_rooted_listing_path", "reason": "shopping_facet_product_taxonomy_not_collection", } ) if dry_run: would += 1 continue path.is_active = False path.notes = _append_path_note(path.notes, cleanup_note) deactivated_paths += 1 for target in session.scalars( select(ChannelListingPathTarget).where(ChannelListingPathTarget.listing_path_id == path.id) ).all(): if target.is_active: target.is_active = False deactivated_targets += 1 landing_stmt = select(CollectionLandingPage).where(CollectionLandingPage.is_active.is_(True)) if hub: landing_stmt = landing_stmt.where(CollectionLandingPage.path_slug.like(f"{hub}/%")) for landing in session.scalars(landing_stmt).all(): slug = normalize_plp_path(landing.path_slug) if not is_shopping_facet_rooted_path(slug): continue _sample( { "path_slug": slug, "action": "deactivate_facet_rooted_collection_landing", "reason": "shopping_facet_product_taxonomy_not_finish_collection", } ) if dry_run: would += 1 continue landing.is_active = False landing.notes = _append_path_note(landing.notes, cleanup_note) deactivated_landings += 1 node_stmt = select(MasterTaxonomyNode).where(MasterTaxonomyNode.is_active.is_(True)) if hub: node_stmt = node_stmt.where(MasterTaxonomyNode.path_slug.like(f"{hub}/%")) for node in session.scalars(node_stmt).all(): slug = normalize_plp_path(node.path_slug) kind = str(node.node_kind or "").strip().lower() if not is_shopping_facet_rooted_path(slug): continue parts = [p for p in slug.split("/") if p] # Wrong finish-collection nodes under accessories/..., or SEO depth >= 4. if kind == "collection" or len(parts) >= 4: _sample( { "path_slug": slug, "node_kind": kind, "action": "deactivate_facet_rooted_taxonomy_node", } ) if dry_run: would += 1 continue node.is_active = False deactivated_taxonomy_nodes += 1 rule_stmt = select(ListingIntersectionRule).where(ListingIntersectionRule.is_active.is_(True)) if hub: rule_stmt = rule_stmt.where( (ListingIntersectionRule.seo_path_slug.like(f"{hub}/%")) | (ListingIntersectionRule.group_path_slug.like(f"{hub}/%")) ) for rule in session.scalars(rule_stmt).all(): seo = normalize_plp_path(rule.seo_path_slug) group = normalize_plp_path(rule.group_path_slug) if not (is_shopping_facet_rooted_path(seo) or is_shopping_facet_rooted_path(group)): continue if dry_run: if not any(s.get("path_slug") == seo for s in samples): _sample( { "path_slug": seo, "group_path_slug": group, "action": "deactivate_facet_rooted_intersection_rule", } ) would += 1 continue rule.is_active = False deactivated_rules += 1 if not dry_run and ( deactivated_paths or deactivated_rules or deactivated_targets or deactivated_landings or deactivated_taxonomy_nodes ): session.flush() return { "deactivated_paths": deactivated_paths, "deactivated_rules": deactivated_rules, "deactivated_targets": deactivated_targets, "deactivated_landings": deactivated_landings, "deactivated_taxonomy_nodes": deactivated_taxonomy_nodes, "would_deactivate": would, "samples": samples, } def _append_path_note(existing: Optional[str], note: str) -> str: text = str(existing or "").strip() if not text: return note if note in text: return text return f"{text}; {note}" def _magento_target_category_id(collection_row: Dict[str, Any]) -> Optional[str]: channel = collection_row.get("channel") or {} magento = channel.get("magento") if isinstance(channel, dict) else {} if isinstance(magento, dict): remote_id = magento.get("category_id") or magento.get("remote_id") if remote_id: return str(remote_id) targets = collection_row.get("channel_targets") or {} magento_target = targets.get("magento") if isinstance(targets, dict) else None if isinstance(magento_target, dict): remote_id = magento_target.get("remote_id") or magento_target.get("category_id") if remote_id: return str(remote_id) return None def _shopify_target_collection_id(collection_row: Dict[str, Any]) -> Optional[str]: channel = collection_row.get("channel") or {} shopify = channel.get("shopify") if isinstance(channel, dict) else {} if isinstance(shopify, dict): remote_id = shopify.get("collection_id") or shopify.get("remote_id") if remote_id: return str(remote_id) targets = collection_row.get("channel_targets") or {} shopify_target = targets.get("shopify") if isinstance(targets, dict) else None if isinstance(shopify_target, dict): remote_id = shopify_target.get("remote_id") or shopify_target.get("collection_id") if remote_id: return str(remote_id) return None def _shopify_target_handle( collection_row: Dict[str, Any], hub: str, group_path: str, ) -> Optional[str]: channel = collection_row.get("channel") or {} shopify = channel.get("shopify") if isinstance(channel, dict) else {} if isinstance(shopify, dict): handle = shopify.get("handle") or shopify.get("remote_path") if handle: return str(handle) collection = str(collection_row.get("collection") or group_path.split("/")[-1] or "").strip() if collection: return slugify_segment(collection) return slugify_segment(group_path.replace("/", "-")) def shopping_intersections_for_collection(session: Session, path_slug: str) -> List[Dict[str, Any]]: out: List[Dict[str, Any]] = [] collection_path = normalize_plp_path(path_slug) if not collection_path: return out rows = session.scalars( select(ListingIntersectionRule).where( ListingIntersectionRule.is_active.is_(True), ( (ListingIntersectionRule.group_path_slug == collection_path) | (ListingIntersectionRule.group_path_slug.like(f"{collection_path}/%")) ), ) ).all() for row in rows: item = listing_intersection_dict(row) channel = item.get("channel_payload") or {} shopify_url = _shopify_intersection_url(channel) out.append( { "title": item["facet_label"], "label": item["facet_label"], "seo_path": item["seo_path_slug"], "url": build_magento_plp_url(item["seo_path_slug"]), "shopify_url": shopify_url, "filter_type": (item.get("filter_payload") or {}).get("type") or "attribute", "filter_attribute": (item.get("filter_payload") or {}).get("attribute") or item["facet_attribute"], "filter_value": (item.get("filter_payload") or {}).get("value") or item["facet_value"], "filter_slug": item["facet_slug"], "filter_category_path": (item.get("filter_payload") or {}).get("category_path"), "filter_category_id": (item.get("filter_payload") or {}).get("category_id"), "channel": channel, } ) return out def _shopify_intersection_url(channel: Dict[str, Any]) -> Optional[str]: shopify = channel.get("shopify") if isinstance(channel, dict) else None if not isinstance(shopify, dict): return None handle = str(shopify.get("handle") or "").strip() if not handle: return None url = f"/collections/{handle}" filt = shopify.get("filter") if isinstance(shopify.get("filter"), dict) else {} attribute = str(filt.get("attribute") or "").strip() value = str(filt.get("value") or "").strip() if attribute and value: url = f"{url}?{attribute}={value}" return url def listing_intersection_dict(row: ListingIntersectionRule) -> Dict[str, Any]: filter_payload = row.filter_payload or {} return { "id": row.id, "base_path_slug": row.base_path_slug, "group_path_slug": row.group_path_slug, "group_kind": row.group_kind, "facet_attribute": row.facet_attribute, "facet_label": row.facet_label, "facet_value": row.facet_value, "facet_slug": row.facet_slug, "seo_path_slug": row.seo_path_slug, "url": build_plp_url(row.seo_path_slug), "magento_url": build_magento_plp_url(row.seo_path_slug), "title": row.title, "listing_path_id": row.listing_path_id, "target_listing_path_id": row.target_listing_path_id, "filter_type": filter_payload.get("type") or "attribute", "filter_attribute": filter_payload.get("attribute") or row.facet_attribute, "filter_value": filter_payload.get("value") or row.facet_value, "filter_category_path": filter_payload.get("category_path"), "filter_category_id": filter_payload.get("category_id"), "filter_payload": filter_payload, "channel_payload": row.channel_payload or {}, "is_indexable": row.is_indexable, "is_active": row.is_active, "notes": row.notes, } def _ensure_path( session: Session, path_slug: str, *, title: str, path_kind: Optional[str] = None, parent_path_slug: Optional[str] = None, hub_path_slug: Optional[str] = None, facet_terms: Optional[List[str]] = None, notes: Optional[str] = None, ) -> Dict[str, Any]: slug = normalize_plp_path(path_slug) path = session.scalar(select(ChannelListingPath).where(ChannelListingPath.path_slug == slug)) if path is not None: changed = False if path_kind and path.path_kind != path_kind: path.path_kind = path_kind changed = True if parent_path_slug and path.parent_path_slug != normalize_plp_path(parent_path_slug): path.parent_path_slug = normalize_plp_path(parent_path_slug) changed = True if hub_path_slug and path.hub_path_slug != normalize_plp_path(hub_path_slug): path.hub_path_slug = normalize_plp_path(hub_path_slug) changed = True if facet_terms: existing = path.facet_terms if isinstance(path.facet_terms, dict) else {} merged = {**existing, "terms": facet_terms} if merged != path.facet_terms: path.facet_terms = merged changed = True if changed: session.flush() return { "id": path.id, "path_slug": path.path_slug, "title": path.title, } return upsert_listing_path( session, path_slug=path_slug, title=title, path_kind=path_kind, parent_path_slug=parent_path_slug, hub_path_slug=hub_path_slug, facet_terms=facet_terms, notes=notes, )