"""Relink Magento channel_listing_path_target rows to registry categories that cover the PLP slug. Problem: taxonomy apply can assign SKUs to the correct PLP path while the path's Magento remote_id still points at a leaf-name collision (e.g. base-cabinets/single-door → Wall/Single Door). Push then uses those IDs, so would_push_paths diverge from expected_paths. This module: 1) audits active Magento listing-path targets against magento_category_registry 2) finds a better registry category when the current link does not cover the PLP path 3) deactivates wrong targets and upserts the correct remote_id 4) optionally deactivates stale product_channel_taxonomy_assignment remotes that are not among the SKU's corrected listing-path targets """ from __future__ import annotations from datetime import datetime, timezone from typing import Any, Dict, List, Optional, Sequence, Set, Tuple from sqlalchemy import select from sqlalchemy.orm import Session from channel.url_canonical import normalize_plp_path from db.channel_listing_path import upsert_listing_path_target from db.magento_path_names import MAGENTO_PATH_SEGMENT_ALIASES, plp_slug_to_magento_path_names from db.models import ( ChannelListingPath, ChannelListingPathTarget, MagentoCategoryRegistry, MasterTaxonomyNode, ProductChannelTaxonomyAssignment, ProductListingPathAssignment, ) from db.master_taxonomy_product_paths import SHOPPING_FACET_SEGMENT_SLUGS ACTIVE = "active" INACTIVE = "inactive" MAGENTO_ROOT_PREFIXES = ("default-category", "root-catalog", "default") # SEO intersection targets intentionally point at the collection/page category. # Product push ignores these kinds; do not treat them as leaf-collision mismatches. SKIP_TAXONOMY_KINDS = frozenset({"category_filter", "collection_filter"}) SKIP_PATH_KINDS = frozenset({"intersection"}) COLLECTION_FACET_SEGMENTS = frozenset(SHOPPING_FACET_SEGMENT_SLUGS) | {"tall", "tall-cabinets"} # Hubs that share leaf names across Magento trees — always require hub in path_names. PROTECTED_MULTI_CATALOG_HUBS = frozenset( { "kitchen-cabinets", "bathroom-vanities", "bathroom", "vanities", } ) # Stale PLP parent accessories/{leaf} may still point at Magento Mouldings/{leaf}. # Cover accepts mouldings; suggestions prefer mouldings over Accessories duplicates. MOULDING_LEAF_SLUGS = frozenset( { "toe-kick", "bead-molding", "crown-molding", "cove-crown", "light-molding", "light-rail", "scribe", "outside-corner", "additional-moldings", "decorative-trim", "quarter-round", "shoe-molding", } ) MOULDING_PARENT_SLUGS = frozenset({"mouldings", "moldings"}) def repair_magento_listing_path_targets( session: Session, *, connection_id: Optional[int] = None, dry_run: bool = True, prune_stale_assignments: bool = True, deactivate_invalid_collection_facet_masters: bool = True, provision_unmatched: bool = False, path_slug: Optional[str] = None, note: Optional[str] = None, ) -> Dict[str, Any]: cleanup_note = note or f"repair_magento_listing_path_targets {datetime.now(timezone.utc).isoformat()}" registry = _registry_by_connection(session, connection_id=connection_id) # Drop invalid SEO trees under hub/accessories/... before auditing remotes. # Remotes on clearance-kit itself are fine; nested .../base-cabinets/... are not. from db.listing_intersections import deactivate_shopping_facet_rooted_intersections facet_intersection_cleanup = deactivate_shopping_facet_rooted_intersections( session, dry_run=dry_run, note=cleanup_note, ) targets = _active_magento_targets(session, connection_id=connection_id, path_slug=path_slug) mismatches: List[Dict[str, Any]] = [] relinks: List[Dict[str, Any]] = [] unmatched: List[Dict[str, Any]] = [] skipped_intersections: List[Dict[str, Any]] = [] deactivated_targets = 0 upserted_targets = 0 for target, path in targets: conn_id = int(target.connection_id) if target.connection_id is not None else None rows = registry.get(conn_id or 0) or registry.get(connection_id or 0) or [] if connection_id is not None and conn_id not in (None, connection_id): continue kind = str(target.taxonomy_kind or "category").strip().lower() path_kind = str(path.path_kind or "").strip().lower() if kind in SKIP_TAXONOMY_KINDS or path_kind in SKIP_PATH_KINDS: from db.master_taxonomy_product_paths import is_shopping_facet_rooted_path # Nested under accessories/clearance-kit/... — cleanup owns these; remotes on # clearance-kit itself stay correct and must not look like repair mismatches. if is_shopping_facet_rooted_path(path.path_slug): continue skipped_intersections.append( { "path_slug": path.path_slug, "target_id": target.id, "taxonomy_kind": kind, "path_kind": path_kind, "remote_id": str(target.remote_id or ""), "remote_path": str(target.remote_path or ""), "reason": "seo_intersection_page_category", } ) continue current_path = _registry_path_for_id(rows, target.remote_id) or str(target.remote_path or "") covers = _magento_path_covers_expected(path.path_slug, [current_path]) if current_path else False if covers: continue # Collection∩facet PLP slugs sometimes got a plain "category" target pointing at the # collection only. That is still SEO-page semantics, not a Magento leaf to relink. if _is_hub_collection_facet_slug(path.path_slug): skipped_intersections.append( { "path_slug": path.path_slug, "target_id": target.id, "taxonomy_kind": kind, "path_kind": path_kind, "remote_id": str(target.remote_id or ""), "remote_path": current_path, "reason": "hub_collection_facet_is_intersection_only", } ) continue suggestion = _suggest_registry_match(path.path_slug, rows) issue = { "path_slug": path.path_slug, "target_id": target.id, "connection_id": conn_id if conn_id is not None else connection_id, "current_remote_id": str(target.remote_id or ""), "current_remote_path": current_path or str(target.remote_path or ""), "suggested_remote_id": suggestion["remote_id"] if suggestion else None, "suggested_remote_path": suggestion["path_names"] if suggestion else None, "match_reason": suggestion["match_reason"] if suggestion else None, "planned_magento_path": _plp_slug_to_magento_path_names(path.path_slug), } mismatches.append(issue) if not suggestion: unmatched.append(issue) continue relinks.append(issue) if dry_run: continue # Deactivate every active Magento target on this path/connection that is not the winner. for sibling, _ in _targets_for_path( session, listing_path_id=path.id, connection_id=conn_id if conn_id is not None else connection_id, ): if str(sibling.remote_id) == str(suggestion["remote_id"]) and sibling.is_active: sibling.remote_path = suggestion["path_names"] continue if sibling.is_active and str(sibling.taxonomy_kind or "category").lower() not in SKIP_TAXONOMY_KINDS: sibling.is_active = False deactivated_targets += 1 upsert_listing_path_target( session, path_slug=path.path_slug, channel_code="magento", remote_id=str(suggestion["remote_id"]), taxonomy_kind=str(target.taxonomy_kind or "category"), connection_id=conn_id if conn_id is not None else connection_id, remote_path=suggestion["path_names"], ) upserted_targets += 1 provision_stats = { "provisioned": 0, "would_provision": 0, "errors": [], "samples": [], } if provision_unmatched and unmatched: from db.taxonomy_invent_policy import can_provision_remotes if not can_provision_remotes(): provision_stats = { "provisioned": 0, "would_provision": 0, "errors": [ "Remote taxonomy provision is frozen (MASTER_TAXONOMY_AUTO_INVENT=false); " "refusing --provision-unmatched. Relink existing remotes only." ], "samples": [], "invent_frozen": True, } else: provision_stats = _provision_unmatched_magento_categories( session, unmatched, connection_id=connection_id, dry_run=dry_run, ) if not dry_run: upserted_targets += int(provision_stats.get("provisioned") or 0) invalid_master_stats = {"deactivated_paths": 0, "would_deactivate_paths": 0, "samples": []} if deactivate_invalid_collection_facet_masters: invalid_master_stats = _deactivate_invalid_hub_collection_facet_masters( session, dry_run=dry_run, note=cleanup_note, path_slug=path_slug, ) assignment_stats = {"deactivated": 0, "would_deactivate": 0, "samples": []} if prune_stale_assignments: assignment_stats = _prune_stale_taxonomy_assignments( session, connection_id=connection_id, dry_run=dry_run, note=cleanup_note, ) return { "status": "ok", "dry_run": dry_run, "connection_id": connection_id, "mismatch_count": len(mismatches), "relink_count": len(relinks), "unmatched_count": len(unmatched), "skipped_intersection_count": len(skipped_intersections), "deactivated_targets": deactivated_targets if not dry_run else 0, "would_deactivate_targets": sum(1 for _ in relinks) if dry_run else 0, "upserted_targets": upserted_targets if not dry_run else 0, "would_upsert_targets": len(relinks) if dry_run else 0, "provision_unmatched": provision_stats, "invalid_collection_facet_masters": invalid_master_stats, "facet_rooted_intersection_cleanup": facet_intersection_cleanup, "stale_assignments": assignment_stats, "sample_mismatches": mismatches[:40], "sample_unmatched": unmatched[:20], "sample_skipped_intersections": skipped_intersections[:20], "note": cleanup_note, } def _plp_slug_to_magento_path_names(path_slug: str) -> str: """kitchen-cabinets/tall/oven-cabinet → Kitchen Cabinets/Tall Cabinets/Oven Cabinet.""" return plp_slug_to_magento_path_names(path_slug) def _provision_unmatched_magento_categories( session: Session, unmatched: Sequence[Dict[str, Any]], *, connection_id: Optional[int], dry_run: bool, ) -> Dict[str, Any]: """Create Magento categories for PLP slugs and relink targets off legacy Accessories paths.""" from db.channel_catalog_provision import _ensure_magento_category_path from db.collection_landing_push import _build_magento_api if connection_id is None: return { "provisioned": 0, "would_provision": len(unmatched), "errors": ["connection_id is required to provision Magento categories"], "samples": [ { "path_slug": row["path_slug"], "planned_magento_path": row.get("planned_magento_path"), "current_remote_path": row.get("current_remote_path"), } for row in unmatched[:20] ], } samples: List[Dict[str, Any]] = [] errors: List[str] = [] would = 0 provisioned = 0 api = None native_id = None if not dry_run: api, native_id = _build_magento_api(session, connection_id) if api is None or native_id is None: return { "provisioned": 0, "would_provision": len(unmatched), "errors": [f"Magento connection {connection_id} not available for API calls"], "samples": [], } seen_slugs: Set[str] = set() for row in unmatched: slug = normalize_plp_path(str(row.get("path_slug") or "")) if not slug or slug in seen_slugs: continue seen_slugs.add(slug) planned = str(row.get("planned_magento_path") or _plp_slug_to_magento_path_names(slug)) sample = { "path_slug": slug, "planned_magento_path": planned, "current_remote_id": row.get("current_remote_id"), "current_remote_path": row.get("current_remote_path"), "action": "create_or_ensure_magento_category_and_relink", } if len(samples) < 40: samples.append(sample) if dry_run: would += 1 continue try: category_id = _ensure_magento_category_path( session, api, connection_id=int(native_id), path=planned, ) if not category_id: errors.append(f"{slug}: ensure returned no category id for {planned}") continue # Deactivate wrong active category targets on this path for this connection. path_row = session.scalar(select(ChannelListingPath).where(ChannelListingPath.path_slug == slug)) if path_row is not None: for sibling, _ in _targets_for_path( session, listing_path_id=path_row.id, connection_id=connection_id, ): if str(sibling.remote_id) == str(category_id) and sibling.is_active: sibling.remote_path = planned continue if sibling.is_active and str(sibling.taxonomy_kind or "category").lower() not in SKIP_TAXONOMY_KINDS: sibling.is_active = False 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=planned, ) sample["new_remote_id"] = str(int(category_id)) provisioned += 1 except Exception as exc: errors.append(f"{slug}: {exc}") return { "provisioned": provisioned, "would_provision": would if dry_run else 0, "errors": errors[:30], "samples": samples, } def _is_hub_collection_facet_slug(path_slug: str) -> bool: """True for hub/collection/shopping-facet — intersection-only, never a master category tree.""" parts = [p for p in normalize_plp_path(path_slug).split("/") if p] if len(parts) != 3: return False hub, middle, leaf = parts if not hub or middle in COLLECTION_FACET_SEGMENTS: # hub/base-cabinets/single-door is a real taxonomy detail path, not collection∩facet. return False return leaf in COLLECTION_FACET_SEGMENTS def _deactivate_invalid_hub_collection_facet_masters( session: Session, *, dry_run: bool, note: str, path_slug: Optional[str] = None, ) -> Dict[str, Any]: """Deactivate wrong master listing paths / taxonomy nodes shaped like hub/collection/facet. Correct model: - master/product paths: hub/collection, hub/facet, hub/facet/detail - SEO only: hub/collection/facet (path_kind=intersection) """ stmt = select(ChannelListingPath).order_by(ChannelListingPath.path_slug) if path_slug: stmt = stmt.where(ChannelListingPath.path_slug == normalize_plp_path(path_slug)) samples: List[Dict[str, Any]] = [] would = 0 deactivated = 0 deactivated_nodes = 0 deactivated_assignments = 0 deactivated_category_targets = 0 for path in session.scalars(stmt).all(): if not _is_hub_collection_facet_slug(path.path_slug): continue path_kind = str(path.path_kind or "").strip().lower() # Valid intersection SEO path — keep, but strip mistaken product-category targets. if path_kind == "intersection": for target in session.scalars( select(ChannelListingPathTarget).where(ChannelListingPathTarget.listing_path_id == path.id) ).all(): kind = str(target.taxonomy_kind or "category").strip().lower() if kind in SKIP_TAXONOMY_KINDS or not target.is_active: continue sample = { "path_slug": path.path_slug, "path_kind": path_kind, "action": "demote_category_target_to_inactive", "target_id": target.id, "remote_id": target.remote_id, } if len(samples) < 30: samples.append(sample) if dry_run: would += 1 continue target.is_active = False deactivated_category_targets += 1 continue # Wrong master category/collection path — deactivate path + assignments + nodes. sample = { "path_slug": path.path_slug, "path_kind": path_kind or "category", "is_active": bool(path.is_active), "action": "deactivate_invalid_master_hub_collection_facet", } if len(samples) < 30: samples.append(sample) if dry_run: would += 1 continue if path.is_active: path.is_active = False path.notes = _append_note(path.notes, note) deactivated += 1 for assignment in session.scalars( select(ProductListingPathAssignment).where(ProductListingPathAssignment.listing_path_id == path.id) ).all(): if assignment.assignment_status == ACTIVE: assignment.assignment_status = INACTIVE assignment.notes = _append_note(assignment.notes, note) deactivated_assignments += 1 for target in session.scalars( select(ChannelListingPathTarget).where(ChannelListingPathTarget.listing_path_id == path.id) ).all(): if target.is_active and str(target.taxonomy_kind or "").lower() not in SKIP_TAXONOMY_KINDS: target.is_active = False deactivated_category_targets += 1 node = session.scalar( select(MasterTaxonomyNode).where(MasterTaxonomyNode.path_slug == path.path_slug) ) if node is not None and node.is_active: node.is_active = False deactivated_nodes += 1 return { "deactivated_paths": deactivated, "would_deactivate_paths": would, "deactivated_taxonomy_nodes": deactivated_nodes, "deactivated_listing_assignments": deactivated_assignments, "deactivated_category_targets": deactivated_category_targets, "samples": samples, } def _active_magento_targets( session: Session, *, connection_id: Optional[int], path_slug: Optional[str], ) -> List[Tuple[ChannelListingPathTarget, ChannelListingPath]]: stmt = ( select(ChannelListingPathTarget, ChannelListingPath) .join(ChannelListingPath, ChannelListingPath.id == ChannelListingPathTarget.listing_path_id) .where(ChannelListingPathTarget.channel_code == "magento") .where(ChannelListingPathTarget.is_active.is_(True)) .where(ChannelListingPath.is_active.is_(True)) .order_by(ChannelListingPath.path_slug, ChannelListingPathTarget.id) ) if connection_id is not None: stmt = stmt.where( (ChannelListingPathTarget.connection_id == connection_id) | (ChannelListingPathTarget.connection_id.is_(None)) ) if path_slug: stmt = stmt.where(ChannelListingPath.path_slug == normalize_plp_path(path_slug)) return list(session.execute(stmt).all()) def _targets_for_path( session: Session, *, listing_path_id: int, connection_id: Optional[int], ) -> List[Tuple[ChannelListingPathTarget, ChannelListingPath]]: stmt = ( select(ChannelListingPathTarget, ChannelListingPath) .join(ChannelListingPath, ChannelListingPath.id == ChannelListingPathTarget.listing_path_id) .where(ChannelListingPathTarget.listing_path_id == listing_path_id) .where(ChannelListingPathTarget.channel_code == "magento") ) if connection_id is not None: stmt = stmt.where(ChannelListingPathTarget.connection_id == connection_id) return list(session.execute(stmt).all()) def _registry_by_connection( session: Session, *, connection_id: Optional[int], ) -> Dict[int, List[Dict[str, Any]]]: stmt = select(MagentoCategoryRegistry) if connection_id is not None: stmt = stmt.where(MagentoCategoryRegistry.connection_id == connection_id) out: Dict[int, List[Dict[str, Any]]] = {} for row in session.scalars(stmt).all(): out.setdefault(int(row.connection_id), []).append( { "category_id": int(row.category_id), "parent_id": row.parent_id, "name": row.name, "path_names": str(row.path_names or row.name or ""), } ) return out def _registry_path_for_id(rows: Sequence[Dict[str, Any]], remote_id: Any) -> str: rid = str(remote_id or "").strip() if not rid: return "" for row in rows: if str(row.get("category_id")) == rid: return str(row.get("path_names") or "") return "" def _suggest_registry_match(path_slug: str, rows: Sequence[Dict[str, Any]]) -> Optional[Dict[str, Any]]: """Pick the Magento category whose path covers the PLP slug; prefer deepest cover. Never map a shopping-facet tree (e.g. panels-and-fillers/fillers) onto Digisilk's legacy Accessories / Accessories & Trim duplicates. """ expected = normalize_plp_path(path_slug) if not expected or not rows: return None expected_parts = [p for p in expected.split("/") if p] leaf = expected_parts[-1] if expected_parts else "" parent = expected_parts[-2] if len(expected_parts) >= 3 else None scored: List[Tuple[int, Dict[str, Any]]] = [] for row in rows: path_names = str(row.get("path_names") or "") pushed = _normalize_channel_category_path(path_names) if _is_legacy_accessories_collision(expected_parts, pushed): continue if _magento_path_covers_expected(expected, [pushed, path_names]): reason = "path_covers" score = 100 + len(pushed) elif _soft_parent_leaf_match(expected_parts, pushed): reason = "parent_leaf_contains" score = 50 + len(pushed) elif _soft_flat_leaf_match(expected_parts, pushed): reason = "flat_leaf_match" score = 40 + len(pushed) else: continue # Prefer Magento Mouldings over Accessories for moulding leaves (even if PLP # still says accessories/toe-kick from a stale listing path). pushed_parts = [_canonical_magento_segment(p) for p in pushed.split("/") if p] if leaf in MOULDING_LEAF_SLUGS and any(p in MOULDING_PARENT_SLUGS for p in pushed_parts): score += 25 if leaf in MOULDING_LEAF_SLUGS and any(p == "accessories" for p in pushed_parts): score -= 15 scored.append( ( score, { "remote_id": str(row["category_id"]), "path_names": path_names, "match_reason": reason, "parent": parent, "leaf": leaf, }, ) ) if not scored: return None scored.sort(key=lambda item: (-item[0], item[1]["path_names"])) return scored[0][1] def _is_legacy_accessories_collision(expected_parts: Sequence[str], pushed_path: str) -> bool: """Block Accessories* Magento paths when the PLP parent is a shopping facet (not accessories).""" if len(expected_parts) < 2: return False parent = expected_parts[-2] if len(expected_parts) >= 3 else expected_parts[-1] if parent not in COLLECTION_FACET_SEGMENTS or parent in {"accessories"}: return False pushed_parts = [p for p in normalize_plp_path(pushed_path).split("/") if p] accessories_markers = {"accessories", "accessories-trim", "storage-accessories", "cabinet-accessories"} return any(part in accessories_markers for part in pushed_parts) def _soft_parent_leaf_match(expected_parts: Sequence[str], pushed_path: str) -> bool: """Allow Magento leaf renames under the same parent facet. Example: base-cabinets/single-door ↔ Base Cabinets/Single Door Base Cabinet Does NOT allow fillers ↔ Panels And Fillers, or fillers under Accessories. """ if len(expected_parts) < 3: return False normalized_expected = [_canonical_magento_segment(part) for part in expected_parts] hub = normalized_expected[0] leaf = expected_parts[-1] parent = normalized_expected[-2] pushed_parts = [_canonical_magento_segment(p) for p in normalize_plp_path(pushed_path).split("/") if p] if len(pushed_parts) < 2: return False if hub not in pushed_parts: return False if not _parent_segment_matches(parent, leaf, pushed_parts[:-1]): return False return _leaf_token_matches(leaf, pushed_parts[-1]) def _soft_flat_leaf_match(expected_parts: Sequence[str], pushed_path: str) -> bool: """Match depth-1/2 PLPs to flat Magento categories that omit the hub folder. Example: doors/exterior-doors ↔ Exterior Doors Never used for protected multi-catalog hubs (kitchen-cabinets vs bathroom). """ if len(expected_parts) > 2 or len(expected_parts) < 1: return False hub = _canonical_magento_segment(expected_parts[0]) leaf = expected_parts[-1] if hub in PROTECTED_MULTI_CATALOG_HUBS: return False pushed = _normalize_channel_category_path(pushed_path) pushed_parts = [_canonical_magento_segment(p) for p in pushed.split("/") if p] if not pushed_parts: return False foreign_hubs = set(pushed_parts) & PROTECTED_MULTI_CATALOG_HUBS if foreign_hubs: return False return _leaf_token_matches(leaf, pushed_parts[-1]) def _parent_segment_matches(expected_parent: str, leaf: str, pushed_parents: Sequence[str]) -> bool: """True when Magento parents include the PLP facet (or mouldings alias for stale accessories).""" if expected_parent in pushed_parents: return True # Stale accessories/toe-kick (etc.) may still correctly target Magento Mouldings/*. if ( leaf in MOULDING_LEAF_SLUGS and expected_parent == "accessories" and any(p in MOULDING_PARENT_SLUGS for p in pushed_parents) ): return True return False def _leaf_token_matches(expected_leaf: str, magento_leaf: str) -> bool: if not expected_leaf or not magento_leaf: return False if expected_leaf == magento_leaf: return True # single-door → single-door-base-cabinet / single-door-wall-cabinet if magento_leaf.startswith(expected_leaf + "-"): return True return False def _magento_path_covers_expected(expected: str, pushed_paths: Sequence[str]) -> bool: expected_path = normalize_plp_path(expected) if not expected_path: return True expected_parts = [_canonical_magento_segment(part) for part in expected_path.split("/") if part] if not expected_parts: return True hub = expected_parts[0] leaf = expected_parts[-1] # For hub/facet (2 segments) parent is None; protected hubs still require hub in # Magento path so kitchen-cabinets/panels-and-fillers never matches Bathroom/... . parent = expected_parts[-2] if len(expected_parts) >= 3 else None for raw in pushed_paths: pushed = _normalize_channel_category_path(raw) if not pushed: continue normalized_pushed = "/".join(_canonical_magento_segment(part) for part in pushed.split("/") if part) normalized_expected_path = "/".join(expected_parts) if normalized_pushed == normalized_expected_path: return True pushed_parts = [_canonical_magento_segment(part) for part in pushed.split("/") if part] if not pushed_parts: continue if _is_legacy_accessories_collision(expected_parts, pushed): continue magento_leaf = pushed_parts[-1] if not _leaf_token_matches(leaf, magento_leaf): continue if hub in pushed_parts: if parent is not None and not _parent_segment_matches(parent, leaf, pushed_parts[:-1]): continue return True # Flat Magento category without hub folder (doors/exterior-doors → Exterior Doors). if parent is None and _soft_flat_leaf_match(expected_parts, pushed): return True return False def _canonical_magento_segment(segment: str) -> str: from db.magento_path_names import canonical_magento_path_segment return canonical_magento_path_segment(segment) def _prune_stale_taxonomy_assignments( session: Session, *, connection_id: Optional[int], dry_run: bool, note: str, ) -> Dict[str, Any]: """Deactivate Magento category assignments whose remote_id is not an active listing-path target for any of the SKU's active listing paths. """ allowed_by_slug = _allowed_remote_ids_by_path(session, connection_id=connection_id) sku_paths = _active_listing_paths_by_sku(session) stmt = ( select(ProductChannelTaxonomyAssignment) .where(ProductChannelTaxonomyAssignment.channel_code == "magento") .where(ProductChannelTaxonomyAssignment.assignment_status == ACTIVE) .where(ProductChannelTaxonomyAssignment.taxonomy_kind.in_(("category", "collection"))) ) if connection_id is not None: stmt = stmt.where( (ProductChannelTaxonomyAssignment.connection_id == connection_id) | (ProductChannelTaxonomyAssignment.connection_id.is_(None)) ) deactivated = 0 would = 0 samples: List[Dict[str, Any]] = [] for row in session.scalars(stmt).all(): sku = str(row.master_sku or "").strip() rid = str(row.remote_id or "").strip() if not sku or not rid: continue paths = sku_paths.get(sku) or [] if not paths: continue allowed: Set[str] = set() for slug in paths: allowed.update(allowed_by_slug.get(slug, set())) if not allowed: continue if rid in allowed: continue sample = { "id": row.id, "sku": sku, "remote_id": rid, "remote_path": row.remote_path, "listing_paths": paths[:5], } if len(samples) < 25: samples.append(sample) if dry_run: would += 1 continue row.assignment_status = INACTIVE row.notes = _append_note(row.notes, note) deactivated += 1 return { "deactivated": deactivated, "would_deactivate": would, "samples": samples, } def _allowed_remote_ids_by_path( session: Session, *, connection_id: Optional[int], ) -> Dict[str, Set[str]]: out: Dict[str, Set[str]] = {} for target, path in _active_magento_targets(session, connection_id=connection_id, path_slug=None): out.setdefault(path.path_slug, set()).add(str(target.remote_id or "").strip()) return out def _active_listing_paths_by_sku(session: Session) -> Dict[str, List[str]]: rows = session.execute( select(ProductListingPathAssignment.master_sku, ChannelListingPath.path_slug) .join(ChannelListingPath, ChannelListingPath.id == ProductListingPathAssignment.listing_path_id) .where(ProductListingPathAssignment.assignment_status == ACTIVE) .where(ChannelListingPath.is_active.is_(True)) ).all() out: Dict[str, List[str]] = {} for sku, slug in rows: out.setdefault(str(sku), []).append(str(slug)) return out def _normalize_channel_category_path(value: Any) -> str: path = normalize_plp_path(str(value or "")) for prefix in ("default-category/", "root-catalog/"): if path.startswith(prefix): return path[len(prefix) :] return path def _append_note(existing: Optional[str], note: str) -> str: text = str(existing or "").strip() if not text: return note if note in text: return text return f"{text}; {note}"