"""Plan/apply Magento or Shopify product removals (RTA / Unassembled cleanup). Examples: # Preview RTA-* still on Magento (from pulled catalog state) python -m app.jobs.remove_channel_products --channel magento --connection-id 1 \\ --selection sku_filter --sku-filter RTA- --dry-run # Also scan Magento live for SKUs starting with RTA- (catches stale catalog_state gaps) python -m app.jobs.remove_channel_products --channel magento --connection-id 1 \\ --selection sku_filter --sku-filter RTA- --scan-remote --dry-run # Hard-delete them python -m app.jobs.remove_channel_products --channel magento --connection-id 1 \\ --selection sku_filter --sku-filter RTA- --scan-remote --apply --magento-action hard # Unassembled via master filter (assembly_type / name) python -m app.jobs.remove_channel_products --channel magento --connection-id 1 \\ --selection master_filter --master-filter 'assembled or rta=Unassembled' --dry-run """ from __future__ import annotations import argparse import json import logging import sys from typing import Any, Dict, List, Optional from urllib.parse import urlencode from db.session import get_session logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def _parse_master_filters(values: Optional[List[str]]) -> Optional[List[Dict[str, str]]]: if not values: return None out: List[Dict[str, str]] = [] for raw in values: text = str(raw or "").strip() if not text: continue if "=" not in text: raise ValueError(f"master-filter must be code=value, got {text!r}") code, value = text.split("=", 1) out.append({"code": code.strip(), "value": value.strip(), "match": "exact"}) return out or None def _magento_api(session, connection_id: int): from db.magento_repositories import SqlAlchemyMagentoConnectionRepository from magento.magento_api import MagentoRestClient from magento.oauth_client import MagentoOAuthClient, build_magento_oauth_kwargs conn = SqlAlchemyMagentoConnectionRepository(session).get_for_sync(connection_id) if not conn: raise ValueError(f"Magento connection {connection_id} not found") oauth = MagentoOAuthClient(**build_magento_oauth_kwargs(conn)) return MagentoRestClient(oauth) def _scan_magento_sku_prefix(api, prefix: str, *, page_size: int = 200, limit: Optional[int] = None) -> List[Dict[str, Any]]: """Live Magento search for SKUs starting with prefix (like RTA-%).""" from magento.magento_api import _rfc3986_quote needle = str(prefix or "").strip() if not needle: return [] like = f"{needle}%" page = 1 found: List[Dict[str, Any]] = [] while True: params = { "searchCriteria[filter_groups][0][filters][0][field]": "sku", "searchCriteria[filter_groups][0][filters][0][value]": like, "searchCriteria[filter_groups][0][filters][0][condition_type]": "like", "searchCriteria[pageSize]": str(page_size), "searchCriteria[currentPage]": str(page), "fields": "items[id,sku,name,type_id,status,visibility],total_count", } query = urlencode(params, quote_via=_rfc3986_quote) url = f"{api._client._url('products')}?{query}" status, body, err = api._client._request_url("GET", url) if status != 200: raise RuntimeError(err or f"Magento product search failed HTTP {status}") items = (body or {}).get("items") if isinstance(body, dict) else [] total = int((body or {}).get("total_count") or 0) if isinstance(body, dict) else 0 for item in items or []: sku = str(item.get("sku") or "").strip() if not sku: continue found.append( { "channel_sku": sku, "remote_id": str(item.get("id") or ""), "name": item.get("name"), "type_id": item.get("type_id"), "status": item.get("status"), "visibility": item.get("visibility"), "reason": f"scan_remote:starts_with:{needle}", } ) if limit and len(found) >= limit: return found if not items or page * page_size >= total: break page += 1 return found def _merge_scan_into_plan(plan: Dict[str, Any], scanned: List[Dict[str, Any]]) -> Dict[str, Any]: from db.channel_product_removal import RemovalCandidate by_remote = {str(c.get("remote_id")): c for c in (plan.get("candidates") or []) if c.get("remote_id")} by_sku = {str(c.get("channel_sku")): c for c in (plan.get("candidates") or []) if c.get("channel_sku")} merged = list(plan.get("candidates") or []) added = 0 for row in scanned: remote_id = str(row.get("remote_id") or "") sku = str(row.get("channel_sku") or "") if (remote_id and remote_id in by_remote) or (sku and sku in by_sku): continue candidate = RemovalCandidate( master_sku=None, channel_sku=sku, remote_id=remote_id or sku, reason=str(row.get("reason") or "scan_remote"), product_type=str(row.get("type_id") or "") or None, ) merged.append(candidate.as_dict()) added += 1 if remote_id: by_remote[remote_id] = merged[-1] if sku: by_sku[sku] = merged[-1] plan = dict(plan) plan["candidates"] = merged plan["candidate_count"] = len(merged) plan["samples"] = merged[:25] plan["scan_remote_added"] = added plan["scan_remote_count"] = len(scanned) return plan def run_remove( *, channel: str, connection_id: int, selection: str, sku_filter: Optional[str] = None, sku_match_mode: str = "starts_with", master_filters: Optional[List[Dict[str, str]]] = None, magento_action: Optional[str] = None, scan_remote: bool = False, cleanup_local: bool = True, limit: Optional[int] = None, apply: bool = False, ) -> Dict[str, Any]: from db.channel_product_removal import ( RemovalCandidate, execute_magento_removals, execute_shopify_removals, plan_product_removals, ) with get_session() as session: plan = plan_product_removals( session, channel, connection_id=connection_id, selection=selection, sku_filter=sku_filter, sku_match_mode=sku_match_mode, master_filters=master_filters, limit=limit, ) if scan_remote and channel == "magento" and selection == "sku_filter" and sku_filter: api = _magento_api(session, connection_id) scanned = _scan_magento_sku_prefix(api, sku_filter, limit=limit) plan = _merge_scan_into_plan(plan, scanned) logger.info( "Live Magento scan for %s*: found=%s added_to_plan=%s", sku_filter, plan.get("scan_remote_count"), plan.get("scan_remote_added"), ) candidates = [RemovalCandidate(**item) for item in (plan.get("candidates") or [])] if not apply: return { "status": "ok", "dry_run": True, "would_delete": len(candidates), "plan": plan, "samples": [c.as_dict() for c in candidates[:50]], } if channel == "magento": if not magento_action: raise ValueError("--magento-action hard|disable is required with --apply") execution = execute_magento_removals( session, connection_id=connection_id, candidates=candidates, dry_run=False, cleanup_local=cleanup_local, magento_action=magento_action, ) else: execution = execute_shopify_removals( session, connection_id=connection_id, candidates=candidates, dry_run=False, cleanup_local=cleanup_local, ) session.commit() return { "status": "ok", "dry_run": False, "plan": plan, "execution": execution, } def main() -> int: parser = argparse.ArgumentParser(description="Remove Magento/Shopify products (RTA / Unassembled cleanup)") parser.add_argument("--channel", choices=["magento", "shopify"], default="magento") parser.add_argument("--connection-id", type=int, required=True) parser.add_argument( "--selection", choices=["sku_filter", "master_filter", "not_in_master"], default="sku_filter", ) parser.add_argument("--sku-filter", default="RTA-", help="SKU prefix/text for selection=sku_filter") parser.add_argument( "--sku-match-mode", choices=["starts_with", "contains", "exact"], default="starts_with", ) parser.add_argument( "--master-filter", action="append", dest="master_filters", help="code=value filter for selection=master_filter (repeatable)", ) parser.add_argument("--scan-remote", action="store_true", help="Also live-scan Magento for matching SKUs") parser.add_argument("--magento-action", choices=["hard", "disable"], help="Required with --apply for Magento") parser.add_argument("--no-cleanup-local", action="store_true", help="Keep local catalog/publish state rows") parser.add_argument("--limit", type=int, default=None) parser.add_argument("--apply", action="store_true", help="Execute deletions (default is dry-run)") parser.add_argument("--dry-run", action="store_true", help="Explicit dry-run (default)") args = parser.parse_args() try: result = run_remove( channel=args.channel, connection_id=args.connection_id, selection=args.selection, sku_filter=args.sku_filter, sku_match_mode=args.sku_match_mode, master_filters=_parse_master_filters(args.master_filters), magento_action=args.magento_action, scan_remote=args.scan_remote, cleanup_local=not args.no_cleanup_local, limit=args.limit, apply=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__": raise SystemExit(main())