from __future__ import annotations import hashlib import json import re from collections import Counter from typing import Any, Callable, Dict, List, Optional, Tuple import pandas as pd import sqlalchemy as sa from sqlalchemy import func, select from sqlalchemy.dialects.postgresql import insert from sqlalchemy.orm import Session from channel.url_canonical import build_pdp_slug from db.master_relations import delete_relations_touching_skus, extract_relations_from_records, upsert_relations from db.models import ( MasterAttributeDefinition, MasterCollectionRegistry, MasterProduct, MasterProductAttributeValue, MasterProductLocationAvailability, MasterProductPrice, MasterProductRelation, ProductChannelAssignment, ) from db.collection_landing_pages import canonical_collection from db.source_imports import normalize_column_key from db.tribeca_sku_parse import collection_name_for_style_code CORE_FIELD_KEYS = { "sku", "item", "stocked", "display_name", "category", "product_category", "cabinet_brand", "product_collection_series_name", "item_size", "item_style", "manufacturer", } LOCATION_COLUMNS = { "baltimore": "12", "bellmawr": "2", "brentwood": "10", "keyport": "1", "north_bergen": "3", } PRICE_COLUMNS = { "price": "base", "rta_price": "rta", "assembled_price": "assembled", "shopify_price": "shopify", "price_starting_at": "starting_at", } OPERATIONAL_FIELD_KEYS = set(LOCATION_COLUMNS) | set(PRICE_COLUMNS) SEO_IMPORT_ALIASES = { "seo_title": { "seo_title", "seofied_title", "seoified_title", "seo_optimized_title", "meta_title", "title_tag", }, "description": { "description", "seo_description_body", "product_description", }, "meta_description": { "meta_description", "seo_description", "seofied_description", "seoified_description", }, "meta_keywords": { "meta_keywords", "meta_keyword", "seo_keywords", "seo_keyword", "keywords", }, "product_url_slug": { "product_url_slug", "url_key", "handle", "seo_url", "seo_url_key", }, } ADMIN_ONLY_KEYS = { "item_style", "item_style_1", "item_style_prefix", "notes", "plytix_notes", "source_image_file_s", "product_url_slug", "product_url_portandbell_com", "lookup_key_auto", } ADMIN_ATTRIBUTE_KEYS = ADMIN_ONLY_KEYS | {"grab_n_go"} ATTRIBUTE_CODE_ALIASES = { "country_of_origin_coo": "country_of_origin", "composite_wood_product_cwp": "composite_wood_product", } ATTRIBUTE_CODE_SKIP_KEYS = { "item_style_1", "grab_n_go", "grab_go", "grab_ngo", } PRESENTATION_FIELD_KEYS = { "cabinet_style_short_description", "cabinet_description", "short_description", "cabinet_collection_image", "cabinet_category_image", "cabinet_detail_image", "l_shape_layout_price", "u_shape_layout_price", "average_price_per_linear_foot", } BULK_WRITE_BATCH_SIZE = 1000 CABINET_CATEGORY_ROOTS = frozenset({"Kitchen Cabinets", "Bathroom Vanities", "Showers"}) NON_COLLECTION_CATEGORY_LEAVES = frozenset({"Vanity Tops"}) _ASSEMBLY_SUFFIXES = (" - Assembled", " - Unassembled", " - RTA") _ITEM_STYLE_NOISE_PHRASES = ( "value select", "north point stock", ) _MSI_LOCATION_PHRASES = ( "baltimore", "bellmawr", "brentwood", "keyport", "north bergen", ) _BRAND_ONLY_ITEM_STYLE_KEYS = frozenset( { "legionmanor", "portbell", "portandbell", "castlecraft", "fabuwood", "echelon", "sunnywood", "legacy", "bridgewaterwholesalers", } ) _NON_COLLECTION_ITEM_STYLE_KEYS = frozenset( { "windows", "window", "interiordoors", "exteriordoors", "doors", "door", "showerdoors", "bathtubdoors", "accessories", "accessory", "mouldings", "moulding", "moldings", "molding", "panelsandfillers", "panelsfillers", "fillers", "filler", "hardware", "northpointstock", "valueselect", } ) _ACCESSORY_ITEM_STYLE_RE = re.compile(r"^([^:]+):\s*(.+?)\s*-\s*\1\s*$", re.IGNORECASE) _CABINET_ACCESSORY_RE = re.compile( r"\b(touch\s*up|touch-up|kit|tray|rollout|roll\s*out|insert|organizer|divider|lazy susan)\b", re.IGNORECASE, ) _CABINET_MOULDING_RE = re.compile( r"\b(mould|mold|moulding|molding|crown|scribe|toe\s*kick|toe-kick|light\s*rail|quarter\s*round|shoe)\b", re.IGNORECASE, ) _CABINET_PANEL_FILLER_RE = re.compile( r"\b(filler|panel|skin|end\s*panel|dishwasher\s*end|return\s*end|refrigerator\s*panel)\b", re.IGNORECASE, ) def registry_intent_records_from_dataframe(df: pd.DataFrame) -> tuple[List[Dict[str, Any]], List[str]]: """Parse a master SKU dataframe into registry-intent rows without writing to the DB.""" records = _records(df) import_rows = [{**_master_fields(r), "attributes": r.get("payload") or {}} for r in records] attribute_codes: set[str] = set() for record in records: for key in (record.get("payload") or {}): attribute_codes.add(normalize_column_key(key)) return import_rows, sorted(attribute_codes) def build_master_sku_import_preview(df: pd.DataFrame, *, sample_limit: int = 25) -> Dict[str, Any]: """Derive canonical master fields from an import file without writing anything.""" records = _records(df) fields_by_sku = [(record["sku"], record, _master_fields(record)) for record in records] def counts_for(key: str) -> Dict[str, int]: return dict(Counter((fields.get(key) or "(blank)") for _, _, fields in fields_by_sku).most_common()) missing: Dict[str, int] = {} for key in ("category_l1", "brand", "collection", "product_family"): count = sum(1 for _, _, fields in fields_by_sku if not fields.get(key)) if count: missing[key] = count suspect_rows = [] for sku, record, fields in fields_by_sku: collection = str(fields.get("collection") or "").strip() category_l1 = str(fields.get("category_l1") or "").strip() category_l2 = str(fields.get("category_l2") or "").strip() if _suspect_collection_name(collection, category_l1, category_l2): suspect_rows.append( { "sku": sku, "collection": collection, "category": record["payload"].get("Category"), "product_collection_series_name": record["payload"].get("Product Collection/Series Name"), "item_style": record["payload"].get("Item Style"), } ) return { "input_rows": len(df), "valid_skus": len(records), "missing_sku": len(df) - len(records), "columns": list(df.columns), "category_l1_counts": counts_for("category_l1"), "category_l2_counts": counts_for("category_l2"), "category_l3_counts": counts_for("category_l3"), "brand_counts": counts_for("brand"), "collection_counts": counts_for("collection"), "product_family_counts": counts_for("product_family"), "missing_canonical_fields": missing, "suspect_collections": { "count": len(suspect_rows), "samples": suspect_rows[:sample_limit], }, } def import_master_sku_dataframe( session: Session, df: pd.DataFrame, *, source_label: str = "master_sku", progress_callback: Optional[Callable[[Dict[str, Any]], None]] = None, resolve_registries: bool = False, reconcile_taxonomy: bool = True, ) -> Dict[str, int]: from db.master_product_images import import_master_product_images_dataframe, is_images_only_csv if is_images_only_csv(df): stats = import_master_product_images_dataframe(session, df, source_label=source_label) return {"source": "master_product_images", **stats} _progress(progress_callback, stage="parse", input_rows=len(df), message="Parsing master CSV rows") records = _records(df) _progress(progress_callback, stage="load_existing", valid_skus=len(records), message="Loading existing master SKUs") current_by_sku = { row.sku: row for row in session.scalars(select(MasterProduct).where(MasterProduct.sku.in_([r["sku"] for r in records]))).all() } inserted = 0 updated = 0 unchanged = 0 attr_values: List[Dict[str, Any]] = [] assignments: List[Dict[str, Any]] = [] prices: List[Dict[str, Any]] = [] availability: List[Dict[str, Any]] = [] attribute_defs: Dict[str, Dict[str, Any]] = {} touched_product_ids: List[int] = [] total_records = len(records) from db.manual_attributes import list_manual_attributes locked_vocabs = [ item for item in (list_manual_attributes(session).get("attributes") or []) if item.get("is_locked") and item.get("is_active") ] for index, record in enumerate(records, start=1): sku = record["sku"] payload = dict(record["payload"] or {}) record = {**record, "payload": payload} _apply_manual_locked_attributes_to_payload(session, record) _apply_manual_attribute_vocabularies_to_payload(session, record, locked_vocabs=locked_vocabs) row_hash = _stable_hash(payload) product = current_by_sku.get(sku) fields = _master_fields(record, session=session) if product is None: product = MasterProduct(**fields, row_hash=row_hash, raw_payload=payload) session.add(product) session.flush() inserted += 1 current_by_sku[sku] = product elif product.row_hash == row_hash and not _canonical_fields_changed(product, fields): unchanged += 1 else: for key, value in fields.items(): setattr(product, key, value) product.row_hash = row_hash product.raw_payload = payload product.is_active = True updated += 1 session.flush() touched_product_ids.append(product.id) attr_values.extend(_attribute_rows(product.id, sku, payload, source_label)) assignments.extend(_assignment_rows(product.id, sku, payload)) prices.extend(_price_rows(product.id, sku, payload, source_label)) availability.extend(_availability_rows(product.id, sku, payload, source_label)) for definition in _attribute_definitions(payload): attribute_defs[definition["attribute_code"]] = definition if index == 1 or index % 500 == 0 or index == total_records: _progress( progress_callback, stage="prepare_rows", processed=index, total=total_records, attributes=len(attr_values), message=f"Prepared {index}/{total_records} master rows", ) _progress(progress_callback, stage="upsert_attribute_definitions", count=len(attribute_defs)) _upsert_attribute_definitions(session, list(attribute_defs.values())) _progress(progress_callback, stage="delete_derived_rows", product_count=len(touched_product_ids)) _delete_master_derived_rows(session, touched_product_ids) _progress(progress_callback, stage="upsert_attribute_values", count=len(attr_values)) _upsert_attribute_values(session, attr_values) _progress(progress_callback, stage="upsert_assignments", count=len(assignments)) _upsert_assignments(session, assignments) _progress(progress_callback, stage="upsert_prices", count=len(prices)) _upsert_prices(session, prices) _progress(progress_callback, stage="upsert_availability", count=len(availability)) _upsert_availability(session, availability) import_skus = [r["sku"] for r in records] _progress(progress_callback, stage="relations", sku_count=len(import_skus)) delete_relations_touching_skus(session, import_skus) relation_rows = extract_relations_from_records(records, source_label=source_label) relations_upserted = upsert_relations(session, relation_rows) from db.master_associations import apply_manual_overrides_from_records, discover_associations _progress(progress_callback, stage="association_overrides", sku_count=len(import_skus)) association_overrides = apply_manual_overrides_from_records( session, records, source_label=source_label ) _progress(progress_callback, stage="association_discovery", sku_count=len(import_skus)) association_discovery = discover_associations( session, skus=import_skus, dry_run=False, source_label=source_label, bidirectional=True, ) taxonomy_reconciliation: Dict[str, Any] = {} if reconcile_taxonomy: _progress(progress_callback, stage="taxonomy_reconciliation", sku_count=len(import_skus)) taxonomy_reconciliation = _reconcile_imported_product_taxonomy(session, import_skus) result = { "input_rows": len(df), "valid_skus": len(records), "inserted": inserted, "updated": updated, "unchanged": unchanged, "attribute_values": len(attr_values), "channel_assignments": len(assignments), "prices": len(prices), "location_availability": len(availability), "attribute_definitions": len(attribute_defs), "relations_upserted": relations_upserted, "association_overrides": association_overrides, "association_discovery": association_discovery, "taxonomy_reconciliation": taxonomy_reconciliation, "import_skus": import_skus, "missing_sku": len(df) - len(records), } if resolve_registries: from db.master_taxonomy_registry import apply_registry_fks_to_product, summarize_import_registry_intent import_rows = [{**_master_fields(r, session=session), "attributes": r.get("payload") or {}} for r in records] result["registry_intent"] = summarize_import_registry_intent(session, import_rows) from db.master_taxonomy_intent import summarize_import_taxonomy_intent result["taxonomy_intent"] = summarize_import_taxonomy_intent(session, import_rows) result["attribute_codes"] = sorted(attribute_defs.keys()) linked = 0 for product in current_by_sku.values(): if apply_registry_fks_to_product(session, product, create_missing=False): linked += 1 result["registry_fks_linked"] = linked return result def import_master_sku_seo_dataframe( session: Session, df: pd.DataFrame, *, source_label: str = "product_seo", dry_run: bool = False, ) -> Dict[str, Any]: """Overlay SEO fields by SKU without replacing taxonomy, prices, assignments, or other attrs.""" records = _records(df) skus = [record["sku"] for record in records] products_by_sku = { product.sku: product for product in session.scalars(select(MasterProduct).where(MasterProduct.sku.in_(skus))).all() } updates: List[Dict[str, Any]] = [] missing_skus: List[str] = [] skipped_no_seo = 0 for record in records: sku = record["sku"] product = products_by_sku.get(sku) if product is None: missing_skus.append(sku) continue seo_fields = _seo_fields_from_payload(record["payload"], sku=sku) if not seo_fields: skipped_no_seo += 1 continue updates.append({"sku": sku, "product": product, "fields": seo_fields}) explicit_seo_skus = {item["sku"] for item in updates} if not dry_run: for item in updates: product = item["product"] fields = item["fields"] if fields.get("meta_title"): product.name = fields["meta_title"] raw_payload = {**(product.raw_payload or {}), **fields} # Explicit SEO on variation parents wins over derived title refresh. if _looks_like_variation_builder_parent(product, []): raw_payload["explicit_seo_title"] = True product.raw_payload = raw_payload product.row_hash = _stable_hash(product.raw_payload or {}) product.is_active = True _upsert_single_attribute_values(session, product, fields, source_label) parent_title_refresh = _refresh_variation_parent_titles_for_children( session, [item["sku"] for item in updates], source_label=source_label, dry_run=dry_run, skip_parent_skus=explicit_seo_skus, ) return { "input_rows": len(df), "valid_skus": len(records), "matched_skus": len(updates), "missing_sku": len(df) - len(records), "missing_master_sku_count": len(missing_skus), "skipped_no_seo": skipped_no_seo, "updated": 0 if dry_run else len(updates), "dry_run": dry_run, "sample_missing_master_skus": missing_skus[:25], "sample_updates": [ {"sku": item["sku"], "fields": item["fields"]} for item in updates[:25] ], "variation_parent_titles": parent_title_refresh, } def _reconcile_imported_product_taxonomy(session: Session, skus: List[str]) -> Dict[str, Any]: """Make master taxonomy the source of truth immediately after every master import.""" if not skus: return {"product_count": 0, "assignment_count": 0, "path_count": 0} from db.master_taxonomy_sync import link_products_to_taxonomy_from_master from db.models import MagentoConnection, ShopifyConnection magento_connection_id = session.scalar( select(MagentoConnection.id) .where(MagentoConnection.status == "active") .order_by(MagentoConnection.id) .limit(1) ) shopify_connection_id = session.scalar( select(ShopifyConnection.id) .where(ShopifyConnection.status == "active") .order_by(ShopifyConnection.id) .limit(1) ) return link_products_to_taxonomy_from_master( session, skus=skus, replace_existing=True, sync_channels=True, magento_connection_id=int(magento_connection_id) if magento_connection_id else None, shopify_connection_id=int(shopify_connection_id) if shopify_connection_id else None, dry_run=False, ) def _progress(callback: Optional[Callable[[Dict[str, Any]], None]], **payload: Any) -> None: if callback is None: return callback(payload) def reprocess_master_sku_raw_payloads( session: Session, *, source_label: str = "master_sku_reprocess", ) -> Dict[str, int]: """Rebuild canonical master catalog tables from stored raw per-SKU payloads.""" payloads = [ dict(row.raw_payload or {}) for row in session.scalars( select(MasterProduct) .where(MasterProduct.is_active.is_(True)) .where(MasterProduct.raw_payload.is_not(None)) .order_by(MasterProduct.sku) ).all() if row.raw_payload ] if not payloads: return { "input_rows": 0, "valid_skus": 0, "inserted": 0, "updated": 0, "unchanged": 0, "attribute_values": 0, "channel_assignments": 0, "prices": 0, "location_availability": 0, "attribute_definitions": 0, "relations_upserted": 0, "missing_sku": 0, } return import_master_sku_dataframe(session, pd.DataFrame(payloads), source_label=source_label) def build_master_catalog_validation(session: Session) -> Dict[str, Any]: products = session.scalars(select(MasterProduct).where(MasterProduct.is_active.is_(True))).all() skus = [p.sku for p in products] product_ids = [p.id for p in products] summary: Dict[str, Any] = { "total_skus": len(products), "missing_name": sum(1 for p in products if not _clean(p.name)), "missing_product_family": sum(1 for p in products if not _clean(p.product_family)), "missing_category_l1": sum(1 for p in products if not _clean(p.category_l1)), "missing_brand": sum(1 for p in products if not _clean(p.brand)), "missing_collection": sum(1 for p in products if not _clean(p.collection)), "missing_assembly_type": sum(1 for p in products if not _clean(p.assembly_type)), } family_counts = _count_rows(session, MasterProduct.product_family, MasterProduct.is_active.is_(True)) assembly_counts = _count_rows(session, MasterProduct.assembly_type, MasterProduct.is_active.is_(True)) price_counts = _count_rows_for_ids(session, MasterProductPrice.price_type, MasterProductPrice.product_id, product_ids) location_counts = _availability_counts(session, product_ids) attr_purpose_counts = _count_rows(session, MasterAttributeDefinition.purpose, MasterAttributeDefinition.is_active.is_(True)) from db.master_product_images import master_image_summary image_summary = master_image_summary(session) summary["skus_with_images"] = image_summary["skus_with_images"] summary["skus_without_images"] = image_summary["skus_without_images"] summary["total_images"] = image_summary["total_images"] missing_price_rows = _price_validation_issues(len(products), price_counts) issue_rows = [ {"check": "missing_name", "count": summary["missing_name"], "severity": "error"}, {"check": "missing_product_family", "count": summary["missing_product_family"], "severity": "error"}, {"check": "missing_category_l1", "count": summary["missing_category_l1"], "severity": "warn"}, {"check": "missing_brand", "count": summary["missing_brand"], "severity": "warn"}, {"check": "missing_collection", "count": summary["missing_collection"], "severity": "warn"}, {"check": "missing_assembly_type", "count": summary["missing_assembly_type"], "severity": "info"}, *missing_price_rows, ] if summary.get("skus_without_images"): issue_rows.append( { "check": "skus_without_images", "count": summary["skus_without_images"], "severity": "warn", } ) sample_issues = _sample_product_issues(products) no_channel_assignment = _skus_without_channel_assignment(session, skus) if no_channel_assignment: issue_rows.append( { "check": "skus_without_channel_assignment", "count": len(no_channel_assignment), "severity": "info", } ) return { "summary": summary, "family_counts": family_counts, "assembly_counts": assembly_counts, "price_counts": price_counts, "price_attribute_candidates": _price_attribute_candidates(session, product_ids), "location_availability_counts": location_counts, "attribute_purpose_counts": attr_purpose_counts, "issues": issue_rows, "sample_product_issues": sample_issues[:100], "sample_skus_without_channel_assignment": no_channel_assignment[:100], } def flatten_master_catalog_validation(report: Dict[str, Any]) -> List[Dict[str, Any]]: rows: List[Dict[str, Any]] = [] for key, value in report.get("summary", {}).items(): rows.append({"section": "summary", "key": key, "value": value, "extra": ""}) for section in ("family_counts", "assembly_counts", "price_counts", "attribute_purpose_counts"): for key, value in report.get(section, {}).items(): rows.append({"section": section, "key": key or "(blank)", "value": value, "extra": ""}) for row in report.get("price_attribute_candidates", []): rows.append( { "section": "price_attribute_candidates", "key": row["attribute_code"], "value": row["product_count"], "extra": row.get("sample_value") or "", } ) for key, value in report.get("location_availability_counts", {}).items(): rows.append({"section": "location_availability_counts", "key": key, "value": value, "extra": ""}) for issue in report.get("issues", []): rows.append( { "section": "issues", "key": issue.get("check"), "value": issue.get("count"), "extra": issue.get("severity"), } ) for issue in report.get("sample_product_issues", []): rows.append( { "section": "sample_product_issues", "key": issue.get("sku"), "value": ",".join(issue.get("issues", [])), "extra": issue.get("name") or "", } ) for sku in report.get("sample_skus_without_channel_assignment", []): rows.append({"section": "sample_skus_without_channel_assignment", "key": sku, "value": "", "extra": ""}) return rows def _records(df: pd.DataFrame) -> List[Dict[str, Any]]: records: List[Dict[str, Any]] = [] for raw in df.to_dict(orient="records"): payload = {str(k): _clean(v) for k, v in raw.items()} payload = _augment_master_payload(payload) sku = _first_text(payload, "SKU", "Item", "sku", "item") if not sku: continue records.append({"sku": sku, "payload": payload}) return records def _count_rows(session: Session, column: Any, *where_clauses: Any) -> Dict[str, int]: stmt = select(column, func.count()).group_by(column) for clause in where_clauses: stmt = stmt.where(clause) return {str(key or ""): int(count) for key, count in session.execute(stmt).all()} def _count_rows_for_ids(session: Session, column: Any, id_column: Any, ids: List[int]) -> Dict[str, int]: if not ids: return {} stmt = select(column, func.count()).where(id_column.in_(ids)).group_by(column) return {str(key or ""): int(count) for key, count in session.execute(stmt).all()} def _availability_counts(session: Session, product_ids: List[int]) -> Dict[str, int]: if not product_ids: return {} stmt = ( select(MasterProductLocationAvailability.location_code, func.count()) .where(MasterProductLocationAvailability.product_id.in_(product_ids)) .where(MasterProductLocationAvailability.is_available.is_(True)) .group_by(MasterProductLocationAvailability.location_code) ) return {str(key or ""): int(count) for key, count in session.execute(stmt).all()} def _price_attribute_candidates(session: Session, product_ids: List[int]) -> List[Dict[str, Any]]: if not product_ids: return [] rows = session.execute( select( MasterProductAttributeValue.attribute_code, func.count(func.distinct(MasterProductAttributeValue.product_id)), func.min(MasterProductAttributeValue.value), ) .where(MasterProductAttributeValue.product_id.in_(product_ids)) .where(MasterProductAttributeValue.attribute_code.ilike("%price%")) .group_by(MasterProductAttributeValue.attribute_code) .order_by(func.count(func.distinct(MasterProductAttributeValue.product_id)).desc()) .limit(25) ).all() return [ {"attribute_code": code, "product_count": int(count), "sample_value": sample} for code, count, sample in rows ] def _price_validation_issues(total_products: int, price_counts: Dict[str, int]) -> List[Dict[str, Any]]: sell_price_types = ("base", "rta", "assembled") if not any(int(price_counts.get(price_type, 0)) for price_type in sell_price_types): return [ { "check": "sell_prices_not_imported", "count": total_products, "severity": "info", } ] return [ { "check": f"missing_price_{price_type}", "count": max(total_products - int(price_counts.get(price_type, 0)), 0), "severity": "warn", } for price_type in sell_price_types ] def _sample_product_issues(products: List[MasterProduct]) -> List[Dict[str, Any]]: rows = [] for p in products: issues = [] if not _clean(p.name): issues.append("missing_name") if not _clean(p.product_family): issues.append("missing_product_family") if not _clean(p.category_l1): issues.append("missing_category_l1") if not _clean(p.brand): issues.append("missing_brand") if not _clean(p.collection): issues.append("missing_collection") if issues: rows.append({"sku": p.sku, "name": p.name, "issues": issues}) return rows def _skus_without_channel_assignment(session: Session, skus: List[str]) -> List[str]: if not skus: return [] assigned = { sku for (sku,) in session.execute( select(ProductChannelAssignment.sku).where(ProductChannelAssignment.sku.in_(skus)) ).all() } return [sku for sku in skus if sku not in assigned] def _canonical_fields_changed(product: MasterProduct, fields: Dict[str, Any]) -> bool: return any(getattr(product, key) != value for key, value in fields.items()) def _delete_master_derived_rows(session: Session, product_ids: List[int]) -> None: if not product_ids: return for batch in _batches(product_ids, BULK_WRITE_BATCH_SIZE): session.execute( sa.delete(MasterProductAttributeValue).where(MasterProductAttributeValue.product_id.in_(batch)) ) session.execute( sa.delete(ProductChannelAssignment) .where(ProductChannelAssignment.product_id.in_(batch)) .where(ProductChannelAssignment.reason == "product_websites") ) def _master_fields(record: Dict[str, Any], *, session: Optional[Session] = None) -> Dict[str, Any]: payload = record["payload"] category = _first_text(payload, "Category") cat_l1, raw_cat_l2, raw_cat_l3 = _category_parts(category) item_style = _first_text(payload, "Item Style") manufacturer = _first_text(payload, "Manufacturer") brand = _brand_from_payload(payload, category, manufacturer) product_category = _first_text(payload, "Product Category") or cat_l1 category_l2, category_l3 = _canonical_product_category_children(payload, product_category, raw_cat_l2, raw_cat_l3) collection = _collection_from_payload(payload, category) collection_registry_id = None source_item = _manual_taxonomy_source_sku(payload) if session is not None: from db.manual_taxonomy_assignments import resolve_manual_taxonomy_import_override manual_override = resolve_manual_taxonomy_import_override( session, master_sku=record["sku"], source_sku=source_item, ) if manual_override: product_category = str(manual_override.get("category_l1") or product_category or "").strip() or None category_l2 = str(manual_override.get("category_l2") or category_l2 or "").strip() or None category_l3 = str(manual_override.get("category_l3") or category_l3 or "").strip() or None if manual_override.get("match_scope") == "exact": collection = str(manual_override.get("collection") or collection or "").strip() or None collection_registry_id = manual_override.get("collection_registry_id") else: resolved_collection = _resolve_master_collection(session, payload, category) if resolved_collection: collection = str(resolved_collection.get("name") or "").strip() or None collection_registry_id = resolved_collection.get("id") else: collection = None collection_registry_id = None else: resolved_collection = _resolve_master_collection(session, payload, category) if resolved_collection: collection = str(resolved_collection.get("name") or "").strip() or None collection_registry_id = resolved_collection.get("id") else: collection = None return { "sku": record["sku"], "source_item": source_item, "name": _first_text(payload, "Display Name", "Label", "Name"), "product_family": _family_from_category(product_category or category), "category_l1": product_category, "category_l2": category_l2, "category_l3": category_l3, "brand": brand, "manufacturer": manufacturer, "collection": collection, "collection_registry_id": collection_registry_id, "item_size": _first_text(payload, "Item Size"), "item_style": item_style, "assembly_type": _assembly_type( record["sku"], item_style, _first_text(payload, "Assembled or RTA"), ), "stocked": _to_bool(_first_text(payload, "Stocked")), "status": "active", } def _apply_manual_locked_attributes_to_payload(session: Session, record: Dict[str, Any]) -> None: from db.manual_attribute_locks import resolve_manual_attribute_locks from db.manual_taxonomy_assignments import resolve_manual_taxonomy_import_override payload = record.get("payload") if not isinstance(payload, dict): return manual_taxonomy = resolve_manual_taxonomy_import_override( session, master_sku=record.get("sku") or "", source_sku=_manual_taxonomy_source_sku(payload), ) if not manual_taxonomy: return locked_values = resolve_manual_attribute_locks( session, canonical_taxonomy_path_slug=manual_taxonomy.get("canonical_taxonomy_path_slug"), collection_code=manual_taxonomy.get("collection_code") if manual_taxonomy.get("match_scope") == "exact" else None, ) if not locked_values: return for attribute_code, config in locked_values.items(): normalized_code = normalize_column_key(attribute_code) conflicting_keys = [key for key in list(payload.keys()) if normalize_column_key(key) == normalized_code] current_value = None for key in conflicting_keys: value = str(payload.get(key) or "").strip() if value: current_value = value break for key in conflicting_keys: payload.pop(key, None) values = [str(item).strip() for item in (config.get("attribute_values") or []) if str(item).strip()] if not values: continue assignment_mode = str(config.get("assignment_mode") or "").strip().lower() or ("allow_list" if len(values) > 1 else "force") if assignment_mode == "force" or len(values) == 1: payload[attribute_code] = values[0] continue matched = next((item for item in values if current_value and item.lower() == current_value.lower()), None) if matched: payload[attribute_code] = matched def _apply_manual_attribute_vocabularies_to_payload( session: Session, record: Dict[str, Any], *, locked_vocabs: Optional[List[Dict[str, Any]]] = None, ) -> None: """Canonicalize/drop payload values using locked attribute vocabularies.""" from db.manual_attributes import apply_manual_attribute_vocabularies_to_payload payload = record.get("payload") if not isinstance(payload, dict): return apply_manual_attribute_vocabularies_to_payload(session, payload, locked_vocabs=locked_vocabs) def _attribute_rows(product_id: int, sku: str, payload: Dict[str, Any], source_label: str) -> List[Dict[str, Any]]: rows = [] seen_codes: set[str] = set() for column_name, value in payload.items(): value_text = _clean(value) if value_text is None: continue key = _canonical_attribute_code(column_name) if not key or key in seen_codes: continue if key in CORE_FIELD_KEYS or key in OPERATIONAL_FIELD_KEYS: continue seen_codes.add(key) rows.append( { "product_id": product_id, "sku": sku, "attribute_code": key, "value": value_text, "source_label": source_label, } ) return rows def _seo_fields_from_payload(payload: Dict[str, Any], *, sku: str) -> Dict[str, str]: fields: Dict[str, str] = {} for canonical_code, aliases in SEO_IMPORT_ALIASES.items(): value = _first_by_normalized_key(payload, aliases) if value: fields[canonical_code] = value if fields.get("seo_title") and not fields.get("meta_title"): fields["meta_title"] = fields.pop("seo_title") if fields.get("meta_title") and not fields.get("product_url_slug"): fields["product_url_slug"] = build_pdp_slug(fields["meta_title"], sku) return fields def _first_by_normalized_key(payload: Dict[str, Any], aliases: set[str]) -> Optional[str]: wanted = {normalize_column_key(alias) for alias in aliases} for key, value in payload.items(): if normalize_column_key(key) in wanted: return _clean(value) return None def _upsert_single_attribute_values( session: Session, product: MasterProduct, fields: Dict[str, str], source_label: str, ) -> None: for code, value in fields.items(): if code == "seo_title": continue row = session.scalar( select(MasterProductAttributeValue) .where(MasterProductAttributeValue.product_id == product.id) .where(MasterProductAttributeValue.attribute_code == code) ) if row is None: row = MasterProductAttributeValue( product_id=product.id, sku=product.sku, attribute_code=code, ) session.add(row) row.value = value row.source_label = source_label def _refresh_variation_parent_titles_for_children( session: Session, child_skus: Sequence[str], *, source_label: str, dry_run: bool, skip_parent_skus: Optional[Sequence[str]] = None, ) -> Dict[str, Any]: wanted = sorted({str(sku or "").strip() for sku in child_skus if str(sku or "").strip()}) skipped_explicit = sorted({str(sku or "").strip() for sku in (skip_parent_skus or []) if str(sku or "").strip()}) if not wanted: return { "matched_parent_count": 0, "would_update": 0, "updated": 0, "skipped_explicit_seo": 0, "sample_updates": [], } relations = list( session.scalars( select(MasterProductRelation) .where(MasterProductRelation.child_sku.in_(wanted)) .order_by(MasterProductRelation.parent_sku, MasterProductRelation.sort_order) ).all() ) by_parent: Dict[str, List[MasterProductRelation]] = {} for relation in relations: by_parent.setdefault(relation.parent_sku, []).append(relation) raw_payload_parents = _variation_parent_payload_children(session, wanted) for parent_sku in raw_payload_parents: by_parent.setdefault(parent_sku, []) parent_skus = sorted(by_parent) if not parent_skus: return { "matched_parent_count": 0, "would_update": 0, "updated": 0, "skipped_explicit_seo": 0, "sample_updates": [], } all_child_skus = sorted({relation.child_sku for relation in relations} | {sku for children in raw_payload_parents.values() for sku in children}) products = list( session.scalars( select(MasterProduct).where(MasterProduct.sku.in_(parent_skus + all_child_skus)) ).all() ) product_by_sku = {product.sku: product for product in products} attrs_by_sku = _attribute_values_by_sku(session, all_child_skus) skip_set = set(skipped_explicit) from db.variation_builder import variation_parent_name_for_product updates: List[Dict[str, str]] = [] skipped_explicit_count = 0 for parent_sku in parent_skus: parent = product_by_sku.get(parent_sku) if parent is None: continue parent_relations = by_parent.get(parent_sku) or [] if not _looks_like_variation_builder_parent(parent, parent_relations): continue raw_payload = parent.raw_payload if isinstance(parent.raw_payload, dict) else {} if parent_sku in skip_set or raw_payload.get("explicit_seo_title"): skipped_explicit_count += 1 continue relation_child_skus = [relation.child_sku for relation in parent_relations] child_skus_for_parent = relation_child_skus or raw_payload_parents.get(parent_sku, []) children = [product_by_sku[sku] for sku in child_skus_for_parent if sku in product_by_sku] group_values = raw_payload.get("group_values") if isinstance(raw_payload, dict) else {} new_name = variation_parent_name_for_product( parent, children, attrs_by_sku=attrs_by_sku, group_values=group_values if isinstance(group_values, dict) else {}, ) old_name = str(parent.name or "").strip() if not new_name or new_name == old_name: continue updates.append({"sku": parent_sku, "old_name": old_name, "new_name": new_name}) if dry_run: continue parent.name = new_name next_payload = dict(raw_payload or {}) refreshes = list(next_payload.get("seo_parent_title_refreshes") or []) refreshes.append({"source": source_label, "old_name": old_name, "new_name": new_name}) next_payload["seo_parent_title_refreshes"] = refreshes[-5:] parent.raw_payload = next_payload parent.row_hash = _stable_hash(next_payload) _upsert_single_attribute_values( session, parent, { "meta_title": new_name, "product_url_slug": build_pdp_slug(new_name, parent_sku), }, source_label, ) return { "matched_parent_count": len(parent_skus), "relation_parent_count": sum(1 for parent_sku in parent_skus if by_parent.get(parent_sku)), "raw_payload_parent_count": len(raw_payload_parents), "skipped_explicit_seo": skipped_explicit_count, "would_update": len(updates), "updated": 0 if dry_run else len(updates), "sample_updates": updates[:25], } def _variation_parent_payload_children(session: Session, child_skus: Sequence[str]) -> Dict[str, List[str]]: wanted = {str(sku or "").strip() for sku in child_skus if str(sku or "").strip()} if not wanted: return {} out: Dict[str, List[str]] = {} candidates = session.scalars( select(MasterProduct) .where(MasterProduct.raw_payload.is_not(None)) .where(MasterProduct.is_active.is_(True)) .order_by(MasterProduct.sku) ).all() for product in candidates: payload = product.raw_payload if isinstance(product.raw_payload, dict) else {} if payload.get("generated_by") != "variation_builder": continue children = [ str(sku or "").strip() for sku in (payload.get("children") or []) if str(sku or "").strip() ] if wanted.intersection(children): out[product.sku] = children return out def _attribute_values_by_sku(session: Session, skus: Sequence[str]) -> Dict[str, Dict[str, str]]: wanted = sorted({str(sku or "").strip() for sku in skus if str(sku or "").strip()}) if not wanted: return {} rows: Dict[str, Dict[str, str]] = {} for value in session.scalars( select(MasterProductAttributeValue).where(MasterProductAttributeValue.sku.in_(wanted)) ).all(): if value.value is None or value.value == "": continue rows.setdefault(value.sku, {})[normalize_column_key(value.attribute_code)] = str(value.value) return rows def _looks_like_variation_builder_parent( product: MasterProduct, relations: Sequence[MasterProductRelation], ) -> bool: raw_payload = product.raw_payload if isinstance(product.raw_payload, dict) else {} if raw_payload.get("generated_by") == "variation_builder": return True if any(str(relation.source_label or "").startswith("variation_builder") for relation in relations): return True sku = str(product.sku or "").upper() return sku.endswith("-PARENT") or sku.endswith("-CONFIG") def _attribute_definitions(payload: Dict[str, Any]) -> List[Dict[str, Any]]: definitions = [] seen_codes: set[str] = set() for column_name in payload: key = _canonical_attribute_code(column_name) if not key or key in seen_codes: continue if key in CORE_FIELD_KEYS or key in OPERATIONAL_FIELD_KEYS: continue seen_codes.add(key) purpose = _attribute_purpose(key) definitions.append( { "attribute_code": key, "label": column_name, "purpose": purpose, "data_type": _attribute_data_type(key), "target_scopes": _target_scopes_for_purpose(purpose), "is_required": False, "is_active": True, "notes": None, } ) return definitions def _augment_master_payload(payload: Dict[str, Any]) -> Dict[str, Any]: payload = dict(payload) category = _first_text(payload, "Category") collection = _collection_from_payload(payload, category) if collection and not _first_text(payload, "Master Collection"): payload["Master Collection"] = collection color = _color_from_collection(collection) if color and not _first_text(payload, "Collection Color"): payload["Collection Color"] = color style = _first_text(payload, "Cabinet Door Style") if style and not _first_text(payload, "Style Family"): payload["Style Family"] = style window_frame = _window_frame_from_payload(payload) if window_frame and not _first_text(payload, "Frame"): payload["Frame"] = window_frame window_glass = _window_glass_from_payload(payload) if window_glass and not _first_text(payload, "Glass"): payload["Glass"] = window_glass sub1, sub2 = _derive_cabinet_subcategories(payload) glass_wall_override = _derive_glass_wall_cabinet_subcategories(payload) if sub1 and (glass_wall_override or not _first_text(payload, "Cabinet Type (Sub-Category 1)")): payload["Cabinet Type (Sub-Category 1)"] = sub1 if sub2 and (glass_wall_override or not _first_text(payload, "Cabinet Type (Sub-Category 2)")): payload["Cabinet Type (Sub-Category 2)"] = sub2 dimensions = _variation_dimensions_from_payload(payload) for key, label in ( ("width", "Variation Width In"), ("height", "Variation Height In"), ("depth", "Variation Depth In"), ): value = dimensions.get(key) if value and not _first_text(payload, label): payload[label] = value axis = _variation_axis(dimensions) if axis and not _first_text(payload, "Variation Axis"): payload["Variation Axis"] = axis option_label = _variation_option_label(dimensions) if option_label and not _first_text(payload, "Variation Option Label"): payload["Variation Option Label"] = option_label group_code = _variation_group_code_from_payload(payload) if group_code and not _first_text(payload, "Variation Group Code"): payload["Variation Group Code"] = group_code return payload def _assignment_rows(product_id: int, sku: str, payload: Dict[str, Any]) -> List[Dict[str, Any]]: rows = [] websites = _first_text(payload, "Product Websites") if websites: for part in websites.replace("|", ",").split(","): channel = normalize_column_key(part) if channel: rows.append( { "product_id": product_id, "sku": sku, "channel_code": channel, "assignment_status": "active", "reason": "product_websites", "payload": {"source_value": websites}, } ) return rows def _price_rows(product_id: int, sku: str, payload: Dict[str, Any], source_label: str) -> List[Dict[str, Any]]: rows = [] for column_key, price_type in PRICE_COLUMNS.items(): raw = _value_by_key(payload, column_key) amount = _to_float(raw) if amount is None: continue rows.append( { "product_id": product_id, "sku": sku, "price_type": price_type, "amount": amount, "currency": "USD", "source_label": source_label, } ) return rows def _availability_rows(product_id: int, sku: str, payload: Dict[str, Any], source_label: str) -> List[Dict[str, Any]]: rows = [] for column_key, location_code in LOCATION_COLUMNS.items(): raw = _value_by_key(payload, column_key) rows.append( { "product_id": product_id, "sku": sku, "location_code": location_code, "is_available": _truthy(raw), "source_value": raw, "source_label": source_label, } ) return rows def _upsert_attribute_definitions(session: Session, rows: List[Dict[str, Any]]) -> None: for batch in _batches(rows, BULK_WRITE_BATCH_SIZE): stmt = insert(MasterAttributeDefinition).values(batch) stmt = stmt.on_conflict_do_update( constraint="uq_master_attribute_definition_code", set_={ "label": stmt.excluded.label, "purpose": stmt.excluded.purpose, "data_type": stmt.excluded.data_type, "target_scopes": stmt.excluded.target_scopes, "is_active": stmt.excluded.is_active, "updated_at": sa.func.now(), }, ) session.execute(stmt) def _upsert_attribute_values(session: Session, rows: List[Dict[str, Any]]) -> None: for batch in _batches(rows, BULK_WRITE_BATCH_SIZE): stmt = insert(MasterProductAttributeValue).values(batch) stmt = stmt.on_conflict_do_update( constraint="uq_master_product_attr_product_code", set_={ "sku": stmt.excluded.sku, "value": stmt.excluded.value, "source_label": stmt.excluded.source_label, "updated_at": sa.func.now(), }, ) session.execute(stmt) def _upsert_prices(session: Session, rows: List[Dict[str, Any]]) -> None: for batch in _batches(rows, BULK_WRITE_BATCH_SIZE): stmt = insert(MasterProductPrice).values(batch) stmt = stmt.on_conflict_do_update( constraint="uq_master_product_price_product_type", set_={ "sku": stmt.excluded.sku, "amount": stmt.excluded.amount, "currency": stmt.excluded.currency, "source_label": stmt.excluded.source_label, "updated_at": sa.func.now(), }, ) session.execute(stmt) def _upsert_availability(session: Session, rows: List[Dict[str, Any]]) -> None: for batch in _batches(rows, BULK_WRITE_BATCH_SIZE): stmt = insert(MasterProductLocationAvailability).values(batch) stmt = stmt.on_conflict_do_update( constraint="uq_master_product_location_product_location", set_={ "sku": stmt.excluded.sku, "is_available": stmt.excluded.is_available, "source_value": stmt.excluded.source_value, "source_label": stmt.excluded.source_label, "updated_at": sa.func.now(), }, ) session.execute(stmt) def _upsert_assignments(session: Session, rows: List[Dict[str, Any]]) -> None: for batch in _batches(rows, BULK_WRITE_BATCH_SIZE): stmt = insert(ProductChannelAssignment).values(batch) stmt = stmt.on_conflict_do_update( constraint="uq_product_channel_assignment_product_channel", set_={ "sku": stmt.excluded.sku, "assignment_status": stmt.excluded.assignment_status, "reason": stmt.excluded.reason, "payload": stmt.excluded.payload, "updated_at": sa.func.now(), }, ) session.execute(stmt) def _batches(rows: List[Any], size: int): for index in range(0, len(rows), size): yield rows[index : index + size] def _category_parts(category: Optional[str]) -> Tuple[Optional[str], Optional[str], Optional[str]]: parts = [p.strip() for p in str(category or "").split(":") if not _is_blank_marker(p)] parts = parts[:3] + [None] * (3 - len(parts)) return parts[0], parts[1], parts[2] def _canonical_product_category_children( payload: Dict[str, Any], product_category: Optional[str], raw_l2: Optional[str], raw_l3: Optional[str], ) -> Tuple[Optional[str], Optional[str]]: sub1 = _first_text(payload, "Cabinet Type (Sub-Category 1)") sub2 = _first_text(payload, "Cabinet Type (Sub-Category 2)") if not sub1 or not sub2: inferred1, inferred2 = _derive_cabinet_subcategories(payload) sub1 = sub1 or inferred1 sub2 = sub2 or inferred2 category_l2 = _canonical_category_label(sub1) category_l3 = _canonical_category_label(sub2) product_category_text = str(product_category or "").strip() if not category_l2 and product_category_text in {"Interior Doors", "Locks & Hardware"}: category_l2 = raw_l2 if not category_l2 and product_category_text == "Bathroom Vanities" and raw_l3 == "Vanity Tops": category_l2 = "Vanity Tops" return category_l2, category_l3 def _canonical_category_label(value: Optional[str]) -> Optional[str]: if not _clean(value): return None from db.catalog_intent_planner import _normalize_customer_category_label return _normalize_customer_category_label(value) def _family_from_category(category: Optional[str]) -> Optional[str]: first, _, _ = _category_parts(category) text = str(first or "").lower() if "kitchen cabinet" in text: return "Cabinets" if "vanit" in text: return "Vanities" if "door" in text: return "Doors" if "window" in text: return "Windows" if "shower" in text: return "Showers" if first: return first return None def _brand_from_category(category: Optional[str]) -> Optional[str]: _, brand, _ = _category_parts(category) return brand def _brand_from_payload(payload: Dict[str, Any], category: Optional[str], manufacturer: Optional[str]) -> Optional[str]: cat_l1, cat_l2, _ = _category_parts(category) product_category = _first_text(payload, "Product Category") or cat_l1 brand = None if cat_l2 and (cat_l1 in CABINET_CATEGORY_ROOTS or product_category in CABINET_CATEGORY_ROOTS): brand = cat_l2 if _category_segment_is_brand(cat_l2) else None if cat_l2 and cat_l1 == "Kitchen Cabinets" and not brand and _category_segment_is_brand(cat_l2): brand = cat_l2 if not brand: brand = _first_text(payload, "Cabinet Brand", "Brand", "brand") if not brand: brand = manufacturer if not brand: brand = _brand_from_product_identity(payload) return _normalize_brand(brand) def _normalize_brand(brand: Optional[str]) -> Optional[str]: text = str(brand or "").strip() if not text: return None key = normalize_column_key(text).replace("_", "") known = { "castlecraft": "CastleCraft", "castlecraftcabinetry": "CastleCraft", "portbell": "Port & Bell", "portandbell": "Port & Bell", "fabuwood": "Fabuwood", "echelon": "Echelon", "legionmanor": "Legion Manor", "sunnywood": "Sunnywood", "legacy": "Legacy", "northpointcabinetry": "NorthPoint Cabinetry", "kountrywood": "Kountry Wood", "kountrywoodproducts": "Kountry Wood", "bridgewaterwholesalers": "Bridgewater Wholesalers", "bridgewaterwholesalersinc": "Bridgewater Wholesalers", "madeparindustriaecomerciodemadeiras": "Madepar", "yijieglass": "Yi Jie Glass", "yijieglassbathroomcoltd": "Yi Jie Glass", "richelieuamericaltd": "Richelieu", "hardwareresources": "Hardware Resources", } return known.get(key, text) def _category_segment_is_brand(value: Optional[str]) -> bool: key = normalize_column_key(value).replace("_", "") return key in { "castlecraft", "castlecraftcabinetry", "portbell", "portandbell", "fabuwood", "echelon", "legionmanor", "sunnywood", "legacy", "northpointcabinetry", "kountrywood", "kountrywoodproducts", } def _brand_from_product_identity(payload: Dict[str, Any]) -> Optional[str]: sku = _first_text(payload, "SKU", "Item", "sku", "item") or "" name = _first_text(payload, "Display Name", "Name", "Label") or "" text = f"{sku} {name}".lower() if sku.upper().startswith("CRY-") or "crystal 100 series" in text: return "CRYSTAL" if sku.upper().startswith("NE-") or "northeast pro series" in text: return "Northeast Pro" return None def _window_frame_from_payload(payload: Dict[str, Any]) -> Optional[str]: sku = (_first_text(payload, "SKU", "Item", "sku", "item") or "").upper() name = (_first_text(payload, "Display Name", "Name", "Label") or "").lower() category = (_first_text(payload, "Category", "Product Category") or "").lower() if "window" not in f"{category} {name}" and not sku.startswith(("CRY-", "NE-")): return None if "-DH-" in sku or "double hung" in name: return "Double Hung" if "-SL-" in sku or "slider" in name or "sliding" in name: return "Slider" return None def _window_glass_from_payload(payload: Dict[str, Any]) -> Optional[str]: sku = (_first_text(payload, "SKU", "Item", "sku", "item") or "").upper() name = (_first_text(payload, "Display Name", "Name", "Label") or "").lower() category = (_first_text(payload, "Category", "Product Category") or "").lower() if "window" not in f"{category} {name}" and not sku.startswith(("CRY-", "NE-")): return None text = f"{sku.lower()} {name}" if "frost" in text: return "Frosted Glass" if "temper" in text: return "Tempered Glass" if "grid" in text or "grille" in text: return "Grids" return "Clear Glass" def _collection_from_payload(payload: Dict[str, Any], category: Optional[str]) -> Optional[str]: cat_l1, _, cat_l3 = _category_parts(category) product_category = _first_text(payload, "Product Category") or cat_l1 from_prefix = _collection_from_style_prefix(_first_text(payload, "Item Style Prefix")) if cat_l3 in NON_COLLECTION_CATEGORY_LEAVES: return None if cat_l3 and (cat_l1 in CABINET_CATEGORY_ROOTS or product_category in CABINET_CATEGORY_ROOTS): for candidate in (from_prefix,): if _collection_candidate_extends_category(candidate, cat_l3): return canonical_collection(candidate) return canonical_collection(cat_l3) if from_prefix: return from_prefix series = _first_text(payload, "Product Collection/Series Name") if series and cat_l3 and _norm_collection_key(series) != _norm_collection_key(cat_l3): return canonical_collection(cat_l3) return canonical_collection(series) or canonical_collection(cat_l3) def _manual_taxonomy_source_sku(payload: Dict[str, Any]) -> Optional[str]: for value in ( _first_text(payload, "Item Size"), _first_text(payload, "Item"), _first_text(payload, "SKU", "sku"), ): text = str(value or "").strip().upper() if text and text not in {"-", "N/A"}: return text return None def _canonical_attribute_code(column_name: str) -> Optional[str]: key = normalize_column_key(column_name) if not key: return None key = ATTRIBUTE_CODE_ALIASES.get(key, key) if key in ATTRIBUTE_CODE_SKIP_KEYS: return None return key def _resolve_master_collection( session: Session, payload: Dict[str, Any], category: Optional[str], ) -> Optional[Dict[str, Any]]: candidates: list[str] = [] prefix_collection = _collection_from_style_prefix(_first_text(payload, "Item Style Prefix")) if prefix_collection: candidates.append(prefix_collection) series = _first_text(payload, "Product Collection/Series Name") if series: candidates.append(series) _, _, cat_l3 = _category_parts(category) if cat_l3: candidates.append(cat_l3) rows = list( session.scalars(select(MasterCollectionRegistry).where(MasterCollectionRegistry.is_active.is_(True))).all() ) by_name: dict[str, MasterCollectionRegistry] = {} by_code: dict[str, MasterCollectionRegistry] = {} for row in rows: name_key = _norm_collection_key(row.name) if name_key: by_name.setdefault(name_key, row) for raw_code in ( str(row.code or "").strip().upper(), str((row.aliases or {}).get("source_collection_code") or "").strip().upper() if isinstance(row.aliases, dict) else "", ): if raw_code: by_code.setdefault(raw_code, row) prefix = str(_first_text(payload, "Item Style Prefix") or "").strip().upper() if prefix and prefix not in {"-"}: match = by_code.get(prefix) if match: return {"id": int(match.id), "name": str(match.name or "").strip()} for candidate in candidates: match = by_name.get(_norm_collection_key(candidate)) if match: return {"id": int(match.id), "name": str(match.name or "").strip()} return None def _collection_from_style_prefix(prefix: Optional[str]) -> Optional[str]: return collection_name_for_style_code(prefix) def _collection_from_item_style(item_style: Optional[str]) -> Optional[str]: candidate = _collection_candidate_from_item_style(item_style) if not candidate: return None return canonical_collection(candidate) def _collection_candidate_extends_category(candidate: Optional[str], category_leaf: Optional[str]) -> bool: candidate_key = _norm_collection_key(candidate) category_key = _norm_collection_key(category_leaf) if not candidate_key or not category_key or candidate_key == category_key: return False if len(candidate_key) <= len(category_key): return False return category_key in candidate_key def _collection_candidate_from_item_style(item_style: Optional[str]) -> Optional[str]: text = _strip_item_style_noise(item_style) if not text: return None lowered = text.lower() for suffix in _ASSEMBLY_SUFFIXES: if lowered.endswith(suffix.lower()): candidate = text[: -len(suffix)].strip() return candidate if _is_collection_style_candidate(candidate) else None if _is_collection_style_candidate(text): return text return None def _strip_item_style_noise(item_style: Optional[str]) -> Optional[str]: text = str(item_style or "").strip() if not text: return None parts = [part.strip() for part in re.split(r"\s*-\s*", text) if part.strip()] kept: List[str] = [] for part in parts: lowered = part.lower() if any(phrase in lowered for phrase in _ITEM_STYLE_NOISE_PHRASES): continue if any(location in lowered for location in _MSI_LOCATION_PHRASES): continue if lowered in {"assembled", "unassembled", "rta"}: continue kept.append(part) return " - ".join(kept) if kept else None def _normalized_item_style_label(item_style: Optional[str]) -> Optional[str]: text = _strip_item_style_noise(item_style) if not text: return None lowered = text.lower() for suffix in _ASSEMBLY_SUFFIXES: if lowered.endswith(suffix.lower()): return text[: -len(suffix)].strip() or None accessory = _ACCESSORY_ITEM_STYLE_RE.match(text) if accessory: return accessory.group(1).strip() if _item_style_key(text) in _NON_COLLECTION_ITEM_STYLE_KEYS: return text.strip() return text.strip() def _derive_cabinet_subcategories(payload: Dict[str, Any]) -> Tuple[Optional[str], Optional[str]]: sub1 = _first_text(payload, "Cabinet Type (Sub-Category 1)") sub2 = _first_text(payload, "Cabinet Type (Sub-Category 2)") glass_wall_override = _derive_glass_wall_cabinet_subcategories(payload) if glass_wall_override: return glass_wall_override if sub1 and sub2: return sub1, sub2 item_style = _first_text(payload, "Item Style") text = _strip_item_style_noise(item_style) if not text: return sub1 or _infer_cabinet_subcategory(payload), sub2 accessory = _ACCESSORY_ITEM_STYLE_RE.match(text) if accessory: return sub1 or accessory.group(1).strip(), sub2 lowered = text.lower() for suffix in _ASSEMBLY_SUFFIXES: if lowered.endswith(suffix.lower()): return sub1, sub2 if _item_style_key(text) in _NON_COLLECTION_ITEM_STYLE_KEYS: return sub1 or text.strip(), sub2 return sub1 or _infer_cabinet_subcategory(payload), sub2 def _derive_glass_wall_cabinet_subcategories(payload: Dict[str, Any]) -> Optional[Tuple[str, str]]: raw_codes = [ str(value or "").strip().upper() for value in ( _first_text(payload, "Item Size"), _first_text(payload, "sku", "SKU"), _first_text(payload, "Item"), ) if str(value or "").strip() ] if any(code.startswith("GD-W") for code in raw_codes): return None if not any(re.match(r"^(?:GDW|WGD)(?:09|12|15|18|21|24|27|30|33|36|39|42)(30|36|42)(?:MI|GD4|GD6)?$", code) for code in raw_codes): return None descriptor = " ".join( str(value or "") for value in ( _first_text(payload, "Display Name", "Name", "Label"), _first_text(payload, "Cabinet Description"), _first_text(payload, "Item Size"), _first_text(payload, "sku", "SKU"), ) ) descriptor_key = descriptor.lower() if "wall" not in descriptor_key: return None if "glass door" not in descriptor_key and "loose glass door" not in descriptor_key: return None width = _dimensions_from_item_code(_first_text(payload, "Item Size") or _first_text(payload, "sku", "SKU")).get("width") width_value: Optional[float] = None if width: try: width_value = float(width) except (TypeError, ValueError): width_value = None if "double door" in descriptor_key: detail = "Double Door Glass Wall Cabinet" elif "single door" in descriptor_key: detail = "Single Door Glass Wall Cabinet" elif width_value is not None and width_value >= 30: detail = "Double Door Glass Wall Cabinet" else: detail = "Single Door Glass Wall Cabinet" return "Wall", detail def _infer_cabinet_subcategory(payload: Dict[str, Any]) -> Optional[str]: descriptor = " ".join( str(value or "") for value in ( _first_text(payload, "Display Name", "Name", "Label"), _first_text(payload, "Cabinet Description"), _first_text(payload, "Item Size"), _first_text(payload, "sku", "SKU"), ) ) if _CABINET_ACCESSORY_RE.search(descriptor): return "Accessories" if _CABINET_MOULDING_RE.search(descriptor): return "Mouldings" if _CABINET_PANEL_FILLER_RE.search(descriptor): return "Panels and Fillers" for code in _cabinet_item_codes(payload): if re.match(r"^(?:TUKIT|TUKITS|KIT)", code): return "Accessories" if re.match(r"^(?:SCM|SM|CM|OCM|ICM|LRB|LRS|QR|SHM|TKC?|BM|CMC|CMV)", code): return "Mouldings" if re.match(r"^(?:BF|F|WF|WEP|DWEP|REP|PH)", code): return "Panels and Fillers" if re.match(r"^(?:GDW|WCOR|WDC|W)", code): return "Wall" if re.match(r"^(?:T|U)", code): return "Tall" if re.match(r"^(?:DB|SB|BBC|BLS|LS|BC|B)", code): return "Base" return None def _cabinet_item_codes(payload: Dict[str, Any]) -> List[str]: codes: List[str] = [] for value in ( _first_text(payload, "Item Size"), _first_text(payload, "sku", "SKU"), _first_text(payload, "Item"), ): text = str(value or "").strip().upper() if not text: continue candidates = [text] if "-" in text: candidates.append(text.split("-", 1)[-1]) for candidate in candidates: code = re.sub(r"[^A-Z0-9]+", "", candidate) if code and code not in codes: codes.append(code) return codes def _is_collection_style_candidate(value: Optional[str]) -> bool: text = str(value or "").strip() if not text or _is_blank_marker(text): return False if ":" in text: return False key = _item_style_key(text) if key in _NON_COLLECTION_ITEM_STYLE_KEYS or key in _BRAND_ONLY_ITEM_STYLE_KEYS: return False if any(phrase in text.lower() for phrase in _ITEM_STYLE_NOISE_PHRASES): return False return True def _item_style_key(value: Optional[str]) -> str: return re.sub(r"[^a-z0-9]+", "", str(value or "").lower()) def _norm_collection_key(value: Optional[str]) -> str: return _item_style_key(value) def _suspect_collection_name(collection: str, category_l1: str, category_l2: str) -> bool: if not collection: return False lowered = collection.lower() if ":" in collection: return True if category_l1 and lowered == category_l1.lower(): return True if category_l2 and lowered == category_l2.lower(): return True return any(lowered.startswith(root.lower()) for root in CABINET_CATEGORY_ROOTS) def _color_from_collection(collection: Optional[str]) -> Optional[str]: text = str(collection or "").strip().lower() if not text: return None rules = [ ("Black", ("black",)), ("Espresso", ("espresso",)), ("Blue", ("blue", "ocean")), ("Grey", ("gray", "grey", "slate", "mist", "stone")), ("Cream / Ivory", ("cream", "ivory", "sandstone")), ("Brown", ("brown", "caramel", "toffee", "pecan", "honey", "golden oak")), ("Natural Oak", ("natural rift white oak", "natural oak")), ("Wood Tones", ("oak", "rift", "natural", "maple")), ("White", ("white", "snow", "pearl", "frost", "polar")), ] for color, tokens in rules: if any(token in text for token in tokens): return color return None def _variation_dimensions_from_payload(payload: Dict[str, Any]) -> Dict[str, str]: dimensions: Dict[str, str] = {} width = _dimension_number(_first_text(payload, "Assembled Width")) height = _dimension_number(_first_text(payload, "Assembled Height")) depth = _dimension_number(_first_text(payload, "Assembled Depth")) if width: dimensions["width"] = width if height: dimensions["height"] = height if depth: dimensions["depth"] = depth text = " ".join( str(value or "") for value in ( _first_text(payload, "Cabinet Description"), _first_text(payload, "Display Name"), _first_text(payload, "Item Size"), _first_text(payload, "sku", "SKU"), ) ) if "width" not in dimensions: width = _dimension_from_text(text, ("wide", "width", "w")) if width: dimensions["width"] = width if "height" not in dimensions: height = _dimension_from_text(text, ("tall", "high", "height", "h")) if height: dimensions["height"] = height if "depth" not in dimensions: depth = _dimension_from_text(text, ("deep", "depth", "d")) if depth: dimensions["depth"] = depth if "width" not in dimensions or "height" not in dimensions: code_dims = _dimensions_from_item_code(_first_text(payload, "Item Size") or _first_text(payload, "sku", "SKU")) for key, value in code_dims.items(): dimensions.setdefault(key, value) return dimensions def _variation_group_code_from_payload(payload: Dict[str, Any]) -> Optional[str]: text = _first_text(payload, "Item Size") from_sku = False if not text: text = _first_text(payload, "sku", "SKU") from_sku = True text = str(text or "").strip().upper() if not text: return None if from_sku and "-" in text: text = text.split("-", 1)[-1] text = re.sub(r"(?:L/R|L|R|LEFT|RIGHT)$", "", text) text = re.sub(r"\d+(?:\.\d+)?(?:/\d+)?", "", text) text = re.sub(r"[^A-Z]+", "-", text).strip("-") return text or None def _dimension_from_text(text: str, labels: Tuple[str, ...]) -> Optional[str]: blob = str(text or "").lower() label_pattern = "|".join(re.escape(label) for label in labels) patterns = [ rf"(\d+(?:\.\d+)?)\s*(?:\"|in|inch|inches)?\s*(?:{label_pattern})\b", rf"\b(\d+(?:\.\d+)?)\s*(?:{label_pattern})\b", ] for pattern in patterns: match = re.search(pattern, blob) if match: return _dimension_number(match.group(1)) return None def _dimensions_from_item_code(value: Optional[str]) -> Dict[str, str]: text = str(value or "").upper() match = re.search(r"([A-Z]+)[-/]?(\d{2})(\d{2})(?:\D|$)", text) if match: return {"width": _dimension_number(match.group(2)), "height": _dimension_number(match.group(3))} match = re.search(r"([A-Z]+)[-/]?(\d{2})(?:\D|$)", text) if match: return {"width": _dimension_number(match.group(2))} return {} def _dimension_number(value: Optional[str]) -> Optional[str]: text = str(value or "").strip() if not text: return None match = re.search(r"\d+(?:\.\d+)?", text) if not match: return None number = float(match.group(0)) return str(int(number)) if number.is_integer() else str(number).rstrip("0").rstrip(".") def _variation_axis(dimensions: Dict[str, str]) -> Optional[str]: keys = [key for key in ("width", "height", "depth") if dimensions.get(key)] return "_".join(keys) if keys else None def _variation_option_label(dimensions: Dict[str, str]) -> Optional[str]: parts = [] if dimensions.get("width"): parts.append(f'{dimensions["width"]} in W') if dimensions.get("height"): parts.append(f'{dimensions["height"]} in H') if dimensions.get("depth"): parts.append(f'{dimensions["depth"]} in D') return " x ".join(parts) if parts else None def _assembly_type(sku: str, item_style: Optional[str], explicit: Optional[str] = None) -> Optional[str]: explicit_text = str(explicit or "").strip().lower() if explicit_text in {"rta", "unassembled", "unassembled/rta", "ready-to-assemble", "ready to assemble"}: return "RTA" if "unassembled" in explicit_text or explicit_text.endswith("/rta"): return "RTA" if explicit_text == "assembled": return "Assembled" text = f"{sku} {item_style or ''}".lower() if "unassembled" in text or "rta-" in text or text.startswith("rta"): return "RTA" if "assembled" in text: return "Assembled" return None def _attribute_purpose(attribute_code: str) -> str: key = normalize_column_key(attribute_code) if key in ADMIN_ATTRIBUTE_KEYS: return "admin" if key in OPERATIONAL_FIELD_KEYS or key in {"stocked"}: return "operational" if key.startswith(("meta_", "seo_")): return "seo" if key.startswith("shopify_"): return "channel_specific" if key in PRESENTATION_FIELD_KEYS: return "presentation" return "presentation" def _attribute_data_type(attribute_code: str) -> str: key = normalize_column_key(attribute_code) if any(token in key for token in ("price", "width", "height", "depth", "weight", "count", "length")): return "number" if key.endswith("_included") or key.endswith("_compliant") or key in {"easy_clean", "scratch_resistant"}: return "boolean" if "image" in key or "thumbnail" in key: return "image" return "text" def _target_scopes_for_purpose(purpose: str) -> Dict[str, bool]: if purpose == "admin": return {"admin": True} if purpose == "operational": return {"admin": True, "magento": True, "shopify": True} if purpose == "seo": return {"admin": True, "magento": True, "shopify": True, "seo": True} if purpose == "channel_specific": return {"admin": True, "shopify": True} return {"admin": True, "magento": True, "shopify": True} def _first_text(payload: Dict[str, Any], *names: str) -> Optional[str]: exact = {str(k): v for k, v in payload.items()} keyed = {normalize_column_key(k): v for k, v in payload.items()} for name in names: if name in exact: return _clean(exact[name]) key = normalize_column_key(name) if key in keyed: return _clean(keyed[key]) return None def _value_by_key(payload: Dict[str, Any], column_key: str) -> Optional[str]: wanted = normalize_column_key(column_key) for key, value in payload.items(): if normalize_column_key(key) == wanted: return _clean(value) return None def _truthy(value: Optional[str]) -> bool: text = str(value or "").strip().lower() return text not in {"", "-", "0", "no", "false", "none", "null"} def _to_bool(value: Optional[str]) -> Optional[bool]: if value is None: return None text = value.strip().lower() if text in {"1", "yes", "true", "y"}: return True if text in {"0", "no", "false", "n"}: return False return None def _to_float(value: Optional[str]) -> Optional[float]: text = _clean(value) if text is None: return None try: return float(text.replace("$", "").replace(",", "")) except (TypeError, ValueError): return None def _clean(value: Any) -> Optional[str]: if value is None: return None text = str(value).strip() if _is_blank_marker(text): return None return text def _is_blank_marker(value: Any) -> bool: text = str(value or "").strip() return not text or text.lower() in {"nan", "none", "null", "-", "--", "\u2014"} def _stable_hash(payload: Dict[str, Any]) -> str: blob = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str) return hashlib.sha256(blob.encode("utf-8")).hexdigest()