""" Incremental sync planner: compares snapshot rows vs magento_sync_state hashes and produces minimal action list per SKU with execution ordering. Sync order (enforced via pass_number): 1. Create/upsert Parent products (configurable) first 2. Create/upsert Children products (simples) second 3. Create product images (UPSERT_MEDIA) for all 4. Set relations (SET_RELATIONS) — links children to configurable parents 5. Set associations (SET_ASSOCIATIONS) — related / upsell / crosssell product links A3: When MEDIA_REQUIRE_PRODUCT_EXISTENCE, never plan UPSERT_MEDIA alone for SKUs absent in Magento; add UPSERT_PRODUCT first so media does not 404. """ from __future__ import annotations import logging import os from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any, Callable, Dict, List, Optional from magento.payload_mapper import snapshot_to_magento_product from magento.sync_hashes import compute_data_hash, compute_images_hash, compute_relations_hash, _parse_children_from_row from magento.association_sync import compute_associations_hash_from_row from magento.sync_overrides import OverrideSet, apply_overrides_to_payload from normalize import _relation_sku logger = logging.getLogger(__name__) ACTION_UPSERT_PRODUCT = "UPSERT_PRODUCT" ACTION_UPSERT_MEDIA = "UPSERT_MEDIA" ACTION_SET_RELATIONS = "SET_RELATIONS" ACTION_SET_ASSOCIATIONS = "SET_ASSOCIATIONS" ACTION_DISABLE_PRODUCT = "DISABLE_PRODUCT" @dataclass class PlannedAction: sku: str action: str payload: Optional[Dict[str, Any]] = None idempotency_key: Optional[str] = None pass_number: int = 0 @dataclass class SyncPlan: """Ordered list of actions. Pass 1=parents, 2=children, 3=media, 4=relations.""" actions: List[PlannedAction] = field(default_factory=list) def add(self, action: PlannedAction) -> None: self.actions.append(action) def _is_configurable(row: Dict[str, Any]) -> bool: pt = str(row.get("product_type", "")).strip().lower() return pt == "configurable" or "configurable" in pt def _has_images(row: Dict[str, Any]) -> bool: base = row.get("base_image") or row.get("base_image_url") addl = row.get("additional_images") or row.get("additional_images_url") if base and str(base).strip(): return True if addl and str(addl).strip(): return True return False def _is_configurable_parent(row: Dict[str, Any]) -> bool: """True if this row is a configurable parent (has children to link).""" if not _is_configurable(row): return False vl = row.get("variant_list") or "" cv = row.get("configurable_variations") or "" if vl and str(vl).strip(): return True if cv and str(cv).strip(): for seg in str(cv).split("|"): for pair in seg.split(","): if str(pair).strip().lower().startswith("sku="): return True return False def _is_child_row(row: Dict[str, Any]) -> bool: """True if this row only has variant_of (child of a configurable).""" return bool(_relation_sku(row.get("variant_of"))) def summarize_plan_actions(plan: SyncPlan) -> Dict[str, Any]: """Human-readable planner stats: actions are per-SKU passes (product, media, relations).""" by_action: Dict[str, int] = {} skus: set[str] = set() for action in plan.actions: by_action[action.action] = by_action.get(action.action, 0) + 1 if action.sku: skus.add(action.sku) return { "total_actions": len(plan.actions), "unique_skus": len(skus), "by_action": by_action, } def plan_incremental_sync( connection_id: int, snapshot_rows: List[Dict[str, Any]], state_by_sku: Dict[str, Dict[str, Any]], *, last_seen_at: Optional[datetime] = None, overrides: Optional[OverrideSet] = None, catalog_state_media_by_sku: Optional[Dict[str, str]] = None, catalog_state_by_sku: Optional[Dict[str, Dict[str, Any]]] = None, force_relations: bool = False, force_associations: bool = False, force_upsert_skus: Optional[set[str]] = None, default_attribute_set_id: Optional[int] = None, resolve_attribute_set_id: Optional[Callable[[Dict[str, Any]], int]] = None, images_only: bool = False, products_only: bool = False, force_images: bool = False, selected_attribute_codes: Optional[set[str]] = None, ) -> SyncPlan: """ Compare snapshot rows to state hashes. Output minimal actions. Rules: - No state: schedule UPSERT_PRODUCT, UPSERT_MEDIA (if has images), SET_RELATIONS (if parent) - Hash differs: schedule only that part - Pass 1: simples first, then configurables (product + media only) - Pass 2: relations for configurables (after children exist) - Overrides applied to payload before hashing; final payload stored in action. """ plan = SyncPlan() seen_at = last_seen_at or datetime.now(timezone.utc) ov = overrides or OverrideSet(force={}, lock={}) catalog_media = catalog_state_media_by_sku or {} catalog_state = catalog_state_by_sku or {} force_rel = force_relations force_assoc = force_associations selected_codes = {str(code).strip().lower() for code in (selected_attribute_codes or set()) if str(code).strip()} rows_by_sku: Dict[str, Dict[str, Any]] = {} for row in snapshot_rows: sku = str(row.get("sku", "")).strip() if sku: rows_by_sku[sku] = dict(row) # Build child->parent so we can inject variant_of for children that lack it (e.g. product_online=0) child_to_parent: Dict[str, str] = {} for rsku, rrow in list(rows_by_sku.items()): if _is_configurable(rrow): children, _, _ = _parse_children_from_row(rrow) for c in children: if c and c not in child_to_parent: child_to_parent[c] = rsku for csku, parent in child_to_parent.items(): if csku in rows_by_sku and not _relation_sku(rows_by_sku[csku].get("variant_of")): rows_by_sku[csku]["variant_of"] = parent simple_skus: List[str] = [] configurable_skus: List[str] = [] for sku, row in rows_by_sku.items(): if _is_configurable(row): configurable_skus.append(sku) else: simple_skus.append(sku) def _price_override(sku: str, row: Dict[str, Any], in_magento: bool, state: Optional[Dict[str, Any]]) -> Optional[float]: """$1 for new products or when Magento has price $0.""" cat = catalog_state.get(sku) effectively_in_magento = in_magento or (state is not None and state.get("data_hash")) if not effectively_in_magento: plytix_price = row.get("price") try: p = float(plytix_price) if plytix_price is not None else None except (TypeError, ValueError): p = None return max(p or 1, 0.01) if p is not None else 1 if cat and (cat.get("price") == 0 or cat.get("price") == 0.0): return 1.0 return None def _payload_for(sku: str, row: Dict[str, Any], is_new: bool, state: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: from magento.payload_mapper import get_axis_attribute_codes, _compute_parent_neutral_name cat = catalog_state.get(sku) in_magento = bool(cat and cat.get("magento_product_id")) price_ov = _price_override(sku, row, in_magento, state) is_configurable_parent = _is_configurable(row) and ( bool(row.get("variant_list")) or "sku=" in str(row.get("configurable_variations", "")).lower() ) axis_exclude = None parent_name_override = None if is_configurable_parent: axis_exclude = get_axis_attribute_codes(row) parent_name_override = _compute_parent_neutral_name(row, axis_exclude, rows_by_sku) if resolve_attribute_set_id is not None: attr_set_id = resolve_attribute_set_id(row) else: attr_set_id = default_attribute_set_id if default_attribute_set_id is not None else 4 base = snapshot_to_magento_product( row, is_new=is_new, include_stock=False, price_override=price_ov, axis_exclude=axis_exclude, parent_name_override=parent_name_override, rows_by_sku=rows_by_sku, attribute_set_id=attr_set_id, selected_attribute_codes=selected_codes or None, ) force = ov.force.get(sku, {}) locked = ov.lock.get(sku, set()) return apply_overrides_to_payload(base, force, locked) force_upsert_set = {str(s).strip().lower() for s in (force_upsert_skus or []) if s} PASS_PARENTS = 1 PASS_CHILDREN = 2 PASS_MEDIA = 3 PASS_RELATIONS = 4 PASS_ASSOCIATIONS = 5 def _plan_sku( sku: str, row: Dict[str, Any], product_pass: int, ) -> None: state = state_by_sku.get(sku) payload = _payload_for(sku, row, state is None, state) data_hash = compute_data_hash(row, payload, rows_by_sku=rows_by_sku) images_hash = compute_images_hash(row) needs_product = ( not images_only and ( (force_upsert_set and sku.strip().lower() in force_upsert_set) or state is None or state.get("data_hash") != data_hash ) ) magento_has_images = bool(catalog_media.get(sku)) state_has_images = state is not None and state.get("images_hash") is not None needs_media = ( not products_only and _has_images(row) and ( force_images or state is None or state.get("images_hash") != images_hash ) and not (magento_has_images and not state_has_images and not force_images) ) # A3: If media planned but product not, and MEDIA_REQUIRE_PRODUCT_EXISTENCE, # product must exist in Magento. If not, add UPSERT_PRODUCT first — unless images_only. if needs_media and not needs_product: media_require = os.getenv("MAGENTO_MEDIA_REQUIRE_PRODUCT_EXISTENCE", "true").strip().lower() in ("1", "true", "yes") if media_require: product_exists = bool(catalog_state.get(sku) and catalog_state.get(sku).get("magento_product_id")) if not product_exists and not images_only: needs_product = True if needs_product: plan.add(PlannedAction( sku=sku, action=ACTION_UPSERT_PRODUCT, payload=payload, idempotency_key=f"{sku}:UPSERT_PRODUCT:{data_hash}", pass_number=product_pass, )) if needs_media: plan.add(PlannedAction( sku=sku, action=ACTION_UPSERT_MEDIA, payload=row, idempotency_key=f"{sku}:UPSERT_MEDIA:{images_hash}", pass_number=PASS_MEDIA, )) # Pass 1: Create/upsert parent products (configurables) first for sku in configurable_skus: row = rows_by_sku[sku] _plan_sku(sku, row, product_pass=PASS_PARENTS) # Pass 2: Create/upsert children products (simples) second for sku in simple_skus: row = rows_by_sku[sku] _plan_sku(sku, row, product_pass=PASS_CHILDREN) # Pass 3: UPSERT_MEDIA is planned above for all (parents + children) # Pass 4: Set relations — link children to configurable parents if not images_only: for sku in configurable_skus: row = rows_by_sku[sku] if not _is_configurable_parent(row): continue state = state_by_sku.get(sku) relations_hash = compute_relations_hash(row) needs_relations = force_rel or state is None or state.get("relations_hash") != relations_hash if needs_relations: plan.add(PlannedAction( sku=sku, action=ACTION_SET_RELATIONS, payload=row, idempotency_key=f"{sku}:SET_RELATIONS:{relations_hash}", pass_number=PASS_RELATIONS, )) # Pass 5: related / upsell / crosssell product links (all product roles) for sku, row in rows_by_sku.items(): associations = row.get("association_fields") or {} has_links = any(associations.get(t) for t in ("related", "upsell", "crosssell")) if not has_links and not force_assoc: # Still clear stale links when hash changes to empty — only if state had associations. state = state_by_sku.get(sku) if not state or not state.get("associations_hash"): continue state = state_by_sku.get(sku) associations_hash = compute_associations_hash_from_row(row) needs_associations = ( force_assoc or state is None or state.get("associations_hash") != associations_hash ) if needs_associations: plan.add(PlannedAction( sku=sku, action=ACTION_SET_ASSOCIATIONS, payload=row, idempotency_key=f"{sku}:SET_ASSOCIATIONS:{associations_hash}", pass_number=PASS_ASSOCIATIONS, )) plan.actions.sort(key=lambda a: (a.pass_number, a.sku)) return plan def compute_planner_diagnostics( connection_id: int, snapshot_rows: List[Dict[str, Any]], state_by_sku: Dict[str, Dict[str, Any]], plan: SyncPlan, *, overrides: Optional[OverrideSet] = None, force_relations: bool = False, trace_skus: Optional[List[str]] = None, ) -> Dict[str, Any]: """Diagnostic info for debugging planned actions. Includes hash breakdown, axis mapping for trace_skus.""" trace_set = {s.strip().upper() for s in (trace_skus or []) if s} rows_by_sku = {str(r.get("sku", "")).strip(): r for r in snapshot_rows if str(r.get("sku", "")).strip()} planned_actions = [{"sku": a.sku, "action": a.action, "pass": a.pass_number} for a in plan.actions] planned_skus = {a.sku for a in plan.actions} skipped = [] trace_details: Dict[str, Dict[str, Any]] = {} for row in snapshot_rows: sku = str(row.get("sku", "")).strip() if not sku: continue state = state_by_sku.get(sku) if sku in planned_skus: if trace_set and sku.upper() in trace_set: payload = next((a.payload for a in plan.actions if a.sku == sku and a.payload), None) data_hash = compute_data_hash(row, payload or {}, rows_by_sku=rows_by_sku) axis_codes: List[str] = [] if _is_configurable(row): from magento.payload_mapper import get_axis_attribute_codes axis_codes = sorted(get_axis_attribute_codes(row)) children, mappings, child_to_mapping = _parse_children_from_row(row) child_axis: Dict[str, Dict[str, str]] = {} for i, c_sku in enumerate(children): m = (child_to_mapping.get(c_sku) if child_to_mapping else None) or ( mappings[i] if i < len(mappings) else None ) if m: child_axis[c_sku] = {k: v for k, v in m.items() if k != "sku"} custom_attrs = (payload or {}).get("custom_attributes") or [] trace_details[sku] = { "data_hash": data_hash, "data_hash_inputs": { "name": row.get("name"), "status": (payload or {}).get("status"), "visibility": (payload or {}).get("visibility"), "custom_attrs_count": len(custom_attrs), }, "axis_codes": axis_codes, "child_axis_mapping": child_axis, "custom_attributes_sample": [ {"code": a.get("attribute_code"), "value": str(a.get("value", ""))[:50]} for a in custom_attrs[:15] ], } else: reason = "all hashes match" if state else "unknown" if trace_set and sku.upper() in trace_set: skipped.append({"sku": sku, "reason": reason, "state": state}) elif len(skipped) < 20: skipped.append({"sku": sku, "reason": reason}) return { "planned_count": len(plan.actions), "planned_actions_sample": planned_actions[:50], "trace_skus_skipped": skipped[:20], "trace_skus_detail": trace_details, }