""" Enable & Inventory Backfill — Add MSI sources/qty to simple/children, enable all products on website. 1. Adds source-items with qty to simple and configurable-children (no MSI for configurable parents). 2. Enables all products (simple, children, parents) with status=1 and appropriate visibility. WARNING: Step 1 writes MSI / source-items. Do not use when NetSuite (or another system) owns Magento inventory. For status+visibility only, use: python -m app.jobs.magento_fix_status_visibility --connection-id X --apply CLI: python -m app.jobs.magento_enable_inventory_backfill --connection-id X [--stock-id N] [--dry-run] Env: Same as magento_msi_source_backfill for MSI; uses MAGENTO_MSI_* vars. """ from __future__ import annotations import argparse import logging import os import sys import time from typing import Any, Dict, List, Optional logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Magento visibility values VISIBILITY_NOT_VISIBLE = 1 # Not Visible Individually (configurable children) VISIBILITY_CATALOG_SEARCH = 4 # Catalog + Search (standalone simples, configurable parents) STATUS_ENABLED = 1 def run_enable_inventory_backfill( connection_id: int, *, stock_id: Optional[int] = None, dry_run: bool = False, session=None, skus: Optional[List[str]] = None, ) -> Dict[str, Any]: """ Backfill MSI source items for simple/children, then enable all products on website. """ from db.session import get_session from db.magento_repositories import ( SqlAlchemyMagentoConnectionRepository, SqlAlchemySourceRegistryRepository, SqlAlchemyStockSourceLinksRepository, ) from db.repositories import SqlAlchemyIngestRepository from magento.oauth_client import MagentoOAuthClient, build_magento_oauth_kwargs from magento.magento_api import MagentoRestClient from settings import load_magento_sync_runtime_config runtime_cfg = load_magento_sync_runtime_config() api_delay_s = max(0, runtime_cfg.api_delay_ms) / 1000.0 own_session = session is None ctx = get_session() if own_session else _noop_ctx(session) with ctx as sess: conn_repo = SqlAlchemyMagentoConnectionRepository(sess) source_repo = SqlAlchemySourceRegistryRepository(sess) stock_links_repo = SqlAlchemyStockSourceLinksRepository(sess) ingest_repo = SqlAlchemyIngestRepository(sess) conn = conn_repo.get_for_sync(connection_id) if not conn: return {"status": "failed", "error": f"Connection {connection_id} not found"} # Resolve source codes for MSI source_codes: List[str] = [] if stock_id: source_codes = stock_links_repo.get_sources_for_stock(connection_id, stock_id) if not source_codes: logger.warning("Stock %s has no linked sources; falling back to all sources", stock_id) if not source_codes: source_codes = source_repo.get_enabled_source_codes(connection_id) if not source_codes: source_codes = source_repo.get_all_source_codes(connection_id) if not source_codes: configured = os.getenv("MAGENTO_DEFAULT_SOURCE_CODE", "").strip() if configured: source_codes = [configured] else: logger.warning("No sources in registry; MSI backfill will be skipped. Run baseline sync first.") oauth = MagentoOAuthClient(**build_magento_oauth_kwargs(conn)) api = MagentoRestClient(oauth) # Load catalog SKUs and product types if skus: all_skus = list(skus) snapshots = ingest_repo.load_current_snapshots(all_skus) else: all_skus = ingest_repo.load_all_current_skus() snapshots = ingest_repo.load_current_snapshots(all_skus) if all_skus else [] sku_to_snap: Dict[str, Dict[str, Any]] = {s["sku"]: s for s in snapshots if s.get("sku")} simple_skus = [ s["sku"] for s in snapshots if (str(s.get("product_type") or "").lower() == "simple") ] configurable_parent_skus = [ s["sku"] for s in snapshots if (str(s.get("product_type") or "").lower() == "configurable") ] # Children have variant_of set; used for visibility def _is_configurable_child(snap: Dict) -> bool: return bool(snap.get("variant_of")) if dry_run: return { "status": "dry_run", "skus_total": len(all_skus), "simple_skus_count": len(simple_skus), "configurable_parents_count": len(configurable_parent_skus), "sources": source_codes, "message": "Would add MSI sources to simples and enable all products", } # Step 1: MSI source backfill for simple products (incl. configurable children) inv_status, sources_api, _ = api.get_inventory_sources() msi_applied = 0 msi_errors: List[str] = [] if inv_status == 200 and sources_api and source_codes: from settings import load_magento_msi_config from magento.inventory_service import ensure_simple_sku_inventory msi_cfg = load_magento_msi_config() needs_backfill: List[str] = [] chunk_size = min(100, max(20, len(simple_skus))) for i in range(0, len(simple_skus), chunk_size): chunk = simple_skus[i : i + chunk_size] time.sleep(api_delay_s) status, by_sku, _ = api.get_source_items_for_skus(chunk) if status != 200: continue for sku in chunk: items = by_sku.get(sku, []) if not items: needs_backfill.append(sku) logger.info("MSI: %d simple SKUs need source backfill", len(needs_backfill)) for sku in needs_backfill: time.sleep(api_delay_s) result = ensure_simple_sku_inventory( api, connection_id, sku, source_codes, msi_cfg ) if result.success and result.after_source_items: msi_applied += len(result.after_source_items) elif result.error: msi_errors.append(f"{sku}: {result.error}") else: if inv_status != 200: logger.warning("MSI not available (HTTP %s); skipping source backfill", inv_status) elif not source_codes: logger.warning("No source codes; skipping MSI backfill") # Step 2: Enable all products (status=1, visibility) enable_ok = 0 enable_errors: List[str] = [] for sku in all_skus: time.sleep(api_delay_s) snap = sku_to_snap.get(sku) is_child = snap and _is_configurable_child(snap) visibility = VISIBILITY_NOT_VISIBLE if is_child else VISIBILITY_CATALOG_SEARCH payload = {"status": STATUS_ENABLED, "visibility": visibility} try: status, body = api.put_product(sku, payload) if status in (200, 201): enable_ok += 1 else: enable_errors.append(f"{sku}: HTTP {status}") except Exception as e: enable_errors.append(f"{sku}: {e}") return { "status": "success", "skus_total": len(all_skus), "msi_source_items_posted": msi_applied, "msi_errors": msi_errors[:20], "products_enabled": enable_ok, "enable_errors": enable_errors[:20], } class _noop_ctx: def __init__(self, obj): self._obj = obj def __enter__(self): return self._obj def __exit__(self, *args): pass def main() -> None: parser = argparse.ArgumentParser(description="Enable products and backfill MSI sources") parser.add_argument("--connection-id", type=int, required=True) parser.add_argument("--stock-id", type=int, default=None) parser.add_argument("--dry-run", action="store_true") parser.add_argument("--skus", type=str, default=None, help="Comma-separated SKUs to limit scope") args = parser.parse_args() skus = [x.strip() for x in args.skus.split(",") if x.strip()] if args.skus else None result = run_enable_inventory_backfill( args.connection_id, stock_id=args.stock_id, dry_run=args.dry_run, skus=skus, ) if result.get("status") == "failed": logger.error("Backfill failed: %s", result.get("error")) sys.exit(1) logger.info("Done: %s", result) if args.dry_run: print(result) if __name__ == "__main__": main()