"""Parse a Product CSV into staged rows without writing catalog records.""" from __future__ import annotations from collections import Counter from dataclasses import dataclass, field from typing import Any, Iterable import pandas as pd from db.v2_import.matching import attribute_code_from_header, display_lookup_value, normalize_lookup_value PRODUCT_FIELD_ALIASES: dict[str, tuple[str, ...]] = { "sku": ("sku", "item sku", "product sku", "item_sku", "product_sku"), "title": ("title", "name", "product name", "product title", "product_name"), "description": ("description", "product description", "long description"), "product_type": ("product type", "product_type", "type"), "lifecycle_status": ("lifecycle status", "lifecycle_status", "status", "product status"), "brand": ("brand", "brand name", "manufacturer", "brand_name"), "hub": ("hub", "category l1", "category_l1", "l1", "department"), "category": ("category", "category l2", "category_l2", "l2", "primary category"), "collection": ("collection", "master collection", "product collection", "series", "series name"), } CORE_FIELD_ROLES = frozenset(PRODUCT_FIELD_ALIASES) LOCKED_LOOKUP_ROLES = ("brand", "hub", "category", "collection") def _normalize_header(value: object) -> str: return normalize_lookup_value(value).replace("_", " ").replace("-", " ") def _header_role_map(columns: Iterable[object]) -> dict[str, str]: role_by_header: dict[str, str] = {} used_roles: set[str] = set() normalized_to_original = {_normalize_header(col): str(col) for col in columns} for role, aliases in PRODUCT_FIELD_ALIASES.items(): for alias in aliases: original = normalized_to_original.get(_normalize_header(alias)) if original and role not in used_roles: role_by_header[original] = role used_roles.add(role) break return role_by_header def _cell_text(value: object) -> str: if value is None: return "" try: if pd.isna(value): return "" except Exception: pass text = str(value).strip() if text.lower() in {"nan", "none", "null"}: return "" return display_lookup_value(text) def _split_category_blob(value: str) -> dict[str, str]: parts = [display_lookup_value(part) for part in str(value or "").split(":") if display_lookup_value(part)] if not parts: return {} if len(parts) == 1: return {"category": parts[0]} if len(parts) == 2: return {"hub": parts[0], "category": parts[1]} return {"hub": parts[0], "brand": parts[1], "category": parts[-1]} @dataclass class ParsedImportRow: row_number: int raw: dict[str, Any] sku: str title: str description: str product_type: str lifecycle_status: str brand: str hub: str category: str collection: str attributes: dict[str, str] = field(default_factory=dict) error: str | None = None def as_normalized(self) -> dict[str, Any]: return { "sku": self.sku, "title": self.title, "description": self.description, "product_type": self.product_type or "simple", "lifecycle_status": self.lifecycle_status or "draft", "brand": self.brand, "hub": self.hub, "category": self.category, "collection": self.collection, "attributes": self.attributes, "error": self.error, } @dataclass class ParsedImportFile: columns: list[str] rows: list[ParsedImportRow] attribute_headers: list[str] missing_required: list[str] def parse_product_csv_dataframe(df: pd.DataFrame) -> ParsedImportFile: columns = [str(col) for col in df.columns] role_by_header = _header_role_map(columns) roles = set(role_by_header.values()) missing_required = [] if "sku" in roles else ["sku"] attribute_headers = [col for col in columns if col not in role_by_header] header_by_role = {role: header for header, role in role_by_header.items()} parsed_rows: list[ParsedImportRow] = [] for index, series in df.iterrows(): raw = {str(col): _cell_text(series.get(col)) for col in columns} fields = {role: _cell_text(series.get(header)) for role, header in header_by_role.items()} category_blob = fields.get("category") or "" if ":" in category_blob: split = _split_category_blob(category_blob) if split.get("hub") and not fields.get("hub"): fields["hub"] = split["hub"] if split.get("brand") and not fields.get("brand"): fields["brand"] = split["brand"] if split.get("category"): fields["category"] = split["category"] sku = fields.get("sku") or "" title = fields.get("title") or sku attributes = { header: _cell_text(series.get(header)) for header in attribute_headers if _cell_text(series.get(header)) } error = None if not sku: error = "missing_sku" parsed_rows.append( ParsedImportRow( row_number=int(index) + 2, raw=raw, sku=sku, title=title, description=fields.get("description") or "", product_type=(fields.get("product_type") or "simple").strip().lower() or "simple", lifecycle_status=_lifecycle(fields.get("lifecycle_status")), brand=fields.get("brand") or "", hub=fields.get("hub") or "", category=fields.get("category") or "", collection=fields.get("collection") or "", attributes=attributes, error=error, ) ) return ParsedImportFile( columns=columns, rows=parsed_rows, attribute_headers=attribute_headers, missing_required=missing_required, ) def unique_lookup_counts(rows: list[ParsedImportRow], role: str) -> Counter[str]: counts: Counter[str] = Counter() for row in rows: if row.error: continue value = display_lookup_value(getattr(row, role, "") or "") if value: counts[value] += 1 return counts def unique_attribute_headers(rows: list[ParsedImportRow]) -> Counter[str]: counts: Counter[str] = Counter() for row in rows: if row.error: continue for header in row.attributes: counts[header] += 1 return counts def unique_attribute_values(rows: list[ParsedImportRow], header: str) -> Counter[str]: counts: Counter[str] = Counter() for row in rows: if row.error: continue value = display_lookup_value(row.attributes.get(header) or "") if value: counts[value] += 1 return counts def _lifecycle(value: str | None) -> str: text = normalize_lookup_value(value) aliases = { "active": "active", "enabled": "active", "online": "active", "draft": "draft", "pending": "draft", "disabled": "disabled", "inactive": "disabled", "archived": "archived", } return aliases.get(text, "draft")