"""One-time cleanup for generated shell parents in master + outbound channels. Examples: python -m app.jobs.remove_shell_parents --dry-run python -m app.jobs.remove_shell_parents --apply python -m app.jobs.remove_shell_parents --apply --magento-connection-id 1 --magento-action hard python -m app.jobs.remove_shell_parents --apply --magento-connection-id 1 --shopify-connection-id 1 python -m app.jobs.remove_shell_parents --parent-skus ACH-B-PARENT,ASG-B-PARENT --dry-run """ from __future__ import annotations import argparse import json import logging import sys from datetime import datetime, timezone from typing import Dict, Iterable, List, Optional, Sequence import sqlalchemy as sa from sqlalchemy import select from db.session import get_session logger = logging.getLogger(__name__) DEFAULT_GENERATED_BY = ("variation_builder", "manual_variation") def run_remove_shell_parents( *, parent_skus: Optional[Sequence[str]] = None, generated_by: Sequence[str] = DEFAULT_GENERATED_BY, magento_connection_id: Optional[int] = None, magento_action: Optional[str] = None, shopify_connection_id: Optional[int] = None, export_variation_rules_path: Optional[str] = None, apply: bool = False, ) -> Dict[str, object]: from db.channel_product_removal import RemovalCandidate, execute_magento_removals, execute_shopify_removals from db.channel_sku_mapping import channel_sku_for_master from db.models import ( CatalogResolvedField, CatalogReviewQueueItem, ChannelProjectionRow, ChannelPublishState, ChannelSkuMapping, MagentoCatalogState, MagentoSyncPending, MagentoSyncState, MasterProduct, MasterProductAssociation, MasterProductAttributeValue, MasterProductImage, MasterProductLocationAvailability, MasterProductPrice, MasterProductRelation, ProductChannelAssignment, ) wanted = {str(sku or "").strip().upper() for sku in (parent_skus or []) if str(sku or "").strip()} generated = {str(item or "").strip() for item in generated_by if str(item or "").strip()} with get_session() as session: parent_rows = _load_target_parent_rows(session, wanted=wanted or None, generated_by=generated) target_parent_skus = [row.sku for row in parent_rows] parent_ids = [int(row.id) for row in parent_rows] variation_rule_export = _build_variation_rule_export(session, parent_rows) mapping_rows = session.execute( select( ChannelSkuMapping.master_sku, ChannelSkuMapping.channel_code, ChannelSkuMapping.connection_id, ChannelSkuMapping.channel_sku, ChannelSkuMapping.remote_id, ).where(ChannelSkuMapping.master_sku.in_(target_parent_skus)) ).all() if target_parent_skus else [] channel_skus_by_master: Dict[str, Dict[str, str]] = {} for master_sku, channel_code, connection_id, channel_sku, _remote_id in mapping_rows: code = str(channel_code or "").strip().lower() master = str(master_sku or "").strip().upper() sku_text = str(channel_sku or "").strip() if not master or not code or not sku_text: continue key = f"{code}:{connection_id or ''}" channel_skus_by_master.setdefault(master, {})[key] = sku_text magento_candidates: List[RemovalCandidate] = [] if magento_connection_id and target_parent_skus: for parent_sku in target_parent_skus: channel_sku = channel_skus_by_master.get(parent_sku, {}).get(f"magento:{magento_connection_id}") if not channel_sku: channel_sku = channel_sku_for_master( session, parent_sku, "magento", connection_id=magento_connection_id, ) remote_id = session.scalar( select(MagentoCatalogState.magento_product_id).where( MagentoCatalogState.connection_id == magento_connection_id, MagentoCatalogState.sku == channel_sku, ) ) if channel_sku: magento_candidates.append( RemovalCandidate( master_sku=parent_sku, channel_sku=channel_sku, remote_id=str(remote_id or channel_sku), reason="remove_shell_parent", product_type="configurable", ) ) shopify_candidates: List[RemovalCandidate] = [] if shopify_connection_id and target_parent_skus: for parent_sku in target_parent_skus: mapping = session.execute( select(ChannelSkuMapping.channel_sku, ChannelSkuMapping.remote_id) .where(ChannelSkuMapping.master_sku == parent_sku) .where(ChannelSkuMapping.channel_code == "shopify") .where( sa.or_( ChannelSkuMapping.connection_id == shopify_connection_id, ChannelSkuMapping.connection_id.is_(None), ) ) .order_by(ChannelSkuMapping.connection_id.is_(None)) ).first() channel_sku = str(mapping[0] or "").strip() if mapping else parent_sku remote_id = str(mapping[1] or "").strip() if mapping else "" if not remote_id: remote_id = str( session.scalar( select(ChannelPublishState.remote_id).where( ChannelPublishState.channel_code == "shopify", ChannelPublishState.sku == parent_sku, ) ) or "" ).strip() if channel_sku and remote_id: shopify_candidates.append( RemovalCandidate( master_sku=parent_sku, channel_sku=channel_sku, remote_id=remote_id, reason="remove_shell_parent", product_type="configurable", ) ) summary: Dict[str, object] = { "status": "ok", "dry_run": not apply, "parent_count": len(target_parent_skus), "parent_skus": target_parent_skus[:200], "generated_by": sorted(generated), "magento_candidate_count": len(magento_candidates), "shopify_candidate_count": len(shopify_candidates), "manual_variation_rule_export": { "template_count": variation_rule_export["template_count"], "source_parent_sku_count": len(variation_rule_export["source_parent_skus"]), "source_parent_skus": variation_rule_export["source_parent_skus"][:200], }, "samples": { "magento": [candidate.as_dict() for candidate in magento_candidates[:25]], "shopify": [candidate.as_dict() for candidate in shopify_candidates[:25]], }, } if export_variation_rules_path: _write_json(export_variation_rules_path, variation_rule_export) summary["manual_variation_rule_export"]["path"] = export_variation_rules_path if not apply: summary["local_delete_plan"] = _local_delete_plan( session, parent_ids=parent_ids, parent_skus=target_parent_skus, magento_connection_id=magento_connection_id, ) return summary if magento_connection_id and magento_candidates: if not magento_action: raise ValueError("--magento-action hard|disable is required when applying Magento shell parent removal") summary["magento_execution"] = execute_magento_removals( session, connection_id=magento_connection_id, candidates=magento_candidates, dry_run=False, cleanup_local=True, magento_action=magento_action, ) else: summary["magento_execution"] = {"deleted": 0, "failed": 0, "results": []} if shopify_connection_id and shopify_candidates: summary["shopify_execution"] = execute_shopify_removals( session, connection_id=shopify_connection_id, candidates=shopify_candidates, dry_run=False, cleanup_local=True, ) else: summary["shopify_execution"] = {"deleted": 0, "failed": 0, "results": []} deleted_counts = { "master_product_relation": _delete_stmt(session, sa.delete(MasterProductRelation).where( sa.or_( MasterProductRelation.parent_sku.in_(target_parent_skus), MasterProductRelation.child_sku.in_(target_parent_skus), ) )), "master_product_association": _delete_stmt(session, sa.delete(MasterProductAssociation).where( sa.or_( MasterProductAssociation.source_sku.in_(target_parent_skus), MasterProductAssociation.target_sku.in_(target_parent_skus), ) )), "master_product_attribute_value": _delete_stmt(session, sa.delete(MasterProductAttributeValue).where( sa.or_( MasterProductAttributeValue.product_id.in_(parent_ids), MasterProductAttributeValue.sku.in_(target_parent_skus), ) )), "master_product_price": _delete_stmt(session, sa.delete(MasterProductPrice).where( sa.or_( MasterProductPrice.product_id.in_(parent_ids), MasterProductPrice.sku.in_(target_parent_skus), ) )), "master_product_location_availability": _delete_stmt(session, sa.delete(MasterProductLocationAvailability).where( sa.or_( MasterProductLocationAvailability.product_id.in_(parent_ids), MasterProductLocationAvailability.sku.in_(target_parent_skus), ) )), "master_product_image": _delete_stmt(session, sa.delete(MasterProductImage).where( sa.or_( MasterProductImage.product_id.in_(parent_ids), MasterProductImage.sku.in_(target_parent_skus), ) )), "product_channel_assignment": _delete_stmt(session, sa.delete(ProductChannelAssignment).where( sa.or_( ProductChannelAssignment.product_id.in_(parent_ids), ProductChannelAssignment.sku.in_(target_parent_skus), ) )), "catalog_resolved_field": _delete_stmt(session, sa.delete(CatalogResolvedField).where( sa.or_( CatalogResolvedField.master_product_id.in_(parent_ids), CatalogResolvedField.sku.in_(target_parent_skus), ) )), "catalog_review_queue_item": _delete_stmt(session, sa.delete(CatalogReviewQueueItem).where( sa.or_( CatalogReviewQueueItem.master_product_id.in_(parent_ids), CatalogReviewQueueItem.sku.in_(target_parent_skus), ) )), "channel_projection_row": _delete_stmt(session, sa.delete(ChannelProjectionRow).where( ChannelProjectionRow.master_sku.in_(target_parent_skus) )), "channel_publish_state": _delete_stmt(session, sa.delete(ChannelPublishState).where( ChannelPublishState.sku.in_(target_parent_skus) )), "channel_sku_mapping": _delete_stmt(session, sa.delete(ChannelSkuMapping).where( ChannelSkuMapping.master_sku.in_(target_parent_skus) )), "magento_sync_pending": _delete_stmt(session, sa.delete(MagentoSyncPending).where( MagentoSyncPending.sku.in_(_magento_sku_candidates(target_parent_skus, channel_skus_by_master)) )), "magento_sync_state": _delete_stmt(session, sa.delete(MagentoSyncState).where( MagentoSyncState.sku.in_(_magento_sku_candidates(target_parent_skus, channel_skus_by_master)) )), "magento_catalog_state": _delete_stmt(session, sa.delete(MagentoCatalogState).where( MagentoCatalogState.sku.in_(_magento_sku_candidates(target_parent_skus, channel_skus_by_master)) )), "master_product": _delete_stmt(session, sa.delete(MasterProduct).where(MasterProduct.id.in_(parent_ids))), } session.commit() summary["local_deleted"] = deleted_counts return summary def _load_target_parent_rows(session, *, wanted: Optional[set[str]], generated_by: set[str]): from db.models import MasterProduct rows = session.scalars( select(MasterProduct) .where(MasterProduct.is_active.is_(True)) .where(MasterProduct.raw_payload.is_not(None)) .order_by(MasterProduct.sku) ).all() out = [] for row in rows: payload = row.raw_payload if isinstance(row.raw_payload, dict) else {} marker = str(payload.get("generated_by") or "").strip() sku = str(row.sku or "").strip().upper() if wanted and sku not in wanted: continue if generated_by and marker not in generated_by: continue out.append(row) return out def _magento_sku_candidates(parent_skus: Sequence[str], channel_skus_by_master: Dict[str, Dict[str, str]]) -> List[str]: values = set(str(sku).strip() for sku in parent_skus if str(sku).strip()) for mapping in channel_skus_by_master.values(): for key, channel_sku in mapping.items(): if key.startswith("magento:") and str(channel_sku).strip(): values.add(str(channel_sku).strip()) return sorted(values) def _delete_stmt(session, stmt) -> int: result = session.execute(stmt) return int(result.rowcount or 0) def _local_delete_plan(session, *, parent_ids: Sequence[int], parent_skus: Sequence[str], magento_connection_id: Optional[int]) -> Dict[str, int]: from db.models import ( CatalogResolvedField, CatalogReviewQueueItem, ChannelProjectionRow, ChannelPublishState, ChannelSkuMapping, MagentoCatalogState, MagentoSyncPending, MagentoSyncState, MasterProduct, MasterProductAssociation, MasterProductAttributeValue, MasterProductImage, MasterProductLocationAvailability, MasterProductPrice, MasterProductRelation, ProductChannelAssignment, ) magento_skus = sorted({ str(sku).strip() for (sku,) in session.execute( select(ChannelSkuMapping.channel_sku).where( ChannelSkuMapping.master_sku.in_(list(parent_skus)), ChannelSkuMapping.channel_code == "magento", ) ).all() if str(sku).strip() } | {str(sku).strip() for sku in parent_skus if str(sku).strip()}) return { "master_product": session.scalar(select(sa.func.count()).select_from(MasterProduct).where(MasterProduct.id.in_(list(parent_ids)))) or 0, "master_product_relation": session.scalar(select(sa.func.count()).select_from(MasterProductRelation).where( sa.or_(MasterProductRelation.parent_sku.in_(list(parent_skus)), MasterProductRelation.child_sku.in_(list(parent_skus))) )) or 0, "master_product_association": session.scalar(select(sa.func.count()).select_from(MasterProductAssociation).where( sa.or_(MasterProductAssociation.source_sku.in_(list(parent_skus)), MasterProductAssociation.target_sku.in_(list(parent_skus))) )) or 0, "master_product_attribute_value": session.scalar(select(sa.func.count()).select_from(MasterProductAttributeValue).where( sa.or_(MasterProductAttributeValue.product_id.in_(list(parent_ids)), MasterProductAttributeValue.sku.in_(list(parent_skus))) )) or 0, "master_product_price": session.scalar(select(sa.func.count()).select_from(MasterProductPrice).where( sa.or_(MasterProductPrice.product_id.in_(list(parent_ids)), MasterProductPrice.sku.in_(list(parent_skus))) )) or 0, "master_product_location_availability": session.scalar(select(sa.func.count()).select_from(MasterProductLocationAvailability).where( sa.or_(MasterProductLocationAvailability.product_id.in_(list(parent_ids)), MasterProductLocationAvailability.sku.in_(list(parent_skus))) )) or 0, "master_product_image": session.scalar(select(sa.func.count()).select_from(MasterProductImage).where( sa.or_(MasterProductImage.product_id.in_(list(parent_ids)), MasterProductImage.sku.in_(list(parent_skus))) )) or 0, "product_channel_assignment": session.scalar(select(sa.func.count()).select_from(ProductChannelAssignment).where( sa.or_(ProductChannelAssignment.product_id.in_(list(parent_ids)), ProductChannelAssignment.sku.in_(list(parent_skus))) )) or 0, "catalog_resolved_field": session.scalar(select(sa.func.count()).select_from(CatalogResolvedField).where( sa.or_(CatalogResolvedField.master_product_id.in_(list(parent_ids)), CatalogResolvedField.sku.in_(list(parent_skus))) )) or 0, "catalog_review_queue_item": session.scalar(select(sa.func.count()).select_from(CatalogReviewQueueItem).where( sa.or_(CatalogReviewQueueItem.master_product_id.in_(list(parent_ids)), CatalogReviewQueueItem.sku.in_(list(parent_skus))) )) or 0, "channel_projection_row": session.scalar(select(sa.func.count()).select_from(ChannelProjectionRow).where(ChannelProjectionRow.master_sku.in_(list(parent_skus)))) or 0, "channel_publish_state": session.scalar(select(sa.func.count()).select_from(ChannelPublishState).where(ChannelPublishState.sku.in_(list(parent_skus)))) or 0, "channel_sku_mapping": session.scalar(select(sa.func.count()).select_from(ChannelSkuMapping).where(ChannelSkuMapping.master_sku.in_(list(parent_skus)))) or 0, "magento_sync_pending": session.scalar(select(sa.func.count()).select_from(MagentoSyncPending).where(MagentoSyncPending.sku.in_(magento_skus))) or 0, "magento_sync_state": session.scalar(select(sa.func.count()).select_from(MagentoSyncState).where(MagentoSyncState.sku.in_(magento_skus))) or 0, "magento_catalog_state": session.scalar(select(sa.func.count()).select_from(MagentoCatalogState).where(MagentoCatalogState.sku.in_(magento_skus))) or 0, } def _parse_csv(raw: Optional[str]) -> List[str]: if not raw: return [] return [part.strip() for part in str(raw).split(",") if part.strip()] def _build_variation_rule_export(session, parent_rows: Sequence[object]) -> Dict[str, object]: from db.manual_variation_assignments import list_manual_variation_assignments source_parent_skus = sorted( { str((row.raw_payload or {}).get("source_parent_sku") or "").strip().upper() for row in parent_rows if isinstance(getattr(row, "raw_payload", None), dict) and str((row.raw_payload or {}).get("generated_by") or "").strip() == "manual_variation" and str((row.raw_payload or {}).get("source_parent_sku") or "").strip() } ) listed = list_manual_variation_assignments(session) templates = listed.get("templates") or [] if source_parent_skus: wanted = set(source_parent_skus) templates = [ item for item in templates if str(item.get("source_parent_sku") or "").strip().upper() in wanted ] return { "source": "manual_variation_assignment_export", "exported_at": datetime.now(timezone.utc).isoformat(), "source_parent_skus": source_parent_skus, "template_count": len(templates), "templates": templates, } def _write_json(path: str, payload: Dict[str, object]) -> None: with open(path, "w", encoding="utf-8") as fh: json.dump(payload, fh, indent=2, default=str) fh.write("\n") def main() -> int: parser = argparse.ArgumentParser(description="Remove generated shell parents from master + outbound channels") parser.add_argument("--parent-skus", default=None, help="Optional comma-separated parent SKU filter") parser.add_argument( "--generated-by", default="variation_builder,manual_variation", help="Comma-separated raw_payload.generated_by markers to target", ) parser.add_argument("--magento-connection-id", type=int, default=None) parser.add_argument("--magento-action", choices=["hard", "disable"], default=None) parser.add_argument("--shopify-connection-id", type=int, default=None) parser.add_argument( "--export-variation-rules", default=None, help="Optional JSON path to export manual base-SKU variation templates before deleting parent shells", ) parser.add_argument("--apply", action="store_true", help="Execute deletes (default is dry-run)") parser.add_argument("--dry-run", action="store_true", help="Explicit dry-run") args = parser.parse_args() try: result = run_remove_shell_parents( parent_skus=_parse_csv(args.parent_skus) or None, generated_by=_parse_csv(args.generated_by) or list(DEFAULT_GENERATED_BY), magento_connection_id=args.magento_connection_id, magento_action=args.magento_action, shopify_connection_id=args.shopify_connection_id, export_variation_rules_path=args.export_variation_rules, apply=bool(args.apply and not args.dry_run), ) except Exception as exc: print(json.dumps({"status": "failed", "error": str(exc)}, indent=2)) return 1 print(json.dumps(result, indent=2, default=str)) return 0 if result.get("status") == "ok" else 1 if __name__ == "__main__": logging.basicConfig(level=logging.INFO) sys.exit(main())