from __future__ import annotations from typing import Any, Dict, List, Optional from sqlalchemy import func, select from sqlalchemy.orm import Session from channel.url_canonical import build_pdp_url, build_plp_url from db.channel_exports import resolve_pipeline_channel_code from db.channel_listing_path import ( build_pdp_preview, list_listing_paths, resolve_listing_paths_for_sku, ) from db.channel_taxonomy import export_taxonomy_csv_rows, list_taxonomy_mappings from db.models import ChannelListingPath, MasterProduct, ProductChannelTaxonomyAssignment, ProductListingPathAssignment from db.product_channel_taxonomy import ( MAGENTO_KIND, SHOPIFY_COLLECTION_KIND, SHOPIFY_PRODUCT_CATEGORY_KIND, resolve_filtered_master_skus, resolve_magento_category_targets, resolve_shopify_collection_ids, resolve_shopify_product_category_id, ) from shopify.collection_sync import list_shopify_collections from shopify.taxonomy_sync import list_taxonomy_registry_rows from db.product_taxonomy_workspace import enrich_workspace_sku_rows, filter_skus_without_channel_products def build_taxonomy_workspace( session: Session, *, magento_connection_id: Optional[int] = None, shopify_connection_id: Optional[int] = None, search: Optional[str] = None, ) -> Dict[str, Any]: """Single dashboard payload: taxonomy options, listing paths, URL rules, assignment counts.""" listing_paths = list_listing_paths(session, search=search, limit=500) magento_categories = export_taxonomy_csv_rows(session, "magento", connection_id=magento_connection_id) shopify_collections = list_shopify_collections( session, connection_id=shopify_connection_id, search=search, limit=500, ) shopify_product_categories = list_taxonomy_registry_rows(session, search=search, limit=500) return { "workspace_kind": "taxonomy_assignment", "url_rules": { "pdp_pattern": "/{pretty-name}-{sku}", "pdp_example": "/anna-snow-white-b15-aswb15", "plp_pattern": "/{path_slug}/", "plp_examples": [ "/kitchen-cabinets/", "/kitchen-cabinets/shaker/", "/kitchen-cabinets/base-cabinets/", "/kitchen-cabinets/white-shaker/", ], "notes": [ "PDP is unique per SKU at site root.", "PLP paths are many-to-many: one SKU can appear on multiple listing paths.", "Assign PLP paths for storefront canonicals; channel targets map paths to Magento categories / Shopify collections.", ], }, "filters": { "fields": [ "skus", "search", "category_l1", "category_l2", "category_l3", "collection", "product_family", "only_assigned", ], }, "counts": _assignment_counts(session), "listing_paths": listing_paths, "magento": { "connection_id": magento_connection_id, "categories": magento_categories[:500], "taxonomy_mappings": list_taxonomy_mappings( session, channel_code="magento", connection_id=magento_connection_id, )[:200], }, "shopify": { "connection_id": shopify_connection_id, "collections": shopify_collections, "product_categories": shopify_product_categories, "taxonomy_mappings": list_taxonomy_mappings( session, channel_code="shopify", connection_id=shopify_connection_id, )[:200], }, "assignment_kinds": { "magento": [ {"kind": MAGENTO_KIND, "label": "Magento Category", "multi": True}, {"kind": "listing_path", "label": "PLP Listing Path", "multi": True}, ], "shopify": [ {"kind": SHOPIFY_PRODUCT_CATEGORY_KIND, "label": "Shopify Product Category", "multi": False}, {"kind": SHOPIFY_COLLECTION_KIND, "label": "Shopify Collection", "multi": True}, {"kind": "listing_path", "label": "PLP Listing Path", "multi": True}, ], }, } def resolve_workspace_filtered_skus( session: Session, *, channel_code: Optional[str] = None, magento_connection_id: Optional[int] = None, shopify_connection_id: Optional[int] = None, only_assigned: bool = True, category_l1: Optional[str] = None, category_l2: Optional[str] = None, category_l3: Optional[str] = None, collection: Optional[str] = None, product_family: Optional[str] = None, search: Optional[str] = None, master_filters: Optional[Any] = None, without_listing_paths: bool = False, without_magento: bool = False, without_shopify: bool = False, ) -> List[str]: """All master SKUs matching the same filters as the taxonomy workspace grid.""" skus = resolve_filtered_master_skus( session, channel_code=channel_code, only_assigned=only_assigned, category_l1=category_l1, category_l2=category_l2, category_l3=category_l3, collection=collection, product_family=product_family, search=search, master_filters=master_filters, ) if without_listing_paths: skus = [sku for sku in skus if not resolve_listing_paths_for_sku(session, sku)] if without_magento or without_shopify: skus = filter_skus_without_channel_products( session, skus, without_magento=without_magento, without_shopify=without_shopify, magento_connection_id=magento_connection_id, shopify_connection_id=shopify_connection_id, ) return skus def build_workspace_sku_rows( session: Session, *, channel_code: Optional[str] = None, magento_connection_id: Optional[int] = None, shopify_connection_id: Optional[int] = None, only_assigned: bool = True, category_l1: Optional[str] = None, category_l2: Optional[str] = None, category_l3: Optional[str] = None, collection: Optional[str] = None, product_family: Optional[str] = None, search: Optional[str] = None, master_filters: Optional[Any] = None, without_listing_paths: bool = False, without_magento: bool = False, without_shopify: bool = False, limit: int = 100, offset: int = 0, ) -> Dict[str, Any]: """Paginated SKU rows for assignment UI with PDP/PLP previews and current placements.""" skus = resolve_workspace_filtered_skus( session, channel_code=channel_code, magento_connection_id=magento_connection_id, shopify_connection_id=shopify_connection_id, only_assigned=only_assigned, category_l1=category_l1, category_l2=category_l2, category_l3=category_l3, collection=collection, product_family=product_family, search=search, master_filters=master_filters, without_listing_paths=without_listing_paths, without_magento=without_magento, without_shopify=without_shopify, ) total = len(skus) page_skus = skus[offset : offset + limit] if not page_skus: return {"total": total, "offset": offset, "limit": limit, "rows": []} products = { p.sku: p for p in session.scalars( select(MasterProduct).where(MasterProduct.sku.in_(page_skus)) ).all() } channel = resolve_pipeline_channel_code(channel_code) if channel_code else None rows: List[Dict[str, Any]] = [] for sku in page_skus: product = products.get(sku) if product is None: continue fields = { "sku": product.sku, "title": product.name, "name": product.name, "category_l1": product.category_l1, "category_l2": product.category_l2, "category_l3": product.category_l3, "collection": product.collection, "product_family": product.product_family, } pdp = build_pdp_preview(session, product) listing_paths = resolve_listing_paths_for_sku(session, sku) row: Dict[str, Any] = { "master_sku": sku, "name": product.name, "category_l1": product.category_l1, "category_l2": product.category_l2, "category_l3": product.category_l3, "collection": product.collection, "product_family": product.product_family, "pdp": pdp, "plp_paths": [ { "path_slug": p["path_slug"], "plp_url": p["plp_url"], "title": p["title"], "path_kind": p["path_kind"], "targets": p.get("targets") or [], } for p in listing_paths ], "plp_urls": [p["plp_url"] for p in listing_paths], "direct_assignments": _direct_assignments_for_sku(session, sku, channel_code=channel), } if channel == "magento": category_ids, category_paths = resolve_magento_category_targets( session, master_sku=sku, canonical_fields=fields, connection_id=magento_connection_id, ) row["magento"] = { "category_ids": category_ids, "category_paths": category_paths, } elif channel == "shopify": row["shopify"] = { "product_category_id": resolve_shopify_product_category_id( session, master_sku=sku, canonical_fields=fields, connection_id=shopify_connection_id, ), "collection_ids": resolve_shopify_collection_ids( session, master_sku=sku, connection_id=shopify_connection_id, ), } rows.append(row) enrich_workspace_sku_rows( session, rows, magento_connection_id=magento_connection_id, shopify_connection_id=shopify_connection_id, ) return {"total": total, "offset": offset, "limit": limit, "rows": rows} def _assignment_counts(session: Session) -> Dict[str, Any]: active_master = session.scalar( select(func.count()).select_from(MasterProduct).where(MasterProduct.is_active.is_(True)) ) or 0 taxonomy_assignments = session.scalar( select(func.count()) .select_from(ProductChannelTaxonomyAssignment) .where(ProductChannelTaxonomyAssignment.assignment_status == "active") ) or 0 listing_assignments = session.scalar( select(func.count()) .select_from(ProductListingPathAssignment) .where(ProductListingPathAssignment.assignment_status == "active") ) or 0 listing_paths = session.scalar( select(func.count()).select_from(ChannelListingPath).where(ChannelListingPath.is_active.is_(True)) ) or 0 return { "active_master_products": active_master, "taxonomy_assignments": taxonomy_assignments, "listing_path_assignments": listing_assignments, "listing_paths": listing_paths, } def _direct_assignments_for_sku( session: Session, master_sku: str, *, channel_code: Optional[str], ) -> List[Dict[str, Any]]: stmt = ( select(ProductChannelTaxonomyAssignment) .where(ProductChannelTaxonomyAssignment.master_sku == master_sku) .where(ProductChannelTaxonomyAssignment.assignment_status == "active") .order_by(ProductChannelTaxonomyAssignment.sort_order, ProductChannelTaxonomyAssignment.id) ) if channel_code: stmt = stmt.where(ProductChannelTaxonomyAssignment.channel_code == channel_code) return [ { "channel_code": row.channel_code, "taxonomy_kind": row.taxonomy_kind, "remote_id": row.remote_id, "remote_path": row.remote_path, "plp_url": build_plp_url(row.remote_path) if row.remote_path and "/" in row.remote_path else None, } for row in session.scalars(stmt).all() ]