"""End-to-end master taxonomy sync: create nodes, link remotes, assign SKUs, push channels. CLI: python -m app.jobs.sync_master_taxonomy_pipeline --dry-run python -m app.jobs.sync_master_taxonomy_pipeline --apply python -m app.jobs.sync_master_taxonomy_pipeline --apply --skip-product-push python -m app.jobs.sync_master_taxonomy_pipeline --apply --magento-connection-id 1 --magento-connection-id 2 """ from __future__ import annotations import argparse import json import logging import math import os import time import uuid from typing import Any, Callable, Dict, List, Optional from sqlalchemy import select from db.master_taxonomy_hierarchy import build_hierarchy_from_master_catalog, sync_channel_links_from_registries from db.master_taxonomy_registry import backfill_master_taxonomy_from_products from db.master_taxonomy_sync import ( link_products_to_taxonomy_from_master, push_taxonomy_nodes_to_channels, ) from db.channel_assignments import active_skus_for_channel from db.models import MagentoConnection, MasterProduct, ShopifyConnection from db.session import get_session logger = logging.getLogger(__name__) DEFAULT_SKU_BATCH_SIZE = int(os.getenv("MASTER_TAXONOMY_SKU_BATCH_SIZE", "500")) DEFAULT_PUSH_BATCH_SIZE = int(os.getenv("MASTER_TAXONOMY_PUSH_BATCH_SIZE", "500")) def _active_connection_id(session, model, connection_id: Optional[int]) -> Optional[int]: if connection_id is not None: row = session.get(model, int(connection_id)) return int(row.id) if row is not None else None row = session.scalar( select(model).where(model.status == "active").order_by(model.id).limit(1) ) if row is None: row = session.scalar(select(model).order_by(model.id).limit(1)) return int(row.id) if row is not None else None def _import_rows_from_master(session) -> List[Dict[str, Any]]: from db.master_taxonomy_hierarchy import _attrs_by_sku products = list(session.scalars(select(MasterProduct).where(MasterProduct.is_active.is_(True))).all()) attrs_by_sku = _attrs_by_sku(session, [p.id for p in products]) rows: List[Dict[str, Any]] = [] for product in products: rows.append( { "sku": product.sku, "brand": product.brand, "product_family": product.product_family, "category_l1": product.category_l1, "category_l2": product.category_l2, "category_l3": product.category_l3, "collection": product.collection, "attributes": attrs_by_sku.get(product.sku, {}), } ) return rows def _load_target_skus( session, skus: Optional[List[str]], *, magento_connection_id: Optional[int] = None, shopify_connection_id: Optional[int] = None, ) -> List[str]: if skus: return [str(s).strip() for s in skus if str(s).strip()] assigned: set[str] = set() if magento_connection_id is not None: assigned |= active_skus_for_channel(session, "magento") if shopify_connection_id is not None: assigned |= active_skus_for_channel(session, "shopify") if assigned: return sorted(str(s).strip() for s in assigned if str(s).strip()) return [ str(row) for row in session.scalars(select(MasterProduct.sku).where(MasterProduct.is_active.is_(True))).all() ] def _apply_missing_taxonomy_proposals( session, *, dry_run: bool, sync_channels: bool, magento_connection_id: Optional[int], shopify_connection_id: Optional[int], magento_api: Any, shopify_client: Any, shopify_shop_code: Optional[str], ) -> Dict[str, Any]: from db.master_taxonomy_intent import apply_taxonomy_proposals, summarize_import_taxonomy_intent intent = summarize_import_taxonomy_intent(session, _import_rows_from_master(session)) proposals = intent.get("proposed_nodes") or [] if not proposals: return {"proposal_count": 0, "intent": {"matched_counts": intent.get("matched_counts")}} applied = apply_taxonomy_proposals( session, proposals, dry_run=dry_run, sync_channels=sync_channels, magento_connection_id=magento_connection_id, shopify_connection_id=shopify_connection_id, magento_api=magento_api, shopify_client=shopify_client, shopify_shop_code=shopify_shop_code, ) return { "proposal_count": len(proposals), "intent": { "matched_counts": intent.get("matched_counts"), "ambiguous_count": len(intent.get("ambiguous") or []), }, "applied": applied, } def _pull_remote_registries( session, *, magento_connection_id: Optional[int], shopify_connection_id: Optional[int], dry_run: bool, ) -> Dict[str, Any]: out: Dict[str, Any] = {} if dry_run: return {"dry_run": True} if magento_connection_id: from app.jobs.magento_baseline_sync import run_baseline_sync out["magento_baseline"] = run_baseline_sync(int(magento_connection_id), force=True, session=session) magento_status = str((out["magento_baseline"] or {}).get("status") or "").lower() if magento_status == "failed": # Shared session may already be rolled back by baseline sync; surface failure # so the pipeline stops instead of continuing with a poisoned transaction. err = (out["magento_baseline"] or {}).get("error") or "Magento baseline sync failed" raise RuntimeError(f"Magento baseline sync failed for connection {magento_connection_id}: {err}") if shopify_connection_id: from shopify.collection_sync import pull_shopify_collections from shopify.connections import build_client, get_active_connection connection = session.get(ShopifyConnection, int(shopify_connection_id)) if connection is None: connection = get_active_connection(session) if connection is not None: client = build_client(connection) out["shopify_collections"] = pull_shopify_collections( client, shop_code=connection.shop_code, connection_id=connection.id, session=session, ) return out def _push_products_for_skus( session, skus: List[str], *, magento_connection_id: Optional[int], shopify_connection_id: Optional[int], dry_run: bool, inline_threshold: int = 50, batch_size: int = DEFAULT_PUSH_BATCH_SIZE, ) -> Dict[str, Any]: if dry_run or not skus: return {"dry_run": dry_run, "sku_count": len(skus)} out: Dict[str, Any] = {"sku_count": len(skus), "batches": []} if magento_connection_id: from db.magento_repositories import SqlAlchemyMagentoSyncQueueRepository magento_batches: List[Dict[str, Any]] = [] for offset in range(0, len(skus), batch_size): batch = skus[offset : offset + batch_size] label = f"taxonomy-pipeline-{uuid.uuid4().hex[:8]}" queue_id = SqlAlchemyMagentoSyncQueueRepository(session).enqueue( int(magento_connection_id), label, options={ "dry_run": False, "use_master_catalog": True, "limit_skus": batch, "expand_relations": False, }, supersede_queued=False, ) magento_batches.append( {"status": "queued", "queue_id": queue_id, "label": label, "sku_count": len(batch)} ) out["magento"] = {"status": "queued", "batches": magento_batches, "batch_count": len(magento_batches)} if shopify_connection_id: if len(skus) <= inline_threshold: from shopify.product_sync import push_products out["shopify"] = push_products( session, dry_run=False, skus=skus, connection_id=int(shopify_connection_id), force=True, ) else: from app.jobs.channel_jobs import enqueue_channel_job from main import _compat_id shopify_batches: List[Dict[str, Any]] = [] for offset in range(0, len(skus), batch_size): batch = skus[offset : offset + batch_size] job = enqueue_channel_job( session, channel_connection_id=_compat_id("shopify", int(shopify_connection_id)), channel_type="shopify", native_connection_id=int(shopify_connection_id), channel_code="shopify", job_type="push", dry_run=False, mode="incremental", notes="master taxonomy pipeline", options={"skus": batch, "force": True}, ) shopify_batches.append({"status": "queued", "job_id": job.id, "sku_count": len(batch)}) out["shopify"] = {"status": "queued", "batches": shopify_batches, "batch_count": len(shopify_batches)} return out def _link_products_in_batches( *, target_skus: List[str], dry_run: bool, native_magento: Optional[int], native_shopify: Optional[int], sku_batch_size: int, report: Callable[..., None], ) -> Dict[str, Any]: """Link products in SKU batches with a commit after each batch.""" total_skus = len(target_skus) batch_total = max(1, math.ceil(total_skus / sku_batch_size)) if total_skus else 1 aggregate: Dict[str, Any] = { "dry_run": dry_run, "sku_total": total_skus, "batch_size": sku_batch_size, "batch_count": batch_total, "batches": [], "assignment_count": 0, "upserted": 0, } if not target_skus: report("link_products", notes="No SKUs to link") return aggregate for batch_index, offset in enumerate(range(0, total_skus, sku_batch_size), start=1): batch = target_skus[offset : offset + sku_batch_size] report( "link_products", notes=f"Linking products batch {batch_index}/{batch_total} ({len(batch)} SKUs)", extra={ "sku_processed": min(offset + len(batch), total_skus), "sku_total": total_skus, "batch": batch_index, "batch_total": batch_total, }, ) with get_session() as session: batch_result = link_products_to_taxonomy_from_master( session, skus=batch, replace_existing=True, sync_channels=not dry_run, magento_connection_id=native_magento, shopify_connection_id=native_shopify, dry_run=dry_run, ) if not dry_run: session.commit() aggregate["batches"].append( { "batch": batch_index, "sku_count": len(batch), "assignment_count": batch_result.get("assignment_count"), "upserted": batch_result.get("upserted"), } ) aggregate["assignment_count"] += int(batch_result.get("assignment_count") or 0) aggregate["upserted"] += int(batch_result.get("upserted") or 0) logger.info( "link_products batch %s/%s: skus=%s upserted=%s", batch_index, batch_total, len(batch), batch_result.get("upserted"), ) return aggregate def run_master_taxonomy_pipeline( *, dry_run: bool = True, magento_connection_id: Optional[int] = None, shopify_connection_id: Optional[int] = None, pull_remotes: bool = True, create_missing_remotes: bool = False, push_products: bool = True, skus: Optional[List[str]] = None, reprocess_master: bool = False, sku_batch_size: int = DEFAULT_SKU_BATCH_SIZE, progress_callback: Optional[Callable[[str, Optional[Dict[str, Any]]], None]] = None, taxonomy_mode: str = "link_only", ) -> Dict[str, Any]: """Create taxonomy nodes, link/create remotes, assign SKUs in batches, optionally push products. taxonomy_mode: - full: backfill + hierarchy rebuild + proposals + remotes + link products - link_only: only assign the given SKUs to existing taxonomy (default; invent frozen) - skip: no taxonomy work When MASTER_TAXONOMY_AUTO_INVENT is off (default), full is coerced to link_only. """ from app.jobs.master_taxonomy_enqueue import _resolve_connections from db.taxonomy_invent_policy import coerce_remote_provision_when_frozen, coerce_taxonomy_mode_when_frozen mode, requested_mode, coerced = coerce_taxonomy_mode_when_frozen(taxonomy_mode) if mode not in {"full", "link_only", "skip"}: raise ValueError("taxonomy_mode must be 'full', 'link_only', or 'skip'") create_missing_remotes, remotes_coerced = coerce_remote_provision_when_frozen(create_missing_remotes) summary: Dict[str, Any] = {"dry_run": dry_run, "taxonomy_mode": mode, "steps": {}} if coerced: summary["taxonomy_mode_requested"] = requested_mode summary["taxonomy_mode_enforced"] = mode summary["note"] = ( "Taxonomy invent is frozen (MASTER_TAXONOMY_AUTO_INVENT=false); " "full mode coerced to link_only." ) if remotes_coerced: summary["create_missing_remotes_requested"] = True summary["create_missing_remotes_enforced"] = False summary["remote_provision_note"] = ( "Remote taxonomy provision is frozen; create_missing_remotes coerced off." ) if mode == "skip": summary["status"] = "ok" summary["steps"]["taxonomy"] = {"status": "skipped", "reason": "taxonomy_mode=skip"} return summary step_index = 0 rebuild = mode == "full" planned_steps = [ name for name, enabled in ( ("reprocess_master", rebuild and reprocess_master and not dry_run), ("pull_remotes", rebuild and pull_remotes), ("backfill", rebuild), ("build_hierarchy", rebuild), ("apply_proposals", rebuild), ("build_hierarchy_after_proposals", rebuild and not dry_run), ("sync_links", rebuild), ("push_taxonomy_remotes", rebuild and create_missing_remotes), ("link_products", True), ("push_products", push_products), ) if enabled ] total_steps = len(planned_steps) def report(stage: str, *, notes: Optional[str] = None, extra: Optional[Dict[str, Any]] = None) -> None: nonlocal step_index if stage in planned_steps: step_index = planned_steps.index(stage) + 1 message = notes or stage.replace("_", " ") logger.info( "Taxonomy pipeline [%s/%s]: %s", step_index or "?", total_steps, message, ) if progress_callback: progress_callback( stage, { "completed": step_index, "total": total_steps, "percent": round(100 * step_index / total_steps, 1) if total_steps else None, "notes": notes or stage.replace("_", " "), **(extra or {}), }, ) with get_session() as session: magento_id = _active_connection_id(session, MagentoConnection, magento_connection_id) shopify_id = _active_connection_id(session, ShopifyConnection, shopify_connection_id) summary["connections"] = {"magento": magento_id, "shopify": shopify_id} target_skus = _load_target_skus( session, skus, magento_connection_id=magento_id, shopify_connection_id=shopify_id, ) summary["sku_total"] = len(target_skus) native_magento, native_shopify, magento_api, shopify_client, shopify_shop_code = _resolve_connections( { "dry_run": dry_run, "create_missing": create_missing_remotes, "magento_connection_id": magento_id, "shopify_connection_id": shopify_id, "include_magento": bool(magento_id), "include_shopify": bool(shopify_id), } ) if rebuild and reprocess_master and not dry_run: report("reprocess_master", notes="Reprocessing master SKU raw payloads") step_started = time.perf_counter() with get_session() as session: from db.master_catalog import reprocess_master_sku_raw_payloads summary["steps"]["reprocess_master"] = reprocess_master_sku_raw_payloads(session) session.commit() logger.info("Taxonomy step reprocess_master finished in %.1fs", time.perf_counter() - step_started) if rebuild and pull_remotes: report("pull_remotes", notes="Pulling Magento and Shopify category/collection registries") step_started = time.perf_counter() with get_session() as session: summary["steps"]["pull_remotes"] = _pull_remote_registries( session, magento_connection_id=magento_id, shopify_connection_id=shopify_id, dry_run=dry_run, ) if not dry_run: session.commit() logger.info("Taxonomy step pull_remotes finished in %.1fs", time.perf_counter() - step_started) if rebuild: report("backfill", notes="Backfilling taxonomy from master catalog") step_started = time.perf_counter() with get_session() as session: summary["steps"]["backfill"] = backfill_master_taxonomy_from_products(session, dry_run=dry_run) if not dry_run: session.commit() logger.info("Taxonomy step backfill finished in %.1fs", time.perf_counter() - step_started) report("build_hierarchy", notes="Building taxonomy hierarchy") step_started = time.perf_counter() with get_session() as session: summary["steps"]["build_hierarchy"] = build_hierarchy_from_master_catalog(session, dry_run=dry_run) if not dry_run: session.commit() logger.info("Taxonomy step build_hierarchy finished in %.1fs", time.perf_counter() - step_started) report("apply_proposals", notes="Applying missing taxonomy proposals") step_started = time.perf_counter() with get_session() as session: summary["steps"]["apply_proposals"] = _apply_missing_taxonomy_proposals( session, dry_run=dry_run, sync_channels=False, magento_connection_id=native_magento, shopify_connection_id=native_shopify, magento_api=magento_api, shopify_client=shopify_client, shopify_shop_code=shopify_shop_code, ) if not dry_run: session.commit() logger.info("Taxonomy step apply_proposals finished in %.1fs", time.perf_counter() - step_started) if not dry_run: report("build_hierarchy_after_proposals", notes="Rebuilding hierarchy after proposals") step_started = time.perf_counter() with get_session() as session: summary["steps"]["build_hierarchy_after_proposals"] = build_hierarchy_from_master_catalog( session, dry_run=False ) session.commit() logger.info( "Taxonomy step build_hierarchy_after_proposals finished in %.1fs", time.perf_counter() - step_started, ) report("sync_links", notes="Syncing channel links from registries") step_started = time.perf_counter() with get_session() as session: summary["steps"]["sync_links"] = sync_channel_links_from_registries( session, magento_connection_id=native_magento, shopify_connection_id=native_shopify, dry_run=dry_run, write_listing_path_targets=True, ) if not dry_run: session.commit() logger.info("Taxonomy step sync_links finished in %.1fs", time.perf_counter() - step_started) if create_missing_remotes: report("push_taxonomy_remotes", notes="Creating or updating remote taxonomy on channels") step_started = time.perf_counter() with get_session() as session: summary["steps"]["push_taxonomy_remotes"] = push_taxonomy_nodes_to_channels( session, magento_connection_id=native_magento, shopify_connection_id=native_shopify, magento_api=magento_api if not dry_run else None, shopify_client=shopify_client if not dry_run else None, shopify_shop_code=shopify_shop_code, dry_run=dry_run, create_missing=create_missing_remotes, ) if not dry_run: session.commit() push_stats = summary["steps"].get("push_taxonomy_remotes") or {} logger.info( "Taxonomy step push_taxonomy_remotes finished in %.1fs (nodes=%s magento_created=%s magento_updated=%s errors=%s)", time.perf_counter() - step_started, push_stats.get("node_count"), push_stats.get("magento_created"), push_stats.get("magento_updated"), push_stats.get("error_count"), ) else: logger.info("Taxonomy mode=link_only: skipping hierarchy/proposals/remote rebuild") summary["steps"]["link_products"] = _link_products_in_batches( target_skus=target_skus, dry_run=dry_run, native_magento=native_magento, native_shopify=native_shopify, sku_batch_size=sku_batch_size, report=report, ) if push_products: report("push_products", notes="Queueing Magento and Shopify product pushes") with get_session() as session: summary["steps"]["push_products"] = _push_products_for_skus( session, list(target_skus), magento_connection_id=native_magento, shopify_connection_id=native_shopify, dry_run=dry_run, ) if not dry_run: session.commit() summary["status"] = "ok" return summary def _normalize_magento_connection_ids(values: Optional[List[int]]) -> List[Optional[int]]: if not values: return [None] out: List[Optional[int]] = [] seen: set[int] = set() for value in values: conn_id = int(value) if conn_id in seen: continue seen.add(conn_id) out.append(conn_id) return out or [None] def _multi_connection_run_plan( *, magento_connection_ids: Optional[List[int]], shopify_connection_id: Optional[int], reprocess_master: bool, ) -> List[Dict[str, Any]]: plan: List[Dict[str, Any]] = [] for index, magento_id in enumerate(_normalize_magento_connection_ids(magento_connection_ids)): plan.append( { "magento_connection_id": magento_id, "shopify_connection_id": shopify_connection_id if index == 0 else None, "reprocess_master": bool(reprocess_master and index == 0), } ) return plan def main() -> int: logging.basicConfig(level=logging.INFO) parser = argparse.ArgumentParser(description="Sync master taxonomy nodes, remote links, and SKU assignments") parser.add_argument("--apply", action="store_true", help="Persist changes (default is dry-run)") parser.add_argument("--sku", action="append", dest="skus", help="Limit SKU assignment/push; repeatable") parser.add_argument( "--magento-connection-id", type=int, action="append", default=None, help="Magento connection id; repeat to sync multiple Magento stores", ) parser.add_argument("--shopify-connection-id", type=int, default=None) parser.add_argument("--skip-pull-remotes", action="store_true") parser.add_argument("--skip-create-remotes", action="store_true", help="Only link existing remotes; do not create") parser.add_argument("--skip-product-push", action="store_true") parser.add_argument( "--taxonomy-mode", choices=["full", "link-only", "link_only", "skip"], default="link_only", help="link-only=SKU assign only (default); full=rebuild hierarchy (requires invent unlock); skip=no taxonomy", ) parser.add_argument("--reprocess-master", action="store_true", help="Rebuild master fields from stored raw payloads") parser.add_argument("--sku-batch-size", type=int, default=DEFAULT_SKU_BATCH_SIZE) args = parser.parse_args() run_plan = _multi_connection_run_plan( magento_connection_ids=args.magento_connection_id, shopify_connection_id=args.shopify_connection_id, reprocess_master=args.reprocess_master, ) if len(run_plan) == 1: item = run_plan[0] result = run_master_taxonomy_pipeline( dry_run=not args.apply, magento_connection_id=item["magento_connection_id"], shopify_connection_id=item["shopify_connection_id"], pull_remotes=not args.skip_pull_remotes, create_missing_remotes=not args.skip_create_remotes, push_products=not args.skip_product_push, skus=args.skus, reprocess_master=item["reprocess_master"], sku_batch_size=args.sku_batch_size, taxonomy_mode=args.taxonomy_mode, ) else: result = { "status": "ok", "dry_run": not args.apply, "multi_magento": True, "run_plan": run_plan, "runs": [], } for item in run_plan: run_result = run_master_taxonomy_pipeline( dry_run=not args.apply, magento_connection_id=item["magento_connection_id"], shopify_connection_id=item["shopify_connection_id"], pull_remotes=not args.skip_pull_remotes, create_missing_remotes=not args.skip_create_remotes, push_products=not args.skip_product_push, skus=args.skus, reprocess_master=item["reprocess_master"], sku_batch_size=args.sku_batch_size, taxonomy_mode=args.taxonomy_mode, ) result["runs"].append(run_result) if run_result.get("status") != "ok": result["status"] = "failed" print(json.dumps(result, indent=2, default=str)) return 0 if result.get("status") == "ok" else 1 if __name__ == "__main__": raise SystemExit(main())