from __future__ import annotations import re from pathlib import PurePosixPath from typing import Any, Dict, Iterable, List, Optional, Sequence, Set, Tuple import pandas as pd from sqlalchemy import delete, func, select from sqlalchemy.dialects.postgresql import insert from sqlalchemy.orm import Session from db.image_urls import normalize_external_image_url from db.models import MasterProduct, MasterProductImage from db.source_imports import normalize_column_key from db.tribeca_sku_parse import ( collection_asset_key, expand_upload_reference_to_master_keys, master_style_codes_for_image_prefix, match_master_skus_for_catalog_image_item, parse_image_item_code, parse_master_raw_sku, resolve_collection_asset_skus, resolve_universal_diagram_skus, split_styled_sku, tribeca_image_matches_master, ) IMAGE_VIEW_SUFFIXES = frozenset( { "close", "closed", "open", "opened", "open1", "open2", "open3", "angle", "side", "front", "back", "detail", "left", "right", "top", "bottom", "interior", "exterior", "lifestyle", "swatch", "thumb", "thumbnail", } ) IMAGE_ROLE_UNKNOWN = "unknown" IMAGE_ROLE_DEFAULT = "default" IMAGE_ROLE_OPEN = "open" IMAGE_ROLE_CLOSED = "closed" IMAGE_ROLE_GENERIC = "generic" IMAGE_ROLE_COLLECTION_HERO = "collection_hero" IMAGE_ROLE_VIGNETTE = "vignette" IMAGE_ROLE_LIFESTYLE = "lifestyle" IMAGE_ROLE_SAMPLE_DOOR = "sample_door" IMAGE_ROLE_SAMPLE_SWATCH = "sample_swatch" IMAGE_ROLE_DIAGRAM = "diagram" IMAGE_ROLE_DETAIL = "detail" IMAGE_ROLE_ANGLE = "angle" PRODUCT_GALLERY_IMAGE_ROLES = frozenset( { IMAGE_ROLE_DEFAULT, IMAGE_ROLE_OPEN, IMAGE_ROLE_CLOSED, IMAGE_ROLE_GENERIC, IMAGE_ROLE_DETAIL, IMAGE_ROLE_ANGLE, IMAGE_ROLE_DIAGRAM, # SKU-owned diagram beats collection vignettes for base_image "front", "side", "back", "top", "bottom", "interior", "exterior", } ) COLLECTION_IMAGE_ROLES = frozenset({IMAGE_ROLE_COLLECTION_HERO, IMAGE_ROLE_VIGNETTE, IMAGE_ROLE_LIFESTYLE}) SAMPLE_IMAGE_ROLES = frozenset({IMAGE_ROLE_SAMPLE_DOOR, IMAGE_ROLE_SAMPLE_SWATCH}) ROLE_SORT_PRIORITY = { IMAGE_ROLE_SAMPLE_DOOR: -2, IMAGE_ROLE_DEFAULT: -1, "close": 0, IMAGE_ROLE_CLOSED: 0, "closed": 0, IMAGE_ROLE_OPEN: 1, "open": 1, IMAGE_ROLE_ANGLE: 5, "angle": 5, "front": 6, "side": 7, IMAGE_ROLE_DETAIL: 8, "detail": 8, IMAGE_ROLE_GENERIC: 10, IMAGE_ROLE_DIAGRAM: 20, IMAGE_ROLE_COLLECTION_HERO: 30, IMAGE_ROLE_VIGNETTE: 31, IMAGE_ROLE_LIFESTYLE: 32, IMAGE_ROLE_SAMPLE_SWATCH: 40, IMAGE_ROLE_UNKNOWN: 99, } VIEW_SORT_PRIORITY = { "close": 0, "closed": 0, "open": 1, "opened": 1, "open1": 2, "open2": 3, "open3": 4, "angle": 5, "front": 6, "side": 7, "detail": 8, } # Image filenames use compact codes (ASW-BBC42, ASW-BBC45FHL-MC, ASW-BEC1124L) while # master SKUs use width ranges with slashes (ASW-BBC39/42L/R, ASW-BEC1120/1124L/R). SLASH_RANGE_IMAGE_CODE_RE = re.compile( r"^(?P.+?)(?P\d+)(?P[A-Z]*)(?:-(?P[A-Z0-9-]+))?$", re.IGNORECASE, ) SLASH_RANGE_MASTER_SKU_RE = re.compile( r"^(?P.+?)(?P\d+)/(?P\d+)(?P[A-Z]*)/(?P.+)$", re.IGNORECASE, ) SKU_TOKEN_RE = re.compile(r"(? str: if value is None: return "" return str(value).strip() def strip_image_upload_noise(name: str) -> str: """Remove known export folder prefixes from image paths or filenames.""" text = _clean(name).replace("\\", "/") if not text: return "" for prefix in IMAGE_UPLOAD_NOISE_PREFIXES: marker = prefix.replace("\\", "/") if marker in text: text = text.split(marker, 1)[1] break return text def normalize_image_file_name(name: str) -> str: raw = strip_image_upload_noise(name) if not raw: return "" return PurePosixPath(raw).name def _first(normalized: Dict[str, str], *keys: str) -> str: for key in keys: val = _clean(normalized.get(normalize_column_key(key))) if val: return val return "" def _build_image_match_context(normalized: Dict[str, str]) -> str: parts: List[str] = [] seen: Set[str] = set() for key in ( "name", "file_name", "filename", "image_name", "image", "url", "image_url", "imageurl", "link", "key", "object_key", "path", "file_path", ): value = _clean(normalized.get(normalize_column_key(key))) if value and value not in seen: seen.add(value) parts.append(value) return " ".join(parts) def extract_view_suffix(file_name: str) -> str: stem = PurePosixPath(normalize_image_file_name(file_name)).stem if not stem or "-" not in stem: return "" suffix = stem.rsplit("-", 1)[-1].lower() return suffix if suffix in IMAGE_VIEW_SUFFIXES else "" def normalize_image_role(value: object) -> Optional[str]: text = _clean(value).strip().lower().replace(" ", "_") if not text: return None alias_map = { "collection_vignette": IMAGE_ROLE_VIGNETTE, "vignette_image": IMAGE_ROLE_VIGNETTE, "hero": IMAGE_ROLE_COLLECTION_HERO, "door": IMAGE_ROLE_SAMPLE_DOOR, "swatch": IMAGE_ROLE_SAMPLE_SWATCH, "sample": IMAGE_ROLE_SAMPLE_DOOR, "sample_door_4x4": IMAGE_ROLE_SAMPLE_SWATCH, "sample_swatch_4x4": IMAGE_ROLE_SAMPLE_SWATCH, "uni": IMAGE_ROLE_DIAGRAM, "universal": IMAGE_ROLE_DIAGRAM, "closed": IMAGE_ROLE_CLOSED, "open": IMAGE_ROLE_OPEN, "default": IMAGE_ROLE_DEFAULT, "generic": IMAGE_ROLE_GENERIC, "detail": IMAGE_ROLE_DETAIL, "angle": IMAGE_ROLE_ANGLE, } return alias_map.get(text, text) def derive_image_role( *, sku: str, file_name: str, explicit_role: object = None, view_suffix: object = None, ) -> str: role = normalize_image_role(explicit_role) if role: return role normalized_name = normalize_image_file_name(file_name).lower() stem = extract_image_stem(file_name).lower() suffix = _clean(view_suffix).lower() or extract_view_suffix(file_name) upper_sku = _clean(sku).upper() if "sample" in normalized_name and "swatch" in normalized_name: return IMAGE_ROLE_SAMPLE_SWATCH if "4x4" in normalized_name: return IMAGE_ROLE_SAMPLE_SWATCH if "sample" in normalized_name or upper_sku.endswith("-SD") or "-SDP" in upper_sku: return IMAGE_ROLE_SAMPLE_DOOR if "door" in normalized_name and (upper_sku.endswith("-SD") or "_door" in normalized_name or "-door" in normalized_name): return IMAGE_ROLE_SAMPLE_DOOR if collection_asset_key(file_name): if "hero" in normalized_name: return IMAGE_ROLE_COLLECTION_HERO if "vignette" in normalized_name: return IMAGE_ROLE_VIGNETTE if "lifestyle" in normalized_name or "room" in normalized_name: return IMAGE_ROLE_LIFESTYLE if "door" in normalized_name: return IMAGE_ROLE_SAMPLE_DOOR return IMAGE_ROLE_VIGNETTE if "_uni" in normalized_name or stem.endswith("_uni"): return IMAGE_ROLE_DIAGRAM if suffix in {"open", "open1", "open2", "open3", "opened"}: return IMAGE_ROLE_OPEN if suffix in {"close", "closed"}: return IMAGE_ROLE_CLOSED if suffix == "detail": return IMAGE_ROLE_DETAIL if suffix == "angle": return IMAGE_ROLE_ANGLE if suffix: return suffix return IMAGE_ROLE_DEFAULT if file_name else IMAGE_ROLE_UNKNOWN def extract_image_stem(file_name: str) -> str: return PurePosixPath(normalize_image_file_name(file_name)).stem.strip() def extract_image_match_candidates(file_name: str) -> List[str]: """Ordered SKU match keys: full filename stem, then stems with view suffixes stripped.""" stem = extract_image_stem(file_name) if not stem: return [] candidates: List[str] = [] seen: Set[str] = set() for seed in [stem, *[match.group(1) for match in SKU_TOKEN_RE.finditer(stem)]]: parts = seed.split("-") while parts: candidate = "-".join(parts).strip() if candidate and candidate not in seen: seen.add(candidate) candidates.append(candidate) if len(parts) >= 2 and parts[-1].lower() in IMAGE_VIEW_SUFFIXES: parts = parts[:-1] continue break return candidates def extract_sku_prefix_from_image_name(file_name: str) -> str: """Derive SKU prefix from image filename, e.g. CA-B15-CLOSE.png -> CA-B15.""" candidates = extract_image_match_candidates(file_name) if len(candidates) >= 2: return candidates[-1] if candidates: return candidates[0] return "" def extract_sku_from_image_name(file_name: str) -> str: """Backward-compatible alias — value is a prefix, not necessarily an exact master SKU.""" return extract_sku_prefix_from_image_name(file_name) def master_sku_matches_image_prefix(master_sku: str, image_prefix: str) -> bool: """True when filename prefix matches exact SKU or a longer master SKU (e.g. CA-B15 -> CA-B15XXXX).""" if not image_prefix or not master_sku: return False if master_sku == image_prefix: return True if master_sku.startswith(f"{image_prefix}-"): return True if len(master_sku) > len(image_prefix) and master_sku.startswith(image_prefix): tail = master_sku[len(image_prefix) :] # Glued suffix without hyphen, e.g. CA-B15 + XXXX. Require 2+ chars to avoid CA-B150 vs CA-B15. return len(tail) >= 2 and tail[0].isalnum() and not tail.startswith("-") return False def match_master_skus_for_image_prefix( image_prefix: str, known_skus: Set[str], ) -> List[str]: if not image_prefix: return [] matches = [sku for sku in known_skus if master_sku_matches_image_prefix(sku, image_prefix)] return sorted(matches, key=len) def parse_slash_range_image_code(image_key: str) -> Optional[Dict[str, Any]]: match = SLASH_RANGE_IMAGE_CODE_RE.match(_clean(image_key).upper()) if not match: return None return { "head": match.group("head").upper(), "size": int(match.group("size")), "variant": (match.group("variant") or "").upper(), "tag": (match.group("tag") or "").upper(), } def parse_slash_range_master_sku(master_sku: str) -> Optional[Dict[str, Any]]: match = SLASH_RANGE_MASTER_SKU_RE.match(_clean(master_sku).upper()) if not match: return None return { "head": match.group("head").upper(), "low": int(match.group("low")), "high": int(match.group("high")), "variant": (match.group("variant") or "").upper(), "rest": (match.group("rest") or "").upper(), } def master_sku_matches_slash_range_image_code(master_sku: str, image_key: str) -> bool: """Match compact image codes to slash-range master SKUs (any product family).""" image = parse_slash_range_image_code(image_key) master = parse_slash_range_master_sku(master_sku) if not image or not master: return False return _slash_range_parsed_match(image, master) def _slash_range_parsed_match(image: Dict[str, Any], master: Dict[str, Any]) -> bool: if image["head"] != master["head"]: return False if image["size"] not in (master["low"], master["high"]): return False image_variant = image["variant"] master_variant = master["variant"] image_tag = image["tag"] master_rest = master["rest"] if image_variant: if image_variant != master_variant: return False elif len(master_variant) > 1: return False if image_tag: return master_rest == image_tag or master_rest.endswith(f"-{image_tag}") return "-" not in master_rest def match_master_skus_for_slash_range_image(image_key: str, known_skus: Set[str]) -> List[str]: if not parse_slash_range_image_code(image_key): return [] matches = [ sku for sku in known_skus if master_sku_matches_slash_range_image_code(sku, image_key) ] return sorted(matches, key=len) class MasterSkuImageMatcher: """Indexed lookup for bulk image imports (avoids scanning all SKUs per row).""" def __init__( self, known_skus: Set[str], tribeca_by_style_type: Dict[Tuple[str, str], List[Tuple[str, Dict[str, Any]]]], slash_range_by_head: Dict[str, List[Tuple[str, Dict[str, Any]]]], ) -> None: self.known_skus = known_skus self.tribeca_by_style_type = tribeca_by_style_type self.slash_range_by_head = slash_range_by_head @classmethod def from_known_skus(cls, known_skus: Set[str]) -> "MasterSkuImageMatcher": tribeca_by_style_type: Dict[Tuple[str, str], List[Tuple[str, Dict[str, Any]]]] = {} slash_range_by_head: Dict[str, List[Tuple[str, Dict[str, Any]]]] = {} for sku in known_skus: style, raw = split_styled_sku(sku) tribeca_parsed = parse_master_raw_sku(raw) if style else None if style and tribeca_parsed: key = (style, tribeca_parsed["type_code"]) tribeca_by_style_type.setdefault(key, []).append((sku, tribeca_parsed)) slash_parsed = parse_slash_range_master_sku(sku) if slash_parsed: slash_range_by_head.setdefault(slash_parsed["head"], []).append((sku, slash_parsed)) return cls( known_skus=known_skus, tribeca_by_style_type=tribeca_by_style_type, slash_range_by_head=slash_range_by_head, ) def match_tribeca( self, image_key: str, *, allow_render_prefix: bool = True, context: Optional[str] = None, ) -> List[str]: style, item = split_styled_sku(image_key) if not style: return [] image_parsed = parse_image_item_code(item) if not image_parsed: return [] master_styles = ( master_style_codes_for_image_prefix(style, context=context or image_key) if allow_render_prefix else [style.upper()] ) matches: List[str] = [] for master_style in master_styles: for sku, master in self.tribeca_by_style_type.get((master_style, image_parsed["type_code"]), []): if tribeca_image_matches_master(image_parsed, master): matches.append(sku) return sorted(set(matches), key=len) def match_slash_range( self, image_key: str, *, allow_render_prefix: bool = True, context: Optional[str] = None, ) -> List[str]: tribeca_matches = self.match_tribeca( image_key, allow_render_prefix=allow_render_prefix, context=context, ) if tribeca_matches: return tribeca_matches image = parse_slash_range_image_code(image_key) if not image: return [] matches = [ sku for sku, master in self.slash_range_by_head.get(image["head"], []) if _slash_range_parsed_match(image, master) ] return sorted(matches, key=len) # Backward-compatible aliases (BBC-specific names from first iteration) parse_bbc_image_code = parse_slash_range_image_code parse_bbc_master_range_sku = parse_slash_range_master_sku master_sku_matches_bbc_image_code = master_sku_matches_slash_range_image_code match_master_skus_for_bbc_image = match_master_skus_for_slash_range_image def _match_master_keys_for_candidate( candidate: str, *, known_skus: Set[str], matcher: Optional[MasterSkuImageMatcher], prefer_exact_first: bool, allow_reverse_prefix: bool, context: Optional[str] = None, ) -> List[str]: context_text = str(context or "").upper() allow_contextual_render = "TB-STANDARD LINE-" in context_text if prefer_exact_first and candidate in known_skus: return [candidate] prefix_matches = match_master_skus_for_image_prefix(candidate, known_skus) if prefix_matches: return prefix_matches catalog_matches = match_master_skus_for_catalog_image_item( candidate, known_skus, allow_render_prefix=allow_reverse_prefix or allow_contextual_render, context=context, ) if catalog_matches: return catalog_matches if matcher: slash_range_matches = matcher.match_slash_range( candidate, allow_render_prefix=allow_reverse_prefix or allow_contextual_render, context=context, ) if slash_range_matches: return slash_range_matches if candidate in known_skus: return [candidate] return [] def expand_skus_with_rta_mirrors( skus: Sequence[str], known_skus: Optional[Set[str]] = None, ) -> List[str]: """Duplicate image targets to RTA-{sku} counterparts when those rows exist in master catalog.""" if not skus: return [] ordered: List[str] = [] seen: Set[str] = set() for sku in skus: text = str(sku or "").strip() if not text or text in seen: continue seen.add(text) ordered.append(text) if text.upper().startswith(RTA_MIRROR_PREFIX) or not known_skus: continue rta_sku = f"{RTA_MIRROR_PREFIX}{text}" if rta_sku in known_skus and rta_sku not in seen: seen.add(rta_sku) ordered.append(rta_sku) return ordered def _finalize_resolved_skus( matches: List[str], known_skus: Optional[Set[str]], *, include_rta_mirrors: bool = True, ) -> List[str]: if include_rta_mirrors: return expand_skus_with_rta_mirrors(matches, known_skus) ordered: List[str] = [] seen: Set[str] = set() for sku in matches: text = str(sku or "").strip() if text and text not in seen: seen.add(text) ordered.append(text) return ordered def resolve_skus_for_candidates( candidates: Sequence[str], *, known_skus: Optional[Set[str]] = None, matcher: Optional[MasterSkuImageMatcher] = None, channel_prefix_rules: Optional[Sequence[Tuple[str, str]]] = None, context: Optional[str] = None, ) -> List[str]: """Match upload references to master SKUs (master-first, reverse Tribeca/channel as fallback).""" if not candidates: return [] if not known_skus: return [candidates[0]] if matcher is None: matcher = MasterSkuImageMatcher.from_known_skus(known_skus) # Pass 1: treat reference as a master SKU (normal uploads). for index, candidate in enumerate(candidates): matches = _match_master_keys_for_candidate( candidate, known_skus=known_skus, matcher=matcher, prefer_exact_first=index == 0, allow_reverse_prefix=False, context=context, ) if matches: return matches # Pass 2: reverse Tribeca render prefixes (CW→HPW) and channel prefixes (HD-CW→HPW). rules = channel_prefix_rules if rules is None: from db.channel_sku_prefix_mapping import default_shopify_prefix_rule_pairs rules = default_shopify_prefix_rule_pairs() if rules: for candidate in candidates: for master_key in expand_upload_reference_to_master_keys( candidate, channel_prefix_rules=rules, context=context or candidate, ): if master_key == candidate.upper(): continue matches = _match_master_keys_for_candidate( master_key, known_skus=known_skus, matcher=matcher, prefer_exact_first=False, allow_reverse_prefix=True, context=context or candidate, ) if matches: return matches return [] def resolve_skus_from_row( normalized: Dict[str, str], *, known_skus: Optional[Set[str]] = None, matcher: Optional[MasterSkuImageMatcher] = None, channel_prefix_rules: Optional[Sequence[Tuple[str, str]]] = None, collection_by_sku: Optional[Dict[str, str]] = None, ) -> Tuple[List[str], str]: """Return (matched master skus, file_name). Master catalog is the target. Uploads normally reference master SKUs; Tribeca render (CW-…) or channel (HD-CW-…) references are reverse-mapped when direct master match fails. """ file_name = _first(normalized, "name", "file_name", "filename", "image_name", "image") match_context = _build_image_match_context(normalized) or file_name explicit_sku = _first(normalized, "sku", "master_sku", "item", "item_sku") if explicit_sku: explicit_candidates = [explicit_sku, *extract_image_match_candidates(explicit_sku)] deduped: List[str] = [] seen: Set[str] = set() for candidate in explicit_candidates: if candidate and candidate not in seen: seen.add(candidate) deduped.append(candidate) matches = resolve_skus_for_candidates( deduped, known_skus=known_skus, matcher=matcher, channel_prefix_rules=channel_prefix_rules, context=match_context or explicit_sku, ) if matches: return _finalize_resolved_skus(matches, known_skus), file_name if not known_skus and explicit_sku: return [explicit_sku], file_name return [], file_name candidates = extract_image_match_candidates(file_name) matches = resolve_skus_for_candidates( candidates, known_skus=known_skus, matcher=matcher, channel_prefix_rules=channel_prefix_rules, context=match_context, ) if matches: return _finalize_resolved_skus(matches, known_skus), file_name # Pass 3: collection vignette/door assets fan out to all SKUs for that style. if known_skus and file_name: collection_matches = resolve_collection_asset_skus( file_name, known_skus, collection_by_sku=collection_by_sku, include_rta=False, ) if collection_matches: return _finalize_resolved_skus( collection_matches, known_skus, include_rta_mirrors=False, ), file_name # Pass 4: universal line diagrams (B12_Base_..._Uni.png) — collection-agnostic. if known_skus and file_name: diagram_matches = resolve_universal_diagram_skus( file_name, known_skus, matcher=matcher, ) if diagram_matches: return _finalize_resolved_skus(diagram_matches, known_skus), file_name return [], file_name def resolve_sku_from_row( normalized: Dict[str, str], *, known_skus: Optional[Set[str]] = None, ) -> Tuple[str, str]: """Return first matched SKU (legacy helper).""" skus, file_name = resolve_skus_from_row(normalized, known_skus=known_skus) return (skus[0] if skus else ""), file_name def is_images_only_csv(df: pd.DataFrame) -> bool: """Detect Name+URL image sheets without master product columns.""" if df.empty: return False normalized_cols = {normalize_column_key(c) for c in df.columns} has_url = bool(normalized_cols & {"url", "image_url", "imageurl", "link"}) has_name = bool(normalized_cols & {"name", "file_name", "filename", "image_name", "image"}) has_sku = bool(normalized_cols & {"sku", "master_sku", "item"}) has_master_cols = bool( normalized_cols & { "display_name", "category", "product_category", "item_style", "manufacturer", "stocked", } ) return has_url and has_name and not has_master_cols and not has_sku def _image_sort_key(file_name: str, view_suffix: str, image_role: Optional[str] = None) -> Tuple[int, int, str]: suffix = view_suffix or extract_view_suffix(file_name) role = normalize_image_role(image_role) or IMAGE_ROLE_UNKNOWN return ( ROLE_SORT_PRIORITY.get(role, 99), VIEW_SORT_PRIORITY.get(suffix.lower(), 99), file_name.lower(), ) def build_image_canonical_fields(urls: Sequence[str]) -> Dict[str, str]: clean: List[str] = [] seen: Set[str] = set() for value in urls: url = _clean(value) if not url or url in seen: continue seen.add(url) clean.append(url) if not clean: return {} fields: Dict[str, str] = {"base_image": clean[0]} if len(clean) > 1: fields["additional_images"] = ",".join(clean[1:]) fields["image_urls"] = ",".join(clean) else: fields["image_urls"] = clean[0] return fields def build_image_canonical_fields_from_rows(images: Sequence[MasterProductImage]) -> Dict[str, str]: def _resolved_role(image: MasterProductImage) -> str: explicit = normalize_image_role(getattr(image, "image_role", None)) if explicit: return explicit return derive_image_role( sku=str(getattr(image, "sku", "") or ""), file_name=str(getattr(image, "file_name", "") or ""), view_suffix=getattr(image, "view_suffix", None), ) # Role priority first so SKU/packshot/diagram always beat vignettes even if # persisted sort_order still has collection assets at 0. ordered = sorted( images, key=lambda img: ( _image_sort_key( str(getattr(img, "file_name", "") or ""), str(getattr(img, "view_suffix", "") or ""), _resolved_role(img), ), int(getattr(img, "sort_order", 0) or 0), ), ) preferred: List[str] = [] fallback: List[str] = [] seen: Set[str] = set() for image in ordered: url = _clean(getattr(image, "image_url", None)) if not url or url in seen: continue seen.add(url) role = _resolved_role(image) if role in PRODUCT_GALLERY_IMAGE_ROLES or role in SAMPLE_IMAGE_ROLES or role == IMAGE_ROLE_DEFAULT: preferred.append(url) else: fallback.append(url) return build_image_canonical_fields([*preferred, *fallback]) def purge_collection_asset_rows( session: Session, skus: Sequence[str], ) -> int: """Remove existing collection lifestyle/vignette rows for merge-only SKUs. This lets vignette-only imports replace prior collection assets without wiping packshots, diagrams, or other SKU-specific renders already in the gallery. """ wanted = [str(sku).strip() for sku in skus if str(sku).strip()] if not wanted: return 0 rows = session.scalars( select(MasterProductImage).where(MasterProductImage.sku.in_(wanted)) ).all() ids = [ row.id for row in rows if collection_asset_key(row.file_name) ] if not ids: return 0 session.execute(delete(MasterProductImage).where(MasterProductImage.id.in_(ids))) return len(ids) def _dedupe_image_rows(rows: List[Dict[str, Any]]) -> Tuple[List[Dict[str, Any]], int]: """Collapse duplicate (product_id, file_name) rows — required for Postgres ON CONFLICT batch upsert.""" deduped: Dict[Tuple[int, str], Dict[str, Any]] = {} for row in rows: key = (int(row["product_id"]), str(row["file_name"])) deduped[key] = row removed = len(rows) - len(deduped) return list(deduped.values()), removed def import_master_product_images_dataframe( session: Session, df: pd.DataFrame, *, source_label: str = "images_csv", replace_skus: Optional[Iterable[str]] = None, ) -> Dict[str, Any]: from db.channel_sku_prefix_mapping import load_all_active_prefix_rules rows: List[Dict[str, Any]] = [] skipped = 0 unmatched: List[str] = [] missing_url = 0 missing_name = 0 products_by_sku = { row.sku: row for row in session.scalars(select(MasterProduct).where(MasterProduct.is_active.is_(True))).all() } known_skus = set(products_by_sku) collection_by_sku = { sku: str(product.collection or "").strip() for sku, product in products_by_sku.items() if str(product.collection or "").strip() } matcher = MasterSkuImageMatcher.from_known_skus(known_skus) channel_prefix_rules = load_all_active_prefix_rules(session) for raw in df.to_dict(orient="records"): normalized = {normalize_column_key(k): v for k, v in raw.items()} url = _first(normalized, "url", "image_url", "imageurl", "link") if not url: missing_url += 1 skipped += 1 continue matched_skus, file_name = resolve_skus_from_row( normalized, known_skus=known_skus, matcher=matcher, channel_prefix_rules=channel_prefix_rules, collection_by_sku=collection_by_sku, ) if not file_name: missing_name += 1 skipped += 1 continue if not matched_skus: skipped += 1 unmatched.append(file_name) continue view_suffix = extract_view_suffix(file_name) seen_skus: Set[str] = set() for sku in matched_skus: if sku in seen_skus: continue seen_skus.add(sku) product = products_by_sku.get(sku) if product is None: skipped += 1 unmatched.append(f"{sku}:{file_name}") continue rows.append( { "product_id": product.id, "sku": sku, "file_name": file_name, "image_url": normalize_external_image_url(url), "image_role": derive_image_role( sku=sku, file_name=file_name, explicit_role=normalized.get("image_role") or normalized.get("role") or normalized.get("slot_key"), view_suffix=view_suffix, ), "view_suffix": view_suffix or None, "source_label": source_label, } ) touched_skus = sorted({row["sku"] for row in rows}) if replace_skus is not None: replace_set = {str(s).strip() for s in replace_skus if str(s).strip()} if replace_set: session.execute( delete(MasterProductImage).where(MasterProductImage.sku.in_(sorted(replace_set))) ) elif touched_skus: session.execute(delete(MasterProductImage).where(MasterProductImage.sku.in_(touched_skus))) inserted = 0 duplicate_rows = 0 if rows: rows, duplicate_rows = _dedupe_image_rows(rows) rows.sort( key=lambda r: ( r["sku"], _image_sort_key( r["file_name"], r.get("view_suffix") or "", r.get("image_role"), ), _clean(r.get("image_role")), ) ) # Per-SKU gallery positions (0..n-1), not a global batch index. per_sku_index: Dict[str, int] = {} for row in rows: sku = str(row["sku"]) row["sort_order"] = per_sku_index.get(sku, 0) per_sku_index[sku] = row["sort_order"] + 1 stmt = insert(MasterProductImage).values(rows) stmt = stmt.on_conflict_do_update( constraint="uq_master_product_image_product_file", set_={ "image_url": stmt.excluded.image_url, "image_role": stmt.excluded.image_role, "view_suffix": stmt.excluded.view_suffix, "sort_order": stmt.excluded.sort_order, "source_label": stmt.excluded.source_label, "updated_at": func.now(), }, ) result = session.execute(stmt) inserted = result.rowcount or len(rows) skus_with_images = len({row["sku"] for row in rows}) return { "input_rows": len(df), "images_upserted": inserted, "skus_touched": skus_with_images, "skipped_rows": skipped, "missing_url_rows": missing_url, "missing_name_rows": missing_name, "unmatched_rows": len(unmatched), "duplicate_rows_collapsed": duplicate_rows, "sample_unmatched": unmatched[:20], "all_unmatched": unmatched, } def images_for_skus(session: Session, skus: Sequence[str]) -> Dict[str, List[MasterProductImage]]: wanted = [str(s).strip() for s in skus if str(s).strip()] if not wanted: return {} rows = session.scalars( select(MasterProductImage) .where(MasterProductImage.sku.in_(wanted)) .order_by(MasterProductImage.sku, MasterProductImage.sort_order, MasterProductImage.file_name) ).all() grouped: Dict[str, List[MasterProductImage]] = {} for row in rows: grouped.setdefault(row.sku, []).append(row) return grouped def image_fields_by_sku(session: Session, skus: Sequence[str]) -> Dict[str, Dict[str, str]]: grouped = images_for_skus(session, skus) return { sku: build_image_canonical_fields_from_rows(images) for sku, images in grouped.items() } def import_photo_assets_dataframe( session: Session, df: pd.DataFrame, *, source_label: str = "photo_assets_csv", fill_family_vignettes: bool = True, ) -> Dict[str, Any]: """Import wide Photo_Assets_By_SKU CSV into master_product_image. Product-shot rows replace that SKU's gallery; vignette/hero-only rows (including family fill) merge so existing renders are preserved. """ from db.photo_assets_import import ( is_photo_assets_by_sku_csv, long_records_to_dataframe, photo_assets_to_long_records, ) if not is_photo_assets_by_sku_csv(df): raise ValueError( "CSV does not look like Photo_Assets_By_SKU " "(need SKU + Image_Default/Open/Closed/Generic or Collection_* columns)" ) products_by_sku = { row.sku: row for row in session.scalars(select(MasterProduct).where(MasterProduct.is_active.is_(True))).all() } known_skus = set(products_by_sku) long_records, transform_stats = photo_assets_to_long_records( df, known_skus=known_skus, fill_family=fill_family_vignettes, ) long_df = long_records_to_dataframe(long_records) replace_skus = expand_skus_with_rta_mirrors( transform_stats.get("replace_skus") or [], known_skus, ) merge_skus = expand_skus_with_rta_mirrors( transform_stats.get("merge_skus") or [], known_skus, ) merge_rows_purged = purge_collection_asset_rows(session, merge_skus) # Always pass replace_skus (possibly empty) so vignette-only merges do not wipe galleries. import_stats = import_master_product_images_dataframe( session, long_df, source_label=source_label, replace_skus=replace_skus, ) # Drop large SKU lists from API response; keep counts. public_transform = { k: v for k, v in transform_stats.items() if k not in {"replace_skus", "merge_skus"} } return { "import_format": "photo_assets_by_sku", "fill_family_vignettes": bool(fill_family_vignettes), **public_transform, **import_stats, "merge_collection_rows_purged": int(merge_rows_purged), "input_rows": transform_stats.get("input_rows", len(df)), } def master_image_summary(session: Session) -> Dict[str, Any]: total_images = session.scalar(select(func.count()).select_from(MasterProductImage)) or 0 skus_with_images = session.scalar(select(func.count(func.distinct(MasterProductImage.sku)))) or 0 active_skus = session.scalar( select(func.count()).select_from(MasterProduct).where(MasterProduct.is_active.is_(True)) ) or 0 return { "total_images": int(total_images), "skus_with_images": int(skus_with_images), "active_master_skus": int(active_skus), "skus_without_images": max(0, int(active_skus) - int(skus_with_images)), }