from __future__ import annotations import re from typing import Any, Dict, Iterable, List, Optional, Set import pandas as pd import sqlalchemy as sa from sqlalchemy import select from sqlalchemy.dialects.postgresql import insert from sqlalchemy.orm import Session from db.channel_assignments import active_skus_for_channel from db.channel_exports import resolve_pipeline_channel_code from db.compat_connections import pipeline_channel_for_connection from db.models import ChannelSkuMapping, MagentoCatalogState, MasterProduct, ProductSnapshot from db.source_imports import normalize_column_key from normalize import _relation_sku ACTIVE_STATUS = "active" PENDING_STATUS = "pending" INACTIVE_STATUS = "inactive" def channel_sku_for_master( session: Session, master_sku: str, channel_code: str, *, connection_id: Optional[int] = None, cache: Optional[Dict[str, str]] = None, ) -> str: """Resolve the SKU string used on a channel; falls back to master_sku when unmapped.""" master_sku = str(master_sku or "").strip() if not master_sku: return "" cache_key = f"{connection_id or ''}:{master_sku}" if cache is not None and cache_key in cache: return cache[cache_key] channel = resolve_pipeline_channel_code(channel_code) if connection_id: row = session.scalar( select(ChannelSkuMapping.channel_sku) .where(ChannelSkuMapping.master_sku == master_sku) .where(ChannelSkuMapping.connection_id == connection_id) .where(ChannelSkuMapping.mapping_status == ACTIVE_STATUS) ) if row: resolved = str(row).strip() if cache is not None: cache[cache_key] = resolved return resolved row = session.scalar( select(ChannelSkuMapping.channel_sku) .where(ChannelSkuMapping.master_sku == master_sku) .where(ChannelSkuMapping.channel_code == channel) .where(ChannelSkuMapping.connection_id.is_(None)) .where(ChannelSkuMapping.mapping_status == ACTIVE_STATUS) ) if row: resolved = str(row).strip() else: from db.channel_sku_prefix_mapping import resolve_channel_sku_from_prefixes resolved = resolve_channel_sku_from_prefixes( session, master_sku, channel, connection_id=connection_id, ) if cache is not None: cache[cache_key] = resolved return resolved def master_sku_for_channel_sku( session: Session, channel_sku: str, channel_code: str, *, connection_id: Optional[int] = None, ) -> Optional[str]: channel_sku = str(channel_sku or "").strip() if not channel_sku: return None channel = resolve_pipeline_channel_code(channel_code) mapped = None if connection_id: mapped = session.scalar( select(ChannelSkuMapping.master_sku) .where(ChannelSkuMapping.channel_sku == channel_sku) .where(ChannelSkuMapping.connection_id == connection_id) .where(ChannelSkuMapping.mapping_status == ACTIVE_STATUS) ) if not mapped: mapped = session.scalar( select(ChannelSkuMapping.master_sku) .where(ChannelSkuMapping.channel_sku == channel_sku) .where(ChannelSkuMapping.channel_code == channel) .where(ChannelSkuMapping.connection_id.is_(None)) .where(ChannelSkuMapping.mapping_status == ACTIVE_STATUS) ) if mapped: return str(mapped).strip() from db.channel_sku_prefix_mapping import resolve_master_sku_from_prefixes reversed_sku = resolve_master_sku_from_prefixes(session, channel_sku, channel, connection_id=connection_id) if reversed_sku != channel_sku: product = session.scalar( select(MasterProduct.sku) .where(MasterProduct.sku == reversed_sku) .where(MasterProduct.is_active.is_(True)) ) if product: return str(product).strip() product = session.scalar( select(MasterProduct.sku) .where(MasterProduct.sku == channel_sku) .where(MasterProduct.is_active.is_(True)) ) return str(product).strip() if product else None def _master_match_indexes(session: Session) -> Dict[str, Any]: """In-memory indexes for channel-first SKU → master matching.""" master_skus: Set[str] = set() source_item_to_master: Dict[str, str] = {} normalized_to_master: Dict[str, str] = {} for sku, source_item in session.execute( select(MasterProduct.sku, MasterProduct.source_item).where(MasterProduct.is_active.is_(True)) ).all(): master = str(sku or "").strip() if not master: continue master_skus.add(master) token = _normalize_sku_token(master) if token and token not in normalized_to_master: normalized_to_master[token] = master source = str(source_item or "").strip() if source and source not in source_item_to_master: source_item_to_master[source] = master return { "master_skus": master_skus, "source_item_to_master": source_item_to_master, "normalized_to_master": normalized_to_master, } def match_remote_sku_to_master( channel_sku: str, *, indexes: Dict[str, Any], prefix_rules: Optional[List[tuple[str, str]]] = None, ) -> tuple[Optional[str], Optional[str]]: """Pure match: channel catalog SKU → active master SKU + match_source label.""" from db.channel_sku_prefix_mapping import apply_prefix_rules channel_sku = str(channel_sku or "").strip() if not channel_sku: return None, None master_skus: Set[str] = indexes["master_skus"] if channel_sku in master_skus: return channel_sku, "remote_exact_master_sku" rules = prefix_rules or [] reversed_sku = apply_prefix_rules(channel_sku, rules, from_master=False) if reversed_sku != channel_sku and reversed_sku in master_skus: return reversed_sku, "remote_prefix_reverse" source_match = indexes["source_item_to_master"].get(channel_sku) if source_match: return source_match, "remote_source_item" normalized = _normalize_sku_token(channel_sku) if normalized: master = indexes["normalized_to_master"].get(normalized) if master: return master, "remote_normalized" return None, None def resolve_master_for_remote_sku( session: Session, channel_sku: str, channel_code: str, *, connection_id: Optional[int] = None, indexes: Optional[Dict[str, Any]] = None, ) -> tuple[Optional[str], Optional[str]]: """Match a channel catalog SKU to an active master SKU (channel-first linking).""" from db.channel_sku_prefix_mapping import load_prefix_rules channel = resolve_pipeline_channel_code(channel_code) match_indexes = indexes or _master_match_indexes(session) rules = load_prefix_rules(session, channel, connection_id=connection_id) return match_remote_sku_to_master(channel_sku, indexes=match_indexes, prefix_rules=rules) def load_remote_catalog_for_channel( session: Session, channel_code: str, *, connection_id: Optional[int] = None, native_connection_id: Optional[int] = None, compat_connection_id: Optional[int] = None, ) -> tuple[Dict[str, Optional[str]], str]: """Load channel SKU → remote product id map from pulled/live catalogs.""" channel = resolve_pipeline_channel_code(channel_code) if channel == "magento": if not connection_id: return {}, "magento_catalog_state" catalog = { str(sku).strip(): (str(product_id) if product_id is not None else None) for sku, product_id in session.execute( select(MagentoCatalogState.sku, MagentoCatalogState.magento_product_id).where( MagentoCatalogState.connection_id == connection_id ) ).all() if str(sku).strip() } return catalog, "magento_catalog_state" if channel == "shopify": catalog = fetch_shopify_remote_catalog( session, connection_id=compat_connection_id, native_connection_id=native_connection_id, ) return catalog, "shopify_admin_api" if channel == "plytix": catalog = { str(sku).strip(): None for (sku,) in session.execute( select(ProductSnapshot.sku).where(ProductSnapshot.valid_to.is_(None)) ).all() if str(sku).strip() } return catalog, "product_snapshot" return {}, "" def link_channel_catalog_to_masters( session: Session, channel_code: str, *, connection_id: Optional[int] = None, native_connection_id: Optional[int] = None, compat_connection_id: Optional[int] = None, remote_catalog: Optional[Dict[str, Optional[str]]] = None, remote_source: Optional[str] = None, dry_run: bool = False, ) -> Dict[str, Any]: """Channel-first link: scan remote catalog SKUs and map each to a master when matched.""" channel = resolve_pipeline_channel_code(channel_code) if remote_catalog is None: remote_catalog, remote_source = load_remote_catalog_for_channel( session, channel, connection_id=connection_id, native_connection_id=native_connection_id, compat_connection_id=compat_connection_id, ) remote_ids = { str(sku).strip(): remote_id for sku, remote_id in (remote_catalog or {}).items() if str(sku).strip() } indexes = _master_match_indexes(session) from db.channel_sku_prefix_mapping import load_prefix_rules prefix_rules = load_prefix_rules(session, channel, connection_id=connection_id) mapped_by_channel_sku: Dict[str, ChannelSkuMapping] = {} mapped_by_master: Dict[str, ChannelSkuMapping] = {} stmt = select(ChannelSkuMapping).where(ChannelSkuMapping.channel_code == channel).where( ChannelSkuMapping.mapping_status == ACTIVE_STATUS ) if connection_id is not None: stmt = stmt.where(ChannelSkuMapping.connection_id == connection_id) else: stmt = stmt.where(ChannelSkuMapping.connection_id.is_(None)) for row in session.scalars(stmt).all(): mapped_by_channel_sku[str(row.channel_sku).strip()] = row mapped_by_master[str(row.master_sku).strip()] = row candidates: List[Dict[str, Any]] = [] backfill: List[Dict[str, Any]] = [] skipped: List[Dict[str, Any]] = [] already_linked = 0 pending_master_channel: Dict[str, str] = {} for channel_sku in sorted(remote_ids): remote_id = remote_ids[channel_sku] existing = mapped_by_channel_sku.get(channel_sku) if existing: already_linked += 1 if remote_id and not existing.remote_id: backfill.append( { "master_sku": existing.master_sku, "channel_code": channel, "connection_id": connection_id, "channel_sku": channel_sku, "remote_id": remote_id, "mapping_status": ACTIVE_STATUS, "match_source": "remote_id_backfill", "notes": "Channel SKU already linked; backfill remote_id from catalog", } ) continue master_sku, match_source = match_remote_sku_to_master( channel_sku, indexes=indexes, prefix_rules=prefix_rules, ) if not master_sku: skipped.append({"channel_sku": channel_sku, "reason": "no_master_match"}) continue master_mapping = mapped_by_master.get(master_sku) existing_channel_sku = pending_master_channel.get(master_sku) linked_channel_sku = ( str(master_mapping.channel_sku).strip() if master_mapping else (existing_channel_sku or "") ) if linked_channel_sku and linked_channel_sku != channel_sku: skipped.append( { "channel_sku": channel_sku, "master_sku": master_sku, "reason": "master_already_linked_different_sku", "existing_channel_sku": linked_channel_sku, } ) continue candidates.append( { "master_sku": master_sku, "channel_code": channel, "connection_id": connection_id, "channel_sku": channel_sku, "remote_id": remote_id, "mapping_status": ACTIVE_STATUS, "match_source": match_source or "link_from_channel_catalog", "notes": f"Channel catalog SKU {channel_sku} matched to master {master_sku}", } ) pending_master_channel[master_sku] = channel_sku all_rows = candidates + backfill summary = { "link_mode": "channel_first", "channel_code": channel, "connection_id": connection_id, "remote_source": remote_source or "", "remote_sku_count": len(remote_ids), "already_linked_count": already_linked, "candidate_count": len(candidates), "remote_id_backfill_count": len(backfill), "skipped_count": len(skipped), "unmatched_remote_count": len(skipped), "candidates": candidates[:500], "skipped": skipped[:500], } if dry_run: summary["dry_run"] = True return summary upserted = upsert_mappings(session, all_rows) summary["dry_run"] = False summary["upserted"] = upserted return summary def mapping_by_master( session: Session, channel_code: str, master_skus: Iterable[str], *, connection_id: Optional[int] = None, ) -> Dict[str, str]: """master_sku → channel_sku via explicit map, then prefix rules, else identity.""" from db.channel_sku_prefix_mapping import load_prefix_rules, apply_prefix_rules channel = resolve_pipeline_channel_code(channel_code) wanted = sorted({str(s).strip() for s in master_skus if str(s).strip()}) prefix_rules = load_prefix_rules(session, channel, connection_id=connection_id) result = { sku: apply_prefix_rules(sku, prefix_rules, from_master=True) if prefix_rules else sku for sku in wanted } if not wanted: return result for master_sku, channel_sku in session.execute( select(ChannelSkuMapping.master_sku, ChannelSkuMapping.channel_sku) .where(ChannelSkuMapping.channel_code == channel) .where(ChannelSkuMapping.mapping_status == ACTIVE_STATUS) .where(ChannelSkuMapping.master_sku.in_(wanted)) .where( ChannelSkuMapping.connection_id == connection_id if connection_id is not None else ChannelSkuMapping.connection_id.is_(None) ) ).all(): result[str(master_sku).strip()] = str(channel_sku).strip() return result def remote_ids_by_master( session: Session, channel_code: str, master_skus: Iterable[str], *, connection_id: Optional[int] = None, ) -> Dict[str, str]: """master_sku → linked remote product id (active mappings with a remote_id only).""" channel = resolve_pipeline_channel_code(channel_code) wanted = sorted({str(s).strip() for s in master_skus if str(s).strip()}) if not wanted: return {} return { str(master_sku).strip(): str(remote_id).strip() for master_sku, remote_id in session.execute( select(ChannelSkuMapping.master_sku, ChannelSkuMapping.remote_id) .where(ChannelSkuMapping.channel_code == channel) .where(ChannelSkuMapping.mapping_status == ACTIVE_STATUS) .where(ChannelSkuMapping.remote_id.is_not(None)) .where(ChannelSkuMapping.master_sku.in_(wanted)) .where( ChannelSkuMapping.connection_id == connection_id if connection_id is not None else ChannelSkuMapping.connection_id.is_(None) ) ).all() if str(remote_id or "").strip() } def reverse_mapping_by_channel_sku( session: Session, channel_code: str, channel_skus: Iterable[str], *, connection_id: Optional[int] = None, ) -> Dict[str, str]: """channel_sku → master_sku for active rows.""" channel = resolve_pipeline_channel_code(channel_code) wanted = sorted({str(s).strip() for s in channel_skus if str(s).strip()}) if not wanted: return {} return { str(channel_sku).strip(): str(master_sku).strip() for channel_sku, master_sku in session.execute( select(ChannelSkuMapping.channel_sku, ChannelSkuMapping.master_sku) .where(ChannelSkuMapping.channel_code == channel) .where(ChannelSkuMapping.mapping_status == ACTIVE_STATUS) .where(ChannelSkuMapping.channel_sku.in_(wanted)) .where( ChannelSkuMapping.connection_id == connection_id if connection_id is not None else ChannelSkuMapping.connection_id.is_(None) ) ).all() } def apply_sku_map_to_relation_fields( relation_fields: Dict[str, Any], sku_map: Dict[str, str], ) -> Dict[str, Any]: """Rewrite parent/child relation SKUs for channel push (e.g. Magento variant_list).""" if not relation_fields or not sku_map: return dict(relation_fields or {}) out = dict(relation_fields) if _relation_sku(out.get("variant_of")): parent = _relation_sku(out["variant_of"]) out["variant_of"] = sku_map.get(parent, parent) if out.get("variant_list"): mapped = [sku_map.get(part.strip(), part.strip()) for part in str(out["variant_list"]).split(",") if part.strip()] out["variant_list"] = ",".join(mapped) cv = str(out.get("configurable_variations") or "").strip() if cv and "sku=" in cv.lower(): segments = [] for seg in cv.split("|"): seg = seg.strip() if not seg: continue parts = [] for pair in seg.split(","): pair = pair.strip() if "=" not in pair: parts.append(pair) continue key, value = pair.split("=", 1) if normalize_column_key(key) == "sku": value = sku_map.get(value.strip(), value.strip()) parts.append(f"{key}={value}") segments.append(",".join(parts)) out["configurable_variations"] = "|".join(segments) return out def import_channel_sku_mapping_dataframe( session: Session, df: pd.DataFrame, *, source_label: str = "csv_import", ) -> Dict[str, Any]: rows: List[Dict[str, Any]] = [] skipped = 0 for raw in df.to_dict(orient="records"): record = {normalize_column_key(k): v for k, v in raw.items()} master_sku = _clean(record.get("master_sku") or record.get("sku") or record.get("canonical_sku")) channel_sku = _clean( record.get("channel_sku") or record.get("remote_sku") or record.get("external_sku") or record.get("target_sku") ) connection_id = _to_int(record.get("connection_id") or record.get("channel_id")) channel_code = _clean(record.get("channel_code") or record.get("channel")) if connection_id and not channel_code: channel_code = pipeline_channel_for_connection(session, connection_id) if not master_sku or not channel_sku or not (channel_code or connection_id): skipped += 1 continue rows.append( { "master_sku": master_sku, "channel_code": resolve_pipeline_channel_code(channel_code or "magento"), "connection_id": connection_id, "channel_sku": channel_sku, "remote_id": _clean(record.get("remote_id") or record.get("channel_product_id")), "mapping_status": _status(record.get("mapping_status") or record.get("status")), "match_source": _clean(record.get("match_source") or record.get("source")) or source_label, "notes": _clean(record.get("notes")), } ) upserted = upsert_mappings(session, rows) return {"input_rows": len(df), "upserted": upserted, "skipped": skipped} def _mapping_match_priority(match_source: Optional[str]) -> int: """Higher wins when two candidates target the same channel SKU slot.""" priorities = { "suggest_exact_sku": 100, "auto_magento_exact": 98, "remote_id_backfill": 95, "reconciliation_accept": 90, "link_from_remote": 85, "suggest_prefix_channel_sku": 80, "suggest_effective_channel_sku": 75, "auto_magento_source_item": 70, "suggest_source_item": 50, "suggest_normalized": 30, } return priorities.get(str(match_source or "").strip(), 40) def _dedupe_mapping_batch( rows: List[Dict[str, Any]], *, by_connection: bool, ) -> List[Dict[str, Any]]: """Collapse in-batch collisions on master slot and channel SKU slot (keep strongest match).""" def _pick_best(bucket: Dict[Any, Dict[str, Any]], key: Any, row: Dict[str, Any]) -> None: prev = bucket.get(key) if prev is None or _mapping_match_priority(row.get("match_source")) > _mapping_match_priority( prev.get("match_source") ): bucket[key] = row by_master: Dict[Any, Dict[str, Any]] = {} for row in rows: if by_connection: master_key = (row.get("connection_id"), row["master_sku"]) else: master_key = (row["channel_code"], row["master_sku"]) _pick_best(by_master, master_key, row) by_channel_sku: Dict[Any, Dict[str, Any]] = {} for row in by_master.values(): if by_connection: sku_key = (row.get("connection_id"), row["channel_sku"]) else: sku_key = (row["channel_code"], row["channel_sku"]) _pick_best(by_channel_sku, sku_key, row) return list(by_channel_sku.values()) def _reclaim_conflicting_channel_skus( session: Session, rows: List[Dict[str, Any]], *, by_connection: bool, ) -> int: """Remove stale rows that occupy a channel SKU slot for a different master.""" if not rows: return 0 reclaimed = 0 if by_connection: scoped: Dict[int, Dict[str, str]] = {} for row in rows: connection_id = _to_int(row.get("connection_id")) if not connection_id: continue scoped.setdefault(connection_id, {})[row["channel_sku"]] = row["master_sku"] for connection_id, channel_sku_to_master in scoped.items(): existing = session.scalars( select(ChannelSkuMapping) .where(ChannelSkuMapping.connection_id == connection_id) .where(ChannelSkuMapping.channel_sku.in_(sorted(channel_sku_to_master))) ).all() for row in existing: winner = channel_sku_to_master.get(row.channel_sku) if winner and row.master_sku != winner: session.delete(row) reclaimed += 1 else: scoped_codes: Dict[str, Dict[str, str]] = {} for row in rows: channel_code = row["channel_code"] scoped_codes.setdefault(channel_code, {})[row["channel_sku"]] = row["master_sku"] for channel_code, channel_sku_to_master in scoped_codes.items(): existing = session.scalars( select(ChannelSkuMapping) .where(ChannelSkuMapping.channel_code == channel_code) .where(ChannelSkuMapping.connection_id.is_(None)) .where(ChannelSkuMapping.channel_sku.in_(sorted(channel_sku_to_master))) ).all() for row in existing: winner = channel_sku_to_master.get(row.channel_sku) if winner and row.master_sku != winner: session.delete(row) reclaimed += 1 if reclaimed: session.flush() return reclaimed def _filter_weaker_channel_sku_claims( session: Session, rows: List[Dict[str, Any]], *, by_connection: bool, ) -> List[Dict[str, Any]]: """Drop incoming rows that would displace an equal-or-stronger existing channel_sku mapping.""" if not rows: return [] kept: List[Dict[str, Any]] = [] if by_connection: scoped: Dict[int, Dict[str, Dict[str, Any]]] = {} for row in rows: connection_id = _to_int(row.get("connection_id")) if not connection_id: continue scoped.setdefault(connection_id, {})[row["channel_sku"]] = row for connection_id, channel_sku_to_row in scoped.items(): existing_rows = session.scalars( select(ChannelSkuMapping) .where(ChannelSkuMapping.connection_id == connection_id) .where(ChannelSkuMapping.channel_sku.in_(sorted(channel_sku_to_row))) ).all() existing_by_sku = {row.channel_sku: row for row in existing_rows} for channel_sku, incoming in channel_sku_to_row.items(): existing = existing_by_sku.get(channel_sku) if ( existing and existing.master_sku != incoming["master_sku"] and _mapping_match_priority(existing.match_source) >= _mapping_match_priority(incoming.get("match_source")) ): continue kept.append(incoming) else: scoped_codes: Dict[str, Dict[str, Dict[str, Any]]] = {} for row in rows: scoped_codes.setdefault(row["channel_code"], {})[row["channel_sku"]] = row for channel_code, channel_sku_to_row in scoped_codes.items(): existing_rows = session.scalars( select(ChannelSkuMapping) .where(ChannelSkuMapping.channel_code == channel_code) .where(ChannelSkuMapping.connection_id.is_(None)) .where(ChannelSkuMapping.channel_sku.in_(sorted(channel_sku_to_row))) ).all() existing_by_sku = {row.channel_sku: row for row in existing_rows} for channel_sku, incoming in channel_sku_to_row.items(): existing = existing_by_sku.get(channel_sku) if ( existing and existing.master_sku != incoming["master_sku"] and _mapping_match_priority(existing.match_source) >= _mapping_match_priority(incoming.get("match_source")) ): continue kept.append(incoming) return kept def upsert_mappings(session: Session, rows: List[Dict[str, Any]]) -> int: if not rows: return 0 channel_rows = _dedupe_mapping_batch( [row for row in rows if not row.get("connection_id")], by_connection=False, ) connection_rows = _dedupe_mapping_batch( [row for row in rows if row.get("connection_id")], by_connection=True, ) channel_rows = _filter_weaker_channel_sku_claims(session, channel_rows, by_connection=False) connection_rows = _filter_weaker_channel_sku_claims(session, connection_rows, by_connection=True) total = 0 if channel_rows: _reclaim_conflicting_channel_skus(session, channel_rows, by_connection=False) stmt = insert(ChannelSkuMapping).values(channel_rows) stmt = stmt.on_conflict_do_update( index_elements=["master_sku", "channel_code"], index_where=ChannelSkuMapping.connection_id.is_(None), set_={ "channel_sku": stmt.excluded.channel_sku, "remote_id": sa.func.coalesce( stmt.excluded.remote_id, ChannelSkuMapping.remote_id, ), "mapping_status": stmt.excluded.mapping_status, "match_source": stmt.excluded.match_source, "notes": stmt.excluded.notes, "updated_at": sa.func.now(), }, ) session.execute(stmt) total += len(channel_rows) if connection_rows: _reclaim_conflicting_channel_skus(session, connection_rows, by_connection=True) stmt = insert(ChannelSkuMapping).values(connection_rows) stmt = stmt.on_conflict_do_update( index_elements=["master_sku", "connection_id"], index_where=ChannelSkuMapping.connection_id.is_not(None), set_={ "channel_code": stmt.excluded.channel_code, "channel_sku": stmt.excluded.channel_sku, "remote_id": sa.func.coalesce( stmt.excluded.remote_id, ChannelSkuMapping.remote_id, ), "mapping_status": stmt.excluded.mapping_status, "match_source": stmt.excluded.match_source, "notes": stmt.excluded.notes, "updated_at": sa.func.now(), }, ) session.execute(stmt) total += len(connection_rows) return total def upsert_single_mapping( session: Session, *, master_sku: str, channel_sku: str, channel_code: Optional[str] = None, connection_id: Optional[int] = None, remote_id: Optional[str] = None, mapping_status: str = ACTIVE_STATUS, match_source: str = "manual", notes: Optional[str] = None, ) -> Dict[str, Any]: master_sku = str(master_sku or "").strip() channel_sku = str(channel_sku or "").strip() if not master_sku or not channel_sku: raise ValueError("master_sku and channel_sku are required") resolved_channel = channel_code if connection_id and not resolved_channel: resolved_channel = pipeline_channel_for_connection(session, connection_id) if not resolved_channel: raise ValueError("channel_code or connection_id is required") row = { "master_sku": master_sku, "channel_code": resolve_pipeline_channel_code(resolved_channel), "connection_id": connection_id, "channel_sku": channel_sku, "remote_id": remote_id, "mapping_status": mapping_status, "match_source": match_source, "notes": notes, } upsert_mappings(session, [row]) return row def list_mappings( session: Session, *, channel_code: Optional[str] = None, connection_id: Optional[int] = None, master_sku: Optional[str] = None, status: Optional[str] = None, limit: Optional[int] = None, ) -> List[Dict[str, Any]]: stmt = select(ChannelSkuMapping).order_by( ChannelSkuMapping.channel_code, ChannelSkuMapping.connection_id, ChannelSkuMapping.master_sku, ) if channel_code: stmt = stmt.where(ChannelSkuMapping.channel_code == resolve_pipeline_channel_code(channel_code)) if connection_id is not None: stmt = stmt.where(ChannelSkuMapping.connection_id == connection_id) if master_sku: stmt = stmt.where(ChannelSkuMapping.master_sku == master_sku.strip()) if status: stmt = stmt.where(ChannelSkuMapping.mapping_status == status) if limit: stmt = stmt.limit(limit) return [ { "master_sku": row.master_sku, "channel_code": row.channel_code, "connection_id": row.connection_id, "channel_sku": row.channel_sku, "remote_id": row.remote_id, "mapping_status": row.mapping_status, "match_source": row.match_source, "notes": row.notes or "", "updated_at": row.updated_at.isoformat() if row.updated_at else "", } for row in session.scalars(stmt).all() ] def bootstrap_identity_mappings( session: Session, channel_code: str, *, only_assigned: bool = True, match_source: str = "auto_exact", ) -> Dict[str, int]: """Create 1:1 mappings where master_sku == channel_sku for active master products.""" channel = resolve_pipeline_channel_code(channel_code) stmt = select(MasterProduct.sku).where(MasterProduct.is_active.is_(True)) if only_assigned: assigned = active_skus_for_channel(session, channel) if not assigned: return {"candidates": 0, "upserted": 0} stmt = stmt.where(MasterProduct.sku.in_(sorted(assigned))) master_skus = [sku for (sku,) in session.execute(stmt).all()] existing = { master for (master,) in session.execute( select(ChannelSkuMapping.master_sku) .where(ChannelSkuMapping.channel_code == channel) .where(ChannelSkuMapping.master_sku.in_(master_skus)) ).all() } rows = [ { "master_sku": sku, "channel_code": channel, "channel_sku": sku, "mapping_status": ACTIVE_STATUS, "match_source": match_source, "notes": "identity bootstrap", } for sku in master_skus if sku not in existing ] return {"candidates": len(master_skus), "upserted": upsert_mappings(session, rows)} def bootstrap_from_magento_catalog( session: Session, connection_id: int, *, only_assigned: bool = True, ) -> Dict[str, int]: """Suggest mappings from pulled Magento catalog SKUs (exact + source_item matches).""" channel = "magento" stmt = select(MasterProduct.sku, MasterProduct.source_item).where(MasterProduct.is_active.is_(True)) if only_assigned: assigned = active_skus_for_channel(session, channel) if not assigned: return {"catalog_skus": 0, "upserted": 0} stmt = stmt.where(MasterProduct.sku.in_(sorted(assigned))) masters = list(session.execute(stmt).all()) catalog_skus = { str(sku).strip() for (sku,) in session.execute( select(MagentoCatalogState.sku).where(MagentoCatalogState.connection_id == connection_id) ).all() if str(sku).strip() } existing_masters = { master for (master,) in session.execute( select(ChannelSkuMapping.master_sku).where(ChannelSkuMapping.channel_code == channel) ).all() } rows: List[Dict[str, Any]] = [] seen_masters: Set[str] = set(existing_masters) for master_sku, source_item in masters: master_sku = str(master_sku).strip() if not master_sku or master_sku in seen_masters: continue if master_sku in catalog_skus: rows.append( _mapping_row(master_sku, channel, master_sku, "auto_magento_exact", "Magento catalog exact SKU match") ) seen_masters.add(master_sku) continue source_item = str(source_item or "").strip() if source_item and source_item in catalog_skus: rows.append( _mapping_row( master_sku, channel, source_item, "auto_magento_source_item", f"Matched master source_item to Magento SKU {source_item}", ) ) seen_masters.add(master_sku) return {"catalog_skus": len(catalog_skus), "upserted": upsert_mappings(session, rows)} def build_reconciliation_report( session: Session, channel_code: str, *, connection_id: Optional[int] = None, only_assigned: bool = True, remote_catalog: Optional[Dict[str, Optional[str]]] = None, remote_source: Optional[str] = None, ) -> Dict[str, Any]: """Compare master catalog, mappings, and remote catalog/snapshots for gap analysis. remote_catalog maps channel SKU → remote product id (id may be None) and overrides the built-in Magento/Plytix sources — used for live Shopify catalogs. """ channel = resolve_pipeline_channel_code(channel_code) master_stmt = select(MasterProduct.sku, MasterProduct.source_item, MasterProduct.name).where( MasterProduct.is_active.is_(True) ) if only_assigned: assigned = active_skus_for_channel(session, channel) master_stmt = master_stmt.where(MasterProduct.sku.in_(sorted(assigned)) if assigned else MasterProduct.sku == "") masters = [ {"master_sku": sku, "source_item": source_item, "name": name} for sku, source_item, name in session.execute(master_stmt).all() ] master_skus = [item["master_sku"] for item in masters] sku_map = mapping_by_master( session, channel, master_skus, connection_id=connection_id, ) active_maps = { row.master_sku: row for row in session.scalars( select(ChannelSkuMapping) .where(ChannelSkuMapping.channel_code == channel) .where( ChannelSkuMapping.connection_id == connection_id if connection_id is not None else ChannelSkuMapping.connection_id.is_(None) ) .where(ChannelSkuMapping.mapping_status == ACTIVE_STATUS) .where(ChannelSkuMapping.master_sku.in_(master_skus) if master_skus else sa.true()) ).all() } lookup_skus = set(master_skus) lookup_skus.update(str(value).strip() for value in sku_map.values() if str(value).strip()) lookup_skus.update(str(row.channel_sku).strip() for row in active_maps.values() if str(row.channel_sku).strip()) remote_ids: Dict[str, Optional[str]] = {} if remote_catalog is not None: remote_source = remote_source or "remote_catalog" remote_ids = { str(sku).strip(): remote_id for sku, remote_id in remote_catalog.items() if str(sku).strip() } elif channel == "magento" and connection_id: remote_source = "magento_catalog_state" if lookup_skus: remote_ids = { str(sku).strip(): (str(product_id) if product_id is not None else None) for sku, product_id in session.execute( select(MagentoCatalogState.sku, MagentoCatalogState.magento_product_id).where( MagentoCatalogState.connection_id == connection_id, MagentoCatalogState.sku.in_(sorted(lookup_skus)), ) ).all() if str(sku).strip() } else: remote_ids = {} elif channel == "plytix": remote_source = "product_snapshot" if lookup_skus: remote_ids = { str(sku).strip(): None for (sku,) in session.execute( select(ProductSnapshot.sku).where( ProductSnapshot.valid_to.is_(None), ProductSnapshot.sku.in_(sorted(lookup_skus)), ) ).all() if str(sku).strip() } else: remote_ids = {} else: remote_source = remote_source or "" remote_skus: Set[str] = set(remote_ids) effective_channel_skus = {str(v).strip() for v in sku_map.values() if str(v).strip()} mapped_channel_skus = set(effective_channel_skus) mapped_channel_skus.update(row.channel_sku for row in active_maps.values()) suggestions: List[Dict[str, Any]] = [] remote_id_backfill: List[Dict[str, Any]] = [] unmapped_masters: List[Dict[str, Any]] = [] mapped: List[Dict[str, Any]] = [] for item in masters: master_sku = item["master_sku"] effective_channel_sku = sku_map.get(master_sku, master_sku) existing = active_maps.get(master_sku) if existing: catalog_remote_id = remote_ids.get(existing.channel_sku) if remote_ids else None mapped.append( { "master_sku": master_sku, "channel_sku": existing.channel_sku, "effective_channel_sku": effective_channel_sku, "match_source": existing.match_source, "remote_id": existing.remote_id, "remote_present": existing.channel_sku in remote_skus if remote_skus else None, } ) if remote_ids and catalog_remote_id and not existing.remote_id: remote_id_backfill.append( { "master_sku": master_sku, "channel_code": channel, "channel_sku": existing.channel_sku, "remote_id": catalog_remote_id, "match_source": "remote_id_backfill", "reason": "Mapping exists; remote catalog has product id but remote_id was empty", } ) continue suggestion = _suggest_for_master( item, remote_skus, effective_channel_sku=effective_channel_sku, ) if suggestion: suggestion["channel_code"] = channel suggestion["effective_channel_sku"] = effective_channel_sku catalog_remote_id = remote_ids.get(suggestion["channel_sku"]) if catalog_remote_id: suggestion["remote_id"] = catalog_remote_id suggestions.append(suggestion) else: master_sku = str(item["master_sku"]).strip() effective = str(effective_channel_sku or master_sku).strip() source_item = str(item.get("source_item") or "").strip() in_remote_catalog = bool( remote_skus and ( master_sku in remote_skus or (effective and effective in remote_skus) or (source_item and source_item in remote_skus) ) ) unmapped_masters.append( { "master_sku": master_sku, "name": item.get("name"), "effective_channel_sku": effective_channel_sku, "in_remote_catalog": in_remote_catalog, } ) master_sku_set = {m["master_sku"] for m in masters} orphan_remote = ( sorted(remote_skus - mapped_channel_skus - master_sku_set) if remote_skus else [] ) return { "channel_code": channel, "remote_source": remote_source, "master_count": len(masters), "mapped_count": len(mapped), "suggestion_count": len(suggestions), "remote_id_backfill_count": len(remote_id_backfill), "unmapped_master_count": len(unmapped_masters), "orphan_remote_sku_count": len(orphan_remote), "prefix_rules_applied": effective_channel_skus != master_sku_set, "mapped": mapped[:500], "suggestions": suggestions[:500], "remote_id_backfill": remote_id_backfill[:500], "unmapped_masters": unmapped_masters[:500], "orphan_remote_skus": orphan_remote[:500], } def reconcile_connection_id_for_channel( channel: str, *, compat_connection_id: Optional[int] = None, native_connection_id: Optional[int] = None, ) -> Optional[int]: """Connection id for catalog lookup / per-connection mapping rows.""" if channel == "magento": return native_connection_id if channel == "shopify": return compat_connection_id return compat_connection_id def fetch_shopify_remote_catalog( session: Session, *, connection_id: Optional[int] = None, native_connection_id: Optional[int] = None, ) -> Dict[str, Optional[str]]: """Live SKU → product GID map from the Shopify Admin API.""" from db.compat_connections import decode_compat_connection_id from db.models import ShopifyConnection from shopify.catalog import fetch_remote_sku_map from shopify.connections import build_client, get_active_connection connection = None if native_connection_id is not None: connection = session.get(ShopifyConnection, native_connection_id) elif connection_id: channel_type, native_id = decode_compat_connection_id(connection_id) if channel_type == "shopify": connection = session.get(ShopifyConnection, native_id) if connection is None: connection = get_active_connection(session) if connection is None: raise ValueError("No Shopify connection available") if connection.status != "active": raise ValueError("Shopify connection is disabled (kill switch engaged)") client = build_client(connection) return fetch_remote_sku_map(client) def fetch_shopify_remote_catalog_for_readiness( session: Session, *, connection_id: Optional[int] = None, native_connection_id: Optional[int] = None, only_assigned: bool = True, ) -> Dict[str, Optional[str]]: """Lookup only assigned channel SKUs in Shopify (readiness must stay under proxy timeouts).""" from db.compat_connections import decode_compat_connection_id from db.models import ShopifyConnection from shopify.catalog import fetch_remote_sku_map_for_skus from shopify.connections import build_client, get_active_connection connection = None if native_connection_id is not None: connection = session.get(ShopifyConnection, native_connection_id) elif connection_id: channel_type, native_id = decode_compat_connection_id(connection_id) if channel_type == "shopify": connection = session.get(ShopifyConnection, native_id) if connection is None: connection = get_active_connection(session) if connection is None: raise ValueError("No Shopify connection available") if connection.status != "active": raise ValueError("Shopify connection is disabled (kill switch engaged)") channel = "shopify" if not only_assigned: client = build_client(connection) from shopify.catalog import fetch_remote_sku_map return fetch_remote_sku_map(client) assigned = active_skus_for_channel(session, channel) if not assigned: return {} sku_map = mapping_by_master(session, channel, sorted(assigned), connection_id=connection_id) channel_skus = sorted({str(value).strip() for value in sku_map.values() if str(value).strip()}) if not channel_skus: return {} client = build_client(connection) return fetch_remote_sku_map_for_skus(client, channel_skus) def auto_link_skus_after_pull( session: Session, *, channel_type: str, channel_code: str, compat_connection_id: Optional[int] = None, native_connection_id: Optional[int] = None, only_assigned: bool = True, ) -> Dict[str, Any]: """Match assigned master SKUs to pulled/live remote catalogs and persist links.""" channel = resolve_pipeline_channel_code(channel_code, channel_type=channel_type) if channel not in {"magento", "shopify", "plytix"}: return {"status": "skipped", "reason": f"unsupported_channel:{channel}"} reconcile_conn_id = reconcile_connection_id_for_channel( channel, compat_connection_id=compat_connection_id, native_connection_id=native_connection_id, ) try: with session.begin_nested(): remote_catalog = None remote_source = None if channel == "shopify": remote_catalog = fetch_shopify_remote_catalog( session, connection_id=compat_connection_id, native_connection_id=native_connection_id, ) remote_source = "shopify_admin_api" result = link_mappings_from_remote( session, channel, connection_id=reconcile_conn_id, only_assigned=only_assigned, remote_catalog=remote_catalog, remote_source=remote_source, dry_run=False, ) result["status"] = "ok" return result except Exception as exc: return {"status": "failed", "error": str(exc)} def link_mappings_from_remote( session: Session, channel_code: str, *, connection_id: Optional[int] = None, only_assigned: bool = True, remote_catalog: Optional[Dict[str, Optional[str]]] = None, remote_source: Optional[str] = None, dry_run: bool = False, include_remote_id_backfill: bool = True, ) -> Dict[str, Any]: """Bulk-persist SKU mappings (+ remote_id) from prefix-aware reconcile against a live catalog.""" report = build_reconciliation_report( session, channel_code, connection_id=connection_id, only_assigned=only_assigned, remote_catalog=remote_catalog, remote_source=remote_source, ) channel = report["channel_code"] candidates: List[Dict[str, Any]] = [] for item in report.get("suggestions") or []: candidates.append( { "master_sku": item["master_sku"], "channel_code": channel, "connection_id": connection_id, "channel_sku": item.get("channel_sku") or item.get("suggested_channel_sku"), "remote_id": _clean(item.get("remote_id")), "mapping_status": ACTIVE_STATUS, "match_source": item.get("match_source") or "link_from_remote", "notes": item.get("reason"), } ) if include_remote_id_backfill: for item in report.get("remote_id_backfill") or []: candidates.append( { "master_sku": item["master_sku"], "channel_code": channel, "connection_id": connection_id, "channel_sku": item["channel_sku"], "remote_id": _clean(item.get("remote_id")), "mapping_status": ACTIVE_STATUS, "match_source": item.get("match_source") or "remote_id_backfill", "notes": item.get("reason"), } ) if dry_run: return { "dry_run": True, "channel_code": channel, "connection_id": connection_id, "candidate_count": len(candidates), "candidates": candidates[:500], **{k: report[k] for k in ("master_count", "mapped_count", "suggestion_count", "unmapped_master_count")}, } upserted = upsert_mappings(session, candidates) return { "dry_run": False, "channel_code": channel, "connection_id": connection_id, "upserted": upserted, "candidate_count": len(candidates), **{k: report[k] for k in ("master_count", "mapped_count", "unmapped_master_count", "orphan_remote_sku_count")}, } def apply_suggestions( session: Session, suggestions: List[Dict[str, Any]], *, match_source: str = "reconciliation_accept", ) -> int: rows = [] for item in suggestions: master_sku = _clean(item.get("master_sku")) channel_code = _clean(item.get("channel_code")) channel_sku = _clean(item.get("channel_sku") or item.get("suggested_channel_sku")) if not master_sku or not channel_code or not channel_sku: continue rows.append( { "master_sku": master_sku, "channel_code": resolve_pipeline_channel_code(channel_code), "connection_id": _to_int(item.get("connection_id")), "channel_sku": channel_sku, "remote_id": _clean(item.get("remote_id")), "mapping_status": ACTIVE_STATUS, "match_source": item.get("match_source") or match_source, "notes": item.get("reason") or item.get("notes"), } ) return upsert_mappings(session, rows) def _suggest_for_master( item: Dict[str, Any], remote_skus: Set[str], *, effective_channel_sku: Optional[str] = None, ) -> Optional[Dict[str, Any]]: master_sku = str(item["master_sku"]).strip() effective = str(effective_channel_sku or master_sku).strip() if effective and effective in remote_skus and effective != master_sku: return { "master_sku": master_sku, "channel_sku": effective, "suggested_channel_sku": effective, "match_source": "suggest_prefix_channel_sku", "confidence": "high", "reason": f"Remote catalog contains prefix-mapped channel SKU ({effective})", } if master_sku in remote_skus: return { "master_sku": master_sku, "channel_sku": master_sku, "suggested_channel_sku": master_sku, "match_source": "suggest_exact_sku", "confidence": "high", "reason": "Remote catalog contains the same SKU as master", } if effective and effective in remote_skus: return { "master_sku": master_sku, "channel_sku": effective, "suggested_channel_sku": effective, "match_source": "suggest_effective_channel_sku", "confidence": "high", "reason": f"Remote catalog contains effective channel SKU ({effective})", } source_item = str(item.get("source_item") or "").strip() if source_item and source_item in remote_skus: return { "master_sku": master_sku, "channel_sku": source_item, "suggested_channel_sku": source_item, "match_source": "suggest_source_item", "confidence": "medium", "reason": f"Remote catalog SKU matches master source_item ({source_item})", } normalized_master = _normalize_sku_token(master_sku) if normalized_master: for remote in remote_skus: if _normalize_sku_token(remote) == normalized_master and remote != master_sku: return { "master_sku": master_sku, "channel_sku": remote, "suggested_channel_sku": remote, "match_source": "suggest_normalized", "confidence": "low", "reason": f"Normalized SKU token matches remote SKU {remote}", } return None def _mapping_row( master_sku: str, channel_code: str, channel_sku: str, match_source: str, notes: str, ) -> Dict[str, Any]: return { "master_sku": master_sku, "channel_code": channel_code, "channel_sku": channel_sku, "mapping_status": ACTIVE_STATUS, "match_source": match_source, "notes": notes, } def _normalize_sku_token(sku: str) -> str: return re.sub(r"[^a-z0-9]+", "", str(sku or "").strip().lower()) def _status(value: Any) -> str: text = str(value or "").strip().lower() if text in {"inactive", "disabled", "removed"}: return INACTIVE_STATUS if text in {"pending", "review"}: return PENDING_STATUS return ACTIVE_STATUS def _clean(value: Any) -> Optional[str]: if value is None: return None text = str(value).strip() if not text or text.lower() in {"nan", "none", "null"}: return None return text def _to_int(value: Any) -> Optional[int]: text = _clean(value) if text is None: return None try: return int(float(text)) except ValueError: return None