"""Push a small sample of master products that have taxonomy + images to channel(s). Magento live push defaults to **phased** mode: 1. Create/update products + taxonomy (no gallery upload) 2. Upload gallery images (images_only) CLI: python -m app.jobs.push_channel_sample --channels magento --limit 10 --dry-run python -m app.jobs.push_channel_sample --channels magento --limit 10 --no-dry-run --wait --run-worker python -m app.jobs.push_channel_sample --channels magento --products-only --limit 10 --no-dry-run --wait --run-worker python -m app.jobs.push_channel_sample --channels magento --images-only --limit 10 --no-dry-run --wait --run-worker python -m app.jobs.push_channel_sample --channels magento --skus ACH-B12 ACH-B15 --no-dry-run --no-phased """ from __future__ import annotations import argparse import json import logging import sys import time from datetime import datetime from typing import Any, Dict, List, Optional from magento.sku_policy import filter_magento_push_skus logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) SAMPLE_QUEUE_OPTS: Dict[str, Any] = { "run_pull_first": False, "run_pull_after": False, } def _default_connection_ids(session) -> Dict[str, int]: from sqlalchemy import select from db.compat_connections import compat_connection_id from db.models import MagentoConnection, ShopifyConnection out: Dict[str, int] = {} magento = session.scalars(select(MagentoConnection).order_by(MagentoConnection.id)).first() if magento: out["magento"] = compat_connection_id("magento", magento.id) shopify = session.scalars( select(ShopifyConnection).where(ShopifyConnection.status == "active").order_by(ShopifyConnection.id) ).first() if shopify is None: shopify = session.scalars(select(ShopifyConnection).order_by(ShopifyConnection.id)).first() if shopify: out["shopify"] = compat_connection_id("shopify", shopify.id) return out def _wait_for_magento_queue(queue_id: int, *, timeout: int, run_worker: bool) -> Dict[str, Any]: from db.models import MagentoSyncQueue from db.session import get_session started = time.time() deadline = started + timeout last_log = 0.0 claim_misses = 0 while time.time() < deadline: with get_session() as session: row = session.get(MagentoSyncQueue, queue_id) if row is None: raise RuntimeError(f"Magento sync queue row {queue_id} not found") status = row.status if status in {"done", "failed"}: return { "queue_id": queue_id, "status": status, "label": row.label, "job_id": row.job_id, "last_error": row.last_error, "elapsed_seconds": round(time.time() - started, 1), } elapsed = int(time.time() - started) if time.time() - last_log >= 15: logger.info( "Waiting for magento_sync_queue id=%s status=%s (elapsed %ds)", queue_id, status, elapsed, ) last_log = time.time() if run_worker and status == "queued": from app.jobs.magento_sync_worker import run_one if run_one(queue_id): claim_misses = 0 continue claim_misses += 1 if claim_misses == 1: logger.info( "Queue id=%s still queued — waiting for this process or magento_worker to claim it", queue_id, ) time.sleep(2 if status == "running" else 1) raise TimeoutError( f"Timed out after {timeout}s waiting for magento_sync_queue id={queue_id}. " "Check magento_worker logs, or run with --wait only (no --run-worker) if systemd worker is active." ) def _enqueue_magento( session, *, skus: List[str], native_id: int, connection_id: int, dry_run: bool, label_suffix: str, extra_options: Dict[str, Any], ) -> Dict[str, Any]: from db.magento_repositories import SqlAlchemyMagentoSyncQueueRepository label = f"push-channel-sample-magento-{label_suffix}-{datetime.utcnow().strftime('%Y%m%d-%H%M%S')}" options: Dict[str, Any] = { "dry_run": dry_run, "use_master_catalog": True, "limit_skus": skus, **SAMPLE_QUEUE_OPTS, **extra_options, } queue_id = SqlAlchemyMagentoSyncQueueRepository( session ).enqueue(native_id, label, options=options, supersede_queued=True) return { "queue_id": queue_id, "label": label, "options": options, "native_connection_id": native_id, "connection_id": connection_id, "sku_count": len(skus), "skus": skus, } def push_magento_sample( session, *, skus: List[str], connection_id: int, dry_run: bool, images_only: bool, products_only: bool, phased: bool, wait: bool, run_worker: bool, wait_timeout: int, force_images: bool = True, force_products: bool = False, force_relations: bool = False, run_pull_after: bool = False, use_connection_pending_table: bool = True, ) -> Dict[str, Any]: from db.compat_connections import decode_compat_connection_id skus = filter_magento_push_skus(skus) channel_type, native_id = decode_compat_connection_id(connection_id) if channel_type != "magento": raise ValueError(f"connection_id {connection_id} is not a Magento connection") if not skus: return { "channel": "magento", "status": "skipped", "reason": "no_pushable_skus", "message": "All requested Magento SKUs were excluded by push policy.", "phased": False, "phases": [], "dry_run": dry_run, } phases: List[Dict[str, Any]] = [] force_product_opts: Dict[str, Any] = {} if force_products and not images_only: force_product_opts = {"force_upsert": True, "force_upsert_skus": list(skus)} # products_only (e.g. channel_full_publish --skip-images) must still allow # force_relations — otherwise parents upsert but SET_RELATIONS is never forced # and Magento shows Out of stock configurables with no options. force_relation_opts: Dict[str, Any] = {} if force_relations and not images_only: force_relation_opts = {"force_relations": True, "expand_relations": True} pull_after_opts = {"run_pull_after": True} if run_pull_after and not images_only else {} def _run_phase(phase_name: str, extra_options: Dict[str, Any]) -> Dict[str, Any]: queue_extra_options = { **extra_options, "use_connection_pending_table": use_connection_pending_table, } enqueued = _enqueue_magento( session, skus=skus, native_id=native_id, connection_id=connection_id, dry_run=dry_run, label_suffix=phase_name, extra_options=queue_extra_options, ) phase_result: Dict[str, Any] = {"phase": phase_name, **enqueued} session.commit() logger.info( "Enqueued magento_sync_queue id=%s label=%s phase=%s", enqueued["queue_id"], enqueued["label"], phase_name, ) if wait: outcome = _wait_for_magento_queue( enqueued["queue_id"], timeout=wait_timeout, run_worker=run_worker, ) phase_result.update(outcome) else: phase_result["note"] = "Queued — run magento_worker.py or pass --wait --run-worker" phases.append(phase_result) return phase_result if phased and not dry_run and not images_only and not products_only: logger.info("Phased Magento push: %d SKU(s) — phase 1 products, phase 2 images", len(skus)) _run_phase("products", {"products_only": True, **force_product_opts}) _run_phase("images", {"images_only": True, "force_images": True}) return { "channel": "magento", "status": "phased", "phased": True, "phases": phases, "dry_run": dry_run, } extra: Dict[str, Any] = dict(force_product_opts) phase_name = "full" if images_only: extra = {"images_only": True} if not dry_run and force_images: extra["force_images"] = True phase_name = "images" elif products_only: extra["products_only"] = True phase_name = "products" # Keep relation force + pull-after even when skipping gallery uploads. extra.update(force_relation_opts) extra.update(pull_after_opts) else: extra.update(force_relation_opts) extra.update(pull_after_opts) single = _run_phase(phase_name, extra) return { "channel": "magento", "status": single.get("status", "queued"), "phased": False, "phases": phases, "dry_run": dry_run, **single, } def push_shopify_sample( session, *, skus: List[str], connection_id: int, dry_run: bool, force: bool, ) -> Dict[str, Any]: from shopify.product_sync import push_products from db.compat_connections import decode_compat_connection_id from db.models import ShopifyConnection channel_type, native_id = decode_compat_connection_id(connection_id) if channel_type != "shopify": raise ValueError(f"connection_id {connection_id} is not a Shopify connection") shop = session.get(ShopifyConnection, native_id) shop_code = shop.shop_code if shop else None summary = push_products( session, dry_run=dry_run, skus=skus, shop_code=shop_code, connection_id=connection_id, force=force, ) return {"channel": "shopify", "connection_id": connection_id, **summary} def run_push_channel_sample( *, channels: List[str], limit: int = 10, skus: Optional[List[str]] = None, dry_run: bool = True, images_only: bool = False, products_only: bool = False, phased: bool = True, require_all_channels: bool = True, magento_connection_id: Optional[int] = None, shopify_connection_id: Optional[int] = None, wait: bool = False, run_worker: bool = False, wait_timeout: int = 3600, force: bool = False, force_images: bool = True, force_products: bool = False, force_relations: bool = False, run_pull_after: bool = False, only_assigned: bool = True, ) -> Dict[str, Any]: from db.channel_exports import resolve_pipeline_channel_code from db.push_ready_skus import select_push_ready_skus from db.session import get_session selected_channels = [resolve_pipeline_channel_code(ch) for ch in channels] summary: Dict[str, Any] = { "status": "ok", "dry_run": dry_run, "images_only": images_only, "products_only": products_only, "phased": phased and "magento" in selected_channels and not images_only and not products_only, "channels": selected_channels, "push_results": {}, } with get_session() as session: defaults = _default_connection_ids(session) connection_ids = { "magento": magento_connection_id or defaults.get("magento"), "shopify": shopify_connection_id or defaults.get("shopify"), } if skus: selected = sorted({str(s).strip() for s in skus if str(s).strip()})[:limit] selection = { "selected_skus": selected, "per_channel_skus": {ch: selected for ch in selected_channels}, "source": "explicit_skus", } else: selection = select_push_ready_skus( session, selected_channels, limit=limit, require_all_channels=require_all_channels, connection_ids={ ch: connection_ids[ch] for ch in selected_channels if connection_ids.get(ch) is not None }, only_assigned=only_assigned, ) selection["source"] = "auto_select" summary["selection"] = selection if not selection["selected_skus"] and not any( selection.get("per_channel_skus", {}).get(ch) for ch in selected_channels ): summary["status"] = "no_candidates" summary["message"] = ( "No SKUs matched assignment + taxonomy + images for the requested channel(s). " "Assign SKUs, import images (Catalog Import), and set taxonomy on Mappings." ) return summary for channel in selected_channels: channel_skus = (selection.get("per_channel_skus") or {}).get(channel) or selection["selected_skus"] if channel == "magento": channel_skus = filter_magento_push_skus(channel_skus) if not channel_skus: summary["push_results"][channel] = { "status": "skipped", "reason": "no_skus_selected_for_channel", } continue conn_id = connection_ids.get(channel) if not conn_id: summary["push_results"][channel] = { "status": "skipped", "reason": "no_connection_configured", } continue if channel == "magento": summary["push_results"][channel] = push_magento_sample( session, skus=channel_skus, connection_id=conn_id, dry_run=dry_run, images_only=images_only, products_only=products_only, phased=phased and not dry_run, wait=wait, run_worker=run_worker, wait_timeout=wait_timeout, force_images=force_images, force_products=force_products, force_relations=force_relations, run_pull_after=run_pull_after, ) elif channel == "shopify": summary["push_results"][channel] = push_shopify_sample( session, skus=channel_skus, connection_id=conn_id, dry_run=dry_run, force=force, ) else: summary["push_results"][channel] = { "status": "skipped", "reason": f"unsupported_channel:{channel}", } session.commit() return summary def main() -> int: parser = argparse.ArgumentParser( description="Push a sample of master SKUs (taxonomy + images) to Magento/Shopify" ) parser.add_argument( "--channels", nargs="+", required=True, choices=["magento", "shopify"], help="Target channel(s)", ) parser.add_argument("--limit", type=int, default=10, help="Max SKUs per selection (default: 10)") parser.add_argument("--skus", nargs="*", default=None, help="Explicit SKUs (skip auto-select)") parser.add_argument("--dry-run", dest="dry_run", action="store_true", default=True) parser.add_argument("--no-dry-run", dest="dry_run", action="store_false", help="Live push") parser.add_argument( "--phased", dest="phased", action="store_true", default=None, help="Magento: products first, then images (default for live push)", ) parser.add_argument("--no-phased", dest="phased", action="store_false", help="Magento: single combined sync job") parser.add_argument( "--images-only", action="store_true", help="Magento: upload images only (live runs force gallery upload unless --no-force-images)", ) parser.add_argument( "--no-force-images", action="store_true", help="Magento images-only: skip when images hash already matches sync state", ) parser.add_argument( "--products-only", action="store_true", help="Magento: create products + taxonomy only (no gallery upload)", ) parser.add_argument( "--per-channel", action="store_true", help="Pick up to --limit SKUs per channel instead of the same set on all channels", ) parser.add_argument("--magento-connection-id", type=int, default=None) parser.add_argument("--shopify-connection-id", type=int, default=None) parser.add_argument("--wait", action="store_true", help="Wait for Magento queue item(s) to finish") parser.add_argument( "--run-worker", action="store_true", help="While waiting, try to process the queue row in-process (omit if magento_worker systemd service is running)", ) parser.add_argument("--wait-timeout", type=int, default=3600) parser.add_argument("--force", action="store_true", help="Shopify: push even when payload hash unchanged") parser.add_argument( "--force-products", action="store_true", help="Magento: force product re-PUT even when product hashes match", ) parser.add_argument( "--force-relations", action="store_true", help="Magento: force configurable relation SET_RELATIONS in a full non-phased job", ) parser.add_argument( "--run-pull-after", action="store_true", help="Magento: pull after a full product/relation job to refresh catalog state", ) parser.add_argument( "--include-unassigned", action="store_true", help="Auto-select from all active masters, not only channel assignments", ) args = parser.parse_args() phased = args.phased if phased is None: phased = ( not args.dry_run and "magento" in args.channels and not args.images_only and not args.products_only ) if args.images_only and args.products_only: print("Use only one of --images-only or --products-only", file=sys.stderr) return 2 if args.force_relations: phased = False summary = run_push_channel_sample( channels=args.channels, limit=args.limit, skus=args.skus, dry_run=args.dry_run, images_only=args.images_only, products_only=args.products_only, phased=phased, require_all_channels=not args.per_channel, magento_connection_id=args.magento_connection_id, shopify_connection_id=args.shopify_connection_id, wait=args.wait, run_worker=args.run_worker, wait_timeout=args.wait_timeout, force=args.force, force_images=not args.no_force_images, force_products=args.force_products, force_relations=args.force_relations, run_pull_after=args.run_pull_after, only_assigned=not args.include_unassigned, ) print(json.dumps(summary, indent=2, default=str)) if summary.get("status") == "no_candidates": return 1 for channel, result in (summary.get("push_results") or {}).items(): if result.get("status") == "failed": return 1 for phase in result.get("phases") or []: if phase.get("status") == "failed": return 1 if channel == "shopify" and result.get("failed"): return 1 return 0 if __name__ == "__main__": sys.exit(main())