"""Reprocess stored Master SKU raw payloads into canonical master catalog tables. CLI: python -m app.jobs.reprocess_master_skus --source-label master_sku_reprocess """ from __future__ import annotations import argparse import sys def _empty_reprocess_result(*, source_label: str, batch_size: int, start_offset: int, max_batches: int | None) -> dict: return { "status": "ok", "source_label": source_label, "source": "master_sku_raw_payload", "batch_size": batch_size, "start_offset": start_offset, "max_batches": max_batches, "batch_count": 0, "input_rows": 0, "valid_skus": 0, "inserted": 0, "updated": 0, "unchanged": 0, "attribute_values": 0, "channel_assignments": 0, "prices": 0, "location_availability": 0, "attribute_definitions": 0, "relations_upserted": 0, "missing_sku": 0, "association_overrides": {}, "association_discovery": {}, "taxonomy_reconciliation": {}, "batches": [], } def _merge_nested_counts(target: dict, incoming: dict) -> None: if not isinstance(incoming, dict): return for key, value in incoming.items(): if isinstance(value, bool): target[key] = value elif isinstance(value, int): target[key] = int(target.get(key, 0)) + value elif isinstance(value, list): existing = target.setdefault(key, []) if isinstance(existing, list): existing.extend(value) elif isinstance(value, dict): child = target.setdefault(key, {}) if isinstance(child, dict): _merge_nested_counts(child, value) else: target[key] = dict(value) elif key not in target: target[key] = value def run_reprocess_master_skus( *, source_label: str = "master_sku_reprocess", batch_size: int = 500, start_offset: int = 0, max_batches: int | None = None, ) -> dict: import pandas as pd from sqlalchemy import select from db.master_catalog import import_master_sku_dataframe from db.master_taxonomy_sync import link_products_to_taxonomy_from_master from db.models import MagentoConnection, ShopifyConnection from db.models import MasterProduct from db.session import get_session batch_size = max(int(batch_size or 500), 1) start_offset = max(int(start_offset or 0), 0) max_batches = None if max_batches in (None, 0) else max(int(max_batches), 1) payloads = [] with get_session() as session: payloads = [ dict(row.raw_payload or {}) for row in session.scalars( select(MasterProduct) .where(MasterProduct.is_active.is_(True)) .where(MasterProduct.raw_payload.is_not(None)) .order_by(MasterProduct.sku) ).all() if row.raw_payload ] if not payloads: return _empty_reprocess_result( source_label=source_label, batch_size=batch_size, start_offset=start_offset, max_batches=max_batches, ) payloads = payloads[start_offset:] if not payloads: return _empty_reprocess_result( source_label=source_label, batch_size=batch_size, start_offset=start_offset, max_batches=max_batches, ) result = _empty_reprocess_result( source_label=source_label, batch_size=batch_size, start_offset=start_offset, max_batches=max_batches, ) batch_ranges = list(range(0, len(payloads), batch_size)) if max_batches is not None: batch_ranges = batch_ranges[:max_batches] for batch_index, local_start in enumerate(batch_ranges, start=1): start = start_offset + local_start batch_payloads = payloads[local_start : local_start + batch_size] with get_session() as session: stats = import_master_sku_dataframe( session, pd.DataFrame(batch_payloads), source_label=source_label, reconcile_taxonomy=False, ) session.commit() import_skus = list(stats.get("import_skus") or []) taxonomy_reconciliation = {} if import_skus: with get_session() as session: magento_connection_id = session.scalar( select(MagentoConnection.id) .where(MagentoConnection.status == "active") .order_by(MagentoConnection.id) .limit(1) ) shopify_connection_id = session.scalar( select(ShopifyConnection.id) .where(ShopifyConnection.status == "active") .order_by(ShopifyConnection.id) .limit(1) ) taxonomy_reconciliation = link_products_to_taxonomy_from_master( session, skus=import_skus, replace_existing=True, sync_channels=True, magento_connection_id=int(magento_connection_id) if magento_connection_id else None, shopify_connection_id=int(shopify_connection_id) if shopify_connection_id else None, dry_run=False, ) session.commit() result["batch_count"] = batch_index result["input_rows"] += int(stats.get("input_rows", 0)) result["valid_skus"] += int(stats.get("valid_skus", 0)) result["inserted"] += int(stats.get("inserted", 0)) result["updated"] += int(stats.get("updated", 0)) result["unchanged"] += int(stats.get("unchanged", 0)) result["attribute_values"] += int(stats.get("attribute_values", 0)) result["channel_assignments"] += int(stats.get("channel_assignments", 0)) result["prices"] += int(stats.get("prices", 0)) result["location_availability"] += int(stats.get("location_availability", 0)) result["attribute_definitions"] += int(stats.get("attribute_definitions", 0)) result["relations_upserted"] += int(stats.get("relations_upserted", 0)) result["missing_sku"] += int(stats.get("missing_sku", 0)) _merge_nested_counts(result["association_overrides"], stats.get("association_overrides") or {}) _merge_nested_counts(result["association_discovery"], stats.get("association_discovery") or {}) _merge_nested_counts(result["taxonomy_reconciliation"], taxonomy_reconciliation) result["batches"].append( { "batch": batch_index, "start": start, "size": len(batch_payloads), "valid_skus": int(stats.get("valid_skus", 0)), "inserted": int(stats.get("inserted", 0)), "updated": int(stats.get("updated", 0)), "unchanged": int(stats.get("unchanged", 0)), "missing_sku": int(stats.get("missing_sku", 0)), "taxonomy_reconciliation": taxonomy_reconciliation, } ) return result def main() -> int: parser = argparse.ArgumentParser(description="Reprocess stored Master SKU raw payloads") parser.add_argument("--source-label", default="master_sku_reprocess", help="Reprocess source label") parser.add_argument("--batch-size", type=int, default=500, help="Rows per transaction batch") parser.add_argument("--start-offset", type=int, default=0, help="Start from this zero-based row offset") parser.add_argument("--max-batches", type=int, default=0, help="Optional limit on how many batches to process") args = parser.parse_args() result = run_reprocess_master_skus( source_label=args.source_label, batch_size=args.batch_size, start_offset=args.start_offset, max_batches=args.max_batches, ) print( "Master SKU reprocess complete: " f"start_offset={result.get('start_offset', 0)}, " f"max_batches={result.get('max_batches')}, " f"batches={result.get('batch_count', 0)}, " f"valid_skus={result.get('valid_skus', 0)}, " f"inserted={result.get('inserted', 0)}, " f"updated={result.get('updated', 0)}, " f"unchanged={result.get('unchanged', 0)}, " f"attribute_values={result.get('attribute_values', 0)}, " f"prices={result.get('prices', 0)}, " f"location_availability={result.get('location_availability', 0)}" ) return 0 if __name__ == "__main__": sys.exit(main())