"""Create Magento categories, Shopify collections, and channel attributes/options. One-shot provision for assigned master-catalog SKUs (or all active SKUs with ``--all-products``). Includes mapped attributes (create_new aliases) and Magento configurable axis attributes + select options. CLI: python -m app.jobs.channel_catalog_provision python -m app.jobs.channel_catalog_provision --apply python -m app.jobs.channel_catalog_provision --apply --magento-connection-id 2 python -m app.jobs.channel_catalog_provision --apply --channel magento --baseline-first API (async, poll until completed): POST /api/channel-catalogs/provision/apply GET /api/channel-catalogs/provision/{job_id} """ from __future__ import annotations import argparse import json import logging import sys from typing import Any, Dict, List, Optional logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def run_channel_catalog_provision( *, dry_run: bool = True, magento_connection_id: Optional[int] = None, shopify_connection_id: Optional[int] = None, only_assigned: bool = True, include_magento: bool = True, include_shopify: bool = True, create_missing: bool = True, assign_skus: bool = True, delta_only: bool = True, seed_master_listing_paths: bool = True, provision_attributes: bool = True, include_brand_collections: bool = True, shopify_brand_filter: Optional[List[str]] = None, limit: Optional[int] = None, baseline_first: bool = False, ) -> Dict[str, Any]: from app.jobs.channel_catalog_provision_enqueue import _execute_catalog_provision params = { "magento_connection_id": magento_connection_id, "shopify_connection_id": shopify_connection_id, "only_assigned": only_assigned, "dry_run": dry_run, "include_magento": include_magento, "include_shopify": include_shopify, "create_missing": create_missing, "assign_skus": assign_skus, "delta_only": delta_only, "seed_master_listing_paths": seed_master_listing_paths, "provision_attributes": provision_attributes, "include_brand_collections": include_brand_collections, "shopify_brand_filter": shopify_brand_filter or [], "limit": limit, "baseline_first": baseline_first, } return _execute_catalog_provision(params) def _provision_had_errors(result: Dict[str, Any]) -> bool: if not result: return True if str(result.get("status") or "").lower() == "failed": return True for channel in ("magento", "shopify"): block = result.get(channel) if isinstance(block, dict) and int(block.get("error_count") or 0) > 0: return True attrs = result.get("attributes") or {} if isinstance(attrs, dict): for block in attrs.values(): if not isinstance(block, dict): continue errors = block.get("errors") or [] if errors: return True cfg = block.get("configurable_catalog") or {} if isinstance(cfg, dict) and cfg.get("errors"): return True return False def main() -> int: parser = argparse.ArgumentParser( description="Create Magento categories, Shopify collections, and channel attributes/options", ) parser.add_argument("--dry-run", dest="dry_run", action="store_true", default=True) parser.add_argument( "--apply", dest="dry_run", action="store_false", help="Create remotes (default is preview only)", ) parser.add_argument("--magento-connection-id", type=int, default=None) parser.add_argument("--shopify-connection-id", type=int, default=None) parser.add_argument( "--channel", action="append", choices=["magento", "shopify"], default=None, help="Limit to one channel (repeatable). Default: both when connections exist.", ) parser.add_argument( "--all-products", dest="only_assigned", action="store_false", help="Use all active master SKUs, not only channel-assigned SKUs", ) parser.add_argument( "--no-create-missing", dest="create_missing", action="store_false", help="Plan only; do not create missing categories/collections", ) parser.add_argument( "--no-assign-skus", dest="assign_skus", action="store_false", help="Create taxonomy nodes but skip SKU taxonomy assignments", ) parser.add_argument( "--full-assignments", dest="delta_only", action="store_false", help="Re-assign SKUs even when assignments already exist", ) parser.add_argument( "--no-listing-paths", dest="seed_master_listing_paths", action="store_false", help="Skip seeding master channel_listing_path rows", ) parser.add_argument( "--no-attributes", dest="provision_attributes", action="store_false", help="Skip attribute/metafield + configurable option provision", ) parser.add_argument( "--no-brand-collections", dest="include_brand_collections", action="store_false", help="Skip Shopify brand-based collections", ) parser.add_argument( "--shopify-brand", action="append", default=None, help="Limit Shopify collection provision to these brand names (repeatable)", ) parser.add_argument("--limit", type=int, default=None) parser.add_argument( "--baseline-first", action="store_true", help="Pull Magento stores/categories/attributes into registry before creating", ) args = parser.parse_args() channels = [str(ch).strip().lower() for ch in (args.channel or []) if str(ch).strip()] include_magento = not channels or "magento" in channels include_shopify = not channels or "shopify" in channels logger.info( "channel_catalog_provision dry_run=%s magento=%s shopify=%s only_assigned=%s", args.dry_run, include_magento, include_shopify, args.only_assigned, ) result = run_channel_catalog_provision( dry_run=args.dry_run, magento_connection_id=args.magento_connection_id, shopify_connection_id=args.shopify_connection_id, only_assigned=args.only_assigned, include_magento=include_magento, include_shopify=include_shopify, create_missing=args.create_missing, assign_skus=args.assign_skus, delta_only=args.delta_only, seed_master_listing_paths=args.seed_master_listing_paths, provision_attributes=args.provision_attributes, include_brand_collections=args.include_brand_collections, shopify_brand_filter=args.shopify_brand, limit=args.limit, baseline_first=args.baseline_first, ) print(json.dumps(result, indent=2, default=str)) return 1 if _provision_had_errors(result) else 0 if __name__ == "__main__": sys.exit(main())