from __future__ import annotations from decimal import Decimal import re from typing import Any, Dict, Iterable, List, Optional from sqlalchemy import func, select from sqlalchemy.orm import Session from channel.url_canonical import normalize_plp_path from db.channel_listing_path import ( count_active_listing_assignments_except, deactivate_listing_assignments_except, upsert_listing_path, ) from db.collection_channel_placement import effective_parent_path_slug from db.models import ( ChannelAttributeAlias, ChannelListingPath, CollectionCommerceProfile, CollectionLandingPage, MasterAttributeDefinition, MasterProduct, MasterProductAttributeValue, MasterProductLocationAvailability, MasterProductPrice, MasterProductRelation, MasterTaxonomyNode, ) from db.collection_landing_schema import EMPTY_BADGES_PAYLOAD, normalize_collection_badges from db.shopping_facet_catalog import COLLECTION_AVAILABILITY_OPTIONS from db.source_imports import normalize_column_key DEFAULT_CATEGORY_L1 = "Kitchen Cabinets" COLLECTION_PAGE = "collection" START_SHOPPING_L1_KEYS = { "base-cabinets", "wall-cabinets", "tall-cabinets", "panels-and-fillers", "mouldings", "accessories", } START_SHOPPING_ATTRIBUTE_DEFINITIONS = ( ("shopping_collection", "Shopping Collection"), ("shopping_l1", "Shopping Level 1"), ("shopping_l2", "Shopping Level 2"), ) START_SHOPPING_SYNC_FLAG = "sync_options_from_master" PASTED_COLLECTION_TEXT_FIELDS = ( "subtitle", "short_description", "description_html", "specifications_html", "materials_html", "assembly_html", ) def canonical_category(value: Optional[str]) -> Optional[str]: text = _clean(value) if not text: return None lowered = _norm(text) aliases = { "kitchen cabinet": "Kitchen Cabinets", "kitchen cabinets": "Kitchen Cabinets", "panels & fillers": "Panels and Fillers", "panels and fillers": "Panels and Fillers", } return aliases.get(lowered, text) def canonical_collection(value: Optional[str]) -> Optional[str]: text = _clean(value) if not text: return None replacements = { "grey": "gray", } words = [replacements.get(word.lower(), word) for word in text.split()] fixed = " ".join(words) title_aliases = { "anna stone gray": "Anna Stone Gray", "anna snow white": "Anna Snow White", "anna caramel harvest": "Anna Caramel Harvest", } return title_aliases.get(_norm(fixed), fixed) def collection_path_slug(category_l1: Optional[str], collection: Optional[str]) -> str: category = canonical_category(category_l1) or DEFAULT_CATEGORY_L1 coll = canonical_collection(collection) if not coll: return normalize_plp_path(category) return normalize_plp_path(f"{category}/{coll}") def list_collection_landing_pages( session: Session, *, search: Optional[str] = None, category_l1: Optional[str] = None, include_inferred: bool = True, limit: int = 500, ) -> List[Dict[str, Any]]: metadata_rows = _metadata_by_slug(session, search=search, category_l1=category_l1, limit=limit) rows = dict(metadata_rows) if include_inferred: for inferred in infer_collection_pages(session, search=search, category_l1=category_l1, limit=limit): if not _taxonomy_backed_collection_slug(session, inferred["path_slug"]): continue existing = rows.get(inferred["path_slug"]) if existing: existing["sku_count"] = inferred["sku_count"] existing["computed_starting_price"] = inferred.get("computed_starting_price") existing["inferred"] = False else: rows[inferred["path_slug"]] = inferred rows = _collapse_collections_by_canonical_slug(session, rows) from db.master_taxonomy_product_paths import is_shopping_facet_rooted_path # Product taxonomy under shopping facets is not a finish collection landing. rows = { slug: item for slug, item in rows.items() if not is_shopping_facet_rooted_path(str(item.get("path_slug") or slug)) } for item in rows.values(): _attach_shopping_intersections(session, item) return sorted( rows.values(), key=lambda item: (int(item.get("sort_order") or 100), str(item.get("title") or item.get("path_slug") or "")), )[:limit] def get_collection_landing_page(session: Session, path_slug: str) -> Optional[Dict[str, Any]]: slug = normalize_plp_path(path_slug) if not slug: return None from db.master_taxonomy_merge import resolve_canonical_path_slug slug = resolve_canonical_path_slug(session, slug) or slug row = session.scalar(select(CollectionLandingPage).where(CollectionLandingPage.path_slug == slug)) if row: data = _row_dict(row) data.update(_collection_stats(session, data.get("category_l1"), data.get("collection"))) _attach_shopping_intersections(session, data) return data category, collection = _split_collection_slug(slug) if not collection: return None inferred = _inferred_dict(session, category, collection) _attach_shopping_intersections(session, inferred) return inferred def upsert_collection_landing_page(session: Session, payload: Dict[str, Any]) -> Dict[str, Any]: category = canonical_category(payload.get("category_l1")) or DEFAULT_CATEGORY_L1 collection = canonical_collection(payload.get("collection") or payload.get("title")) slug = normalize_plp_path(payload.get("path_slug") or collection_path_slug(category, collection)) if not slug: raise ValueError("path_slug or collection is required") title = _clean(payload.get("title")) or collection or slug row = session.scalar(select(CollectionLandingPage).where(CollectionLandingPage.path_slug == slug)) data = { "path_slug": slug, "page_type": str(payload.get("page_type") or COLLECTION_PAGE).strip() or COLLECTION_PAGE, "category_l1": category, "collection": collection, "brand": _clean(payload.get("brand")), "title": title, "subtitle": _clean_collection_text(payload.get("subtitle"), field_name="subtitle"), "short_description": _clean_collection_text(payload.get("short_description"), field_name="short_description"), "description_html": _clean_collection_text(payload.get("description_html"), field_name="description_html"), "specifications_html": _clean_collection_text(payload.get("specifications_html"), field_name="specifications_html"), "materials_html": _clean_collection_text(payload.get("materials_html"), field_name="materials_html"), "assembly_html": _clean_collection_text(payload.get("assembly_html"), field_name="assembly_html"), "pdf_url": _clean(payload.get("pdf_url")), "thumbnail_image_url": _clean(payload.get("thumbnail_image_url")), "category_icon_url": _clean(payload.get("category_icon_url")), "hero_images": _json_list(payload.get("hero_images")), "badges": normalize_collection_badges(payload.get("badges")), "feature_bullets": _json_list(payload.get("feature_bullets")), "cta_rows": _json_list(payload.get("cta_rows")), "matching_collection_slugs": _json_list(payload.get("matching_collection_slugs")), "sku_prefixes": _json_list(payload.get("sku_prefixes")), "start_shopping_config": _normalize_start_shopping_config(payload.get("start_shopping_config")), "parent_path_slug": normalize_plp_path(payload.get("parent_path_slug")) if _clean(payload.get("parent_path_slug")) else None, "starting_price": _optional_decimal(payload.get("starting_price")), "compare_at_price": _optional_decimal(payload.get("compare_at_price")), "sort_order": int(payload.get("sort_order") or 100), "is_active": bool(payload.get("is_active", True)), "notes": _clean(payload.get("notes")), } if payload.get("collection_registry_id") is not None: data["collection_registry_id"] = int(payload["collection_registry_id"]) if row is None: row = CollectionLandingPage(**data) session.add(row) else: for key, value in data.items(): setattr(row, key, value) session.flush() if payload.get("collection_registry_id") is None: from db.collection_taxonomy_bridge import link_collection_registry_to_landing try: link_collection_registry_to_landing(session, path_slug=slug) except ValueError: pass result = _row_dict(row) result.update(_collection_stats(session, result.get("category_l1"), result.get("collection"))) from db.listing_intersections import sync_shopping_intersections_for_collection from db.master_taxonomy_product_paths import is_shopping_facet_rooted_path # Never invent Start Shopping SEO under accessories/clearance-kit/... if not is_shopping_facet_rooted_path(slug): sync_shopping_intersections_for_collection(session, result) _attach_shopping_intersections(session, result) return result def apply_collection_to_skus( session: Session, *, path_slug: Optional[str] = None, category_l1: Optional[str] = None, collection: Optional[str] = None, dry_run: bool = True, only_assigned: bool = False, ) -> Dict[str, Any]: from db.collection_sku_mapping import resolve_collection_skus from db.master_taxonomy_merge import resolve_canonical_path_slug if path_slug: existing = get_collection_landing_page(session, path_slug) if existing: category_l1 = category_l1 or existing.get("category_l1") collection = collection or existing.get("collection") category = canonical_category(category_l1) or DEFAULT_CATEGORY_L1 coll = canonical_collection(collection) if not coll: raise ValueError("collection is required") slug = resolve_canonical_path_slug(session, path_slug) if path_slug else None slug = slug or collection_path_slug(category, coll) resolved = resolve_collection_skus( session, path_slug=slug, category_l1=category, collection=coll, ) sku_items = resolved.get("skus") or [] master_skus = {item["master_sku"] for item in sku_items} parent_slug = None if path_slug: saved = session.scalar( select(CollectionLandingPage).where(CollectionLandingPage.path_slug == slug).limit(1) ) if saved: parent_slug = saved.parent_path_slug parent_slug = effective_parent_path_slug(category_l1=category, parent_path_slug=parent_slug) if dry_run: path = session.scalar( select(ChannelListingPath).where(ChannelListingPath.path_slug == slug).limit(1) ) would_deactivate = 0 if path is not None: would_deactivate = count_active_listing_assignments_except( session, listing_path_id=path.id, keep_skus=master_skus, ) return { "path_slug": slug, "category_l1": category, "collection": coll, "parent_path_slug": parent_slug, "dry_run": True, "sku_count": len(master_skus), "assignment_count": len(master_skus), "would_upsert": len(master_skus), "would_deactivate": would_deactivate, "sku_prefixes": resolved.get("sku_prefixes") or [], "samples": sku_items[:25], } upsert_listing_path( session, path_slug=slug, title=coll, path_kind="collection", parent_path_slug=parent_slug, hub_path_slug=normalize_plp_path(category), notes="collection landing page", ) path = session.scalar( select(ChannelListingPath).where(ChannelListingPath.path_slug == slug).limit(1) ) if path is None: raise ValueError(f"Listing path not found after upsert: {slug}") from db.channel_listing_path import _upsert_listing_assignment upserted = 0 for item in sku_items: _upsert_listing_assignment( session, master_sku=item["master_sku"], listing_path_id=path.id, match_source=item.get("match_source") or "collection_apply", ) upserted += 1 deactivated = deactivate_listing_assignments_except( session, listing_path_id=path.id, keep_skus=master_skus, ) return { "path_slug": slug, "category_l1": category, "collection": coll, "parent_path_slug": parent_slug, "dry_run": dry_run, "sku_count": len(master_skus), "assignment_count": len(master_skus), "upserted": upserted, "deactivated": deactivated, "sku_prefixes": resolved.get("sku_prefixes") or [], "samples": sku_items[:25], } def _matching_collection_skus(session: Session, category_l1: str, collection: str) -> List[str]: stmt = ( select(MasterProduct.sku) .where(MasterProduct.is_active.is_(True)) .where(MasterProduct.category_l1.ilike(category_l1)) .where(MasterProduct.collection.ilike(collection)) .order_by(MasterProduct.sku) ) return [str(sku) for sku in session.scalars(stmt).all()] def infer_collection_pages( session: Session, *, search: Optional[str] = None, category_l1: Optional[str] = None, limit: int = 500, ) -> List[Dict[str, Any]]: category = canonical_category(category_l1) stmt = ( select( MasterProduct.category_l1, MasterProduct.collection, MasterProduct.brand, func.count(MasterProduct.sku).label("sku_count"), func.min(MasterProductPrice.amount).label("starting_price"), ) .join(MasterProductPrice, MasterProductPrice.product_id == MasterProduct.id, isouter=True) .where(MasterProduct.is_active.is_(True)) .where(MasterProduct.collection.is_not(None)) .where(MasterProduct.collection != "") .group_by(MasterProduct.category_l1, MasterProduct.collection, MasterProduct.brand) .order_by(MasterProduct.category_l1, MasterProduct.collection) .limit(limit) ) if category: stmt = stmt.where(MasterProduct.category_l1.ilike(category)) if search: pattern = f"%{search.strip()}%" stmt = stmt.where( (MasterProduct.collection.ilike(pattern)) | (MasterProduct.category_l1.ilike(pattern)) | (MasterProduct.brand.ilike(pattern)) ) out: Dict[str, Dict[str, Any]] = {} from db.master_taxonomy_merge import resolve_canonical_path_slug for row in session.execute(stmt).all(): row_category = canonical_category(row.category_l1) or DEFAULT_CATEGORY_L1 row_collection = canonical_collection(row.collection) if not row_collection: continue slug = resolve_canonical_path_slug( session, collection_path_slug(row_category, row_collection), ) or collection_path_slug(row_category, row_collection) from db.collection_path_guard import is_polluted_hub_collection_path if is_polluted_hub_collection_path(session, slug): continue item = out.setdefault( slug, { "id": None, "path_slug": slug, "plp_url": f"/{slug}.html", "page_type": COLLECTION_PAGE, "category_l1": row_category, "collection": row_collection, "brand": row.brand, "title": row_collection, "sort_order": 100, "is_active": True, "inferred": True, "sku_count": 0, "computed_starting_price": None, }, ) item["sku_count"] += int(row.sku_count or 0) if row.starting_price is not None: current = item.get("computed_starting_price") price = float(row.starting_price) item["computed_starting_price"] = price if current is None else min(float(current), price) return list(out.values()) def _taxonomy_backed_collection_slug(session: Session, path_slug: str) -> bool: """True when an active taxonomy node exists for this collection path.""" from db.master_taxonomy_merge import resolve_canonical_path_slug from db.master_taxonomy_product_paths import is_shopping_facet_rooted_path from db.models import MasterTaxonomyNode canonical = resolve_canonical_path_slug(session, path_slug) or normalize_plp_path(path_slug) if not canonical or is_shopping_facet_rooted_path(canonical): return False node = session.scalar( select(MasterTaxonomyNode) .where(MasterTaxonomyNode.path_slug == canonical) .where(MasterTaxonomyNode.is_active.is_(True)) .limit(1) ) return node is not None and node.node_kind in {"collection", "category"} def _collapse_collections_by_canonical_slug( session: Session, rows: Dict[str, Dict[str, Any]], ) -> Dict[str, Dict[str, Any]]: from db.master_taxonomy_merge import resolve_canonical_node, resolve_canonical_path_slug collapsed: Dict[str, Dict[str, Any]] = {} for slug, item in rows.items(): if not item.get("is_active", True): continue canonical = resolve_canonical_path_slug(session, item.get("path_slug") or slug) or slug existing = collapsed.get(canonical) if existing is None: merged = dict(item) merged["path_slug"] = canonical merged["plp_url"] = f"/{canonical}.html" if canonical != (item.get("path_slug") or slug): merged["merged_from_slugs"] = [item.get("path_slug") or slug] collapsed[canonical] = merged continue collapsed[canonical] = _merge_collection_rows(existing, item, canonical_slug=canonical) for canonical, item in collapsed.items(): node = resolve_canonical_node(session, path_slug=canonical) if node is None: continue coll = canonical_collection(node.name) if coll: item["collection"] = coll if not item.get("title") or item.get("inferred"): item["title"] = coll return collapsed def _merge_collection_rows( primary: Dict[str, Any], secondary: Dict[str, Any], *, canonical_slug: str, ) -> Dict[str, Any]: winner, loser = primary, secondary if _collection_row_preference_score(secondary, canonical_slug) > _collection_row_preference_score( primary, canonical_slug ): winner, loser = secondary, primary merged = dict(winner) merged["path_slug"] = canonical_slug merged["plp_url"] = f"/{canonical_slug}.html" merged_slugs = list(merged.get("merged_from_slugs") or []) for slug in (primary.get("path_slug"), secondary.get("path_slug")): if slug and slug != canonical_slug and slug not in merged_slugs: merged_slugs.append(slug) if merged_slugs: merged["merged_from_slugs"] = merged_slugs merged["sku_count"] = int(merged.get("sku_count") or 0) + int(loser.get("sku_count") or 0) for price_key in ("computed_starting_price", "starting_price"): a = merged.get(price_key) b = loser.get(price_key) if a is None: merged[price_key] = b elif b is not None: merged[price_key] = min(float(a), float(b)) for field in ( "description_html", "thumbnail_image_url", "category_icon_url", "hero_images", "subtitle", "short_description", "brand", "collection_registry_id", ): if not merged.get(field) and loser.get(field): merged[field] = loser[field] merged["inferred"] = bool(merged.get("inferred")) and bool(loser.get("inferred")) return merged def _collection_row_preference_score(item: Dict[str, Any], canonical_slug: str) -> int: score = 0 if item.get("path_slug") == canonical_slug: score += 20 if not item.get("inferred"): score += 10 if item.get("id"): score += 5 if item.get("description_html") or item.get("hero_images"): score += 3 return score def _metadata_by_slug( session: Session, *, search: Optional[str], category_l1: Optional[str], limit: int, ) -> Dict[str, Dict[str, Any]]: stmt = ( select(CollectionLandingPage) .where(CollectionLandingPage.is_active.is_(True)) .order_by(CollectionLandingPage.sort_order, CollectionLandingPage.title) ) if category_l1: stmt = stmt.where(CollectionLandingPage.category_l1.ilike(canonical_category(category_l1) or category_l1)) if search: pattern = f"%{search.strip()}%" stmt = stmt.where( (CollectionLandingPage.title.ilike(pattern)) | (CollectionLandingPage.collection.ilike(pattern)) | (CollectionLandingPage.brand.ilike(pattern)) | (CollectionLandingPage.path_slug.ilike(pattern)) ) if limit: stmt = stmt.limit(limit) out: Dict[str, Dict[str, Any]] = {} for row in session.scalars(stmt).all(): data = _row_dict(row) data.update(_collection_stats(session, data.get("category_l1"), data.get("collection"))) _attach_shopping_intersections(session, data) out[data["path_slug"]] = data return out def _attach_shopping_intersections(session: Session, data: Dict[str, Any]) -> None: path_slug = data.get("path_slug") if not path_slug: return from db.listing_intersections import shopping_intersections_for_collection data["shopping_intersections"] = shopping_intersections_for_collection(session, str(path_slug)) _attach_collection_commerce(session, data) def _attach_collection_commerce(session: Session, data: Dict[str, Any]) -> None: category = data.get("category_l1") collection = data.get("collection") if not collection: return profile = _collection_commerce_profile( session, category_l1=category, collection=collection, collection_registry_id=data.get("collection_registry_id"), ) sample = _sample_product_for_collection(session, category, collection, profile=profile) if sample: data["sample_product"] = sample cta_rows = list(data.get("cta_rows") or []) if not any(str(row.get("action") or "").lower() == "order_sample" for row in cta_rows if isinstance(row, dict)): cta_rows.append( { "label": str((profile.sample_cta_label if profile else None) or sample.get("label") or "Order Sample"), "action": "order_sample", "sku": sample["sku"], "product_name": sample.get("name"), } ) if profile and profile.sample_category_path_slug: data["sample_category_path_slug"] = profile.sample_category_path_slug data["cta_rows"] = cta_rows if profile and profile.appointment_url: cta_rows = list(data.get("cta_rows") or []) if not any(str(row.get("action") or "").lower() == "book_appointment" for row in cta_rows if isinstance(row, dict)): cta_rows.append( { "label": str(profile.appointment_cta_label or "Book Appointment"), "action": "book_appointment", "url": profile.appointment_url, } ) data["cta_rows"] = cta_rows if profile and profile.landing_metadata_json: data["landing_metadata"] = profile.landing_metadata_json data["availability_filters"] = _availability_filters_for_collection(session, category, collection, profile=profile) def _collection_stats(session: Session, category_l1: Optional[str], collection: Optional[str]) -> Dict[str, Any]: category = canonical_category(category_l1) coll = canonical_collection(collection) if not coll: return {"sku_count": 0, "computed_starting_price": None} stmt = ( select(func.count(MasterProduct.sku), func.min(MasterProductPrice.amount)) .join(MasterProductPrice, MasterProductPrice.product_id == MasterProduct.id, isouter=True) .where(MasterProduct.is_active.is_(True)) .where(MasterProduct.collection.ilike(coll)) ) if category: stmt = stmt.where(MasterProduct.category_l1.ilike(category)) sku_count, starting_price = session.execute(stmt).one() return { "sku_count": int(sku_count or 0), "computed_starting_price": float(starting_price) if starting_price is not None else None, } def _collection_product_stmt(category_l1: Optional[str], collection: Optional[str]): category = canonical_category(category_l1) coll = canonical_collection(collection) stmt = select(MasterProduct).where(MasterProduct.is_active.is_(True)) if category: stmt = stmt.where(MasterProduct.category_l1.ilike(category)) if coll: stmt = stmt.where(MasterProduct.collection.ilike(coll)) return stmt def _sample_product_for_collection( session: Session, category_l1: Optional[str], collection: Optional[str], *, profile: Optional[CollectionCommerceProfile] = None, ) -> Optional[Dict[str, Any]]: coll = canonical_collection(collection) if not coll: return None if profile and profile.sample_master_sku: sample = session.scalar( select(MasterProduct) .where(MasterProduct.sku == str(profile.sample_master_sku).strip()) .where(MasterProduct.is_active.is_(True)) .limit(1) ) if sample is not None: return { "master_sku": sample.sku, "channel_sku": sample.sku, "sku": sample.sku, "name": sample.name, "label": str(profile.sample_cta_label or "Order Sample"), "collection": canonical_collection(sample.collection), "brand": sample.brand, "sample_category_path_slug": profile.sample_category_path_slug, } rows = session.scalars(_collection_product_stmt(category_l1, coll)).all() candidates = [] for product in rows: text = f"{product.sku or ''} {product.name or ''}".lower() if ( str(product.sku or "").upper().endswith("-SD") or str(product.sku or "").upper().endswith("SD") or "sample door" in text ): candidates.append(product) if not candidates: return None sample = sorted(candidates, key=lambda item: (0 if str(item.sku or "").upper().endswith("-SD") else 1, item.sku))[0] return { "master_sku": sample.sku, "channel_sku": sample.sku, "sku": sample.sku, "name": sample.name, "label": "Order Sample", "collection": canonical_collection(sample.collection), "brand": sample.brand, } def _availability_filters_for_collection( session: Session, category_l1: Optional[str], collection: Optional[str], *, profile: Optional[CollectionCommerceProfile] = None, ) -> List[Dict[str, Any]]: products = list(session.scalars(_collection_product_stmt(category_l1, collection)).all()) if not products: return [] product_ids = [int(product.id) for product in products] attr_rows = session.execute( select(MasterProductAttributeValue.product_id, MasterProductAttributeValue.attribute_code, MasterProductAttributeValue.value) .where(MasterProductAttributeValue.product_id.in_(product_ids)) ).all() text_by_product: Dict[int, str] = {int(product.id): f"{product.sku or ''} {product.name or ''}" for product in products} for product_id, code, value in attr_rows: text_by_product[int(product_id)] = f"{text_by_product.get(int(product_id), '')} {code or ''} {value or ''}" available_ids = { int(product_id) for product_id in session.scalars( select(MasterProductLocationAvailability.product_id) .where(MasterProductLocationAvailability.product_id.in_(product_ids)) .where(MasterProductLocationAvailability.is_available.is_(True)) .distinct() ).all() } counts = {"grab_and_go": 0, "available": 0, "special_order": 0} for product in products: text = _norm(text_by_product.get(int(product.id), "")) grab_text = text.replace("&", " and ") if "grab go" in grab_text or "grab and go" in _norm(grab_text): counts["grab_and_go"] += 1 if int(product.id) in available_ids or "available" in text: counts["available"] += 1 if "special order" in text: counts["special_order"] += 1 out: List[Dict[str, Any]] = [] for option in COLLECTION_AVAILABILITY_OPTIONS: count = counts.get(str(option["value"]), 0) if count <= 0: continue out.append({**option, "count": count}) explicit_modes = _profile_availability_modes(profile) if explicit_modes: existing = {str(item.get("value")): item for item in out} sku_count = len(products) for option in COLLECTION_AVAILABILITY_OPTIONS: value = str(option["value"]) if value not in explicit_modes: continue if value in existing: existing[value]["count"] = max(int(existing[value].get("count") or 0), sku_count) else: out.append({**option, "count": sku_count}) return out def _collection_commerce_profile( session: Session, *, category_l1: Optional[str], collection: Optional[str], collection_registry_id: Optional[int] = None, ) -> Optional[CollectionCommerceProfile]: coll = canonical_collection(collection) if not coll: return None stmt = select(CollectionCommerceProfile).where(CollectionCommerceProfile.is_active.is_(True)) if collection_registry_id: row = session.scalar( stmt.where(CollectionCommerceProfile.collection_registry_id == int(collection_registry_id)) .order_by(CollectionCommerceProfile.id.desc()) .limit(1) ) if row is not None: return row category = canonical_category(category_l1) stmt = stmt.where(CollectionCommerceProfile.collection_name.ilike(coll)) if category: stmt = stmt.where( (CollectionCommerceProfile.category_l1.is_(None)) | (CollectionCommerceProfile.category_l1.ilike(category)) ) return session.scalar(stmt.order_by(CollectionCommerceProfile.id.desc()).limit(1)) def _profile_availability_modes(profile: Optional[CollectionCommerceProfile]) -> set[str]: raw = profile.availability_modes_json if profile is not None else None if isinstance(raw, dict): values = raw.get("modes") if isinstance(raw.get("modes"), list) else raw.get("items") if isinstance(values, list): return {str(value).strip() for value in values if str(value).strip()} if isinstance(raw, list): return {str(value).strip() for value in raw if str(value).strip()} return set() def _inferred_dict(session: Session, category: str, collection: str) -> Dict[str, Any]: slug = collection_path_slug(category, collection) data = { "id": None, "path_slug": slug, "plp_url": f"/{slug}.html", "page_type": COLLECTION_PAGE, "category_l1": canonical_category(category), "collection": canonical_collection(collection), "title": canonical_collection(collection) or collection, "is_active": True, "sort_order": 100, "inferred": True, } data.update(_collection_stats(session, data["category_l1"], data["collection"])) return data def _split_collection_slug(slug: str) -> tuple[str, Optional[str]]: parts = [part for part in normalize_plp_path(slug).split("/") if part] if len(parts) < 2: return (parts[0] if parts else DEFAULT_CATEGORY_L1), None return parts[0].replace("-", " ").title(), parts[-1].replace("-", " ").title() def collection_start_shopping_options( session: Session, *, path_slug: Optional[str] = None, category_l1: Optional[str] = None, parent_path_slug: Optional[str] = None, ) -> Dict[str, Any]: row = get_collection_landing_page(session, path_slug) if path_slug else None category = canonical_category(category_l1 or (row or {}).get("category_l1")) or DEFAULT_CATEGORY_L1 hub_path = normalize_plp_path(parent_path_slug or (row or {}).get("parent_path_slug") or category) if "/" not in hub_path: hub_path = normalize_plp_path(category) hub = session.scalar( select(MasterTaxonomyNode) .where(MasterTaxonomyNode.path_slug == hub_path) .where(MasterTaxonomyNode.is_active.is_(True)) .limit(1) ) options: List[Dict[str, Any]] = [] if hub is not None: l1_nodes = session.scalars( select(MasterTaxonomyNode) .where(MasterTaxonomyNode.parent_id == hub.id) .where(MasterTaxonomyNode.is_active.is_(True)) .where(MasterTaxonomyNode.node_kind == "category") .order_by(MasterTaxonomyNode.sort_order, MasterTaxonomyNode.name) ).all() for node in l1_nodes: node_path = normalize_plp_path(node.path_slug) node_slug = node_path.split("/")[-1] if node_slug not in START_SHOPPING_L1_KEYS: continue children = session.scalars( select(MasterTaxonomyNode) .where(MasterTaxonomyNode.parent_id == node.id) .where(MasterTaxonomyNode.is_active.is_(True)) .where(MasterTaxonomyNode.node_kind == "category") .order_by(MasterTaxonomyNode.sort_order, MasterTaxonomyNode.name) ).all() options.append( { "path_slug": node_path, "slug": node_slug, "label": node.name, "children": [ { "path_slug": normalize_plp_path(child.path_slug), "slug": normalize_plp_path(child.path_slug).split("/")[-1], "label": child.name, } for child in children ], } ) return { "hub_path_slug": hub_path, "category_l1": category, "options": options, "selected": _normalize_start_shopping_config((row or {}).get("start_shopping_config")), } def apply_start_shopping_config_to_collections( session: Session, *, parent_path_slug: str, collection_path_slugs: List[str], start_shopping_config: Any, category_l1: Optional[str] = None, ) -> Dict[str, Any]: parent_slug = normalize_plp_path(parent_path_slug) if not parent_slug: raise ValueError("parent_path_slug is required") parent_node = session.scalar( select(MasterTaxonomyNode) .where(MasterTaxonomyNode.path_slug == parent_slug) .where(MasterTaxonomyNode.is_active.is_(True)) .limit(1) ) if parent_node is None: raise ValueError(f"Parent taxonomy path not found: {parent_slug}") normalized = _normalize_start_shopping_config(start_shopping_config) normalized["hub_path_slug"] = parent_slug seen: set[str] = set() applied: List[Dict[str, Any]] = [] skipped: List[str] = [] for raw_slug in collection_path_slugs or []: slug = normalize_plp_path(raw_slug) if not slug or slug in seen: continue seen.add(slug) node = session.scalar( select(MasterTaxonomyNode) .where(MasterTaxonomyNode.path_slug == slug) .where(MasterTaxonomyNode.is_active.is_(True)) .limit(1) ) if node is None or str(node.node_kind or "") != "collection": skipped.append(slug) continue existing = get_collection_landing_page(session, slug) or {} inferred_category, inferred_collection = _split_collection_slug(slug) payload = dict(existing) payload.update( { "path_slug": slug, "category_l1": canonical_category( category_l1 or existing.get("category_l1") or inferred_category ) or DEFAULT_CATEGORY_L1, "collection": canonical_collection( existing.get("collection") or inferred_collection or node.name ) or node.name, "title": _clean(existing.get("title")) or node.name or inferred_collection or slug, "parent_path_slug": parent_slug, "start_shopping_config": normalized, "is_active": bool(existing.get("is_active", True)), "sort_order": int(existing.get("sort_order") or 100), } ) if node.collection_registry_id and not payload.get("collection_registry_id"): payload["collection_registry_id"] = int(node.collection_registry_id) applied.append(upsert_collection_landing_page(session, payload)) return { "parent_path_slug": parent_slug, "start_shopping_config": normalized, "applied_count": len(applied), "skipped_count": len(skipped), "collections": applied, "skipped": skipped, } def ensure_start_shopping_master_attributes(session: Session) -> int: existing = { normalize_column_key(code) for (code,) in session.execute( select(MasterAttributeDefinition.attribute_code).where( MasterAttributeDefinition.attribute_code.in_( [code for code, _label in START_SHOPPING_ATTRIBUTE_DEFINITIONS] ) ) ).all() } created = 0 for code, label in START_SHOPPING_ATTRIBUTE_DEFINITIONS: if code in existing: continue session.add( MasterAttributeDefinition( attribute_code=code, label=label, purpose="taxonomy", data_type="select", is_required=False, is_active=True, notes="Start Shopping intersection attribute", ) ) created += 1 for code, label in START_SHOPPING_ATTRIBUTE_DEFINITIONS: row = session.scalar( select(MasterAttributeDefinition).where(MasterAttributeDefinition.attribute_code == code).limit(1) ) if row is None: continue row.label = label row.purpose = "taxonomy" row.data_type = "select" row.is_active = True row.notes = "Start Shopping intersection attribute" return created def ensure_start_shopping_magento_aliases(session: Session) -> int: existing_rows = { ( normalize_column_key(row.canonical_code), normalize_column_key(row.channel_attribute_code), ): row for row in session.scalars( select(ChannelAttributeAlias).where( ChannelAttributeAlias.channel_code == "magento", ChannelAttributeAlias.canonical_code.in_( [code for code, _label in START_SHOPPING_ATTRIBUTE_DEFINITIONS] ), ) ).all() } created = 0 for code, label in START_SHOPPING_ATTRIBUTE_DEFINITIONS: key = (code, code) row = existing_rows.get(key) if row is not None: row.is_active = True row.mapping_scope = row.mapping_scope or "dynamic" row.action = "create_new" row.data_type = "select" row.transform_rule = {START_SHOPPING_SYNC_FLAG: True} row.notes = f"Auto-created for Start Shopping ({label})" continue session.add( ChannelAttributeAlias( channel_code="magento", canonical_code=code, channel_attribute_code=code, mapping_scope="dynamic", action="create_new", data_type="select", transform_rule={START_SHOPPING_SYNC_FLAG: True}, is_active=True, notes=f"Auto-created for Start Shopping ({label})", ) ) created += 1 return created def apply_start_shopping_to_master_skus( session: Session, *, collection_path_slugs: List[str], dry_run: bool = False, ) -> Dict[str, Any]: from db.collection_sku_mapping import resolve_collection_skus from db.master_taxonomy_intent import _NodeIndex ensure_start_shopping_master_attributes(session) selected_slugs: List[str] = [] seen: set[str] = set() for raw_slug in collection_path_slugs or []: slug = normalize_plp_path(raw_slug) if not slug or slug in seen: continue seen.add(slug) selected_slugs.append(slug) if not selected_slugs: raise ValueError("At least one collection_path_slug is required") sku_values: Dict[str, Dict[str, Optional[str]]] = {} collection_configs: Dict[str, List[Dict[str, Any]]] = {} processed: List[Dict[str, Any]] = [] skipped: List[Dict[str, Any]] = [] conflicts: List[Dict[str, Any]] = [] for slug in selected_slugs: row = get_collection_landing_page(session, slug) if not row: skipped.append({"path_slug": slug, "reason": "collection landing page not found"}) continue config = _normalize_start_shopping_config(row.get("start_shopping_config")) items = config.get("items") or [] if not items: skipped.append({"path_slug": slug, "reason": "no saved Start Shopping config"}) continue collection_configs[slug] = items resolved = resolve_collection_skus( session, path_slug=slug, category_l1=canonical_category(row.get("category_l1")) or DEFAULT_CATEGORY_L1, collection=canonical_collection(row.get("collection")), ) sku_items = resolved.get("skus") or [] for sku_item in sku_items: sku = str(sku_item.get("master_sku") or "").strip() if not sku: continue slot = sku_values.setdefault( sku, { "__shopping_collection_path_slug": None, "shopping_collection": None, "shopping_l1": None, "shopping_l2": None, }, ) slot["__shopping_collection_path_slug"] = slug _assign_start_shopping_scalar_value( slot, "shopping_collection", _relative_shopping_collection_value(slug), sku=sku, conflicts=conflicts, ) processed.append( { "path_slug": slug, "sku_count": len({str(item.get('master_sku') or '').strip() for item in sku_items if item.get('master_sku')}), "item_count": len(items), } ) if not sku_values: return { "collection_count": len(processed), "sku_count": 0, "updated_products": 0, "updated_values": 0, "dry_run": dry_run, "processed": processed, "skipped": skipped, } products = { str(row.sku): row for row in session.scalars( select(MasterProduct).where(MasterProduct.sku.in_(sorted(sku_values.keys()))) ).all() } attrs_by_sku: Dict[str, Dict[str, str]] = {} product_ids = [int(product.id) for product in products.values() if getattr(product, "id", None)] if product_ids: for value in session.scalars( select(MasterProductAttributeValue).where(MasterProductAttributeValue.product_id.in_(product_ids)) ).all(): if value.value is None or value.value == "": continue attrs_by_sku.setdefault(str(value.sku), {})[normalize_column_key(value.attribute_code)] = str(value.value).strip() taxonomy_index = _NodeIndex.load(session) for sku, values in sku_values.items(): product = products.get(sku) if product is None: continue collection_path_slug = str(values.get("__shopping_collection_path_slug") or "") config_items = collection_configs.get(collection_path_slug, []) derived_l1, derived_l2 = _derive_start_shopping_paths_for_product( session=session, product=product, attrs=attrs_by_sku.get(sku, {}), config_items=config_items, index=taxonomy_index, ) _assign_start_shopping_scalar_value( values, "shopping_l1", _relative_shopping_l1_value(derived_l1), sku=sku, conflicts=conflicts, ) _assign_start_shopping_scalar_value( values, "shopping_l2", _relative_shopping_l2_value(derived_l2), sku=sku, conflicts=conflicts, ) child_to_parent = { str(row.child_sku or "").strip(): str(row.parent_sku or "").strip() for row in session.scalars( select(MasterProductRelation).where(MasterProductRelation.child_sku.in_(sorted(sku_values.keys()))) ).all() if str(row.child_sku or "").strip() and str(row.parent_sku or "").strip() } for child_sku, parent_sku in child_to_parent.items(): child_values = sku_values.get(child_sku) if not child_values: continue parent_values = sku_values.setdefault( parent_sku, { "__shopping_collection_path_slug": None, "shopping_collection": None, "shopping_l1": None, "shopping_l2": None, }, ) if not _clean(parent_values.get("__shopping_collection_path_slug")): parent_values["__shopping_collection_path_slug"] = child_values.get("__shopping_collection_path_slug") for code, _label in START_SHOPPING_ATTRIBUTE_DEFINITIONS: _assign_start_shopping_scalar_value( parent_values, code, child_values.get(code), sku=parent_sku, conflicts=conflicts, ) missing_product_skus = sorted(set(sku_values.keys()) - set(products.keys())) if missing_product_skus: for row in session.scalars( select(MasterProduct).where(MasterProduct.sku.in_(missing_product_skus)) ).all(): products[str(row.sku)] = row updated_values = 0 samples: List[Dict[str, Any]] = [] for sku, values in sku_values.items(): product = products.get(sku) encoded = { code: _clean(values.get(code)) for code, _label in START_SHOPPING_ATTRIBUTE_DEFINITIONS } if len(samples) < 25: samples.append({"sku": sku, **encoded}) if dry_run: continue if product is None: continue for code, value in encoded.items(): _upsert_master_product_attribute_value( session, product=product, attribute_code=code, value=value, source_label="start_shopping_apply", ) updated_values += 1 return { "collection_count": len(processed), "sku_count": len(sku_values), "updated_products": 0 if dry_run else len(products), "updated_values": updated_values, "dry_run": dry_run, "processed": processed, "skipped": skipped, "conflicts": conflicts, "samples": samples, } def _row_dict(row: CollectionLandingPage) -> Dict[str, Any]: starting_price = row.starting_price compare_at_price = row.compare_at_price return { "id": row.id, "path_slug": row.path_slug, "plp_url": f"/{row.path_slug}.html", "page_type": row.page_type, "category_l1": row.category_l1, "collection": row.collection, "collection_registry_id": row.collection_registry_id, "brand": row.brand, "title": row.title, "subtitle": row.subtitle, "short_description": row.short_description, "description_html": row.description_html, "specifications_html": row.specifications_html, "materials_html": row.materials_html, "assembly_html": row.assembly_html, "pdf_url": row.pdf_url, "thumbnail_image_url": row.thumbnail_image_url, "category_icon_url": row.category_icon_url, "hero_images": _list_from_json(row.hero_images), "badges": normalize_collection_badges(row.badges) or dict(EMPTY_BADGES_PAYLOAD), "feature_bullets": _list_from_json(row.feature_bullets), "cta_rows": _list_from_json(row.cta_rows), "matching_collection_slugs": _list_from_json(row.matching_collection_slugs), "sku_prefixes": _list_from_json(row.sku_prefixes), "start_shopping_config": _normalize_start_shopping_config(row.start_shopping_config), "parent_path_slug": row.parent_path_slug, "effective_parent_path_slug": effective_parent_path_slug( category_l1=row.category_l1, parent_path_slug=row.parent_path_slug, ), "starting_price": float(starting_price) if starting_price is not None else None, "compare_at_price": float(compare_at_price) if compare_at_price is not None else None, "sort_order": row.sort_order, "is_active": row.is_active, "notes": row.notes, "updated_at": row.updated_at.isoformat() if row.updated_at else None, "inferred": False, } def _json_list(value: Any) -> Optional[Dict[str, Any]]: if value is None: return None if isinstance(value, dict): if "items" in value: return {"items": [item for item in value.get("items") or [] if _clean(item) or isinstance(item, dict)]} return value if isinstance(value, str): items = [item.strip() for item in value.splitlines() if item.strip()] return {"items": items} if items else None if isinstance(value, Iterable): return {"items": [item for item in value if item not in (None, "")]} return None def _list_from_json(value: Any) -> List[Any]: if isinstance(value, dict): items = value.get("items") return items if isinstance(items, list) else [] if isinstance(value, list): return value return [] def _normalize_start_shopping_config(value: Any) -> Dict[str, Any]: if not isinstance(value, dict): return {"hub_path_slug": None, "items": []} hub_path_slug = normalize_plp_path(value.get("hub_path_slug")) if _clean(value.get("hub_path_slug")) else None items: List[Dict[str, Any]] = [] seen_l1: set[str] = set() raw_items = value.get("items") for item in raw_items if isinstance(raw_items, list) else []: if not isinstance(item, dict): continue l1_path_slug = normalize_plp_path(item.get("l1_path_slug")) if not l1_path_slug or l1_path_slug in seen_l1: continue seen_l1.add(l1_path_slug) l2_values = item.get("l2_path_slugs") l2_path_slugs: List[str] = [] seen_l2: set[str] = set() for raw_l2 in l2_values if isinstance(l2_values, list) else []: l2_path = normalize_plp_path(raw_l2) if not l2_path or l2_path in seen_l2: continue seen_l2.add(l2_path) l2_path_slugs.append(l2_path) items.append({"l1_path_slug": l1_path_slug, "l2_path_slugs": l2_path_slugs}) return {"hub_path_slug": hub_path_slug, "items": items} def _assign_start_shopping_scalar_value( slot: Dict[str, Optional[str]], attribute_code: str, value: Optional[str], *, sku: str, conflicts: List[Dict[str, Any]], ) -> None: cleaned = normalize_plp_path(value) if _clean(value) else None if not cleaned: return current = _clean(slot.get(attribute_code)) if not current: slot[attribute_code] = cleaned return if current == cleaned: return conflicts.append( { "sku": sku, "attribute_code": attribute_code, "current": current, "incoming": cleaned, } ) def _relative_shopping_collection_value(path_slug: Optional[str]) -> Optional[str]: path = normalize_plp_path(path_slug) if _clean(path_slug) else None if not path: return None parts = [part for part in path.split("/") if part] return parts[-1] if parts else None def _relative_shopping_l1_value(path_slug: Optional[str]) -> Optional[str]: path = normalize_plp_path(path_slug) if _clean(path_slug) else None if not path: return None parts = [part for part in path.split("/") if part] return parts[-1] if parts else None def _relative_shopping_l2_value(path_slug: Optional[str]) -> Optional[str]: path = normalize_plp_path(path_slug) if _clean(path_slug) else None if not path: return None parts = [part for part in path.split("/") if part] if len(parts) >= 2: return "/".join(parts[-2:]) return parts[-1] if parts else None def _derive_start_shopping_paths_for_product( *, session: Session, product: MasterProduct, attrs: Dict[str, str], config_items: List[Dict[str, Any]], index: Any, ) -> tuple[Optional[str], Optional[str]]: from db.master_taxonomy_product_paths import canonical_master_taxonomy_path_slugs if not config_items: return None, None product_paths = { normalize_plp_path(path) for path in canonical_master_taxonomy_path_slugs(session, product, attrs or {}, index=index) if normalize_plp_path(path) } by_l1_slug = [ { "l1_path_slug": normalize_plp_path(item.get("l1_path_slug")), "l2_path_slugs": [ normalize_plp_path(raw_l2) for raw_l2 in (item.get("l2_path_slugs") or []) if normalize_plp_path(raw_l2) ], } for item in config_items if normalize_plp_path(item.get("l1_path_slug")) ] l1_match: Optional[str] = None l2_match: Optional[str] = None for item in by_l1_slug: l1_path = item["l1_path_slug"] if l1_path in product_paths or any( product_path.startswith(f"{l1_path}/") for product_path in product_paths ): l1_match = l1_path for l2_path in item["l2_path_slugs"]: if l2_path in product_paths or any( product_path.startswith(f"{l2_path}/") for product_path in product_paths ): l2_match = l2_path break break return l1_match, l2_match def _upsert_master_product_attribute_value( session: Session, *, product: MasterProduct, attribute_code: str, value: str, source_label: str, ) -> None: row = session.scalar( select(MasterProductAttributeValue) .where(MasterProductAttributeValue.product_id == product.id) .where(MasterProductAttributeValue.attribute_code == attribute_code) ) if row is None: row = MasterProductAttributeValue( product_id=product.id, sku=product.sku, attribute_code=attribute_code, ) session.add(row) row.value = value row.source_label = source_label def _optional_decimal(value: Any) -> Optional[Decimal]: text = _clean(value) if text is None: return None try: return Decimal(str(text).replace(",", "").replace("$", "")) except Exception: return None def _clean(value: Any) -> Optional[str]: text = str(value or "").strip() if not text or text.lower() in {"nan", "none", "null"}: return None return text def _clean_collection_text(value: Any, *, field_name: str) -> Optional[str]: text = _clean(value) if not text: return None text = _decode_common_json_escapes(text) text = _extract_pasted_collection_field(text, field_name=field_name) text = _strip_wrapping_quotes(text.strip()) return _clean(text) def _extract_pasted_collection_field(text: str, *, field_name: str) -> str: field_pattern = re.compile(rf'"{re.escape(field_name)}"\s*:\s*"') field_match = field_pattern.search(text) if field_match: text = text[field_match.end() :] other_fields = [key for key in PASTED_COLLECTION_TEXT_FIELDS if key != field_name] if other_fields: field_list = "|".join(re.escape(key) for key in other_fields) boundary_pattern = re.compile(rf'(? str: return ( text.replace("\\r\\n", "\n") .replace("\\n", "\n") .replace("\\r", "\n") .replace("\\t", "\t") .replace('\\"', '"') .replace("\\/", "/") .replace("\\\\", "\\") ) def _strip_wrapping_quotes(text: str) -> str: if len(text) >= 2 and text[0] == text[-1] and text[0] in {'"', "'"}: return text[1:-1] return text def _norm(value: Any) -> str: return " ".join(str(value or "").strip().lower().replace("_", " ").replace("-", " ").split())