"""Sample-door Magento hub: ensure root Samples category and assign sample SKUs.""" from __future__ import annotations import hashlib import logging import re from typing import Any, Dict, List, Optional, Set from sqlalchemy import select from sqlalchemy.orm import Session from channel.url_canonical import normalize_plp_path from magento.sku_policy import filter_magento_push_skus logger = logging.getLogger(__name__) DEFAULT_SAMPLE_PATH_SLUG = "samples" DEFAULT_SAMPLE_REMOTE_PATH = "Samples" DEFAULT_SAMPLE_TITLE = "Samples" LEGACY_SAMPLE_PATH_SLUG = "product-category/sample_doors" LEGACY_SAMPLE_SINGULAR_PATH_SLUG = "sample" SAMPLE_HUB_PATH_SLUGS: Set[str] = { normalize_plp_path(DEFAULT_SAMPLE_PATH_SLUG), normalize_plp_path(LEGACY_SAMPLE_PATH_SLUG), normalize_plp_path(LEGACY_SAMPLE_SINGULAR_PATH_SLUG), } # Real collection codes: ACH, ANRWO, PB-HE, CORONET-HONEY — not path slugs. _COLLECTION_CODE_RE = re.compile(r"^[A-Z0-9]+(?:-[A-Z0-9]+)*$") def is_sample_hub_path(path_slug: str) -> bool: slug = normalize_plp_path(path_slug or "") return bool(slug) and slug in SAMPLE_HUB_PATH_SLUGS def is_sample_master_sku(sku: str, *, name: str = "") -> bool: text_sku = str(sku or "").strip().upper() text_name = str(name or "").strip().lower() if not text_sku and not text_name: return False if text_sku.endswith("-SD") or text_sku.endswith("-SDP"): return True return "sample door" in text_name def is_plausible_collection_sample_sku(sku: str) -> bool: """Accept real collection sample SKUs; reject path-slug inventions. Good: ACH-SD, ANRWO-SD, PB-HE-SD, CORONET-HONEY-SD, SANIBEL-SD Bad: ANNA-CARAMEL-HARVEST-SD, KITCHEN-CABINETS-HALLMARK-FROST-SD, bathroom-vanities-bedford-SD, FROST-ACCESSORIES-SD """ text = str(sku or "").strip() if not text: return False upper = text.upper() if text != upper: return False if upper.endswith("-SDP"): prefix = upper[:-4] elif upper.endswith("-SD"): prefix = upper[:-3] else: return False if not prefix or not _COLLECTION_CODE_RE.fullmatch(prefix): return False parts = prefix.split("-") # Collection codes are short acronyms or at most two tokens (PB-HE, CORONET-HONEY). if len(parts) > 2: return False banned = { "KITCHEN", "CABINETS", "BATHROOM", "VANITIES", "ACCESSORIES", "DOORS", "SHOWER", "VANITY", "TOPS", "INTERIOR", "PRODUCT", "CATEGORY", "SAMPLES", "SAMPLE", } if any(part in banned for part in parts): return False if len(prefix) > 20: return False return True def _sample_product_name(meta: Dict[str, Any]) -> str: collection = str(meta.get("collection") or "").strip() or "Collection" brand = str(meta.get("brand") or "").strip() if brand: return f"SD Sample Door (SD) - {collection} | {brand}" return f"SD Sample Door (SD) - {collection}" def _new_sample_master_product(meta: Dict[str, Any]): from db.models import MasterProduct sku = str(meta["sku"]).strip() payload = { "product_role": "sample_door", "source": "sample_hub_ensure", "sample_cta_label": meta.get("sample_cta_label") or "Order Sample", "collection": meta.get("collection"), } row_hash = hashlib.sha256( f"sample_door|{sku}|{meta.get('collection')}|{meta.get('brand')}".encode("utf-8") ).hexdigest() return MasterProduct( sku=sku, name=_sample_product_name(meta), product_family="Sample Door", category_l1=meta.get("category_l1") or "Kitchen Cabinets", category_l2="Samples", category_l3="Sample Door", brand=meta.get("brand"), brand_id=meta.get("brand_id"), family_id=meta.get("family_id"), collection=meta.get("collection"), collection_registry_id=meta.get("collection_registry_id"), status="enabled", normalization_status="ready", raw_payload=payload, row_hash=row_hash, is_active=True, ) def _brochure_style_collection_code(code: str) -> bool: """True for brochure codes like ACH/ANRWO/PB-HE — false for path slugs.""" text = str(code or "").strip() if not text or any(ch.islower() for ch in text): return False upper = text.upper() if not _COLLECTION_CODE_RE.fullmatch(upper): return False parts = upper.split("-") if len(parts) > 2 or len(upper) > 20: return False banned = { "KITCHEN", "CABINETS", "BATHROOM", "VANITIES", "ACCESSORIES", "DOORS", "SHOWER", "VANITY", "TOPS", "INTERIOR", "PRODUCT", "CATEGORY", "SAMPLES", "SAMPLE", } return not any(part in banned for part in parts) def _sample_sku_from_landing( *, profile_sku: Optional[str], cta_rows: Any, sku_prefixes: Any, registry_code: Optional[str], ) -> Optional[str]: sku = str(profile_sku or "").strip().upper() if sku and is_plausible_collection_sample_sku(sku): return sku for row in cta_rows if isinstance(cta_rows, list) else []: if not isinstance(row, dict): continue if str(row.get("action") or "").strip().lower() != "order_sample": continue candidate = str(row.get("sku") or row.get("master_sku") or "").strip().upper() if candidate and is_plausible_collection_sample_sku(candidate): return candidate prefixes = [] if isinstance(sku_prefixes, dict): prefixes = sku_prefixes.get("items") if isinstance(sku_prefixes.get("items"), list) else [] elif isinstance(sku_prefixes, list): prefixes = sku_prefixes for raw in prefixes: prefix = str(raw or "").strip().upper() if not _brochure_style_collection_code(prefix): continue candidate = f"{prefix}-SD" if is_plausible_collection_sample_sku(candidate): return candidate code = str(registry_code or "").strip().upper() if _brochure_style_collection_code(code): candidate = f"{code}-SD" if is_plausible_collection_sample_sku(candidate): return candidate return None def _sample_intents_from_collections(session: Session) -> Dict[str, Dict[str, Any]]: """Map sample SKU -> metadata for active dashboard collections only (~24). Source of truth: ``collection_landing_page`` rows (page_type=collection). Sample SKU comes from commerce profile / CTA / sku prefix / brochure registry code. Never invent from path slugs. """ from db.collection_landing_pages import COLLECTION_PAGE, _collection_commerce_profile, _list_from_json from db.models import CollectionLandingPage, MasterBrand, MasterCollectionRegistry intents: Dict[str, Dict[str, Any]] = {} registry_rows = { int(row.id): row for row in session.scalars(select(MasterCollectionRegistry)).all() if row.id is not None } brand_rows = { int(row.id): row for row in session.scalars(select(MasterBrand)).all() if row.id is not None } landings = session.scalars( select(CollectionLandingPage) .where(CollectionLandingPage.is_active.is_(True)) .where(CollectionLandingPage.page_type == COLLECTION_PAGE) .order_by(CollectionLandingPage.sort_order, CollectionLandingPage.title) ).all() for landing in landings: registry = ( registry_rows.get(int(landing.collection_registry_id)) if landing.collection_registry_id else None ) profile = _collection_commerce_profile( session, category_l1=landing.category_l1, collection=landing.collection, collection_registry_id=landing.collection_registry_id, ) sku = _sample_sku_from_landing( profile_sku=(profile.sample_master_sku if profile is not None else None), cta_rows=_list_from_json(landing.cta_rows), sku_prefixes=landing.sku_prefixes, registry_code=(registry.code if registry is not None else None), ) if not sku: continue collection_name = ( str(landing.collection or "").strip() or str(landing.title or "").strip() or (str(registry.name).strip() if registry is not None else "") or sku ) brand_name = str(landing.brand or "").strip() or None brand_id = None family_id = None registry_id = int(registry.id) if registry is not None else None if registry is not None: family_id = registry.family_id if registry.brand_id is not None: brand_id = int(registry.brand_id) if not brand_name: brand = brand_rows.get(brand_id) brand_name = str(brand.name).strip() if brand is not None else None intents[sku] = { "sku": sku, "collection": collection_name, "category_l1": str(landing.category_l1 or "").strip() or "Kitchen Cabinets", "brand": brand_name, "brand_id": brand_id, "family_id": family_id, "collection_registry_id": registry_id, "sample_cta_label": str( (profile.sample_cta_label if profile is not None else None) or "Order Sample" ).strip() or "Order Sample", "path_slug": str(landing.path_slug or "").strip() or None, } return intents def ensure_sample_master_products( session: Session, *, skus: Optional[List[str]] = None, dry_run: bool = False, cleanup_bogus: bool = True, ) -> Dict[str, Any]: """Create missing sample-door master products for the real dashboard collections only.""" from db.models import MasterProduct, ProductListingPathAssignment intents = _sample_intents_from_collections(session) if skus is not None: wanted = set(filter_magento_push_skus(sku.upper() for sku in skus)) # Keep only requested SKUs that are either already an intent or a plausible # brochure-style sample SKU (e.g. brochure seed for one collection). filtered: Dict[str, Dict[str, Any]] = { sku: meta for sku, meta in intents.items() if sku in wanted } for sku in wanted: if sku in filtered or not is_plausible_collection_sample_sku(sku): continue filtered[sku] = { "sku": sku, "collection": sku, "category_l1": "Kitchen Cabinets", "brand": None, "brand_id": None, "family_id": None, "collection_registry_id": None, "sample_cta_label": "Order Sample", } intents = filtered existing = { str(row.sku): row for row in session.scalars( select(MasterProduct).where(MasterProduct.sku.in_(list(intents.keys()) or ["__none__"])) ).all() } would_create: List[str] = [] created: List[str] = [] reactivated: List[str] = [] unchanged: List[str] = [] for sku, meta in sorted(intents.items()): row = existing.get(sku) if row is None: would_create.append(sku) if dry_run: continue row = _new_sample_master_product(meta) session.add(row) existing[sku] = row created.append(sku) continue changed = False if not row.is_active: row.is_active = True reactivated.append(sku) changed = True if not str(row.name or "").strip(): row.name = _sample_product_name(meta) changed = True if not str(row.collection or "").strip() and meta.get("collection"): row.collection = meta["collection"] changed = True if not str(row.category_l1 or "").strip() and meta.get("category_l1"): row.category_l1 = meta["category_l1"] changed = True if not str(row.brand or "").strip() and meta.get("brand"): row.brand = meta["brand"] changed = True if row.brand_id is None and meta.get("brand_id") is not None: row.brand_id = meta["brand_id"] changed = True if row.family_id is None and meta.get("family_id") is not None: row.family_id = meta["family_id"] changed = True if row.collection_registry_id is None and meta.get("collection_registry_id") is not None: row.collection_registry_id = meta["collection_registry_id"] changed = True if not str(row.product_family or "").strip(): row.product_family = "Sample Door" changed = True if not str(row.category_l2 or "").strip(): row.category_l2 = "Samples" changed = True if not str(row.category_l3 or "").strip(): row.category_l3 = "Sample Door" changed = True if not changed: unchanged.append(sku) deactivated_bogus: List[str] = [] if cleanup_bogus and not dry_run: legitimate = set(intents.keys()) for row in session.scalars(select(MasterProduct).where(MasterProduct.is_active.is_(True))).all(): sku = str(row.sku or "").strip() payload = row.raw_payload if isinstance(row.raw_payload, dict) else {} if payload.get("source") != "sample_hub_ensure": continue if sku in legitimate: continue row.is_active = False deactivated_bogus.append(sku) for assignment in session.scalars( select(ProductListingPathAssignment).where( ProductListingPathAssignment.master_sku == sku ) ).all(): assignment.assignment_status = "inactive" if not dry_run and (created or reactivated or deactivated_bogus): session.flush() return { "dry_run": dry_run, "intent_count": len(intents), "would_create": would_create, "created": created, "reactivated": reactivated, "unchanged": unchanged[:50], "deactivated_bogus": deactivated_bogus[:100], "created_count": len(created), "would_create_count": len(would_create), "reactivated_count": len(reactivated), "deactivated_bogus_count": len(deactivated_bogus), } def discover_sample_master_skus(session: Session) -> Dict[str, Any]: """Sample SKUs for the real dashboard collections only.""" from db.models import MasterProduct intents = _sample_intents_from_collections(session) active = { str(sku) for sku in session.scalars( select(MasterProduct.sku).where( MasterProduct.sku.in_(list(intents.keys()) or ["__none__"]), MasterProduct.is_active.is_(True), ) ).all() } found = sorted(filter_magento_push_skus(sku for sku in intents if sku in active)) missing_master_skus = sorted( filter_magento_push_skus(sku for sku in intents if sku not in active) ) return { "skus": found, "missing_master_skus": missing_master_skus, "intent_count": len(intents), "intent_skus": sorted(intents.keys()), } def _default_magento_attribute_set_id(session: Session, connection_id: int) -> Optional[int]: from db.magento_repositories import SqlAlchemyBaselineSyncRunRepository return SqlAlchemyBaselineSyncRunRepository(session).get_default_attribute_set_id(connection_id) def _default_magento_website_id(session: Session, connection_id: int) -> Optional[int]: from db.magento_repositories import SqlAlchemyMagentoStoreRegistryRepository from db.models import MagentoConnection conn = session.get(MagentoConnection, connection_id) if conn is not None and conn.default_website_id: return int(conn.default_website_id) return SqlAlchemyMagentoStoreRegistryRepository(session).get_default_website_id(connection_id) def _ensure_magento_sample_product( api: Any, session: Session, *, connection_id: int, channel_sku: str, name: str, category_id: int, ) -> Dict[str, Any]: """Create a minimal Magento simple product for a sample door when missing.""" attribute_set_id = _default_magento_attribute_set_id(session, connection_id) or 4 website_id = _default_magento_website_id(session, connection_id) or 1 payload = { "sku": channel_sku, "name": name or channel_sku, "attribute_set_id": int(attribute_set_id), "price": 0, "status": 1, "visibility": 4, "type_id": "simple", "weight": 1, "extension_attributes": { "website_ids": [int(website_id)], "category_links": [{"position": 0, "category_id": str(int(category_id))}], }, } status, body = api.post_product(payload) return {"http_status": status, "body": body, "sku": channel_sku} def sync_magento_sample_hub( session: Session, *, connection_id: int, path_slug: str = DEFAULT_SAMPLE_PATH_SLUG, remote_path: str = DEFAULT_SAMPLE_REMOTE_PATH, title: str = DEFAULT_SAMPLE_TITLE, dry_run: bool = True, assign_products: bool = True, create_missing_master: bool = True, create_missing_magento: bool = True, cleanup_bogus_master: bool = True, master_skus: Optional[List[str]] = None, ) -> Dict[str, Any]: """Ensure Magento Samples hub exists and assign real collection sample products.""" from db.channel_catalog_provision import _ensure_magento_category_path from db.channel_listing_path import ( bulk_assign_listing_paths, upsert_listing_path, upsert_listing_path_target, ) from db.taxonomy_invent_policy import can_auto_invent, can_provision_remotes if create_missing_magento and not can_provision_remotes(): create_missing_magento = False if create_missing_master and not can_auto_invent(): create_missing_master = False from db.channel_sku_mapping import mapping_by_master from db.collection_landing_push import _build_magento_api, _upsert_taxonomy_channel_link_for_path from db.models import MasterProduct from db.product_channel_taxonomy import upsert_sku_taxonomy_assignment slug = normalize_plp_path(path_slug) or DEFAULT_SAMPLE_PATH_SLUG SAMPLE_HUB_PATH_SLUGS.add(slug) display_path = str(remote_path or title or slug).strip() or DEFAULT_SAMPLE_REMOTE_PATH display_title = str(title or display_path).strip() or DEFAULT_SAMPLE_TITLE ensure_result: Dict[str, Any] = {"skipped": True} if create_missing_master: ensure_result = ensure_sample_master_products( session, skus=master_skus, dry_run=dry_run, cleanup_bogus=cleanup_bogus_master, ) missing_master_skus: List[str] = [] if master_skus is not None: requested = filter_magento_push_skus(master_skus) existing = { str(sku) for sku in session.scalars( select(MasterProduct.sku).where( MasterProduct.sku.in_(requested), MasterProduct.is_active.is_(True), ) ).all() } skus = [sku for sku in requested if sku in existing] missing_master_skus = [sku for sku in requested if sku not in existing] else: discovered = discover_sample_master_skus(session) skus = list(discovered["skus"]) missing_master_skus = list(discovered["missing_master_skus"]) if create_missing_master and dry_run: planned = set(skus) | set(ensure_result.get("would_create") or []) skus = sorted(filter_magento_push_skus(planned)) missing_master_skus = [] result: Dict[str, Any] = { "connection_id": connection_id, "dry_run": dry_run, "path_slug": slug, "remote_path": display_path, "title": display_title, "sample_sku_count": len(skus), "sample_skus": skus[:50], "missing_master_skus": missing_master_skus, "master_ensure": ensure_result, "assign_products": assign_products, } if dry_run: result["would_ensure_category"] = True result["would_assign"] = len(skus) if assign_products else 0 return result api, native_id = _build_magento_api(session, connection_id) if api is None or native_id is None: raise ValueError(f"Magento connection {connection_id} not found") category_id = _ensure_magento_category_path( session, api, connection_id=int(native_id), path=display_path, ) if not category_id: raise RuntimeError(f"Could not resolve Magento Samples category for path={display_path!r}") upsert_listing_path( session, path_slug=slug, title=display_title, path_kind="hub", notes="Samples Magento hub", ) upsert_listing_path_target( session, path_slug=slug, channel_code="magento", remote_id=str(int(category_id)), taxonomy_kind="category", connection_id=connection_id, remote_path=display_path, ) _upsert_taxonomy_channel_link_for_path( session, path_slug=slug, channel_code="magento", connection_id=int(native_id), remote_id=str(int(category_id)), remote_path=display_path, taxonomy_kind="category", ) listing_assign = bulk_assign_listing_paths( session, path_slugs=[slug], skus=skus, dry_run=False, replace_existing=False, ) assigned = 0 skipped_existing = 0 created_magento: List[str] = [] errors: List[Dict[str, Any]] = [] if assign_products and skus: channel_skus = mapping_by_master( session, "magento", skus, connection_id=connection_id, ) name_by_sku = { str(row.sku): str(row.name or row.sku) for row in session.scalars(select(MasterProduct).where(MasterProduct.sku.in_(skus))).all() } status, existing_rows, error = api.get_category_products(int(category_id)) if status != 200: raise RuntimeError(f"get_category_products failed for Samples hub: HTTP {status} {error}") existing = { str(item.get("sku") or "").strip() for item in (existing_rows or []) if isinstance(item, dict) and str(item.get("sku") or "").strip() } for position, master_sku in enumerate(skus): channel_sku = channel_skus.get(master_sku, master_sku) try: upsert_sku_taxonomy_assignment( session, master_sku=master_sku, channel_code="magento", connection_id=connection_id, taxonomy_kind="category", remote_id=str(int(category_id)), remote_path=display_path, match_source="bulk_assign", sort_order=position, ) if channel_sku in existing: skipped_existing += 1 continue link_status, body, link_error = api.assign_product_to_category( int(category_id), channel_sku, position=position, ) if link_status == 404 and create_missing_magento: create_result = _ensure_magento_sample_product( api, session, connection_id=connection_id, channel_sku=channel_sku, name=name_by_sku.get(master_sku, channel_sku), category_id=int(category_id), ) if create_result.get("http_status") in (200, 201): created_magento.append(channel_sku) assigned += 1 existing.add(channel_sku) continue errors.append( { "master_sku": master_sku, "channel_sku": channel_sku, "http_status": create_result.get("http_status"), "error": create_result.get("body"), "stage": "create_magento_product", } ) continue if link_status in (200, 201): assigned += 1 existing.add(channel_sku) else: errors.append( { "master_sku": master_sku, "channel_sku": channel_sku, "http_status": link_status, "error": link_error or body, } ) except Exception as exc: # noqa: BLE001 - collect per-SKU failures errors.append({"master_sku": master_sku, "channel_sku": channel_sku, "error": str(exc)}) result.update( { "category_id": int(category_id), "listing_assign": listing_assign, "assigned": assigned, "skipped_existing": skipped_existing, "created_magento": created_magento, "created_magento_count": len(created_magento), "errors": errors, "error_count": len(errors), } ) logger.info( "Samples hub sync connection_id=%s category_id=%s path=%s skus=%s assigned=%s skipped=%s " "created_master=%s created_magento=%s missing_master=%s errors=%s", connection_id, category_id, slug, len(skus), assigned, skipped_existing, int((ensure_result or {}).get("created_count") or 0), len(created_magento), len(missing_master_skus), len(errors), ) return result