"""Magento product sync orchestration.""" from __future__ import annotations import base64 import hashlib import json import logging import os import re import time from datetime import datetime, timedelta, timezone from typing import Any, Callable, Dict, Iterable, List, Optional, Protocol, Sequence, Set import requests from magento.magento_api import MagentoRestClient from magento.payload_mapper import ( VISIBILITY_CATALOG_SEARCH, VISIBILITY_NOT_VISIBLE, _filter_magento_product_payload, snapshot_to_magento_product, ) from magento.sync_hashes import ( _parse_children_from_row, build_media_payload_from_row, build_relations_payload_from_row, compute_data_hash, compute_images_hash, compute_relations_hash, ) from magento.inventory_service import ( InventoryUpsertResult, ensure_simple_sku_inventory, resolve_source_code, upsert_source_items_for_simple, ) from magento.sync_overrides import OverrideSet, apply_overrides_to_payload from magento.sync_planner import ( ACTION_SET_RELATIONS, ACTION_SET_ASSOCIATIONS, ACTION_UPSERT_MEDIA, ACTION_UPSERT_PRODUCT, PlannedAction, compute_planner_diagnostics, plan_incremental_sync, summarize_plan_actions, ) from channel.attribute_value_resolver import ( OPTION_FALLBACK_VALUE_INDEX, OPTION_LABEL_ALIASES, norm_option_label, resolve_select_value_index, resolve_select_value_indices, ) from normalize import _relation_sku logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _extract_magento_error_message(body: Any) -> str: """Extract human-readable error from Magento API response body.""" if body is None: return "Unknown error" if isinstance(body, str): return body[:500] if isinstance(body, dict): msg = body.get("message") or body.get("error") or body.get("err") if isinstance(msg, str): return msg[:500] if msg is not None: return str(msg)[:500] return str(body)[:500] def _magento_error_text(body: Any) -> str: """Normalize Magento error body to a lowercase searchable string.""" if not body: return "" if isinstance(body, dict): return (str(body.get("message", "")) + str(body.get("parameters", []))).lower() return str(body).lower() def _is_url_key_exists_error(status: int, body: Any) -> bool: """True if response indicates 'URL key for specified store already exists'.""" if status != 400 or not body: return False err_text = _magento_error_text(body) return "url key" in err_text and ("already exists" in err_text or "exists" in err_text) def _is_sku_already_exists_error(status: int, body: Any) -> bool: """True if Magento reports the product/SKU already exists (not URL-key conflicts).""" if status not in (400, 422) or not body: return False err_text = _magento_error_text(body) if "url key" in err_text: return False return "already exists" in err_text and ("sku" in err_text or "product" in err_text) def _is_ambiguous_missing_product_put_error(status: int, body: Any) -> bool: """True when PUT 400 may mean 'missing' OR 'invalid data on existing product'.""" if status != 400 or not body: return False err_text = _magento_error_text(body) return ( "invalid attribute set entity type" in err_text or "invalid product data" in err_text ) def _custom_attr_value(payload: Dict[str, Any], code: str) -> Optional[Any]: for attr in payload.get("custom_attributes") or []: if isinstance(attr, dict) and str(attr.get("attribute_code") or "").strip() == code: return attr.get("value") return None def _summarize_product_payload(payload: Dict[str, Any]) -> Dict[str, Any]: """Compact view of what we POST/PUT to Magento (for console debugging).""" ext = payload.get("extension_attributes") or {} category_links = ext.get("category_links") if isinstance(ext, dict) else None category_ids: List[Any] = [] if isinstance(category_links, list): category_ids = [ link.get("category_id") for link in category_links if isinstance(link, dict) and link.get("category_id") is not None ] custom_codes = [ str(a.get("attribute_code") or "") for a in (payload.get("custom_attributes") or []) if isinstance(a, dict) and a.get("attribute_code") ] return { "sku": payload.get("sku"), "name": payload.get("name"), "type_id": payload.get("type_id"), "status": payload.get("status"), "visibility": payload.get("visibility"), "price": payload.get("price"), "url_key": _custom_attr_value(payload, "url_key"), "meta_title": _custom_attr_value(payload, "meta_title"), "category_ids": category_ids, "custom_attribute_codes": custom_codes[:40], "custom_attribute_count": len(custom_codes), } def _log_magento_product_post(method: str, sku: str, payload: Dict[str, Any]) -> None: """Print the outbound Magento product payload summary (and full JSON when enabled).""" summary = _summarize_product_payload(payload) logger.info("Magento %s product %s payload: %s", method, sku, json.dumps(summary, default=str)) if os.getenv("MAGENTO_LOG_PAYLOAD_FULL", "").strip().lower() in ("1", "true", "yes"): # Avoid dumping huge base64 media blobs if somehow present. safe = dict(payload) if isinstance(safe.get("media_gallery_entries"), list): safe["media_gallery_entries"] = f"<{len(safe['media_gallery_entries'])} entries omitted>" logger.info( "Magento %s product %s full payload:\n%s", method, sku, json.dumps(safe, indent=2, default=str)[:20000], ) def _magento_error_text(body: Any) -> str: if body is None: return "" if isinstance(body, dict): return " ".join( str(body.get(key) or "") for key in ("message", "parameters", "error", "body") ).lower() return str(body).lower() def _is_option_label_exists_error(status: int, body: Any) -> bool: """True if Magento reports the select attribute option label already exists.""" if status != 400 or not body: return False err_text = _magento_error_text(body) return "option label" in err_text and "already exists" in err_text def _parse_magento_option_rows(options: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """Parse Magento attribute options API list into bulk_upsert rows.""" rows: List[Dict[str, Any]] = [] for o in options or []: if not isinstance(o, dict): continue lb = o.get("label") val = o.get("value") if o.get("value") is not None else o.get("value_index") if lb is None or val is None: continue label = str(lb).strip() if not label: continue try: rows.append({"label": label, "value_index": int(val)}) except (ValueError, TypeError): continue return rows def _option_map_from_rows(rows: List[Dict[str, Any]]) -> Dict[str, int]: return { norm_option_label(str(r["label"])): int(r["value_index"]) for r in rows if r.get("label") is not None and r.get("value_index") is not None } def _image_content_type(url: str, data: bytes) -> tuple[Optional[str], Optional[str], Optional[str]]: """Return (extension, mime, error) after validating downloaded image bytes.""" if not data: return None, None, "empty image response" head = data[:64].lstrip().lower() if head.startswith(b"= 12 and data[:4] == b"RIFF" and data[8:12] == b"WEBP": return "webp", "image/webp", None return None, None, f"unsupported image content for {url}" def _ordered_unique_media_urls(row: Dict[str, Any]) -> List[str]: urls: List[str] = [] seen: set[str] = set() base = row.get("base_image") or row.get("base_image_url") addl_raw = row.get("additional_images") or row.get("additional_images_url") for candidate in [base, *str(addl_raw or "").split(",")]: url = str(candidate or "").strip() if not url or url in seen: continue seen.add(url) urls.append(url) return urls def _duplicate_axis_signatures( children: Sequence[str], mappings: Sequence[Dict[str, str]], attr_codes: Sequence[str], child_to_mapping: Optional[Dict[str, Dict[str, str]]] = None, ) -> List[Dict[str, Any]]: """Find children that would collide under Magento configurable axis values.""" axes = [str(code or "").strip().lower() for code in attr_codes if str(code or "").strip()] if not axes: return [] seen: Dict[tuple[str, ...], List[str]] = {} for idx, child_sku in enumerate(children): child = str(child_sku or "").strip() if not child: continue mapping = (child_to_mapping or {}).get(child) if mapping is None and idx < len(mappings): mapping = mappings[idx] mapping = {str(k or "").strip().lower(): str(v or "").strip() for k, v in dict(mapping or {}).items()} signature = tuple(mapping.get(axis, "") for axis in axes) seen.setdefault(signature, []).append(child) return [ {"axis_values": dict(zip(axes, signature)), "children": child_skus} for signature, child_skus in seen.items() if len(child_skus) > 1 ] def _title_variants_for_url_key_retry(sku: str, title: str) -> List[str]: """Title variants to force unique Magento url_key. Magento REST does not accept url_key; it auto-generates from name. Appending SKU and/or numeric suffixes produces unique slugs.""" sku_s = str(sku or "").strip() title_s = str(title or "").strip() or "product" seen: Set[str] = set() result: List[str] = [] # SKU-based variants (only if SKU not already in title) if sku_s and sku_s not in title_s: for candidate in [ f"{title_s} {sku_s}", # "Name SKU" f"{title_s} - {sku_s}", # "Name - SKU" f"{sku_s} {title_s}", # "SKU Name" f"{title_s} ({sku_s})", # "Name (SKU)" f"{title_s} {sku_s} _1", # "Name SKU _1" f"{title_s} {sku_s} _2", # "Name SKU _2" ]: if candidate.strip() and candidate.strip() not in seen: seen.add(candidate.strip()) result.append(candidate.strip()) # Numeric suffix fallbacks (always added for extra retry options) for suffix in ("_1", "_2", "_3"): candidate = f"{title_s} {suffix}" if candidate.strip() not in seen: seen.add(candidate.strip()) result.append(candidate.strip()) return result def _extract_needed_attribute_codes_from_rows( snapshot_rows: List[Dict[str, Any]] ) -> tuple[set[str], Dict[str, str]]: """Extract Magento channel attribute codes and optional labels from configurable rows.""" from magento.configurable_requirements import extract_configurable_requirements codes, code_to_label, _ = extract_configurable_requirements(snapshot_rows) return codes, code_to_label def _extract_all_needed_option_labels( snapshot_rows: List[Dict[str, Any]], attr_codes: set[str] ) -> Dict[str, set[str]]: """Extract {attr_code: set of option labels} from configurable_variations.""" from magento.configurable_requirements import extract_configurable_requirements _, _, option_labels = extract_configurable_requirements(snapshot_rows) return {code: set(option_labels.get(code, set())) for code in attr_codes} # Backward-compatible aliases for tests and callers. _norm_option_label = norm_option_label _resolve_option_value_index = resolve_select_value_index def _is_configurable_parent_static(row: Dict[str, Any]) -> bool: """Static check: is this a configurable parent with children.""" pt = str(row.get("product_type", "")).strip().lower() if pt != "configurable" and "configurable" not in pt: 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 _make_sync_item_create(it: Dict[str, Any]) -> Any: """Lazy import to avoid circular dependency on db.schemas.""" from db.schemas import MagentoSyncItemCreate magento_response = {} if it.get("msi_mode_detected") is not None: magento_response["msi_mode_detected"] = it["msi_mode_detected"] if it.get("source_items_written") is not None: magento_response["source_items_written"] = it["source_items_written"] if it.get("legacy_stock_written") is not None: magento_response["legacy_stock_written"] = it["legacy_stock_written"] if it.get("inventory_error"): magento_response["inventory_error"] = it["inventory_error"] return MagentoSyncItemCreate( job_id=it["job_id"], sku=it["sku"], action=it["action"], status=it["status"], error=it.get("error"), duration_ms=it.get("duration_ms"), http_status=it.get("http_status"), idempotency_key=it.get("idempotency_key"), remote_image_urls=it.get("remote_image_urls"), version_id=it.get("version_id"), mode=it.get("mode", "sync"), magento_response=magento_response if magento_response else None, ) # --------------------------------------------------------------------------- # Protocol interfaces # --------------------------------------------------------------------------- class SyncStateRepository(Protocol): def get_states_for_skus(self, connection_id: int, skus: List[str]) -> dict: ... def upsert_state( self, connection_id: int, sku: str, *, data_hash: Optional[str] = None, images_hash: Optional[str] = None, relations_hash: Optional[str] = None, last_seen_in_feed_at: Optional[datetime] = None, last_pushed_at: Optional[datetime] = None, last_error: Optional[str] = None, ) -> None: ... class MediaMapRepository(Protocol): def get_by_sku_url(self, connection_id: int, sku: str, remote_url: str) -> Optional[dict]: ... def get_by_sha256(self, connection_id: int, sku: str, content_sha256: str) -> Optional[dict]: ... def get_media_maps_for_sku(self, connection_id: int, sku: str) -> Dict[str, dict]: ... def upsert( self, connection_id: int, sku: str, remote_url: str, content_sha256: Optional[str] = None, magento_entry_id: Optional[int] = None, magento_file: Optional[str] = None, ) -> None: ... class SyncJobRepository(Protocol): def create_sync_job(self, payload: Any) -> int: ... def update_sync_job(self, job_id: int, **kwargs: object) -> None: ... def add_sync_items(self, items: Iterable[Any]) -> None: ... def get_sync_job(self, job_id: int) -> Optional[dict]: ... def get_pending_items(self, job_id: int) -> List[dict]: ... def update_sync_item(self, item_id: int, **kwargs: object) -> None: ... class SyncPendingRepository(Protocol): """Connection-scoped pending work. UPSERT on plan; update existing with new payload.""" def upsert_pending( self, connection_id: int, sku: str, action: str, *, payload: Optional[dict] = None, pass_number: int = 1, idempotency_key: Optional[str] = None, ) -> None: ... def get_pending_items(self, connection_id: int, *, skus: Optional[Iterable[str]] = None) -> List[dict]: ... def prune_pending_not_in_plan( self, connection_id: int, planned_keys: Iterable[tuple[str, str]], ) -> int: ... def update_pending_item( self, item_id: int, *, status: Optional[str] = None, error: Optional[str] = None, duration_ms: Optional[int] = None, http_status: Optional[int] = None, ) -> None: ... class OverrideRepository(Protocol): def get_overrides_for_skus(self, connection_id: int, skus: List[str]) -> OverrideSet: ... class NotificationRepository(Protocol): def enqueue_email( self, connection_id: int, job_id: int, to_email: str, subject: str, body: str ) -> int: ... class AttributeCacheRepository(Protocol): """Cache for Magento attribute_id and option label->value_index. TTL-controlled.""" def get_attributes( self, connection_id: int, codes: List[str], ttl_cutoff: datetime ) -> Dict[str, dict]: ... def upsert_attribute( self, connection_id: int, attribute_code: str, attribute_id: int, *, frontend_label: Optional[str] = None, frontend_input: Optional[str] = None, backend_type: Optional[str] = None, is_user_defined: Optional[bool] = None, fetched_at: datetime, ) -> None: ... def get_options( self, connection_id: int, attribute_code: str, ttl_cutoff: datetime ) -> Dict[str, int]: ... def bulk_upsert_options( self, connection_id: int, attribute_code: str, options: List[Dict[str, Any]], fetched_at: datetime, ) -> None: ... class CategoryCacheRepository(Protocol): """Cache for Magento category tree. Resolves paths/names to category_ids.""" def get_all(self, connection_id: int, ttl_cutoff: datetime) -> List[dict]: ... def resolve_path(self, connection_id: int, path_names: str, ttl_cutoff: datetime) -> Optional[int]: ... def resolve_name(self, connection_id: int, name: str, ttl_cutoff: datetime) -> Optional[int]: ... def bulk_replace(self, connection_id: int, categories: List[dict], fetched_at: datetime) -> int: ... class VersionRepository(Protocol): def create_version( self, connection_id: int, sku: str, *, job_id: Optional[int] = None, source: str = "sync", data_hash: Optional[str] = None, images_hash: Optional[str] = None, relations_hash: Optional[str] = None, product_payload_json: Optional[dict] = None, media_payload_json: Optional[dict] = None, relations_payload_json: Optional[dict] = None, ) -> int: ... def update_version_status(self, version_id: int, status: str) -> None: ... # --------------------------------------------------------------------------- # Service # --------------------------------------------------------------------------- class StoreRegistryRepository(Protocol): """Read default website_id from baseline store registry.""" def get_default_website_id(self, connection_id: int) -> Optional[int]: ... def get_all(self, connection_id: int) -> List[Dict[str, Any]]: ... class BaselineSyncRunRepository(Protocol): """Read default_attribute_set_id from last successful baseline run.""" def get_default_attribute_set_id(self, connection_id: int) -> Optional[int]: ... class AttributeSetAttributesRepository(Protocol): """Attributes assigned to an attribute set: set_id -> {attribute_code: attribute_id}.""" def get_attributes_for_set( self, connection_id: int, attribute_set_id: int ) -> Dict[str, int]: ... def attribute_in_set( self, connection_id: int, attribute_set_id: int, attribute_code: str ) -> bool: ... class AttributeSetRegistryRepository(Protocol): """Valid attribute sets from magento_attribute_set_registry. Resolve feed values against this.""" def get_by_name(self, connection_id: int, name: str) -> Optional[int]: ... def get_by_id(self, connection_id: int, attribute_set_id: int) -> Optional[int]: ... def get_first_by_id(self, connection_id: int) -> Optional[int]: ... class MagentoSyncService: """Orchestrates product sync: build payloads, upsert to Magento, persist audit.""" def __init__( self, api_client: MagentoRestClient, connection: Dict[str, Any], *, sync_state_repo: Optional[SyncStateRepository] = None, media_map_repo: Optional[MediaMapRepository] = None, sync_job_repo: Optional[SyncJobRepository] = None, override_repo: Optional[OverrideRepository] = None, version_repo: Optional[VersionRepository] = None, notification_repo: Optional[NotificationRepository] = None, attr_cache_repo: Optional[AttributeCacheRepository] = None, category_cache_repo: Optional[CategoryCacheRepository] = None, store_registry_repo: Optional[StoreRegistryRepository] = None, baseline_repo: Optional[BaselineSyncRunRepository] = None, attr_set_attrs_repo: Optional[AttributeSetAttributesRepository] = None, attr_set_registry_repo: Optional[AttributeSetRegistryRepository] = None, resolve_image_url: Optional[Callable[[str], str]] = None, source_codes_provider: Optional[Callable[[int], List[str]]] = None, sync_pending_repo: Optional[SyncPendingRepository] = None, ) -> None: from settings import load_magento_sync_runtime_config self._api = api_client self._connection = connection self._state_repo = sync_state_repo self._media_repo = media_map_repo self._job_repo = sync_job_repo self._override_repo = override_repo self._version_repo = version_repo self._notification_repo = notification_repo self._attr_cache_repo = attr_cache_repo self._cat_cache_repo = category_cache_repo self._store_repo = store_registry_repo self._baseline_repo = baseline_repo self._attr_set_attrs_repo = attr_set_attrs_repo self._attr_set_registry_repo = attr_set_registry_repo self._resolve_url = resolve_image_url or (lambda u: u) self._sync_pending_repo = sync_pending_repo self._source_codes_provider = source_codes_provider self._conn_id = int(connection.get("id", 0)) runtime_cfg = load_magento_sync_runtime_config() self._api_delay_s = max(0, runtime_cfg.api_delay_ms) / 1000.0 self._retry_max_attempts = max(1, runtime_cfg.api_retry_max_attempts) self._retry_base_s = max(50, runtime_cfg.api_retry_base_ms) / 1000.0 self._media_gallery_cache: Dict[str, List[Dict[str, Any]]] = {} self._default_website_id: Optional[int] = None self._all_products_category_id: Optional[int] = None self._default_attribute_set_id: Optional[int] = None self._default_source_code: Optional[str] = None # MSI source (e.g. homesurplus) self._category_create_failed_cache: Set[str] = set() # path_names we tried to create and failed (per run) self._category_missing_warned: Set[str] = set() # log each missing path once per run def _resolve_attribute_set_id(self, row: Dict[str, Any]) -> int: """ Resolve attribute_set_id for a product row. Uses magento_attribute_set_registry: - If feed has attribute_set_id/code that exists in registry → use it - Otherwise → fall back to Default from registry - MAGENTO_DEFAULT_ATTRIBUTE_SET_ID env overrides Default (use when registry has wrong entity type) """ conn_id = self._conn_id reg = self._attr_set_registry_repo env_override = os.getenv("MAGENTO_DEFAULT_ATTRIBUTE_SET_ID", "").strip() if env_override and env_override.isdigit(): return int(env_override) if not reg: return self._default_attribute_set_id or 4 def _fallback() -> int: default_id = reg.get_by_name(conn_id, "Default") if default_id is not None: return default_id first_id = reg.get_first_by_id(conn_id) if first_id is not None: return first_id return self._default_attribute_set_id or 4 # Try attribute_set_id from feed - must exist in registry raw_id = row.get("attribute_set_id") if raw_id is not None and str(raw_id).strip(): try: as_id = int(raw_id) if reg.get_by_id(conn_id, as_id) is not None: return as_id except (TypeError, ValueError): pass # Try attribute_set_code from feed - must exist in registry raw_code = row.get("attribute_set_code") if raw_code is not None and str(raw_code).strip(): by_name = reg.get_by_name(conn_id, str(raw_code).strip()) if by_name is not None: return by_name return _fallback() # ------------------------------------------------------------------ # API call helpers # ------------------------------------------------------------------ def _sleep_api_delay(self) -> None: if self._api_delay_s > 0: time.sleep(self._api_delay_s) @staticmethod def _is_retryable_status(status: Optional[int]) -> bool: # 5xx / connection-down are already exhausted inside oauth_client's # multi-round recovery (5m/10m/30m). Only rate-limit is retriable here. return status == 429 def _call_with_retry(self, op_name: str, fn: Callable[[], Any]) -> Any: last_err: Optional[Exception] = None last_result: Any = None for attempt in range(1, self._retry_max_attempts + 1): self._sleep_api_delay() try: result = fn() except Exception as exc: last_err = exc if attempt < self._retry_max_attempts: sleep_s = self._retry_base_s * (2 ** (attempt - 1)) logger.warning("%s exception (attempt %d/%d), retrying in %.2fs: %s", op_name, attempt, self._retry_max_attempts, sleep_s, exc) time.sleep(sleep_s) continue raise last_result = result status = result[0] if isinstance(result, tuple) and result and isinstance(result[0], int) else None if self._is_retryable_status(status) and attempt < self._retry_max_attempts: sleep_s = self._retry_base_s * (2 ** (attempt - 1)) logger.warning("%s retryable status=%s (attempt %d/%d), retrying in %.2fs", op_name, status, attempt, self._retry_max_attempts, sleep_s) time.sleep(sleep_s) continue return result if last_err: raise last_err return last_result # ------------------------------------------------------------------ # Category cache # ------------------------------------------------------------------ def _refresh_category_cache(self, connection_id: int, *, ttl_days: int = 7) -> Optional[int]: """Fetch full Magento category tree and populate cache.""" cat_repo = self._cat_cache_repo if not cat_repo: return None now = datetime.now(timezone.utc) ttl_cutoff = now - timedelta(days=ttl_days) existing = cat_repo.get_all(connection_id, ttl_cutoff) if existing: logger.info("Category cache: %d categories fresh in cache for connection %s", len(existing), connection_id) return 0 logger.info("Category cache: refreshing from Magento for connection %s", connection_id) try: status, tree = self._call_with_retry( "get_categories_tree", lambda: self._api.get_categories_tree(), ) if status != 200 or not tree: logger.error("Category cache: failed to fetch tree from Magento (HTTP %s)", status) return None except Exception as e: logger.warning("Category cache: exception fetching tree: %s", e) return None flat: List[dict] = [] self._flatten_category_tree(tree, flat, parent_path_names="") if flat: count = cat_repo.bulk_replace(connection_id, flat, now) logger.info("Category cache: cached %d categories for connection %s", count, connection_id) return count return 0 def _flatten_category_tree( self, node: Dict[str, Any], out: List[dict], parent_path_names: str ) -> None: """Recursively flatten Magento category tree node.""" name = str(node.get("name", "")).strip() cat_id = node.get("id") if not cat_id: return path = str(node.get("path", "")) path_names = f"{parent_path_names}/{name}" if parent_path_names else name out.append({ "magento_category_id": int(cat_id), "parent_id": node.get("parent_id"), "name": name, "path": path, "path_names": path_names, "level": node.get("level", 0), "is_active": node.get("is_active", True), }) for child in node.get("children_data", []): if isinstance(child, dict): self._flatten_category_tree(child, out, path_names) INVALID_CATEGORY_VALUES = frozenset({"nan", "none", "null", "nat"}) DEFAULT_ROOT_CATEGORY_ID = 2 # Magento Default Category; fallback when registry empty def _registry_rows_for_match(self, connection_id: int, ttl_cutoff: datetime) -> List[dict]: cat_repo = self._cat_cache_repo if not cat_repo: return [] try: return list(cat_repo.get_all(connection_id, ttl_cutoff) or []) except Exception: return [] def _adopt_existing_category( self, connection_id: int, *, cat_id: int, parent_id: int, name: str, path_names: str, level: int, magento_path: str = "", is_active: bool = True, ) -> int: """Persist a Magento category into the local registry under the feed path.""" cat_repo = self._cat_cache_repo assert cat_repo is not None and hasattr(cat_repo, "upsert_category") now = datetime.now(timezone.utc) cat_repo.upsert_category(connection_id, { "category_id": int(cat_id), "parent_id": parent_id, "name": name, "path": magento_path, "path_names": path_names, "level": level, "is_active": is_active, }, now) return int(cat_id) def _find_existing_category_under_parent( self, *, name: str, parent_id: int ) -> Optional[Dict[str, Any]]: """Recover a category Magento already has (url_key conflict / prior create).""" from magento.category_paths import humanize_slug_segment, segment_url_key display_name = humanize_slug_segment(name) or name url_key = segment_url_key(name) search_attempts = [ {"url_key": url_key, "parent_id": parent_id}, {"name": display_name, "parent_id": parent_id}, {"name": name, "parent_id": parent_id}, {"url_key": url_key}, ] for kwargs in search_attempts: try: status, items = self._api.search_categories(**kwargs) except TypeError: # Older API stubs may not accept url_key. if "url_key" in kwargs: continue raise if status != 200 or not items: continue for item in items: if not isinstance(item, dict): continue item_parent = item.get("parent_id") if item_parent is not None and int(item_parent) != int(parent_id): continue return item return None def _ensure_category_path_exists( self, connection_id: int, path_names: str, ttl_cutoff: datetime ) -> Optional[int]: """Create missing category path via Magento API. Returns leaf category_id or None on failure.""" from magento.category_paths import humanize_slug_path, humanize_slug_segment, match_category_id cat_repo = self._cat_cache_repo if not cat_repo or not hasattr(cat_repo, "upsert_category"): return None path = str(path_names).strip() if not path: return None segments = [s.strip() for s in path.split("/") if s.strip()] if not segments: return None registry_rows = self._registry_rows_for_match(connection_id, ttl_cutoff) matched = match_category_id(registry_rows, path) if matched is not None: return matched parent_id: Optional[int] = None built_path = "" built_human_path = "" for i, name in enumerate(segments): built_path = f"{built_path}/{name}" if built_path else name display_name = humanize_slug_segment(name) or name built_human_path = ( f"{built_human_path}/{display_name}" if built_human_path else display_name ) cid = cat_repo.resolve_path(connection_id, built_path, ttl_cutoff) if cid is None: cid = cat_repo.resolve_path(connection_id, built_human_path, ttl_cutoff) if cid is None: cid = match_category_id(registry_rows, built_path) if cid is not None: parent_id = cid continue create_parent = parent_id if parent_id is not None else self.DEFAULT_ROOT_CATEGORY_ID if built_path in self._category_create_failed_cache: return None payload: Dict[str, Any] = { "parent_id": create_parent, "name": display_name, "is_active": True, "include_in_menu": True, } status, body = self._api.create_category(payload) if status not in (200, 201) or not isinstance(body, dict) or not body.get("id"): err_detail = body.get("error", "") if isinstance(body, dict) else "" err_lower = (err_detail or "").lower() existing = None if status == 400 and "url key" in err_lower and ("already exists" in err_lower or "exists" in err_lower): existing = self._find_existing_category_under_parent( name=name, parent_id=create_parent ) if existing is not None: cat_id = existing.get("id") or existing.get("entity_id") if cat_id is not None: existing_name = str(existing.get("name") or display_name) human_path = humanize_slug_path(built_path) or built_human_path parent_id = self._adopt_existing_category( connection_id, cat_id=int(cat_id), parent_id=create_parent, name=existing_name, path_names=human_path, level=i + 1, magento_path=str(existing.get("path", "")), is_active=bool(existing.get("is_active", True)), ) registry_rows.append({ "category_id": parent_id, "parent_id": create_parent, "name": existing_name, "path_names": human_path, }) logger.info( "Category '%s' already exists in Magento (id=%s); added to registry for connection %s", human_path, cat_id, connection_id, ) continue self._category_create_failed_cache.add(built_path) logger.warning( "Failed to create category '%s' (parent=%s): HTTP %s%s", display_name, create_parent, status, f" — {err_detail[:200]}" if err_detail else "", ) return None created_id = body.get("id") created_path = str(body.get("path", "")) human_path = humanize_slug_path(built_path) or built_human_path parent_id = self._adopt_existing_category( connection_id, cat_id=int(created_id), parent_id=create_parent, name=display_name, path_names=human_path, level=i + 1, magento_path=created_path, is_active=True, ) registry_rows.append({ "category_id": parent_id, "parent_id": create_parent, "name": display_name, "path_names": human_path, }) logger.info("Created category '%s' (id=%s) for connection %s", human_path, created_id, connection_id) return parent_id def _resolve_category_ids( self, connection_id: int, categories_raw: str, *, ttl_days: int = 7, master_catalog_mode: bool = False ) -> List[int]: """Resolve feed category strings to Magento category IDs using cache. If not found, auto-creates the category path via Magento API (unless master_catalog_mode). Skips invalid parts (nan, none, etc.). Empty/invalid → returns [] (caller uses default). """ from magento.category_paths import match_category_id cat_repo = self._cat_cache_repo if not cat_repo: return [] s = str(categories_raw).strip() if categories_raw is not None else "" if not s or s.lower() in self.INVALID_CATEGORY_VALUES: return [] ttl_cutoff = datetime.now(timezone.utc) - timedelta(days=ttl_days) registry_rows = self._registry_rows_for_match(connection_id, ttl_cutoff) cat_ids: List[int] = [] parts = [p.strip() for p in s.split(",") if p.strip()] for part in parts: if part.lower() in self.INVALID_CATEGORY_VALUES: continue cid = cat_repo.resolve_path(connection_id, part, ttl_cutoff) if cid is None: cid = match_category_id(registry_rows, part) if cid is None and not master_catalog_mode: leaf = part.split("/")[-1].strip() if "/" in part else part cid = cat_repo.resolve_name(connection_id, leaf, ttl_cutoff) if cid is None: from magento.category_paths import humanize_slug_segment human_leaf = humanize_slug_segment(leaf) if human_leaf and human_leaf != leaf: cid = cat_repo.resolve_name(connection_id, human_leaf, ttl_cutoff) if cid is None: try: from settings import load_magento_category_cache_config if not master_catalog_mode and load_magento_category_cache_config().auto_create: cid = self._ensure_category_path_exists(connection_id, part, ttl_cutoff) if cid is not None: registry_rows = self._registry_rows_for_match(connection_id, ttl_cutoff) except Exception as exc: logger.warning( "Category auto-create raised for '%s' (connection %s): %s", part, connection_id, exc, ) if cid is not None: cat_ids.append(cid) elif part not in self._category_missing_warned: self._category_missing_warned.add(part) logger.warning( "Category '%s' not found and could not create for connection %s", part, connection_id, ) return cat_ids # ------------------------------------------------------------------ # Attribute cache # ------------------------------------------------------------------ def _refresh_attribute_cache( self, connection_id: int, snapshot_rows: List[Dict[str, Any]], *, ttl_days: int = 7, ) -> Optional[int]: """ Pre-sync: extract needed attribute codes from rows, refresh cache if stale/missing. Returns count of attributes refreshed, or None if skipped. """ attr_repo = self._attr_cache_repo if not attr_repo: return None needed_codes, code_to_label = _extract_needed_attribute_codes_from_rows(snapshot_rows) if not needed_codes: configurable_count = sum(1 for r in snapshot_rows if _is_configurable_parent_static(r)) logger.info( "Attribute cache: no attrs to refresh (needed_codes empty). " "configurable_parents_in_batch=%d total_rows=%d", configurable_count, len(snapshot_rows), ) return None now = datetime.now(timezone.utc) ttl_cutoff = now - timedelta(days=ttl_days) cached = attr_repo.get_attributes(connection_id, list(needed_codes), ttl_cutoff) to_refresh = [c for c in needed_codes if c not in cached] if not to_refresh: logger.info("Attribute cache: all %d attributes fresh in cache for connection %s", len(needed_codes), connection_id) return 0 logger.info("Attribute cache: refreshing %d attributes for connection %s (codes=%s)", len(to_refresh), connection_id, to_refresh) refreshed_count = 0 for attr_code in to_refresh: try: status, detail = self._call_with_retry( f"get_attribute_detail:{attr_code}", lambda ac=attr_code: self._api.get_attribute_detail(ac), ) if status == 404 and attr_code in code_to_label: label = code_to_label[attr_code] status, detail = self._call_with_retry( f"find_attribute_by_frontend_label:{label}", lambda lb=label: self._api.find_attribute_by_frontend_label(lb), ) if status == 200 and detail: attr_code = str(detail.get("attribute_code", attr_code)).strip().lower() if status != 200 or not detail: logger.error( "Attribute %s not found in Magento (HTTP %s, label=%s). " "Create attribute in Magento or fix feed.", attr_code, status, code_to_label.get(attr_code), ) continue attr_id = detail.get("attribute_id") if attr_id is None: continue attr_repo.upsert_attribute( connection_id, attr_code, int(attr_id), frontend_label=detail.get("frontend_label"), frontend_input=detail.get("frontend_input"), backend_type=detail.get("backend_type"), is_user_defined=detail.get("is_user_defined"), fetched_at=now, ) refreshed_count += 1 opts_status, opts = self._call_with_retry( f"get_attribute_options:{attr_code}", lambda ac=attr_code: self._api.get_attribute_options(ac), ) if opts_status == 200 and opts: option_rows = [] for o in opts: if isinstance(o, dict): lb = o.get("label") val = o.get("value") or o.get("value_index") if lb is not None and val is not None: try: vi = int(val) except (ValueError, TypeError): continue option_rows.append({"label": str(lb).strip(), "value_index": vi}) if option_rows: attr_repo.bulk_upsert_options(connection_id, attr_code, option_rows, now) except Exception as e: logger.warning("Attribute cache refresh for %s failed: %s", attr_code, e) return refreshed_count def _refresh_attribute_options_cache( self, connection_id: int, attr_code: str, now: datetime ) -> Dict[str, int]: """Fetch attribute options from Magento and upsert into option cache.""" attr_repo = self._attr_cache_repo if not attr_repo: return {} status, olist = self._call_with_retry( f"get_attribute_options:{attr_code}", lambda ac=attr_code: self._api.get_attribute_options(ac), ) if status != 200 or not olist: return {} option_rows = _parse_magento_option_rows(olist) if option_rows: attr_repo.bulk_upsert_options(connection_id, attr_code, option_rows, now) return _option_map_from_rows(option_rows) def ensure_attributes_and_options_exist( self, connection_id: int, snapshot_rows: List[Dict[str, Any]], *, ttl_days: int = 7, provision: bool = False, ) -> Dict[str, Any]: """ Pre-sync: ensure all needed attributes and options exist in Magento. When auto_create_attributes/auto_create_options enabled (or provision=True), creates missing ones. Returns {created_attributes: int, created_options: int, errors: [...]}. """ try: from settings import load_magento_sync_auto_create_config auto_cfg = load_magento_sync_auto_create_config() except Exception: auto_cfg = None reserved = set() if auto_cfg: reserved = set(auto_cfg.reserved_attribute_codes) needed_codes, code_to_label = _extract_needed_attribute_codes_from_rows(snapshot_rows) needed_codes = {c for c in needed_codes if c not in reserved} attr_repo = self._attr_cache_repo if not attr_repo: return {"created_attributes": 0, "created_options": 0, "errors": []} now = datetime.now(timezone.utc) ttl_cutoff = now - timedelta(days=ttl_days) cached = attr_repo.get_attributes(connection_id, list(needed_codes), ttl_cutoff) to_create_attrs = [c for c in needed_codes if c not in cached] created_attrs = 0 errors: List[str] = [] create_attributes = bool(provision or (auto_cfg and auto_cfg.auto_create_attributes)) create_options = bool(provision or (auto_cfg and auto_cfg.auto_create_options)) # Create missing attributes default_set_id = self._default_attribute_set_id or 4 for attr_code in to_create_attrs: if not create_attributes: errors.append(f"Attribute '{attr_code}' not in Magento; auto_create disabled") continue if not re.match(r"^[a-z0-9_]{1,64}$", attr_code): errors.append(f"Invalid attribute code '{attr_code}' (lowercase, underscore, max 64)") continue frontend_label = code_to_label.get(attr_code) or attr_code.replace("_", " ").title() try: existing_status, existing_body = self._api.get_attribute(attr_code) if existing_status == 200 and existing_body: aid = existing_body.get("attribute_id") if aid and attr_repo: attr_repo.upsert_attribute( connection_id, attr_code, int(aid), frontend_label=str( existing_body.get("default_frontend_label") or existing_body.get("frontend_label") or frontend_label ), frontend_input=str(existing_body.get("frontend_input") or "select"), fetched_at=now, ) add_status, add_err = self._api.add_attribute_to_attribute_set( default_set_id, attr_code ) if add_status not in (200, 201) and add_err: logger.warning( "Attribute %s exists but add to set failed: %s", attr_code, add_err ) continue status, body = self._call_with_retry( f"create_attribute:{attr_code}", lambda ac=attr_code, fl=frontend_label: self._api.create_attribute( ac, frontend_label=fl, frontend_input="select", is_configurable=True ), ) if status in (200, 201) and body: aid = body.get("attribute_id") if aid: attr_repo.upsert_attribute( connection_id, attr_code, int(aid), frontend_label=frontend_label, frontend_input="select", fetched_at=now, ) created_attrs += 1 add_status, add_err = self._api.add_attribute_to_attribute_set( default_set_id, attr_code ) if add_status not in (200, 201) and add_err: logger.warning("Attribute %s created but add to set failed: %s", attr_code, add_err) else: errors.append(f"Create attribute {attr_code}: HTTP {status}") except Exception as e: errors.append(f"Create attribute {attr_code}: {e}") # Create missing options option_labels = _extract_all_needed_option_labels(snapshot_rows, needed_codes) created_opts = 0 for attr_code, labels in option_labels.items(): if not labels: continue opts = attr_repo.get_options(connection_id, attr_code, ttl_cutoff) needed_norms = { _norm_option_label(label) for label in labels if _norm_option_label(label) } if needed_norms - set(opts.keys()): opts = self._refresh_attribute_options_cache(connection_id, attr_code, now) or opts for label in labels: norm = _norm_option_label(label) if norm in opts: continue if not create_options: errors.append(f"Option '{label}' for {attr_code} not in Magento; auto_create disabled") continue try: result = self._call_with_retry( f"create_option:{attr_code}:{label}", lambda ac=attr_code, lb=label: self._api.create_attribute_option(ac, lb), ) status = result[0] if isinstance(result, tuple) and result else 0 body = result[1] if isinstance(result, tuple) and len(result) > 1 else None err = result[2] if isinstance(result, tuple) and len(result) > 2 else None if status in (200, 201): created_opts += 1 opts = self._refresh_attribute_options_cache(connection_id, attr_code, now) or opts elif _is_option_label_exists_error(status, err or body): opts = self._refresh_attribute_options_cache(connection_id, attr_code, now) or opts if norm in opts: continue # Remote already has this label; stale cache or duplicate POST is OK. continue else: errors.append(f"Create option '{label}' for {attr_code}: {err or status}") except Exception as e: errors.append(f"Create option '{label}' for {attr_code}: {e}") return {"created_attributes": created_attrs, "created_options": created_opts, "errors": errors} # ------------------------------------------------------------------ # Product upsert # ------------------------------------------------------------------ SELECT_MULTISELECT = ("select", "multiselect") BOOLEAN_INPUTS = ("boolean",) @staticmethod def _coerce_magento_boolean_value(value: Any) -> Optional[int]: if value is None: return None if isinstance(value, bool): return 1 if value else 0 if isinstance(value, (int, float)): return 0 if float(value) == 0 else 1 text = str(value).strip() if not text: return None norm = text.lower() if norm in {"1", "true", "yes", "y", "on"}: return 1 if norm in {"0", "false", "no", "n", "off"}: return 0 return 1 if norm else None def _transform_custom_attributes_for_magento( self, payload: Dict[str, Any], connection_id: int ) -> Dict[str, Any]: """ Transform master-catalog attribute values to Magento format. - Only include attributes that exist in Magento (attribute set registry). - For select/multiselect: resolve catalog labels to value_index via option registry. - No hardcoded values; all from catalog, mapped to Magento attributes/options. """ custom = payload.get("custom_attributes") or [] if not custom: return payload attr_repo = self._attr_cache_repo set_attrs_repo = self._attr_set_attrs_repo default_set_id = self._default_attribute_set_id if not attr_repo: return payload allowed_codes: set = set() if set_attrs_repo and default_set_id: allowed_codes = set( set_attrs_repo.get_attributes_for_set(connection_id, default_set_id).keys() ) if not allowed_codes: codes_to_check = [x.get("attribute_code", "") for x in custom if x.get("attribute_code")] reg = attr_repo.get_attributes( connection_id, codes_to_check, datetime.now(timezone.utc) - timedelta(days=30), ) or {} allowed_codes = set(reg.keys()) ttl = datetime.now(timezone.utc) - timedelta(days=7) attr_meta = attr_repo.get_attributes( connection_id, list(allowed_codes), ttl, ) if attr_repo else {} # Never pass attribute_set_code/attribute_set_id via custom_attributes (use payload-level Default). # status/visibility are Magento product top-level ints (1/2 and 1–4); master "active" must not # be resolved as a select option label in custom_attributes. ATTR_SET_DENY = {"attribute_set_code", "attribute_set_id", "status", "visibility", "sku", "name", "type_id"} out: List[Dict[str, Any]] = [] for attr in custom: code = (attr.get("attribute_code") or "").strip().lower() if not code or code in ATTR_SET_DENY or code not in allowed_codes: continue val = attr.get("value") if val is None: continue meta = attr_meta.get(code, {}) frontend_input = (meta.get("frontend_input") or "").strip().lower() if frontend_input in self.SELECT_MULTISELECT: opts = attr_repo.get_options(connection_id, code, ttl) if not opts: logger.debug("No options in registry for %s; passing value as-is", code) out.append({"attribute_code": code, "value": val}) continue value_indices = set(opts.values()) resolved_indices = resolve_select_value_indices(val, opts, code) resolved = [str(vi) for vi in resolved_indices] if not resolved: raw = val if isinstance(val, list) else [v.strip() for v in str(val).split(",") if v.strip()] for part in raw: if str(part).strip().lower() in ("nan", "none", "null"): continue try: as_int = int(str(part).strip()) if as_int in value_indices: resolved.append(str(as_int)) else: logger.warning( "Attribute %s: catalog value '%s' not in Magento option registry; skipping", code, part, ) except (ValueError, TypeError): logger.warning( "Attribute %s: catalog value '%s' not in Magento option registry; skipping", code, part, ) if resolved: out.append({ "attribute_code": code, "value": ",".join(resolved) if frontend_input == "multiselect" else resolved[0], }) elif frontend_input in self.BOOLEAN_INPUTS: coerced = self._coerce_magento_boolean_value(val) if coerced is not None: out.append({"attribute_code": code, "value": coerced}) else: out.append({"attribute_code": code, "value": val}) payload = dict(payload) payload["custom_attributes"] = out if out else None if not payload["custom_attributes"]: del payload["custom_attributes"] return payload def _resolve_image_urls_in_payload( self, payload: Dict[str, Any], sku: str, connection_id: int ) -> Dict[str, Any]: """ Replace Plytix image URLs in custom_attributes with Magento file paths. base_image, small_image, thumbnail_image, additional_images must use Magento paths (e.g. /i/m/image_0_166.jpg) not external URLs. """ if not self._media_repo: return payload media_maps = self._media_repo.get_media_maps_for_sku(connection_id, sku) if not media_maps: return payload image_attrs = {"base_image", "small_image", "thumbnail_image", "additional_images"} custom = payload.get("custom_attributes") or [] out: List[Dict[str, Any]] = [] for attr in custom: code = attr.get("attribute_code") if code not in image_attrs: out.append(attr) continue val = attr.get("value") if not val or not isinstance(val, str): out.append(attr) continue paths: List[str] = [] for url in (u.strip() for u in str(val).split(",") if u.strip()): m = media_maps.get(url) if m and m.get("magento_file"): paths.append(m["magento_file"]) if paths: out.append({"attribute_code": code, "value": ",".join(paths)}) payload = dict(payload) payload["custom_attributes"] = out if out else None if not payload["custom_attributes"]: del payload["custom_attributes"] return payload def _inject_store_and_website_ids( self, payload: Dict[str, Any], connection_id: int ) -> Dict[str, Any]: """ Replace 0 or invalid website_id/store_ids in custom_attributes with baseline values. The feed often has website_id=0; we inject the default website and store IDs from magento_store_registry (populated by baseline sync). """ custom = payload.get("custom_attributes") or [] if not custom: return payload web_id = self._default_website_id store_ids: List[int] = [] if self._store_repo: store_ids = self._store_repo.get_store_ids_for_default_website( connection_id, web_id ) def _is_invalid(val: object) -> bool: if val is None: return True if isinstance(val, list): return not val or all(x == 0 for x in val if isinstance(x, (int, float))) if isinstance(val, (int, float)): return val == 0 s = str(val).strip() return not s or s == "0" or s.lower() in ("nan", "none", "null") out: List[Dict[str, Any]] = [] for attr in custom: code = attr.get("attribute_code") val = attr.get("value") if code == "website_id" and _is_invalid(val) and web_id is not None: out.append({"attribute_code": "website_id", "value": str(web_id)}) continue if code in ("store_ids", "store_id") and _is_invalid(val) and store_ids: out.append({"attribute_code": code, "value": store_ids}) continue out.append(attr) payload = dict(payload) payload["custom_attributes"] = out return payload def _ensure_inventory_for_simple( self, sku: str, row: Dict[str, Any] ) -> InventoryUpsertResult: """ Upsert MSI source items for SIMPLE products only. Required so Admin "Current Variations" shows all children. Uses canonical ensure_simple_sku_inventory when source_codes_provider is available; otherwise falls back to legacy upsert_source_items_for_simple. """ pt = str(row.get("product_type", "")).strip().lower() if "configurable" in pt: return InventoryUpsertResult(msi_mode_detected=False) try: from settings import load_magento_msi_config msi_cfg = load_magento_msi_config() except Exception: msi_cfg = None source_codes: List[str] = [] if self._source_codes_provider: source_codes = self._source_codes_provider(self._conn_id) if not source_codes: # Fallback: single source from baseline config try: from settings import load_magento_baseline_config bl_cfg = load_magento_baseline_config() configured = getattr(bl_cfg, "default_source_code", None) except Exception: configured = None fallback_source = ( [configured] if configured and str(configured).strip() else [resolve_source_code(self._api, self._conn_id, configured)] ) source_codes = fallback_source if fallback_source else ["default"] if msi_cfg: result = ensure_simple_sku_inventory( self._api, self._conn_id, sku, source_codes, msi_cfg ) return InventoryUpsertResult( msi_mode_detected=result.success or bool(result.before_source_items or result.after_source_items), source_items_written=[ {"source_code": si.get("source_code"), "quantity": si.get("quantity"), "status": si.get("status")} for si in (result.after_source_items or []) ], error=result.error, mode=result.mode, before_source_items=result.before_source_items, after_source_items=result.after_source_items, sources_added=result.sources_added, sources_updated=result.sources_updated, ) return upsert_source_items_for_simple( self._api, sku, row, connection_id=self._conn_id, configured_source=self._default_source_code, ) def _build_put_payload_for_upsert( self, sku: str, payload: Dict[str, Any], row: Dict[str, Any], *, overrides: Optional[OverrideSet] = None, selected_attribute_codes: Optional[Set[str]] = None, ) -> Dict[str, Any]: """Build the Magento PUT product payload (status/visibility/overrides applied).""" selected_codes = { str(code).strip().lower() for code in (selected_attribute_codes or set()) if str(code).strip() } partial_attr_update = bool(selected_codes) put_payload = dict(payload) put_payload = self._transform_custom_attributes_for_magento(put_payload, self._conn_id) put_payload = self._resolve_image_urls_in_payload(put_payload, sku, self._conn_id) if not partial_attr_update: put_payload = self._inject_store_and_website_ids(put_payload, self._conn_id) # Never send stock_item (stock_id=1) - breaks MSI; use source-items instead if put_payload.get("extension_attributes"): ext = dict(put_payload["extension_attributes"]) ext.pop("stock_item", None) put_payload["extension_attributes"] = ext # A1: Always enabled unless override explicitly disables or locks status _status_locked = overrides is not None and "status" in overrides.lock.get(sku, set()) if not partial_attr_update and not _status_locked: _status_ov = overrides.force.get(sku, {}).get("status") if overrides else None _explicit_disable = ( _status_ov is not None and str(_status_ov).strip() in ("0", "false", "disabled") ) put_payload["status"] = 0 if _explicit_disable else 1 if overrides: _put_force = overrides.force.get(sku, {}) _put_locked = overrides.lock.get(sku, set()) put_payload = apply_overrides_to_payload(put_payload, _put_force, _put_locked) if "status" in put_payload: put_payload["status"] = int(put_payload["status"]) _visibility_locked = overrides is not None and "visibility" in overrides.lock.get(sku, set()) if not partial_attr_update and not _visibility_locked: variant_of = _relation_sku(row.get("variant_of")) type_id = str(put_payload.get("type_id") or row.get("product_type") or "").strip().lower() if variant_of: put_payload["visibility"] = VISIBILITY_NOT_VISIBLE elif "configurable" in type_id: put_payload["visibility"] = VISIBILITY_CATALOG_SEARCH if "visibility" in put_payload: put_payload["visibility"] = int(put_payload["visibility"]) return put_payload def _build_create_payload_for_upsert( self, sku: str, payload: Dict[str, Any], row: Dict[str, Any], *, overrides: Optional[OverrideSet] = None, rows_by_sku: Optional[Dict[str, Dict[str, Any]]] = None, ) -> Dict[str, Any]: """Build the Magento POST (create) product payload used on PUT 404 fallback.""" price_ov = None if "price" in payload: try: price_ov = float(payload["price"]) except (TypeError, ValueError): pass attr_set_id = self._resolve_attribute_set_id(row) payload_create = snapshot_to_magento_product( row, is_new=True, include_stock=False, price_override=price_ov if price_ov is not None else 1, attribute_set_id=attr_set_id, rows_by_sku=rows_by_sku, ) payload_create = self._transform_custom_attributes_for_magento( payload_create, self._conn_id ) payload_create = self._inject_website_ids(payload_create, is_new=True) payload_create = self._inject_category_ids( payload_create, row, self._conn_id, rows_by_sku=rows_by_sku ) payload_create = self._resolve_image_urls_in_payload( payload_create, sku, self._conn_id ) payload_create = self._inject_store_and_website_ids( payload_create, self._conn_id ) _status_ov = overrides.force.get(sku, {}).get("status") if overrides else None _explicit_disable = ( _status_ov is not None and str(_status_ov).strip() in ("0", "false", "disabled") ) payload_create["status"] = 0 if _explicit_disable else 1 if payload_create.get("extension_attributes"): ext = dict(payload_create["extension_attributes"]) ext.pop("stock_item", None) payload_create["extension_attributes"] = ext payload_create.pop("attribute_set_code", None) if overrides: force = overrides.force.get(sku, {}) locked = overrides.lock.get(sku, set()) payload_create = apply_overrides_to_payload(payload_create, force, locked) _visibility_locked = overrides is not None and "visibility" in overrides.lock.get(sku, set()) if not _visibility_locked: variant_of = _relation_sku(row.get("variant_of")) type_id = str(payload_create.get("type_id") or row.get("product_type") or "").strip().lower() if variant_of: payload_create["visibility"] = VISIBILITY_NOT_VISIBLE elif "configurable" in type_id: payload_create["visibility"] = VISIBILITY_CATALOG_SEARCH if "visibility" in payload_create: payload_create["visibility"] = int(payload_create["visibility"]) return payload_create def _success_upsert_result( self, *, sku: str, row: Dict[str, Any], start: float, http_status: int, payload_sent: Dict[str, Any], is_new: bool, partial_attr_update: bool, ) -> Dict[str, Any]: inv_result = ( InventoryUpsertResult(msi_mode_detected=False) if partial_attr_update else self._ensure_inventory_for_simple(sku, row) ) return { "status": "success", "duration_ms": int((time.perf_counter() - start) * 1000), "http_status": http_status, "is_new": is_new, "inventory_result": inv_result, "payload_sent": payload_sent, } def _retry_put_as_update( self, *, sku: str, put_payload: Dict[str, Any], row: Dict[str, Any], start: float, partial_attr_update: bool, reason: str, ) -> Dict[str, Any]: """PUT again after discovering the product already exists (race / false create).""" logger.info("Product %s: %s; retrying PUT instead of create", sku, reason) _log_magento_product_post("PUT", sku, put_payload) status, body = self._call_with_retry( f"put_product:{sku}", lambda: self._api.put_product(sku, put_payload), ) if status in (200, 201): return self._success_upsert_result( sku=sku, row=row, start=start, http_status=status, payload_sent=put_payload, is_new=False, partial_attr_update=partial_attr_update, ) return { "status": "failed", "duration_ms": int((time.perf_counter() - start) * 1000), "http_status": status, "error": str(body) if body else f"HTTP {status}", "is_new": False, "payload_sent": put_payload, } def execute_upsert_product( self, sku: str, payload: Dict[str, Any], row: Dict[str, Any], *, overrides: Optional[OverrideSet] = None, rows_by_sku: Optional[Dict[str, Dict[str, Any]]] = None, selected_attribute_codes: Optional[Set[str]] = None, ) -> Dict[str, Any]: """Execute UPSERT_PRODUCT. PUT-first; POST only when GET confirms missing. Magento does not enforce a DB unique index on SKU, so optimistic POST-create races can create true duplicate entities. We GET before create and treat "SKU already exists" as update. """ start = time.perf_counter() selected_codes = { str(code).strip().lower() for code in (selected_attribute_codes or set()) if str(code).strip() } partial_attr_update = bool(selected_codes) put_payload = self._build_put_payload_for_upsert( sku, payload, row, overrides=overrides, selected_attribute_codes=selected_attribute_codes, ) _log_magento_product_post("PUT", sku, put_payload) try: status, body = self._call_with_retry( f"put_product:{sku}", lambda: self._api.put_product(sku, put_payload), ) duration_ms = int((time.perf_counter() - start) * 1000) if status in (200, 201): return self._success_upsert_result( sku=sku, row=row, start=start, http_status=status, payload_sent=put_payload, is_new=False, partial_attr_update=partial_attr_update, ) # 404 = missing. Some Magento setups return 400 "Invalid attribute set entity type" # or "Invalid product data" for missing OR for invalid updates on existing products. try_create = status == 404 ambiguous_missing = _is_ambiguous_missing_product_put_error(status, body) if not try_create and ambiguous_missing: logger.info( "PUT %s returned 400 (product may not exist); checking GET before create", sku, ) try_create = True if try_create: existing = None try: existing = self._api.get_product(sku) except Exception as exc: logger.warning( "GET product before create failed for %s: %s", sku, exc ) if existing is not None: if ambiguous_missing and status != 404: # Existing product failed PUT validation — never POST-create a twin. return { "status": "failed", "duration_ms": int((time.perf_counter() - start) * 1000), "http_status": status, "error": ( f"PUT failed on existing product (HTTP {status}); " f"refusing POST create to avoid duplicate SKU. " f"Detail: {body}" )[:500], "is_new": False, "payload_sent": put_payload, } return self._retry_put_as_update( sku=sku, put_payload=put_payload, row=row, start=start, partial_attr_update=partial_attr_update, reason="GET found product after PUT reported missing", ) payload_create = self._build_create_payload_for_upsert( sku, payload, row, overrides=overrides, rows_by_sku=rows_by_sku, ) # Try POST; on "URL key already exists" retry with modified name. # Magento REST does not accept url_key; it auto-generates from name. title = str(payload_create.get("name") or row.get("name") or row.get("title") or "").strip() or "product" title_variants = _title_variants_for_url_key_retry(sku, title) last_status: Optional[int] = None last_body: Any = None last_payload = payload_create for attempt in range(len(title_variants) + 1): if attempt > 0: payload_create = dict(last_payload) payload_create["name"] = title_variants[attempt - 1] logger.info( "Product %s: retrying create with name=%r (attempt %d, Magento will generate unique url_key)", sku, payload_create["name"], attempt, ) # Re-check existence each attempt — concurrent workers may have created it. try: raced = self._api.get_product(sku) except Exception: raced = None if raced is not None: return self._retry_put_as_update( sku=sku, put_payload=put_payload, row=row, start=start, partial_attr_update=partial_attr_update, reason="GET found product before POST create", ) last_payload = payload_create _log_magento_product_post("POST", sku, payload_create) status2, body2 = self._call_with_retry( f"post_product:{sku}", lambda p=payload_create: self._api.post_product(p), ) last_status, last_body = status2, body2 duration_ms = int((time.perf_counter() - start) * 1000) if status2 in (200, 201): if os.getenv("MAGENTO_VERIFY_PRODUCT_AFTER_CREATE", "1").strip().lower() in ("1", "true", "yes"): verified = self._api.get_product(sku) if verified is None: logger.warning( "Product %s: POST returned %s but GET product returned 404; treating as failed", sku, status2, ) return { "status": "failed", "duration_ms": duration_ms, "http_status": status2, "error": f"Product create succeeded (HTTP {status2}) but verification GET returned 404", "is_new": True, } return self._success_upsert_result( sku=sku, row=row, start=start, http_status=status2, payload_sent=payload_create, is_new=True, partial_attr_update=False, ) if _is_sku_already_exists_error(status2, body2): return self._retry_put_as_update( sku=sku, put_payload=put_payload, row=row, start=start, partial_attr_update=partial_attr_update, reason="POST reported SKU already exists", ) if not _is_url_key_exists_error(status2, body2) or attempt >= len(title_variants): break return { "status": "failed", "duration_ms": int((time.perf_counter() - start) * 1000), "http_status": last_status or 400, "error": str(last_body) if last_body else "HTTP 400", "is_new": True, "payload_sent": last_payload, } return { "status": "failed", "duration_ms": duration_ms, "http_status": status, "error": str(body) if body else f"HTTP {status}", "payload_sent": put_payload, } except Exception as e: duration_ms = int((time.perf_counter() - start) * 1000) return {"status": "failed", "duration_ms": duration_ms, "error": str(e)} def execute_upsert_products_bulk( self, items: List[Dict[str, Any]], *, overrides: Optional[OverrideSet] = None, rows_by_sku: Optional[Dict[str, Dict[str, Any]]] = None, batch_size: int = 100, poll_interval_ms: int = 2000, poll_timeout_s: int = 600, ) -> List[Dict[str, Any]]: """Batch UPSERT_PRODUCT via Magento async bulk API. ``items`` entries: ``{sku, payload, row}``. Returns one result per item (same order). SKUs that need URL-key retry fall back to single-path upsert. Inventory enrichment runs per successful SKU after the bulk write. """ from magento.product_bulk_writer import BulkProductOp, MagentoProductBulkWriter if not items: return [] ops: List[BulkProductOp] = [] for it in items: sku = str(it["sku"]) payload = it.get("payload") or {} row = it.get("row") or {} put_payload = self._build_put_payload_for_upsert( sku, payload, row, overrides=overrides, ) create_payload = self._build_create_payload_for_upsert( sku, payload, row, overrides=overrides, rows_by_sku=rows_by_sku, ) _log_magento_product_post("BULK_PUT", sku, put_payload) ops.append( BulkProductOp( sku=sku, put_payload=put_payload, create_payload=create_payload, row=row, source_payload=payload, ) ) writer = MagentoProductBulkWriter( api=self._api, batch_size=batch_size, poll_interval_s=max(0.2, poll_interval_ms / 1000.0), poll_timeout_s=float(poll_timeout_s), call_with_retry=self._call_with_retry, ) results = writer.execute(ops) out: List[Dict[str, Any]] = [] for op, result in zip(ops, results): if result.get("needs_single_fallback"): logger.info( "Bulk create needs single-path fallback for %s: %s", op.sku, (result.get("error") or "")[:160], ) result = self.execute_upsert_product( op.sku, op.source_payload or op.put_payload, op.row, overrides=overrides, rows_by_sku=rows_by_sku, ) elif result.get("status") == "success": inv_result = self._ensure_inventory_for_simple(op.sku, op.row) result = dict(result) result["inventory_result"] = inv_result out.append(result) return out # ------------------------------------------------------------------ # Media helpers # ------------------------------------------------------------------ def _get_magento_media_entries(self, sku: str) -> List[Dict[str, Any]]: """GET /V1/products/{sku}/media. Cached per SKU for the job.""" if sku in self._media_gallery_cache: return self._media_gallery_cache[sku] status, entries, _ = self._call_with_retry( f"get_product_media:{sku}", lambda: self._api.get_product_media(sku), ) result = entries if status == 200 and entries else [] self._media_gallery_cache[sku] = result return result def _invalidate_media_cache(self, sku: str) -> None: self._media_gallery_cache.pop(sku, None) def _entry_exists_in_gallery( self, sku: str, magento_entry_id: Optional[int], magento_file: Optional[str] ) -> bool: """Check if a media entry is present in Magento gallery by entry_id or file path.""" entries = self._get_magento_media_entries(sku) for e in entries: if not isinstance(e, dict): continue if magento_entry_id and e.get("id") == magento_entry_id: return True if magento_file and e.get("file") == magento_file: return True return False def _resolve_media_entry_id( self, sku: str, position: int, entry: Dict[str, Any] ) -> Optional[int]: """Resolve entry_id by GET media; match on position, label, or file suffix.""" entries = self._get_magento_media_entries(sku) if not entries: return None label = (entry.get("label") or "").strip() content_name = (entry.get("content") or {}) if isinstance(content_name, dict): content_name = content_name.get("name", "") or "" our_suffix = str(content_name).strip().lower() for e in entries: if not isinstance(e, dict): continue if e.get("position") == position: return e.get("id") if label and str(e.get("label", "")).strip() == label: return e.get("id") f = str(e.get("file", "")).strip() if our_suffix and f: magento_suffix = f.split("/")[-1].lower() our_base = our_suffix.split(".")[0] if our_suffix in magento_suffix or (our_base and our_base in magento_suffix): return e.get("id") if position < len(entries) and isinstance(entries[position], dict): return entries[position].get("id") return None def _get_media_file_by_entry_id(self, sku: str, entry_id: int) -> Optional[str]: """Lookup Magento media file path by entry id.""" entries = self._get_magento_media_entries(sku) for e in entries: if isinstance(e, dict): try: if int(e.get("id", -1)) == int(entry_id): file_val = e.get("file") if file_val: return str(file_val) except Exception: continue return None def _download_image(self, url: str) -> tuple[Optional[bytes], Optional[str]]: """Download image, return (bytes, content_sha256_hex).""" try: resolved = self._resolve_url(url) resp = requests.get(resolved, timeout=30) resp.raise_for_status() data = resp.content sha = hashlib.sha256(data).hexdigest() return data, sha except Exception as e: logger.warning("Failed to download image %s: %s", url, e) return None, None # ------------------------------------------------------------------ # Media upsert (D - robust) # ------------------------------------------------------------------ def execute_upsert_media( self, sku: str, row: Dict[str, Any], *, overrides: Optional[OverrideSet] = None, rows_by_sku: Optional[Dict[str, Dict[str, Any]]] = None, images_only: bool = False, ) -> Dict[str, Any]: """ Upload images to Magento. Implements robust mapping: - If mapping exists and image in gallery → skip - If mapping exists but not in gallery → re-upload and update mapping - If no mapping → download, dedupe by sha256, upload, store mapping Product must exist in Magento before media upload (POST /media creates gallery entry for a product by SKU). If product is missing and overrides/rows_by_sku are provided, runs UPSERT_PRODUCT first, then retries media upload. """ if not self._media_repo: return {"status": "failed", "error": "media_map_repo not configured"} start = time.perf_counter() # A3: Product must exist before media upload (Magento returns 404 if product missing). # If missing and we have row/overrides/rows_by_sku, create product first then retry. product = self._api.get_product(sku) if product is None: if images_only: logger.warning( "Media upload skipped for %s: product does not exist in Magento (images_only mode).", sku, ) return { "status": "failed", "error": "Product does not exist in Magento. Link SKU or run full product sync first.", "media_skipped_due_to_missing_product": True, "duration_ms": int((time.perf_counter() - start) * 1000), "uploaded_count": 0, "skipped_count": 0, "remote_image_urls": [], } if overrides is not None and rows_by_sku is not None and row: logger.info( "UPSERT_MEDIA %s: product does not exist in Magento, running UPSERT_PRODUCT first", sku, ) upsert_result = self.execute_upsert_product( sku, {}, row, overrides=overrides, rows_by_sku=rows_by_sku, ) if upsert_result.get("status") == "success": product = self._api.get_product(sku) if product is not None: logger.info("UPSERT_MEDIA %s: product created, proceeding with media upload", sku) if product is None: logger.warning( "Media upload skipped for %s: product does not exist in Magento. " "Run UPSERT_PRODUCT first (e.g. force_upsert=True).", sku, ) return { "status": "failed", "error": "Product does not exist in Magento. Create product before uploading media.", "media_skipped_due_to_missing_product": True, "duration_ms": int((time.perf_counter() - start) * 1000), "uploaded_count": 0, "skipped_count": 0, "remote_image_urls": [], } urls = _ordered_unique_media_urls(row) if not urls: return { "status": "success", "duration_ms": int((time.perf_counter() - start) * 1000), "uploaded_count": 0, "skipped_count": 0, "remote_image_urls": [], } self._invalidate_media_cache(sku) existing_maps = self._media_repo.get_media_maps_for_sku(self._conn_id, sku) uploaded = 0 skipped = 0 reordered = 0 def _media_role_types(pos: int) -> List[str]: return ["image", "small_image", "thumbnail"] if pos == 0 else [] def _sync_existing_media_order(entry_id: int, pos: int, *, label: str = "") -> bool: """Ensure gallery position/roles match canonical URL order (SKU image at 0).""" nonlocal reordered desired_types = _media_role_types(pos) entry_payload: Dict[str, Any] = { "id": int(entry_id), "media_type": "image", "label": label if pos == 0 else "", "position": pos, "disabled": False, "types": desired_types, } status, _body, err = self._call_with_retry( f"put_product_media:{sku}:{pos}:{entry_id}", lambda e=entry_payload, eid=int(entry_id): self._api.update_product_media(sku, eid, e), ) self._invalidate_media_cache(sku) if status in (200, 201): reordered += 1 return True logger.warning( "Media reorder failed for %s entry=%s pos=%d: HTTP %s %s", sku, entry_id, pos, status, err, ) return False for position, url in enumerate(urls): mapping = existing_maps.get(url) label = (row.get("name") or sku) if position == 0 else "" if mapping and mapping.get("magento_entry_id"): entry_id = int(mapping["magento_entry_id"]) if self._entry_exists_in_gallery(sku, entry_id, mapping.get("magento_file")): _sync_existing_media_order(entry_id, position, label=str(label or "")) skipped += 1 continue logger.info("Media map exists for %s/%s but not in gallery; re-uploading", sku, url[:60]) data, content_sha = self._download_image(url) if not data or not content_sha: continue ext, media_type, validation_error = _image_content_type(url, data) if validation_error: logger.warning("Media upload skipped for %s pos=%d url=%s: %s", sku, position, url, validation_error) skipped += 1 continue if mapping and mapping.get("content_sha256") == content_sha and mapping.get("magento_entry_id"): entry_id = int(mapping["magento_entry_id"]) if self._entry_exists_in_gallery(sku, entry_id, mapping.get("magento_file")): _sync_existing_media_order(entry_id, position, label=str(label or "")) skipped += 1 continue by_sha = self._media_repo.get_by_sha256(self._conn_id, sku, content_sha) if by_sha and by_sha.get("magento_entry_id"): entry_id = int(by_sha["magento_entry_id"]) if self._entry_exists_in_gallery(sku, entry_id, None): self._media_repo.upsert(self._conn_id, sku, url, content_sha, entry_id) _sync_existing_media_order(entry_id, position, label=str(label or "")) skipped += 1 continue b64 = base64.b64encode(data).decode("ascii") entry = { "media_type": "image", "label": label, "position": position, "disabled": False, "types": _media_role_types(position), "content": { "base64_encoded_data": b64, "type": media_type, "name": f"image_{position}.{ext}", }, } status, body, err = self._call_with_retry( f"post_product_media:{sku}:{position}", lambda e=entry: self._api.post_product_media(sku, e), ) self._invalidate_media_cache(sku) if status in (200, 201): entry_id = body.get("id") if isinstance(body, dict) else None if entry_id is None and isinstance(body, (int, str)): try: entry_id = int(body) except (ValueError, TypeError): pass if entry_id is None: entry_id = self._resolve_media_entry_id(sku, position, entry) magento_file = self._get_media_file_by_entry_id(sku, int(entry_id)) if entry_id else None self._media_repo.upsert( self._conn_id, sku, url, content_sha, magento_entry_id=int(entry_id) if entry_id is not None else None, magento_file=magento_file, ) uploaded += 1 else: logger.warning("Media upload failed for %s pos=%d: HTTP %s %s", sku, position, status, err) duration_ms = int((time.perf_counter() - start) * 1000) return { "status": "success", "duration_ms": duration_ms, "uploaded_count": uploaded, "skipped_count": skipped, "reordered_count": reordered, "remote_image_urls": urls, } # ------------------------------------------------------------------ # Configurable relations # ------------------------------------------------------------------ def execute_set_relations( self, parent_sku: str, row: Dict[str, Any], *, strict: bool = True, replace_options: bool = False, overrides: Optional[OverrideSet] = None, rows_by_sku: Optional[Dict[str, Dict[str, Any]]] = None, ) -> Dict[str, Any]: """ Sync configurable children. Remove stale, add missing, create options. strict=True: fail if attribute_id or option value_index cannot be resolved. replace_options=True: delete ALL existing options first, then create correct ones (use when fixing wrong axis configuration). If parent does not exist and overrides/rows_by_sku are provided, runs UPSERT_PRODUCT first, then retries SET_RELATIONS. """ # Guard: parent must exist before we can create options or link children parent = self._api.get_product(parent_sku) if parent is None: if overrides is not None and rows_by_sku is not None: logger.info( "SET_RELATIONS %s: parent missing in Magento, running UPSERT_PRODUCT first", parent_sku, ) upsert_result = self.execute_upsert_product( parent_sku, {}, row, overrides=overrides, rows_by_sku=rows_by_sku, ) if upsert_result.get("status") != "success": return { "status": "failed", "error": ( upsert_result.get("error") or f"Parent {parent_sku} UPSERT_PRODUCT failed; cannot run SET_RELATIONS" ), "linked_count": 0, "removed_count": 0, "options_created": 0, } parent = self._api.get_product(parent_sku) if parent is None: return { "status": "failed", "error": f"Parent product {parent_sku} does not exist in Magento. Run UPSERT_PRODUCT first.", "linked_count": 0, "removed_count": 0, "options_created": 0, } children, mappings, child_to_mapping = _parse_children_from_row(row) expected_set = {s.strip().lower(): s for s in children if s.strip()} expected_skus = list(expected_set.values()) options_created = 0 attr_codes = self._extract_configurable_attribute_codes(row, mappings) expected_axis_codes = {c.strip().lower() for c in attr_codes if c.strip()} duplicate_signatures = _duplicate_axis_signatures( expected_skus, mappings, attr_codes, child_to_mapping, ) if duplicate_signatures: return { "status": "failed", "error": ( f"Parent {parent_sku} has duplicate child configurable attribute values. " "Magento requires every child under a configurable parent to have a unique " "combination of axis values. Fix the Variation Builder axes or split the parent group." ), "duplicate_axis_signatures": duplicate_signatures, "linked_count": 0, "removed_count": 0, "options_created": options_created, "verified": False, } ttl_cutoff_rel = datetime.now(timezone.utc) - timedelta(days=7) default_set_id = self._default_attribute_set_id set_attrs_map: Dict[str, int] = {} if self._attr_set_attrs_repo and default_set_id: set_attrs_map = self._attr_set_attrs_repo.get_attributes_for_set( self._conn_id, default_set_id ) # attr_id_to_code: map Magento attribute_id -> our attr_code. Used to recognize # existing options and avoid wrongly treating axes as changed (which triggers replace). # Enrich from attr_cache/API for each feed axis - default set may not include custom attrs. attr_id_to_code = {str(v): k for k, v in set_attrs_map.items()} for ac in attr_codes: code_lower = ac.strip().lower() if set_attrs_map.get(code_lower) is not None: continue # Already from set_attrs_map aid = None if self._attr_cache_repo: attrs = self._attr_cache_repo.get_attributes(self._conn_id, [ac], ttl_cutoff_rel) a = attrs.get(code_lower) if a: aid = a.get("attribute_id") if aid is None: aid = self._api.get_attribute_id_by_code(ac) if aid is not None: attr_id_to_code[str(aid)] = code_lower opts_status, existing_opts, _ = self._call_with_retry( f"get_configurable_options:{parent_sku}", lambda: self._api.get_configurable_options(parent_sku), ) existing_attr_ids = set() existing_axis_codes: set[str] = set() if opts_status == 200 and existing_opts: for opt in existing_opts: aid = opt.get("attribute_id") or opt.get("id") if aid is not None: existing_attr_ids.add(str(aid)) code = attr_id_to_code.get(str(aid)) if code: existing_axis_codes.add(code.strip().lower()) # When feed axes differ from Magento, force replace axes_changed = expected_axis_codes != existing_axis_codes # When same axis but existing option has insufficient values (e.g. only width=24, feed has 24,30,36) # we must delete and recreate the option - Magento add_configurable_child requires option to have the value values_changed = self._configurable_option_values_insufficient( existing_opts, attr_id_to_code, mappings, attr_codes ) should_replace = replace_options or axes_changed or values_changed if opts_status == 200 and existing_opts and should_replace: for opt in existing_opts: opt_id = opt.get("id") if opt_id is not None: try: del_status, del_err = self._call_with_retry( f"delete_configurable_option:{parent_sku}:{opt_id}", lambda oid=opt_id: self._api.delete_configurable_option(parent_sku, oid), ) if del_status in (200, 204): logger.info( "SET_RELATIONS %s: deleted old option id=%s (axes_changed=%s values_changed=%s)", parent_sku, opt_id, axes_changed, values_changed, ) else: logger.warning("Failed to delete option %s for %s: %s", opt_id, parent_sku, del_err) except Exception as e: logger.warning("Delete option %s for %s: %s", opt_id, parent_sku, e) existing_attr_ids = set() for idx, attr_code in enumerate(attr_codes): attr_id = None code_lower = attr_code.strip().lower() if set_attrs_map and code_lower in set_attrs_map: attr_id = set_attrs_map[code_lower] if attr_id is None and self._attr_set_attrs_repo and default_set_id: if not self._attr_set_attrs_repo.attribute_in_set( self._conn_id, default_set_id, attr_code ): if strict: return { "status": "failed", "error": f"Attribute '{attr_code}' not in default attribute set (id={default_set_id}). " "Add attribute to the set in Magento or fix configurable_attributes in feed.", "linked_count": 0, "options_created": options_created, } if attr_id is None and self._attr_cache_repo: attrs = self._attr_cache_repo.get_attributes(self._conn_id, [attr_code], ttl_cutoff_rel) a = attrs.get(code_lower) if a: attr_id = a.get("attribute_id") if attr_id is None: attr_id = self._api.get_attribute_id_by_code(attr_code) if attr_id is None and strict: return { "status": "failed", "error": f"Attribute '{attr_code}' not found in Magento (no attribute_id). " f"Create attribute in Magento or fix configurable_attributes in feed.", "linked_count": 0, "options_created": options_created, } if str(attr_id or "") in existing_attr_ids: continue option_payload, err = self._build_configurable_option_payload( parent_sku, attr_code, mappings, idx, set_attrs_map=set_attrs_map ) if err: if strict: return {"status": "failed", "error": err, "linked_count": 0, "options_created": options_created} logger.error("SET_RELATIONS %s: %s", parent_sku, err) continue if not option_payload: continue status, body = self._call_with_retry( f"create_configurable_option:{parent_sku}:{attr_code}", lambda op=option_payload: self._api.create_configurable_option(parent_sku, op), ) if status in (200, 201): options_created += 1 else: err_msg = _extract_magento_error_message(body) logger.warning("Failed to create option %s for %s: %s", attr_code, parent_sku, err_msg) return { "status": "failed", "error": err_msg, "linked_count": 0, "removed_count": 0, "options_created": options_created, "verified": False, } ch_status, current_skus, _ = self._call_with_retry( f"get_configurable_children:{parent_sku}", lambda: self._api.get_configurable_children(parent_sku), ) current_set = {s.strip().lower(): s for s in (current_skus or []) if s and s.strip()} removed = 0 for clower, csku in list(current_set.items()): if clower not in expected_set: try: status, err = self._call_with_retry( f"remove_configurable_child:{parent_sku}:{csku}", lambda sku=csku: self._api.remove_configurable_child(parent_sku, sku), ) if status in (200, 204): removed += 1 del current_set[clower] else: logger.warning("Failed to remove %s from %s: %s", csku, parent_sku, err) except Exception as e: logger.warning("Failed to remove %s from %s: %s", csku, parent_sku, e, exc_info=True) linked = 0 for child_sku in expected_skus: if child_sku.strip().lower() in current_set: continue if self._link_configurable_child( parent_sku, child_sku, attr_codes=attr_codes, mappings=mappings, child_to_mapping=child_to_mapping, rows_by_sku=rows_by_sku, ): linked += 1 verified = self._verify_configurable_links(parent_sku, expected_skus) if not verified: # Retry once: re-fetch and add any still-missing children ch_status, current_skus_retry, _ = self._call_with_retry( f"get_configurable_children:{parent_sku}", lambda: self._api.get_configurable_children(parent_sku), ) current_retry = {s.strip().lower(): s for s in (current_skus_retry or []) if s and s.strip()} for child_sku in expected_skus: if child_sku.strip().lower() not in current_retry: if self._link_configurable_child( parent_sku, child_sku, attr_codes=attr_codes, mappings=mappings, child_to_mapping=child_to_mapping, rows_by_sku=rows_by_sku, retry=True, ): linked += 1 logger.info("SET_RELATIONS retry: linked %s to %s", child_sku, parent_sku) verified = self._verify_configurable_links(parent_sku, expected_skus) logger.info( "SET_RELATIONS %s: linked %d, removed %d, options_created=%d, verified=%s", parent_sku, linked, removed, options_created, verified, ) return { "status": "success", "linked_count": linked, "removed_count": removed, "options_created": options_created, "verified": verified, } def _link_configurable_child( self, parent_sku: str, child_sku: str, *, attr_codes: List[str], mappings: List[Dict[str, str]], child_to_mapping: Dict[str, Dict[str, str]], rows_by_sku: Optional[Dict[str, Dict[str, Any]]], retry: bool = False, ) -> bool: """Ensure axis attrs on child, then addChild. Returns True if linked.""" ensure_err = self._ensure_child_axis_attributes_on_magento( child_sku, attr_codes=attr_codes, mappings=mappings, child_to_mapping=child_to_mapping, rows_by_sku=rows_by_sku, ) if ensure_err: logger.warning( "SET_RELATIONS %s: could not set axis attrs on %s before link: %s", parent_sku, child_sku, ensure_err, ) op = f"add_configurable_child_retry:{parent_sku}:{child_sku}" if retry else f"add_configurable_child:{parent_sku}:{child_sku}" try: status, err = self._call_with_retry( op, lambda sku=child_sku: self._api.add_configurable_child(parent_sku, sku), ) if status in (200, 201): return True err_text = str(err or "") # Magento rejects link when child is missing the configurable attribute value. # Force a second ensure+link once when that specific error appears. if (not retry) and "doesn't have" in err_text.lower() and "attribute value" in err_text.lower(): ensure_err = self._ensure_child_axis_attributes_on_magento( child_sku, attr_codes=attr_codes, mappings=mappings, child_to_mapping=child_to_mapping, rows_by_sku=rows_by_sku, force=True, ) if ensure_err: logger.warning( "SET_RELATIONS %s: re-ensure axis attrs failed for %s: %s", parent_sku, child_sku, ensure_err, ) else: status2, err2 = self._call_with_retry( f"add_configurable_child_after_ensure:{parent_sku}:{child_sku}", lambda sku=child_sku: self._api.add_configurable_child(parent_sku, sku), ) if status2 in (200, 201): return True err = err2 logger.warning("Failed to link %s to %s: %s", child_sku, parent_sku, err) return False except Exception as e: logger.warning("Failed to link %s to %s: %s", child_sku, parent_sku, e, exc_info=True) return False def _resolve_child_axis_value_indexes( self, child_sku: str, *, attr_codes: List[str], mappings: List[Dict[str, str]], child_to_mapping: Dict[str, Dict[str, str]], rows_by_sku: Optional[Dict[str, Dict[str, Any]]], ) -> Dict[str, str]: """Map configurable axis codes → Magento option value_index strings for one child.""" mapping = None if child_to_mapping: mapping = child_to_mapping.get(child_sku) if mapping is None: for key, m in child_to_mapping.items(): if str(key).strip().lower() == child_sku.strip().lower(): mapping = m break if mapping is None: for m in mappings or []: mapped_sku = str(m.get("sku") or "").strip() if mapped_sku and mapped_sku.lower() == child_sku.lower(): mapping = m break child_row = (rows_by_sku or {}).get(child_sku) or {} ttl = datetime.now(timezone.utc) - timedelta(days=7) resolved: Dict[str, str] = {} for attr_code in attr_codes: code = attr_code.strip().lower() if not code: continue value_label = None if mapping: value_label = ( mapping.get(attr_code) or mapping.get(code) or mapping.get(attr_code.replace("_", "")) ) if not value_label: value_label = self._get_axis_value_from_child_row(child_row, code) if not value_label: continue opts: Dict[str, int] = {} if self._attr_cache_repo: opts = self._attr_cache_repo.get_options(self._conn_id, code, ttl) or {} if not opts: status, magento_options = self._call_with_retry( f"get_attribute_options_for_child:{code}", lambda c=code: self._api.get_attribute_options(c), ) if status == 200 and magento_options: for o in magento_options: if isinstance(o, dict): lb = str(o.get("label", "")).strip() val = o.get("value") or o.get("value_index") if lb and val is not None: try: opts[_norm_option_label(lb)] = int(val) except (ValueError, TypeError): pass vi = _resolve_option_value_index(str(value_label), opts, code) if opts else None if vi is not None: resolved[code] = str(vi) else: logger.warning( "Child %s: cannot resolve %s=%r to value_index (cache labels=%s)", child_sku, code, value_label, list(opts.keys())[:15] if opts else "[]", ) return resolved def _ensure_child_axis_attributes_on_magento( self, child_sku: str, *, attr_codes: List[str], mappings: List[Dict[str, str]], child_to_mapping: Dict[str, Dict[str, str]], rows_by_sku: Optional[Dict[str, Dict[str, Any]]], force: bool = False, ) -> Optional[str]: """Write missing configurable axis attributes onto the Magento child product. Magento ``addChild`` fails with "child product doesn't have the attribute value" when height/width/etc. are unset on the simple. Returns error string or None. """ axis_values = self._resolve_child_axis_value_indexes( child_sku, attr_codes=attr_codes, mappings=mappings, child_to_mapping=child_to_mapping, rows_by_sku=rows_by_sku, ) if not axis_values: return ( f"no configurable axis values resolved for {child_sku} " f"(attrs={attr_codes})" ) product = None try: product = self._api.get_product(child_sku) except Exception as exc: return f"get_product failed: {exc}" if product is None: return f"child {child_sku} not found in Magento" existing: Dict[str, str] = {} for attr in product.get("custom_attributes") or []: if not isinstance(attr, dict): continue code = str(attr.get("attribute_code") or "").strip().lower() if code in axis_values and attr.get("value") is not None: existing[code] = str(attr.get("value")).strip() needed = { code: value for code, value in axis_values.items() if force or existing.get(code) != value } if not needed: return None payload = { "sku": child_sku, "custom_attributes": [ {"attribute_code": code, "value": value} for code, value in needed.items() ], } status, body = self._call_with_retry( f"put_child_axis_attrs:{child_sku}", lambda: self._api.put_product(child_sku, payload), ) if status not in (200, 201): return _extract_magento_error_message(body) or f"HTTP {status}" logger.info( "SET_RELATIONS: set axis attrs on %s: %s", child_sku, ", ".join(f"{k}={v}" for k, v in sorted(needed.items())), ) return None def _configurable_option_values_insufficient( self, existing_opts: List[Dict[str, Any]], attr_id_to_code: Dict[str, str], mappings: List[Dict[str, str]], attr_codes: List[str], ) -> bool: """ True if any existing option has fewer values than we need for our children. E.g. Magento has width with [24]; feed has children with width 24,30,36 → replace needed. """ if not existing_opts or not mappings: return False for opt in existing_opts: aid = opt.get("attribute_id") or opt.get("id") if aid is None: continue code = attr_id_to_code.get(str(aid)) if not code: continue expected_codes = {c.strip().lower() for c in attr_codes if c.strip()} if code.strip().lower() not in expected_codes: continue existing_values: set[int] = set() for v in opt.get("values") or []: vi = v.get("value_index") if isinstance(v, dict) else v if vi is not None: try: existing_values.add(int(vi)) except (ValueError, TypeError): pass needed = self._get_needed_value_indexes_for_attr(code.strip().lower(), mappings) if needed is None: continue # Cannot resolve; skip this attr if needed and needed - existing_values: logger.info( "SET_RELATIONS values_changed: attr=%s needed=%s existing=%s", code, needed, existing_values, ) return True return False def _get_needed_value_indexes_for_attr( self, attr_code: str, mappings: List[Dict[str, str]] ) -> Optional[set[int]]: """Resolve mapping labels to value_index; return set or None if unresolved.""" code_lower = attr_code.strip().lower() distinct_labels: set[str] = set() for m in mappings: v = m.get(attr_code) or m.get(attr_code.replace("_", "")) if v and str(v).strip(): distinct_labels.add(str(v).strip()) if not distinct_labels: return set() opts = {} if self._attr_cache_repo: opts = self._attr_cache_repo.get_options( self._conn_id, attr_code, datetime.now(timezone.utc) - timedelta(days=7) ) if not opts: status, magento_options = self._call_with_retry( f"get_attribute_options_values:{attr_code}", lambda: self._api.get_attribute_options(attr_code), ) if status == 200 and magento_options: for o in magento_options: if isinstance(o, dict): lb = str(o.get("label", "")).strip() val = o.get("value") or o.get("value_index") if lb and val is not None: try: opts[_norm_option_label(lb)] = int(val) except (ValueError, TypeError): pass if not opts: return None result: set[int] = set() for label in distinct_labels: norm = _norm_option_label(label) vi = opts.get(norm) if vi is None: for alias in OPTION_LABEL_ALIASES.get(norm, []): vi = opts.get(alias) if vi is not None: break if vi is None: norm_flat = norm.replace(" ", "") for k, val in opts.items(): if k.replace(" ", "") == norm_flat: vi = val break if vi is not None: result.add(int(vi)) return result if len(result) == len(distinct_labels) else None def _build_configurable_option_payload( self, parent_sku: str, attr_code: str, mappings: List[Dict[str, str]], position: int, *, set_attrs_map: Optional[Dict[str, int]] = None, ) -> tuple[Optional[Dict[str, Any]], Optional[str]]: """Build Magento option payload with attribute_id and values[{"value_index":...}].""" attr_repo = self._attr_cache_repo ttl_cutoff = datetime.now(timezone.utc) - timedelta(days=7) detail = None label_to_value_index: Dict[str, int] = {} code_lower = attr_code.strip().lower() attr_id_from_set = (set_attrs_map or {}).get(code_lower) if set_attrs_map else None if attr_repo: attrs = attr_repo.get_attributes(self._conn_id, [attr_code], ttl_cutoff) detail = attrs.get(code_lower) if detail: opts = attr_repo.get_options(self._conn_id, attr_code, ttl_cutoff) label_to_value_index = opts if not detail and not attr_id_from_set: status, detail = self._call_with_retry( f"get_attribute_detail:{attr_code}", lambda: self._api.get_attribute_detail(attr_code), ) if status != 200 or not detail: attr_id = self._api.get_attribute_id_by_code(attr_code) if attr_id is None: return None, f"Attribute {attr_code} not found in Magento for {parent_sku}" detail = {"attribute_id": attr_id, "frontend_label": attr_code.replace("_", " ").title()} if not label_to_value_index and attr_repo: opts = attr_repo.get_options(self._conn_id, attr_code, ttl_cutoff) label_to_value_index = opts if attr_id_from_set is not None and (not detail or detail.get("attribute_id") != attr_id_from_set): detail = {**(detail or {}), "attribute_id": attr_id_from_set} if not label_to_value_index: status, magento_options = self._call_with_retry( f"get_attribute_options:{attr_code}", lambda: self._api.get_attribute_options(attr_code), ) if status == 200 and magento_options: option_rows: List[Dict[str, Any]] = [] for opt in magento_options: if isinstance(opt, dict): lb = str(opt.get("label", "")).strip() val = opt.get("value") or opt.get("value_index") if lb and val is not None: try: vi = int(val) label_to_value_index[_norm_option_label(lb)] = vi option_rows.append({"label": lb, "value_index": vi}) except (ValueError, TypeError): pass if attr_repo and option_rows: attr_repo.bulk_upsert_options( self._conn_id, attr_code, option_rows, datetime.now(timezone.utc), ) attr_id = (attr_id_from_set if attr_id_from_set is not None else None) or (detail.get("attribute_id") if detail else None) if attr_id is None: return None, f"Attribute {attr_code} has no attribute_id for {parent_sku}" label = detail.get("frontend_label") or attr_code.replace("_", " ").title() distinct_values: set[str] = set() for m in mappings: v = m.get(attr_code) or m.get(attr_code.replace("_", "")) if v and str(v).strip(): distinct_values.add(str(v).strip()) if not distinct_values: return {"attribute_id": str(attr_id), "label": label, "position": position, "is_use_default": True, "values": []}, None if not label_to_value_index: return None, f"Cannot get options for attribute {attr_code} (HTTP/empty) for {parent_sku}" value_indices: List[Dict[str, Any]] = [] for feed_label in sorted(distinct_values): norm = _norm_option_label(feed_label) vi = label_to_value_index.get(norm) if vi is None: return None, ( f"Parent {parent_sku}, attribute {attr_code}: " f"feed value '{feed_label}' not found in Magento options. " f"Create this option in Magento or implement auto-provision. " f"Available: {list(label_to_value_index.keys())[:20]}" ) value_indices.append({"value_index": int(vi)}) value_indices.sort(key=lambda x: (x.get("value_index"), 0)) return { "attribute_id": str(attr_id), "label": label, "position": position, "is_use_default": True, "values": value_indices, }, None def _get_axis_value_from_child_row(self, child_row: Dict[str, Any], attr_code: str) -> Optional[str]: """Get axis attribute value from child row (height, width, etc.). Mirrors normalize's _get_child_value logic.""" code = attr_code.strip().lower() for key in (f"{code}_in", code, f"{code}_inches", f"{code} (inches)"): v = child_row.get(key) if v is not None and str(v).strip(): s = str(v).strip() try: from decimal import Decimal, InvalidOperation num = Decimal(s) if num % 1 == 0: return str(int(num)) return s except (ValueError, InvalidOperation): return s if code == "height": try: from normalize import _try_derive_height_from_sku derived = _try_derive_height_from_sku(str(child_row.get("sku", "")).strip()) if derived: return derived except Exception: pass add_attrs = child_row.get("additional_attributes") or "" if add_attrs: for pair in str(add_attrs).replace("|", ",").split(","): if "=" in pair: k, v = pair.split("=", 1) if k.strip().lower() == code and v and str(v).strip(): return str(v).strip() return None def _inject_child_configurable_attributes( self, payload: Dict[str, Any], child_row: Dict[str, Any], rows_by_sku: Dict[str, Dict[str, Any]], connection_id: int, ) -> Dict[str, Any]: """Inject configurable attribute value_index into child payload. Uses parent mapping first; falls back to child row when mapping missing or resolution fails.""" _debug_height = os.getenv("MAGENTO_DEBUG_HEIGHT", "").strip().lower() in ("1", "true", "yes") parent_sku = _relation_sku(child_row.get("variant_of")) if not parent_sku: return payload parent_row = rows_by_sku.get(parent_sku) if not parent_row: logger.warning("Child %s has variant_of=%s but parent row not in batch", child_row.get("sku"), parent_sku) return payload from magento.payload_mapper import get_axis_attribute_codes parent_axis_codes = get_axis_attribute_codes(parent_row) if not parent_axis_codes: return payload children, mappings, child_to_mapping = _parse_children_from_row(parent_row) child_sku = str(child_row.get("sku", "")).strip() mapping = child_to_mapping.get(child_sku) if child_to_mapping else None if mapping is None: try: idx = children.index(child_sku) except ValueError: logger.warning("Child %s not found in parent %s children list", child_sku, parent_sku) return payload if idx < len(mappings): mapping = mappings[idx] if _debug_height and "height" in parent_axis_codes: logger.info( "DEBUG_HEIGHT %s: parent=%s variant_list=%r configurable_variations=%r child_to_mapping[%s]=%r mapping_from_index=%r child_row_height_keys=%s", child_sku, parent_sku, str(parent_row.get("variant_list", ""))[:120], str(parent_row.get("configurable_variations", ""))[:120], child_sku, child_to_mapping.get(child_sku) if child_to_mapping else None, mapping, {k: child_row.get(k) for k in child_row if "height" in str(k).lower()}, ) attr_repo = self._attr_cache_repo if not attr_repo: return payload ttl = datetime.now(timezone.utc) - timedelta(days=7) existing_by_code: Dict[str, str] = {} for a in payload.get("custom_attributes") or []: c = str(a.get("attribute_code", "")).strip().lower() if c in parent_axis_codes and a.get("value") is not None: existing_by_code[c] = str(a.get("value", "")).strip() custom = [ a for a in (payload.get("custom_attributes") or []) if str(a.get("attribute_code", "")).strip().lower() not in parent_axis_codes ] for attr_code in parent_axis_codes: value_from_mapping = (mapping or {}).get(attr_code) value_from_child = self._get_axis_value_from_child_row(child_row, attr_code) if not value_from_mapping else None value_from_existing = existing_by_code.get(attr_code) if not value_from_mapping and not value_from_child else None value_label = value_from_mapping or value_from_child or value_from_existing if not value_label: logger.warning( "Child %s: no value for axis %s (mapping=%s, child_row keys=%s)", child_sku, attr_code, bool(mapping), list(k for k in child_row if "height" in str(k).lower() or "width" in str(k).lower()), ) continue opts = attr_repo.get_options(connection_id, attr_code, ttl) vi = _resolve_option_value_index(str(value_label), opts, attr_code) if vi is not None: custom.append({"attribute_code": attr_code, "value": str(vi)}) if _debug_height and attr_code == "height": logger.info( "DEBUG_HEIGHT %s: height injected label=%r -> value_index=%s (from=%s)", child_sku, value_label, vi, "mapping" if value_from_mapping else ("child_row" if value_from_child else "existing"), ) elif existing_by_code.get(attr_code) and existing_by_code[attr_code].isdigit(): custom.append({"attribute_code": attr_code, "value": existing_by_code[attr_code]}) else: logger.warning( "Child %s: option label '%s' for attribute %s not in cache; skipping injection. Cache labels=%s", child_sku, value_label, attr_code, list(opts.keys())[:15] if opts else "[]", ) if custom != (payload.get("custom_attributes") or []): payload = dict(payload) payload["custom_attributes"] = custom return payload def _inject_category_ids( self, payload: Dict[str, Any], row: Dict[str, Any], connection_id: int, *, rows_by_sku: Optional[Dict[str, Dict[str, Any]]] = None, ) -> Dict[str, Any]: """Inject resolved category_ids (union with all-products category) as extension_attributes.category_links. For configurable parents with no categories, inherit from children when rows_by_sku provided. """ cats_raw = row.get("categories") or row.get("categories_raw") or "" cat_ids = row.get("category_ids") # Configurable parent with no categories: inherit from children if ( not cat_ids and not str(cats_raw).strip() and rows_by_sku and (str(row.get("product_type", "")).strip().lower() == "configurable" or "configurable" in str(row.get("product_type", "")).lower()) ): children, _, _ = _parse_children_from_row(row) child_cat_ids: List[int] = [] child_cats: List[str] = [] for c_sku in children: c_row = rows_by_sku.get(c_sku) if rows_by_sku else None if c_row: c_ids = c_row.get("category_ids") if c_ids: for cid in (c_ids if isinstance(c_ids, list) else [c_ids]): try: vi = int(cid) if vi not in child_cat_ids: child_cat_ids.append(vi) except (TypeError, ValueError): pass c_raw = c_row.get("categories") or c_row.get("categories_raw") or "" if str(c_raw).strip(): for p in str(c_raw).replace("|", ",").split(","): part = p.strip() if part and part not in child_cats: child_cats.append(part) if child_cat_ids: cat_ids = child_cat_ids elif child_cats: cats_raw = ",".join(child_cats) if not cat_ids and self._cat_cache_repo and str(cats_raw).strip(): cat_ids = self._resolve_category_ids( connection_id, str(cats_raw).strip(), master_catalog_mode=bool(str(row.get("master_sku") or "").strip()), ) if isinstance(cat_ids, str): cat_ids = [int(c.strip()) for c in cat_ids.split(",") if c.strip().isdigit()] if not cat_ids: cat_ids = [] all_prod_id = self._all_products_category_id if all_prod_id and all_prod_id not in cat_ids: cat_ids = list(cat_ids) + [all_prod_id] if not cat_ids: return payload cat_links = [{"position": i, "category_id": str(cid)} for i, cid in enumerate(cat_ids)] payload = dict(payload) ext = dict(payload.get("extension_attributes") or {}) ext["category_links"] = cat_links payload["extension_attributes"] = ext return payload def _inject_website_ids( self, payload: Dict[str, Any], is_new: bool, ) -> Dict[str, Any]: """On create, set extension_attributes.website_ids from store registry / env config.""" if not is_new: return payload web_id = self._default_website_id if not web_id: return payload payload = dict(payload) ext = dict(payload.get("extension_attributes") or {}) ext["website_ids"] = [web_id] payload["extension_attributes"] = ext return payload def _verify_configurable_links(self, parent_sku: str, expected_child_skus: List[str]) -> bool: status, child_skus, _ = self._api.get_configurable_children(parent_sku) if status == 200 and child_skus: expected_set = {s.strip().lower() for s in expected_child_skus if s.strip()} actual_set = {s.strip().lower() for s in child_skus if s.strip()} return expected_set <= actual_set return False def _extract_configurable_attribute_codes( self, row: Dict[str, Any], mappings: List[Dict[str, str]] ) -> List[str]: codes: set[str] = set() attrs = str(row.get("configurable_attributes", "")).strip() if attrs: for x in attrs.split(","): part = x.strip() if "=" in part: code = part.split("=", 1)[0].strip().lower() else: code = part.lower() if part else "" if code and code != "sku": codes.add(code) for m in mappings: for k in m: if k != "sku": codes.add(k) return sorted(codes) # ------------------------------------------------------------------ # Incremental sync orchestrator # ------------------------------------------------------------------ def sync_incremental( self, connection_id: int, snapshot_rows: List[Dict[str, Any]], *, job_id: Optional[int] = None, dry_run: bool = False, state_repo: Optional[SyncStateRepository] = None, job_repo: Optional[SyncJobRepository] = None, override_repo: Optional[OverrideRepository] = None, version_repo: Optional[VersionRepository] = None, notification_repo: Optional[NotificationRepository] = None, notify_email: Optional[str] = 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, replace_options: bool = False, force_full_sync: bool = False, debug_output_path: Optional[str] = None, trace_skus: Optional[List[str]] = None, return_plan: bool = False, return_action_details: bool = False, force_upsert_skus: Optional[List[str]] = None, images_only: bool = False, products_only: bool = False, force_images: bool = False, execute_skus: Optional[Iterable[str]] = None, selected_attribute_codes: Optional[Set[str]] = None, use_connection_pending_table: bool = True, progress_callback: Optional[Any] = None, abort_callback: Optional[Any] = None, ) -> Dict[str, Any]: """Run incremental sync: plan actions from hash diff, execute, update state. All sync pipelines use this flow (worker, full sync API, incremental API, debug sync-sku). Guarantees applied to every product: - Status=1 (Enabled) for configurable parents and children (shell products) - Categories: parents with none inherit from children; children keep their categories - website_id / store_ids: invalid (0) values replaced with baseline from store registry - MSI: never send stock_item (stock_id=1); assign to MAGENTO_DEFAULT_SOURCE_CODE via source-items """ repo = state_repo or self._state_repo jobr = job_repo or self._job_repo ov_repo = override_repo or self._override_repo ver_repo = version_repo or self._version_repo notif_repo = notification_repo or self._notification_repo if not repo: return {"status": "failed", "error": "sync_state_repo required for incremental sync"} conn_id = connection_id skus = [str(r.get("sku", "")).strip() for r in snapshot_rows if str(r.get("sku", "")).strip()] state_by_sku = repo.get_states_for_skus(conn_id, skus) overrides = ov_repo.get_overrides_for_skus(conn_id, skus) if ov_repo else OverrideSet(force={}, lock={}) selected_codes = { str(code).strip().lower() for code in (selected_attribute_codes or set()) if str(code).strip() } # ---- Load baseline-derived config (website_id, all-products category, default attribute set) ---- try: from settings import load_magento_baseline_config bl_cfg = load_magento_baseline_config() self._all_products_category_id = bl_cfg.all_products_category_id if bl_cfg.default_website_id: self._default_website_id = bl_cfg.default_website_id elif self._store_repo: self._default_website_id = self._store_repo.get_default_website_id(conn_id) if self._baseline_repo: self._default_attribute_set_id = self._baseline_repo.get_default_attribute_set_id(conn_id) # Env override: MAGENTO_DEFAULT_ATTRIBUTE_SET_ID forces a specific ID (e.g. when baseline stored wrong entity type) env_set_id = os.getenv("MAGENTO_DEFAULT_ATTRIBUTE_SET_ID", "").strip() if env_set_id and env_set_id.isdigit(): self._default_attribute_set_id = int(env_set_id) logger.info("Using MAGENTO_DEFAULT_ATTRIBUTE_SET_ID=%s", env_set_id) if bl_cfg.default_source_code: self._default_source_code = bl_cfg.default_source_code.strip() else: self._default_source_code = None except Exception as e: logger.warning("Could not load baseline config: %s", e) # ---- Resolve category_ids for all rows first (children get IDs from their paths) ---- if not images_only and not selected_codes: self._category_create_failed_cache.clear() self._category_missing_warned.clear() if self._cat_cache_repo: for idx, row in enumerate(snapshot_rows): if (idx + 1) % 500 == 0: logger.info("Category resolution: processed %d/%d rows", idx + 1, len(snapshot_rows)) cats_raw = row.get("categories") or row.get("categories_raw") or "" if str(cats_raw).strip() and not row.get("category_ids"): cat_ids = self._resolve_category_ids(conn_id, str(cats_raw).strip()) if cat_ids: row["category_ids"] = cat_ids logger.info("Category resolution: done (%d rows)", len(snapshot_rows)) # ---- Inherit categories for configurable parents with none (shell products) ---- rows_by_sku_pre = {str(r.get("sku", "")).strip(): r for r in snapshot_rows if str(r.get("sku", "")).strip()} for row in snapshot_rows: if row.get("category_ids"): continue pt = str(row.get("product_type", "")).strip().lower() if "configurable" not in pt: continue children, _, _ = _parse_children_from_row(row) child_cat_ids: List[int] = [] child_cats: List[str] = [] for c_sku in children: c_row = rows_by_sku_pre.get(c_sku) if c_row: c_ids = c_row.get("category_ids") if c_ids: for cid in (c_ids if isinstance(c_ids, list) else [c_ids]): try: vi = int(cid) if vi not in child_cat_ids: child_cat_ids.append(vi) except (TypeError, ValueError): pass c_raw = c_row.get("categories") or c_row.get("categories_raw") or "" if str(c_raw).strip(): for p in str(c_raw).replace("|", ",").split(","): part = p.strip() if part and part not in child_cats: child_cats.append(part) if child_cat_ids: row["category_ids"] = child_cat_ids elif child_cats and self._cat_cache_repo: cat_ids = self._resolve_category_ids(conn_id, ",".join(child_cats)) if cat_ids: row["category_ids"] = cat_ids # Pre-sync: ensure attributes and options exist (auto-create if enabled) if not images_only and not dry_run and (self._attr_cache_repo or True): try: ac_result = self.ensure_attributes_and_options_exist( conn_id, snapshot_rows, provision=True ) if ac_result.get("created_attributes") or ac_result.get("created_options"): logger.info( "ensure_attributes_and_options: created_attributes=%d created_options=%d", ac_result.get("created_attributes", 0), ac_result.get("created_options", 0), ) for err in ac_result.get("errors", [])[:5]: logger.warning("ensure_attributes_and_options: %s", err) except Exception as e: logger.warning("ensure_attributes_and_options failed: %s", e) # replace_options requires SET_RELATIONS to run, so force it # force_full_sync: upsert all products + relations + associations effective_force_relations = force_relations or replace_options or force_full_sync effective_force_associations = force_associations or force_full_sync effective_force_upsert = ( set(force_upsert_skus) if force_upsert_skus else {str(r.get("sku", "")).strip() for r in snapshot_rows if str(r.get("sku", "")).strip()} if force_full_sync else None ) default_set_id = self._default_attribute_set_id if self._default_attribute_set_id is not None else 4 resolve_attr_set = self._resolve_attribute_set_id if self._attr_set_registry_repo else None rows_by_sku = {str(r.get("sku", "")).strip(): r for r in snapshot_rows if str(r.get("sku", "")).strip()} # Resumable sync: use connection-scoped pending table when available; else job-scoped items pending_repo = self._sync_pending_repo plan: Optional[Any] = None pending: List[dict] = [] # Selected-attribute pushes are intentionally allowed to run in parallel. # The connection-scoped pending table is keyed only by (connection_id, sku, action), # so concurrent runs can overwrite/prune each other's rows and trigger # stale/concurrent update failures. Keep resumable pending-table behavior when # explicitly requested, but allow callers such as overnight full publish to fall # back to job-scoped pending items for isolation. use_pending_table = ( pending_repo is not None and not selected_codes and use_connection_pending_table ) if use_pending_table: # Connection-scoped: always plan, upsert to pending (updates existing with new payload) plan = plan_incremental_sync( conn_id, snapshot_rows, state_by_sku, overrides=overrides, catalog_state_media_by_sku=catalog_state_media_by_sku, catalog_state_by_sku=catalog_state_by_sku, force_relations=effective_force_relations, force_associations=effective_force_associations, force_upsert_skus=effective_force_upsert, default_attribute_set_id=default_set_id, resolve_attribute_set_id=resolve_attr_set, images_only=images_only, products_only=products_only, force_images=force_images, selected_attribute_codes=selected_codes or None, ) if len(plan.actions) == 0 and snapshot_rows: logger.info( "Planner produced 0 actions for %d snapshot row(s) " "(images_only=%s products_only=%s force_images=%s force_upsert=%s) — " "hashes match; nothing posted. Pass force_upsert / --force-products to re-push.", len(snapshot_rows), images_only, products_only, force_images, bool(effective_force_upsert), ) if not images_only: for row in snapshot_rows[:25]: sku = str(row.get("sku") or "").strip() if not sku: continue try: would = snapshot_to_magento_product( row, rows_by_sku=rows_by_sku, attribute_set_id=default_set_id, ) _log_magento_product_post("SKIPPED(hash-match)", sku, would) except Exception as exc: logger.info( "Magento SKIPPED(hash-match) product %s: could not build payload (%s); " "row name=%r meta_title=%r", sku, exc, row.get("name"), row.get("meta_title"), ) if dry_run: return { "status": "success", "dry_run": True, "planned_count": len(plan.actions), "plan_summary": summarize_plan_actions(plan), "actions": [{"sku": a.sku, "action": a.action} for a in plan.actions], } planned_keys = {(a.sku, a.action) for a in plan.actions} for a in plan.actions: pending_repo.upsert_pending( conn_id, a.sku, a.action, payload=a.payload if a.action == ACTION_UPSERT_PRODUCT else None, pass_number=a.pass_number, idempotency_key=a.idempotency_key, ) pruned = pending_repo.prune_pending_not_in_plan(conn_id, planned_keys) scope_skus = execute_skus if execute_skus is not None else None pending = pending_repo.get_pending_items(conn_id, skus=scope_skus) plan_summary = summarize_plan_actions(plan) if scope_skus is not None: logger.info( "Scoped pending execution to %d SKU(s): %d pending action(s) (connection=%s) " "plan=%s pruned_stale=%d", len({str(s).strip() for s in scope_skus if str(s).strip()}), len(pending), conn_id, plan_summary, pruned, ) else: logger.info( "Persisted plan to pending table: %d items (connection=%s) plan=%s pruned_stale=%d", len(pending), conn_id, plan_summary, pruned, ) else: # Job-scoped (legacy) if jobr and job_id: pending = jobr.get_pending_items(job_id) if not pending: plan = plan_incremental_sync( conn_id, snapshot_rows, state_by_sku, overrides=overrides, catalog_state_media_by_sku=catalog_state_media_by_sku, catalog_state_by_sku=catalog_state_by_sku, force_relations=effective_force_relations, force_associations=effective_force_associations, force_upsert_skus=effective_force_upsert, default_attribute_set_id=default_set_id, resolve_attribute_set_id=resolve_attr_set, images_only=images_only, products_only=products_only, force_images=force_images, selected_attribute_codes=selected_codes or None, ) if dry_run: return { "status": "success", "dry_run": True, "planned_count": len(plan.actions), "actions": [{"sku": a.sku, "action": a.action} for a in plan.actions], } from db.schemas import MagentoSyncItemCreate jobr.add_sync_items([ MagentoSyncItemCreate( job_id=job_id, sku=a.sku, action=a.action, payload=a.payload if a.action == ACTION_UPSERT_PRODUCT else None, status="pending", idempotency_key=a.idempotency_key, mode="sync", ) for a in plan.actions ]) pending = jobr.get_pending_items(job_id) logger.info("Persisted plan: %d items (job_id=%s)", len(pending), job_id) plan_count = len(jobr.get_job_items(job_id)) if (jobr and job_id and not use_pending_table) else len(pending) if jobr and job_id and plan_count and not dry_run: # Prefer progress_callback for total_count so the worker can commit the # pending plan without also locking magento_sync_jobs in the same txn. if progress_callback: progress_callback(0, plan_count, 0, 0) else: jobr.update_sync_job(job_id, total_count=plan_count) debug_path = debug_output_path debug_file = None if debug_path: import json debug_file = open(debug_path, "a", encoding="utf-8") logger.info("Debug output enabled: writing to %s (no Magento API calls)", debug_path) _h = lambda: datetime.now(timezone.utc).isoformat() if plan: diag = compute_planner_diagnostics( conn_id, snapshot_rows, state_by_sku, plan, overrides=overrides, force_relations=force_relations, trace_skus=trace_skus, ) header = { "__debug_header__": True, "connection_id": conn_id, "job_id": job_id, "planned_count": len(plan.actions), "at": _h(), "planner_diagnostics": diag, } else: header = { "__debug_header__": True, "resume": True, "connection_id": conn_id, "job_id": job_id, "pending_count": len(pending), "at": _h(), } debug_file.write(json.dumps(header, default=str) + "\n") debug_file.flush() def _sanitize(obj: Any) -> Any: if obj is None: return None if isinstance(obj, dict): return {k: _sanitize(v) for k, v in obj.items()} if isinstance(obj, (list, tuple)): return [_sanitize(x) for x in obj] if hasattr(obj, "isoformat"): return obj.isoformat() if obj is not None else None if isinstance(obj, float) and (obj != obj or abs(obj) == float("inf")): return None return obj def _write_debug_record(record: Dict[str, Any]) -> None: if debug_file: import json line = json.dumps(_sanitize(record), default=str, ensure_ascii=False) + "\n" debug_file.write(line) debug_file.flush() now = datetime.now(timezone.utc) success_count = 0 error_count = 0 items: List[Dict[str, Any]] = [] failed_items: List[Dict[str, Any]] = [] pending_count = len(pending) logger.info("Executing plan: %d pending actions (job_id=%s)", pending_count, job_id) from settings import load_magento_bulk_product_config bulk_cfg = load_magento_bulk_product_config() use_bulk_products = ( bulk_cfg.enabled and not debug_path and not selected_codes and not dry_run ) if use_bulk_products: logger.info( "Bulk product upsert enabled: batch_size=%d poll_timeout_s=%d", bulk_cfg.batch_size, bulk_cfg.poll_timeout_s, ) bulk_buffer: List[Dict[str, Any]] = [] def _record_action_outcome( *, item: Dict[str, Any], sku: str, action_type: str, action_payload: Any, idempotency_key: Any, row: Dict[str, Any], result: Dict[str, Any], version_id: Optional[int], action_idx: int, ) -> None: nonlocal success_count, error_count status_str = result.get("status", "failed") if ver_repo and version_id is not None: ver_repo.update_version_status(version_id, "applied" if status_str == "success" else "failed") if status_str == "success": success_count += 1 else: error_count += 1 err_msg = result.get("error") or "unknown" failed_items.append({ "sku": sku, "action": action_type, "error": err_msg, }) if error_count <= 5: payload_sent = result.get("payload_sent") planned = action_payload or {} debug_extra = "" attr_set = (payload_sent or {}).get("attribute_set_id") if isinstance(payload_sent, dict) else None if attr_set is None and isinstance(planned, dict): attr_set = planned.get("attribute_set_id") type_id = (payload_sent or planned or {}).get("type_id") if isinstance(payload_sent, dict) or isinstance(planned, dict) else None if attr_set is not None or type_id is not None: debug_extra = f" attribute_set_id={attr_set} type_id={type_id}" feed_attr = "" if action_type == ACTION_UPSERT_PRODUCT and row: feed_attr = f" feed[attribute_set_code]={row.get('attribute_set_code')!r} feed[attribute_set_id]={row.get('attribute_set_id')!r}" logger.warning( "Action failed #%d: sku=%s action=%s error=%s%s%s", error_count, sku, action_type, (err_msg or "")[:200], debug_extra, feed_attr, ) if repo: repo.upsert_state( conn_id, sku, last_seen_in_feed_at=now, last_error=(result.get("error") or "unknown")[:500], ) if use_pending_table and pending_repo and "id" in item: pending_repo.update_pending_item( item["id"], status="succeeded" if status_str == "success" else "failed", error=result.get("error"), duration_ms=result.get("duration_ms"), http_status=result.get("http_status"), ) elif jobr and job_id and "id" in item: magento_resp = {} if result.get("msi_mode_detected") is not None: magento_resp["msi_mode_detected"] = result["msi_mode_detected"] if result.get("source_items_written") is not None: magento_resp["source_items_written"] = result["source_items_written"] if result.get("legacy_stock_written") is not None: magento_resp["legacy_stock_written"] = result["legacy_stock_written"] if result.get("inventory_error"): magento_resp["inventory_error"] = result["inventory_error"] if result.get("bulk_uuid"): magento_resp["bulk_uuid"] = result["bulk_uuid"] jobr.update_sync_item( item["id"], status="succeeded" if status_str == "success" else "failed", error=result.get("error"), duration_ms=result.get("duration_ms"), http_status=result.get("http_status"), payload=result.get("payload_sent"), remote_image_urls=result.get("remote_image_urls"), version_id=version_id, magento_response=magento_resp if magento_resp else None, ) item_dict = { "job_id": job_id, "sku": sku, "action": action_type, "payload": result.get("payload_sent"), "status": "succeeded" if status_str == "success" else "failed", "error": result.get("error"), "duration_ms": result.get("duration_ms"), "http_status": result.get("http_status"), "idempotency_key": idempotency_key, "remote_image_urls": result.get("remote_image_urls"), "version_id": version_id, "mode": "sync", } inv = result.get("inventory_result") if inv is not None: item_dict["msi_mode_detected"] = inv.msi_mode_detected item_dict["source_items_written"] = inv.source_items_written item_dict["legacy_stock_written"] = inv.legacy_stock_written if inv.error: item_dict["inventory_error"] = inv.error if getattr(inv, "mode", None): item_dict["inventory_action_taken"] = inv.mode if getattr(inv, "before_source_items", None) is not None: item_dict["inventory_before"] = inv.before_source_items if getattr(inv, "after_source_items", None) is not None: item_dict["inventory_after"] = inv.after_source_items if getattr(inv, "sources_added", None) is not None: item_dict["inventory_sources_added"] = inv.sources_added if getattr(inv, "sources_updated", None) is not None: item_dict["inventory_sources_updated"] = inv.sources_updated if result.get("media_skipped_due_to_missing_product"): item_dict["media_skipped_due_to_missing_product"] = True if result.get("bulk_uuid"): item_dict["bulk_uuid"] = result["bulk_uuid"] items.append(item_dict) completed = action_idx + 1 if progress_callback: progress_callback(completed, pending_count, success_count, error_count) def _flush_bulk_products() -> None: if not bulk_buffer: return batch = list(bulk_buffer) bulk_buffer.clear() logger.info("Flushing bulk UPSERT_PRODUCT batch: %d SKUs", len(batch)) bulk_results = self.execute_upsert_products_bulk( [ {"sku": b["sku"], "payload": b["payload"], "row": b["row"]} for b in batch ], overrides=overrides, rows_by_sku=rows_by_sku, batch_size=bulk_cfg.batch_size, poll_interval_ms=bulk_cfg.poll_interval_ms, poll_timeout_s=bulk_cfg.poll_timeout_s, ) for ctx, result in zip(batch, bulk_results): if result.get("status") == "success" and repo: repo.upsert_state( conn_id, ctx["sku"], data_hash=compute_data_hash( ctx["row"], ctx["action_payload"] or {}, rows_by_sku=rows_by_sku ), last_seen_in_feed_at=now, last_pushed_at=now, last_error="", ) _record_action_outcome( item=ctx["item"], sku=ctx["sku"], action_type=ACTION_UPSERT_PRODUCT, action_payload=ctx["action_payload"], idempotency_key=ctx["idempotency_key"], row=ctx["row"], result=result, version_id=ctx.get("version_id"), action_idx=ctx["action_idx"], ) try: for action_idx, item in enumerate(pending): if abort_callback and abort_callback(): _flush_bulk_products() message = "Sync aborted because a newer Magento queue item superseded this run." logger.warning(message) if jobr and job_id: jobr.update_sync_job(job_id, status="cancelled", finished_at=datetime.utcnow()) return { "status": "cancelled", "reason": "superseded_by_newer_queue_item", "message": message, "planned_count": pending_count, "success_count": success_count, "error_count": error_count, } if (action_idx + 1) % 100 == 0: logger.info("Plan progress: %d/%d actions (%d ok, %d fail)", action_idx + 1, pending_count, success_count, error_count) sku = item["sku"] action_type = item["action"] action_payload = item.get("payload") idempotency_key = item.get("idempotency_key") row = rows_by_sku.get(sku, {}) result: Dict[str, Any] = {} version_id: Optional[int] = None # Keep pass ordering: flush product bulk before media/relations/etc. if action_type != ACTION_UPSERT_PRODUCT: _flush_bulk_products() if ver_repo and job_id: if action_type == ACTION_UPSERT_PRODUCT: data_hash = compute_data_hash( row, action_payload or {}, rows_by_sku=rows_by_sku ) version_id = ver_repo.create_version( conn_id, sku, job_id=job_id, source="sync", data_hash=data_hash, product_payload_json=action_payload, ) elif action_type == ACTION_UPSERT_MEDIA: images_hash = compute_images_hash(row) media_payload = build_media_payload_from_row(row) version_id = ver_repo.create_version( conn_id, sku, job_id=job_id, source="sync", images_hash=images_hash, media_payload_json=media_payload, ) elif action_type == ACTION_SET_RELATIONS: relations_hash = compute_relations_hash(row) relations_payload = build_relations_payload_from_row(row) version_id = ver_repo.create_version( conn_id, sku, job_id=job_id, source="sync", relations_hash=relations_hash, relations_payload_json=relations_payload, ) if action_type == ACTION_UPSERT_PRODUCT: payload = action_payload or {} if _relation_sku(row.get("variant_of")) and self._attr_cache_repo and rows_by_sku: payload = self._inject_child_configurable_attributes( payload, row, rows_by_sku, conn_id, ) if selected_codes: payload = _filter_magento_product_payload(payload, selected_codes) if not selected_codes: payload = self._inject_category_ids( payload, row, conn_id, rows_by_sku=rows_by_sku ) if debug_path: create_payload = snapshot_to_magento_product( row, is_new=True, include_stock=False, price_override=1.0 ) _write_debug_record({ "seq": len(items) + 1, "action": "UPSERT_PRODUCT", "sku": sku, "source_row": dict(row), "magento_payload": { "put_payload": payload, "create_payload_fallback": create_payload, "note": "PUT tried first; create_payload used on 404", }, }) result = {"status": "success", "duration_ms": 0} elif use_bulk_products and "/" not in str(sku): bulk_buffer.append({ "item": item, "sku": sku, "payload": payload, "row": row, "action_payload": action_payload, "idempotency_key": idempotency_key, "version_id": version_id, "action_idx": action_idx, }) if len(bulk_buffer) >= bulk_cfg.batch_size: _flush_bulk_products() continue else: result = self.execute_upsert_product( sku, payload, row, overrides=overrides, rows_by_sku=rows_by_sku, selected_attribute_codes=selected_codes or None, ) # Inventory: now handled in execute_upsert_product via _ensure_inventory_for_simple (simples only) if result.get("status") == "success" and repo and not debug_path: repo.upsert_state( conn_id, sku, data_hash=( compute_data_hash(row, action_payload or {}, rows_by_sku=rows_by_sku) if not selected_codes else None ), last_seen_in_feed_at=now, last_pushed_at=now, last_error="", ) elif action_type == ACTION_UPSERT_MEDIA: if debug_path: media_payload = build_media_payload_from_row(row) _write_debug_record({ "seq": len(items) + 1, "action": "UPSERT_MEDIA", "sku": sku, "source_row": dict(row), "magento_payload": { "media_entries": media_payload, "note": "Each entry: content.base64_encoded_data from URL fetch at runtime", }, }) result = {"status": "success", "duration_ms": 0} else: result = self.execute_upsert_media( sku, row, overrides=overrides, rows_by_sku=rows_by_sku, images_only=images_only, ) if result.get("status") == "success" and repo and not debug_path: images_hash = compute_images_hash(row) repo.upsert_state( conn_id, sku, images_hash=images_hash, last_seen_in_feed_at=now, last_pushed_at=now, last_error="", ) elif action_type == ACTION_SET_RELATIONS: if debug_path: rel_payload = build_relations_payload_from_row(row) _write_debug_record({ "seq": len(items) + 1, "action": "SET_RELATIONS", "sku": sku, "source_row": dict(row), "magento_payload": { "relations": rel_payload, "api_actions": [ "create_configurable_option (per attr if missing)", "remove_configurable_child (Magento-only children)", "add_configurable_child (Plytix children)", ], }, }) result = {"status": "success", "linked_count": 0, "removed_count": 0} else: result = self.execute_set_relations( sku, row, strict=True, replace_options=replace_options, overrides=overrides, rows_by_sku=rows_by_sku, ) if result.get("status") == "success" and repo and not debug_path: relations_hash = compute_relations_hash(row) repo.upsert_state( conn_id, sku, relations_hash=relations_hash, last_seen_in_feed_at=now, last_pushed_at=now, last_error="", ) elif action_type == ACTION_SET_ASSOCIATIONS: from magento.association_sync import ( compute_associations_hash_from_row, execute_set_associations, ) if debug_path: _write_debug_record({ "seq": len(items) + 1, "action": "SET_ASSOCIATIONS", "sku": sku, "source_row": dict(row), "magento_payload": { "association_fields": row.get("association_fields") or {}, "api_actions": [ "GET products/{sku}/links/{type}", "DELETE products/{sku}/links/{type}/{linkedSku}", "POST products/{sku}/links", ], }, }) result = {"status": "success", "added": 0, "removed": 0} else: result = execute_set_associations(self._api, sku, row) if result.get("status") == "success" and repo and not debug_path: associations_hash = ( result.get("associations_hash") or compute_associations_hash_from_row(row) ) repo.upsert_state( conn_id, sku, associations_hash=associations_hash, last_seen_in_feed_at=now, last_pushed_at=now, last_error="", ) _record_action_outcome( item=item, sku=sku, action_type=action_type, action_payload=action_payload, idempotency_key=idempotency_key, row=row, result=result, version_id=version_id, action_idx=action_idx, ) _flush_bulk_products() finally: try: _flush_bulk_products() except Exception: logger.exception("Failed flushing remaining bulk UPSERT_PRODUCT buffer") if debug_file: try: debug_file.close() except Exception: pass if jobr and job_id: jobr.update_sync_job( job_id, status="completed" if error_count == 0 else "completed_with_errors", finished_at=now, success_count=success_count, error_count=error_count, ) if use_pending_table and items: jobr.add_sync_items([_make_sync_item_create(it) for it in items]) if notif_repo and notify_email and job_id and (error_count > 0 or plan_count > 0): subject = f"Magento Sync Job {job_id}: {'Completed with errors' if error_count > 0 else 'Completed'}" top_failures = failed_items[:20] failures_text = "\n".join( f" - {f['sku']} ({f['action']}): {f['error'][:100]}" for f in top_failures ) or " (none)" job_info = jobr.get_sync_job(job_id) if jobr and hasattr(jobr, "get_sync_job") else {} started = job_info.get("started_at") if job_info else None finished = job_info.get("finished_at") or now if job_info else now body = f"""Magento Sync Job Report Job ID: {job_id} Started: {started.isoformat() if started and hasattr(started, 'isoformat') else 'N/A'} Finished: {finished.isoformat() if hasattr(finished, 'isoformat') else finished} Planned: {len(plan.actions) if plan else plan_count} Success: {success_count} Failed: {error_count} Top failures: {failures_text} """ notif_repo.enqueue_email(conn_id, job_id, notify_email, subject, body) if repo: for sku in rows_by_sku: repo.upsert_state(conn_id, sku, last_seen_in_feed_at=now) out: Dict[str, Any] = { "status": "success", "planned_count": plan_count, "success_count": success_count, "error_count": error_count, "job_id": job_id, } if return_plan: if plan: out["planned_actions"] = [ {"sku": a.sku, "action": a.action, "idempotency_key": a.idempotency_key} for a in plan.actions ] else: out["planned_actions"] = [ {"sku": i["sku"], "action": i["action"], "idempotency_key": i.get("idempotency_key")} for i in pending ] if return_action_details and items: out["action_details"] = [ { "sku": it.get("sku"), "action": it.get("action"), "status": it.get("status"), "msi_mode_detected": it.get("msi_mode_detected"), "source_items_written": it.get("source_items_written"), "legacy_stock_written": it.get("legacy_stock_written"), "inventory_error": it.get("inventory_error"), "inventory_action_taken": it.get("inventory_action_taken"), "inventory_before": it.get("inventory_before"), "inventory_after": it.get("inventory_after"), "inventory_sources_added": it.get("inventory_sources_added"), "inventory_sources_updated": it.get("inventory_sources_updated"), "media_skipped_due_to_missing_product": it.get("media_skipped_due_to_missing_product"), } for it in items ] return out