"""Full channel publish: taxonomies, attributes, assignments, product upserts, image upserts. Runs end-to-end in one CLI invocation (Magento worker runs in-process when --apply). CLI: python -m app.jobs.channel_full_publish --dry-run python -m app.jobs.channel_full_publish --apply python -m app.jobs.channel_full_publish --apply --channel magento --magento-connection-id 1 python -m app.jobs.channel_full_publish --apply --sku ACH-B12 --wait-timeout 7200 python -m app.jobs.channel_full_publish --apply --no-run-worker # queue only; external magento_worker """ from __future__ import annotations import argparse import json import logging import math import os from typing import Any, Dict, List, Optional from sqlalchemy import select from db.compat_connections import compat_connection_id from db.models import MagentoConnection, MasterProduct, ShopifyConnection from db.session import get_session from magento.sku_policy import filter_magento_push_skus logger = logging.getLogger(__name__) DEFAULT_BATCH_SIZE = int(os.getenv("CHANNEL_FULL_PUBLISH_BATCH_SIZE", "500")) def _enforce_manual_taxonomy_publish_mode(mode: str) -> str: """Manual taxonomy policy: full publish may link to taxonomy, never create it.""" normalized = str(mode or "link_only").strip().lower().replace("-", "_") if normalized == "full": return "link_only" return normalized or "link_only" def _active_native_ids( session, *, magento_connection_id: Optional[int], shopify_connection_id: Optional[int], ) -> Dict[str, Optional[int]]: magento_id = magento_connection_id if magento_id is not None: row = session.get(MagentoConnection, int(magento_id)) magento_id = int(row.id) if row is not None else None else: row = session.scalar( select(MagentoConnection).where(MagentoConnection.status == "active").order_by(MagentoConnection.id).limit(1) ) if row is None: row = session.scalar(select(MagentoConnection).order_by(MagentoConnection.id).limit(1)) magento_id = int(row.id) if row is not None else None shopify_id = shopify_connection_id if shopify_id is not None: row = session.get(ShopifyConnection, int(shopify_id)) shopify_id = int(row.id) if row is not None else None else: row = session.scalar( select(ShopifyConnection).where(ShopifyConnection.status == "active").order_by(ShopifyConnection.id).limit(1) ) if row is None: row = session.scalar(select(ShopifyConnection).order_by(ShopifyConnection.id).limit(1)) shopify_id = int(row.id) if row is not None else None return {"magento": magento_id, "shopify": shopify_id} def _load_target_skus(session, skus: Optional[List[str]], *, only_assigned: bool) -> List[str]: if skus: return [str(s).strip() for s in skus if str(s).strip()] if only_assigned: from db.channel_exports import resolve_push_skus return sorted( resolve_push_skus( session, "magento", only_assigned=True, ) | resolve_push_skus( session, "shopify", only_assigned=True, ) ) return [ str(row) for row in session.scalars(select(MasterProduct.sku).where(MasterProduct.is_active.is_(True))).all() ] def _bootstrap_shopping_facets( session, *, magento_id: Optional[int], dry_run: bool, ) -> Dict[str, Any]: if magento_id is None: return {"status": "skipped", "reason": "no_magento_connection"} from db.collection_landing_push import _build_magento_api from magento.shopping_facet_bootstrap import ensure_magento_shopping_facets api, native_id = _build_magento_api(session, magento_id) if api is None or native_id is None: return {"status": "skipped", "reason": "magento_api_unavailable"} return ensure_magento_shopping_facets( session, api, connection_id=int(native_id), dry_run=dry_run, ) def _provision_attributes( session, *, magento_id: Optional[int], shopify_id: Optional[int], dry_run: bool, ) -> Dict[str, Any]: from db.channel_attribute_provision import provision_pending_for_channels channels: List[str] = [] if magento_id is not None: channels.append("magento") if shopify_id is not None: channels.append("shopify") if not channels: return {"status": "skipped", "reason": "no_connections"} return provision_pending_for_channels( session, channels, magento_connection_id=magento_id, shopify_connection_id=shopify_id, dry_run=dry_run, ) def _push_collection_landings( session, *, channel_code: str, connection_id: Optional[int], dry_run: bool, provision_missing_remote: bool = False, skip_collection_assets: bool = False, ) -> Dict[str, Any]: from db.collection_landing_push import push_collection_landing_channel if connection_id is None: return {"status": "skipped", "reason": "no_connection"} return push_collection_landing_channel( session, channel_code=channel_code, connection_id=connection_id, path_slug=None, only_with_landing_content=True, dry_run=dry_run, provision_missing_remote=provision_missing_remote, apply_listing_paths=True, sync_products=channel_code == "magento", sync_shopping_icons=channel_code == "magento", skip_collection_assets=skip_collection_assets, limit=500, ) def _prime_projection_cache( session, *, channel_code: str, connection_id: Optional[int], skus: Optional[List[str]], only_assigned: bool, ) -> Dict[str, Any]: from db.channel_capabilities import resolve_product_structure_for_connection from db.channel_exports import build_channel_product_payloads from db.channel_projection_cache import store_projection_payloads from db.channel_projection_invalidation import inspect_projection_drift if connection_id is None: return {"status": "skipped", "reason": "no_connection"} drift = inspect_projection_drift( session, channel_code=channel_code, connection_id=connection_id, skus=skus, apply=True, ) structure = resolve_product_structure_for_connection( session, channel_code, connection_id=connection_id, ) payloads = build_channel_product_payloads( session, channel_code, skus=skus, only_assigned=only_assigned, connection_id=connection_id, product_structure=structure, prefer_projection_cache=False, ) stored = store_projection_payloads( session, channel_code=channel_code, connection_id=connection_id, payloads=payloads, product_structure=structure, scope_key="assigned" if only_assigned else "all", notes="Primed by channel_full_publish", replace_scope=skus is None, ) return { "status": "ok", "channel_code": channel_code, "connection_id": connection_id, "product_structure": structure, "sku_count": len(payloads), "drift_inspect": drift, **stored, } def _push_magento_batches( session, *, skus: List[str], compat_connection_id: int, dry_run: bool, batch_size: int, wait: bool, run_worker: bool, wait_timeout: int, push_images: bool, force_products: bool = False, force_relations: bool = False, run_pull_after: bool = False, ) -> Dict[str, Any]: from app.jobs.push_channel_sample import push_magento_sample skus = filter_magento_push_skus(skus) if not skus: return {"status": "skipped", "reason": "no_skus", "batches": []} batches: List[Dict[str, Any]] = [] total_batches = max(1, math.ceil(len(skus) / batch_size)) for batch_index, offset in enumerate(range(0, len(skus), batch_size), start=1): batch = skus[offset : offset + batch_size] logger.info( "Magento push batch %s/%s (%s SKUs)", batch_index, total_batches, len(batch), ) if push_images and not dry_run and not force_relations: result = push_magento_sample( session, skus=batch, connection_id=compat_connection_id, dry_run=dry_run, images_only=False, products_only=False, phased=True, wait=wait, run_worker=run_worker, wait_timeout=wait_timeout, force_images=True, force_products=force_products, run_pull_after=run_pull_after, use_connection_pending_table=False, ) else: result = push_magento_sample( session, skus=batch, connection_id=compat_connection_id, dry_run=dry_run, images_only=False, products_only=not push_images, phased=False, wait=wait and not dry_run, run_worker=run_worker, wait_timeout=wait_timeout, force_images=True, force_products=force_products, force_relations=force_relations, run_pull_after=run_pull_after, use_connection_pending_table=False, ) batches.append({"batch": batch_index, "sku_count": len(batch), **result}) phase_failed = str(result.get("status") or "").lower() == "failed" if not phase_failed: for phase in result.get("phases") or []: if str(phase.get("status") or "").lower() == "failed": phase_failed = True break if phase_failed and not dry_run: from db.channel_projection_invalidation import invalidate_skus from db.compat_connections import resolve_native_connection_id native_id = resolve_native_connection_id("magento", compat_connection_id) invalidate_skus( session, channel_code="magento", connection_id=native_id if native_id is not None else compat_connection_id, skus=batch, reason=f"magento_batch_push_failure:{result.get('last_error') or 'failed'}", ) session.commit() return { "status": "ok", "batch_count": len(batches), "sku_count": len(skus), "batches": batches, } def _push_shopify_batches( session, *, skus: List[str], compat_connection_id: int, dry_run: bool, batch_size: int, push_images: bool, ) -> Dict[str, Any]: from app.jobs.push_channel_sample import push_shopify_sample from shopify.media_sync import push_product_images if not skus: return {"status": "skipped", "reason": "no_skus", "batches": []} batches: List[Dict[str, Any]] = [] total_batches = max(1, math.ceil(len(skus) / batch_size)) for batch_index, offset in enumerate(range(0, len(skus), batch_size), start=1): batch = skus[offset : offset + batch_size] logger.info( "Shopify push batch %s/%s (%s SKUs)", batch_index, total_batches, len(batch), ) products = push_shopify_sample( session, skus=batch, connection_id=compat_connection_id, dry_run=dry_run, force=True, ) images: Dict[str, Any] = {"status": "skipped"} if push_images and not dry_run: images = push_product_images( session, dry_run=False, skus=batch, connection_id=compat_connection_id, force=True, ) batches.append( { "batch": batch_index, "sku_count": len(batch), "products": products, "images": images, } ) failed_skus = [ str(err.get("sku") or "").strip() for err in (products.get("errors") or []) if str(err.get("sku") or "").strip() ] if failed_skus and not dry_run: from db.channel_projection_invalidation import invalidate_skus from db.compat_connections import resolve_native_connection_id native_id = resolve_native_connection_id("shopify", compat_connection_id) invalidate_skus( session, channel_code="shopify", connection_id=native_id if native_id is not None else compat_connection_id, skus=failed_skus, reason="shopify_batch_push_failure", ) session.commit() return { "status": "ok", "batch_count": len(batches), "sku_count": len(skus), "batches": batches, } def run_channel_full_publish( *, dry_run: bool = True, channels: Optional[List[str]] = None, magento_connection_id: Optional[int] = None, shopify_connection_id: Optional[int] = None, skus: Optional[List[str]] = None, only_assigned: bool = True, skip_pull: bool = False, skip_taxonomy: bool = False, skip_attributes: bool = False, skip_products: bool = False, skip_images: bool = False, skip_collection_assets: bool = False, wait: bool = True, run_worker: bool = True, wait_timeout: int = 7200, batch_size: int = DEFAULT_BATCH_SIZE, reprocess_master: bool = False, taxonomy_mode: str = "link_only", fast: bool = False, force_products: bool = False, force_relations: bool = False, run_pull_after: bool = False, ) -> Dict[str, Any]: """Create taxonomies/attributes, assign SKUs, upsert products and images. taxonomy_mode: full | link_only | skip fast: consecutive-push shortcut (skip taxonomy rebuild + attributes + registry pull) force_products: ignore data_hash and re-PUT products (needed when Magento is stale but hashes match) force_relations: run Magento as a full non-phased job and force configurable child/option links """ from app.jobs.sync_master_taxonomy_pipeline import run_master_taxonomy_pipeline requested_mode = str(taxonomy_mode or "link_only").strip().lower().replace("-", "_") mode = _enforce_manual_taxonomy_publish_mode(requested_mode) if fast: mode = "skip" skip_pull = True skip_attributes = True if skip_taxonomy: mode = "skip" selected = [str(ch).strip().lower() for ch in (channels or ["magento", "shopify"]) if str(ch).strip()] summary: Dict[str, Any] = { "dry_run": dry_run, "channels": selected, "taxonomy_mode": mode, "fast": bool(fast or mode == "skip"), "steps": {}, "skip_collection_assets": bool(skip_collection_assets), } if requested_mode != mode: summary["taxonomy_mode_requested"] = requested_mode summary["taxonomy_mode_enforced"] = mode summary["note"] = ( "Manual taxonomy policy enforced: full publish links to existing taxonomy only; " "automatic taxonomy/category/collection creation is disabled." ) with get_session() as session: native = _active_native_ids( session, magento_connection_id=magento_connection_id, shopify_connection_id=shopify_connection_id, ) summary["connections"] = native target_skus = _load_target_skus(session, skus, only_assigned=only_assigned) summary["sku_total"] = len(target_skus) include_magento = "magento" in selected and native.get("magento") is not None include_shopify = "shopify" in selected and native.get("shopify") is not None if not include_magento and not include_shopify: summary["status"] = "failed" summary["error"] = "No active connection for requested channel(s)" return summary if mode != "skip": logger.info("Step: master taxonomy (mode=%s)", mode) summary["steps"]["taxonomy"] = run_master_taxonomy_pipeline( dry_run=dry_run, magento_connection_id=native["magento"] if include_magento else None, shopify_connection_id=native["shopify"] if include_shopify else None, pull_remotes=not skip_pull, create_missing_remotes=False, push_products=False, skus=target_skus or None, reprocess_master=reprocess_master, sku_batch_size=batch_size, taxonomy_mode=mode, ) else: logger.info("Step: master taxonomy skipped (fast/consecutive push)") summary["steps"]["taxonomy"] = {"status": "skipped", "reason": "taxonomy_mode=skip"} if not skip_attributes: logger.info("Step: provision channel attributes (create_new aliases)") with get_session() as session: summary["steps"]["provision_attributes"] = _provision_attributes( session, magento_id=native["magento"] if include_magento else None, shopify_id=native["shopify"] if include_shopify else None, dry_run=dry_run, ) if include_magento: logger.info("Step: bootstrap shopping facets (cabinet_type options + filterable)") summary["steps"]["shopping_facets"] = _bootstrap_shopping_facets( session, magento_id=native["magento"], dry_run=dry_run, ) if not dry_run: session.commit() if include_magento: logger.info("Step: Magento collection landing metadata push") with get_session() as session: summary["steps"]["magento_collection_landings"] = _push_collection_landings( session, channel_code="magento", connection_id=native["magento"], dry_run=dry_run, provision_missing_remote=False, skip_collection_assets=skip_collection_assets, ) if not dry_run: session.commit() if summary["steps"]["magento_collection_landings"].get("errors"): summary["status"] = "failed" summary["error"] = "Magento collection landing push failed" return summary if include_shopify: logger.info("Step: Shopify collection landing metadata push") with get_session() as session: summary["steps"]["shopify_collection_landings"] = _push_collection_landings( session, channel_code="shopify", connection_id=native["shopify"], dry_run=dry_run, provision_missing_remote=False, skip_collection_assets=skip_collection_assets, ) if not dry_run: session.commit() if summary["steps"]["shopify_collection_landings"].get("errors"): summary["status"] = "failed" summary["error"] = "Shopify collection landing push failed" return summary if not dry_run: logger.info("Step: prime channel projection cache") with get_session() as session: projection_steps: Dict[str, Any] = {} if include_magento: projection_steps["magento"] = _prime_projection_cache( session, channel_code="magento", connection_id=native["magento"], skus=target_skus, only_assigned=only_assigned, ) if include_shopify: projection_steps["shopify"] = _prime_projection_cache( session, channel_code="shopify", connection_id=native["shopify"], skus=target_skus, only_assigned=only_assigned, ) summary["steps"]["projection_cache"] = projection_steps session.commit() push_products = not skip_products and not dry_run push_images = not skip_images and not dry_run and push_products if include_magento and (push_products or dry_run): logger.info("Step: Magento product/image upserts") magento_compat = compat_connection_id("magento", int(native["magento"])) with get_session() as session: summary["steps"]["magento_push"] = _push_magento_batches( session, skus=target_skus, compat_connection_id=magento_compat, dry_run=dry_run, batch_size=batch_size, wait=wait, run_worker=run_worker, wait_timeout=wait_timeout, push_images=push_images, force_products=force_products, force_relations=force_relations, run_pull_after=run_pull_after, ) if include_shopify and (push_products or dry_run): logger.info("Step: Shopify product/image upserts") shopify_compat = compat_connection_id("shopify", int(native["shopify"])) with get_session() as session: summary["steps"]["shopify_push"] = _push_shopify_batches( session, skus=target_skus, compat_connection_id=shopify_compat, dry_run=dry_run, batch_size=batch_size, push_images=push_images, ) if dry_run: summary["note"] = ( "Dry-run preview only. Re-run with --apply to create taxonomies, attributes, " "assignments, product upserts, and image upserts." ) summary["status"] = "ok" return summary def main() -> int: logging.basicConfig(level=logging.INFO) parser = argparse.ArgumentParser( description=( "Full channel publish: taxonomies, attributes, category/collection assignments, " "product upserts, and image upserts" ) ) parser.add_argument("--apply", action="store_true", help="Persist and push (default is dry-run)") parser.add_argument( "--channel", action="append", dest="channels", choices=["magento", "shopify"], help="Limit to one channel (repeatable; default: both when configured)", ) parser.add_argument("--sku", action="append", dest="skus", help="Limit to SKU(s); repeatable") parser.add_argument("--magento-connection-id", type=int, default=None, help="Native Magento connection id") parser.add_argument("--shopify-connection-id", type=int, default=None, help="Native Shopify connection id") parser.add_argument("--batch-size", type=int, default=DEFAULT_BATCH_SIZE) parser.add_argument("--skip-pull", action="store_true", help="Skip Magento/Shopify registry pull") parser.add_argument("--skip-taxonomy", action="store_true", help="Skip taxonomy create/link/assign steps") parser.add_argument( "--taxonomy-mode", choices=["full", "link-only", "link_only", "skip"], default="link_only", help="link-only=assign SKUs only; skip=no taxonomy. full is treated as link-only under manual taxonomy policy.", ) parser.add_argument( "--fast", action="store_true", help="Consecutive-push shortcut: skip taxonomy rebuild, attributes, and registry pull", ) parser.add_argument( "--force-products", action="store_true", help="Force Magento product re-PUT even when data hashes match (use when Magento title/url is stale)", ) parser.add_argument( "--force-relations", action="store_true", help="Force Magento configurable child/option links; uses one full non-phased Magento job", ) parser.add_argument( "--run-pull-after", action="store_true", help="Run Magento pull after product/relation push to refresh local catalog state", ) parser.add_argument("--skip-attributes", action="store_true", help="Skip create_new attribute provisioning") parser.add_argument("--skip-products", action="store_true", help="Skip product upserts") parser.add_argument("--skip-images", action="store_true", help="Skip image upserts (products still run)") parser.add_argument( "--skip-collection-assets", action="store_true", help="Skip collection landing media assets (PDFs, hero images, thumbnails, category icons) while still pushing collection metadata/relationships.", ) parser.add_argument("--include-unassigned", action="store_true", help="Include SKUs not channel-assigned") parser.add_argument("--reprocess-master", action="store_true", help="Rebuild master fields from raw payloads first") parser.add_argument("--no-wait", action="store_true", help="Enqueue Magento work only; do not wait for completion") parser.add_argument( "--no-run-worker", action="store_true", help="Do not run magento_sync_worker in-process (use external magento_worker.py)", ) parser.add_argument("--wait-timeout", type=int, default=7200, help="Seconds to wait per Magento queue batch") args = parser.parse_args() dry_run = not args.apply wait = not args.no_wait if args.no_run_worker and wait: logger.info( "--no-run-worker set: waiting for Magento queue but leaving execution to external magento_worker/systemd." ) elif args.no_run_worker and not wait: logger.info( "--no-run-worker and --no-wait set: enqueueing Magento work only; external magento_worker/systemd must process it." ) elif not args.no_run_worker and not wait: logger.info( "--no-wait set: queueing Magento work without waiting for in-process completion." ) result = run_channel_full_publish( dry_run=dry_run, channels=args.channels, magento_connection_id=args.magento_connection_id, shopify_connection_id=args.shopify_connection_id, skus=args.skus, only_assigned=not args.include_unassigned, skip_pull=args.skip_pull, skip_taxonomy=args.skip_taxonomy, skip_attributes=args.skip_attributes, skip_products=args.skip_products, skip_images=args.skip_images, skip_collection_assets=args.skip_collection_assets, wait=wait, run_worker=not args.no_run_worker, wait_timeout=args.wait_timeout, batch_size=args.batch_size, reprocess_master=args.reprocess_master, taxonomy_mode=args.taxonomy_mode, fast=args.fast, force_products=args.force_products, force_relations=args.force_relations, run_pull_after=args.run_pull_after, ) print(json.dumps(result, indent=2, default=str)) return 0 if result.get("status") == "ok" else 1 if __name__ == "__main__": raise SystemExit(main())