#!/usr/bin/env python3 """ Phase 2 — Semantic Normalization / Identity Extraction Compliments Reference DB Pipeline Input: Phase 1 output (phase1_output.parquet) Output: Phase 2 output (phase2_output.parquet) This phase performs: 1. Brand normalization → brand_norm + product_line (uses brand_clean from Phase 1) 2. Identity attribute extraction from ORIGINAL title (BEFORE normalization) 3. Fat-level and fat-percentage extraction 4. Flavour extraction 5. Formulation extraction 6. Variant/size attribute parsing 7. Core-title extraction (strip brand prefix + size) CRITICAL: Identity information is extracted from the ORIGINAL title before any destructive normalization removes it. Phase 1 (data quality) has already cleaned: - brand → brand_clean (whitespace/case normalization) - title → title_clean (whitespace normalization) - upc → upc_raw (preserved, not altered) - external_id → external_id_raw (preserved, not altered) Phase 2 uses brand_clean for semantic normalization, and original title for identity extraction. """ import json import re import sys from datetime import datetime, timezone from pathlib import Path import pandas as pd import numpy as np # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- BASE_DIR = Path(__file__).resolve().parent.parent INPUT_PATH = BASE_DIR.parent / "phase1" / "outputs" / "phase1_output.parquet" OUTPUT_DIR = BASE_DIR / "outputs" VALIDATION_DIR = BASE_DIR / "validation" STATISTICS_DIR = BASE_DIR / "statistics" VERSION = "2.0.0" TIMESTAMP = datetime.now(timezone.utc).isoformat() def log(msg: str) -> None: print(f"[Phase2] {msg}") # --------------------------------------------------------------------------- # 1. Brand Normalization # --------------------------------------------------------------------------- # Maps 11 raw brand variants → (brand_norm, product_line) # Case-insensitive matching BRAND_MAP = { "compliments organic": ("Compliments", "Organic"), "compliments balance": ("Compliments", "Balance"), "compliments naturally simple": ("Compliments", "Naturally Simple"), "compliments green care": ("Compliments", "Green"), "compliments green": ("Compliments", "Green"), "compliments little ones": ("Compliments", "Little Ones"), "sensations": ("Sensations", "Sensations"), "compliments": ("Compliments", "Core"), "compliments ": ("Compliments", "Core"), # trailing space " compliments": ("Compliments", "Core"), # leading space } def normalize_brand(clean_brand: str) -> tuple[str, str]: """Map cleaned brand string to (brand_norm, product_line). Phase 1 has already cleaned the brand field (whitespace trim, case normalization). This function performs SEMANTIC normalization: mapping brand variants to canonical brand + product_line. """ if pd.isna(clean_brand): return ("Unknown", "Core") key = str(clean_brand).strip().lower() if key in BRAND_MAP: return BRAND_MAP[key] # Default: if it contains "compliments", treat as Core Compliments if "compliments" in key: return ("Compliments", "Core") if "sensations" in key: return ("Sensations", "Sensations") return (str(clean_brand).strip(), "Core") # --------------------------------------------------------------------------- # 2. Identity Attribute Extraction # --------------------------------------------------------------------------- # Extracted from the ORIGINAL title BEFORE any normalization def extract_identity_flags(title: str) -> dict: """Extract 9 boolean identity flags from raw product title.""" t = title.lower() return { "is_organic": bool(re.search(r"\borganic\b", t)), "is_gluten_free": bool(re.search(r"\bgluten[\s-]+free\b", t)), "is_naturally_simple": bool(re.search(r"\bnaturally\s+simple\b", t)), "is_sugar_free": bool(re.search( r"\bsugar[\s-]+free\b|\b(?:no sugar added|unsweetened)\b|\bzero\s+sugar\b", t)), "is_unsalted": bool(re.search(r"\bunsalted\b|\bno salt\b", t)), "is_lactose_free": bool(re.search(r"\blactose[\s-]+free\b", t)), "is_peanut_free": bool(re.search(r"\bpeanut[\s-]+free\b", t)), "is_plant_based": bool(re.search(r"\bplant[\s-]*based\b", t)), "is_reduced_sodium": bool(re.search( r"\breduced\s+sodium\b|\blow\s+sodium\b|\bno\s+salt\s+added\b", t)), } # --------------------------------------------------------------------------- # 3. Fat Level and Fat Percentage Extraction # --------------------------------------------------------------------------- def extract_fat_info(title: str) -> dict: """Extract fat_level and fat_percentage from raw product title.""" t = title.lower() # Fat level fat_level = "regular" if re.search(r"\b(?:light|lite|reduced fat|low fat|lean)\b", t): fat_level = "reduced_fat" elif re.search(r"\bfat[\s-]+free\b", t): fat_level = "fat_free" # Fat percentage — extract numeric % from title fat_percentage = None pct_match = re.search(r"(\d+(?:\.\d+)?)\s*%", title) if pct_match: val = float(pct_match.group(1)) # Skip values that are clearly not fat percentages # (cocoa %, alcohol %, etc.) skip_context = any(w in t for w in ["cocoa", "alcohol", "isopropyl", "peanuts"]) if val < 100 and not skip_context: fat_percentage = val # Override fat_level if percentage is very low if 0 < val <= 0.7: fat_level = "fat_free" return { "fat_level": fat_level, "fat_percentage": fat_percentage, } # --------------------------------------------------------------------------- # 4. Flavour Extraction # --------------------------------------------------------------------------- FLAVOUR_KEYWORDS = [ "almond", "apple", "banana", "blueberry", "caramel", "cherry", "chocolate", "cinnamon", "coconut", "cranberry", "honey", "lemon", "lime", "mango", "maple", "mixed berry", "peach", "peanut", "peppermint", "pineapple", "pomegranate", "raspberry", "strawberry", "tropical", "vanilla", "watermelon", "white chocolate", "berry", "espresso", "caramel", "butterscotch", "toffee", ] def extract_flavours(title: str) -> list[str]: """Extract flavour keywords from raw product title.""" t = title.lower() found = [] for kw in FLAVOUR_KEYWORDS: if kw in t: found.append(kw) return sorted(set(found)) # --------------------------------------------------------------------------- # 5. Formulation Extraction # --------------------------------------------------------------------------- FORMULATION_KEYWORDS = [ "smooth", "crunchy", "creamy", "chunky", "whole", "halves", "sliced", "ground", "chopped", "breaded", "fresh", "frozen", "roasted", "smoked", ] def extract_formulation(title: str) -> list[str]: """Extract formulation keywords from raw product title.""" t = title.lower() found = [] for kw in FORMULATION_KEYWORDS: if kw in t: found.append(kw) return sorted(set(found)) # --------------------------------------------------------------------------- # 6. Variant Attributes Parsing # --------------------------------------------------------------------------- def build_variant_attributes(size_str: str) -> dict: """Parse raw size string into structured variant attributes.""" if not size_str or pd.isna(size_str): return {} s = str(size_str).strip() attrs = {} # Try to match: [qty x ] amount unit [x multiplier] m = re.match( r"(\d+(?:\.\d+)?)\s*x\s*" # qty x r"(\d+(?:\.\d+)?)\s*" # amount r"(g|kg|ml|l|oz|lb)s?\s*$", # unit s, re.IGNORECASE ) if m: attrs["qty"] = int(float(m.group(1))) attrs["amount"] = float(m.group(2)) attrs["unit"] = m.group(3).lower() return attrs # Try: amount unit [x multiplier] m = re.match( r"(\d+(?:\.\d+)?)\s*" # amount r"(g|kg|ml|l|oz|lb)s?\s*" # unit r"(?:x\s*(\d+))?\s*$", # optional x multiplier s, re.IGNORECASE ) if m: attrs["amount"] = float(m.group(1)) attrs["unit"] = m.group(2).lower() if m.group(3): attrs["multiplier"] = int(m.group(3)) return attrs # Try: count-based (e.g. "12 per pack", "20 count") m = re.match( r"(\d+)\s*(?:per\s+pack|count|ea|pack|piece|slice|cups?|pound)s?\s*$", s, re.IGNORECASE ) if m: attrs["count"] = int(m.group(1)) return attrs # Fallback: store raw string attrs["raw"] = s return attrs # --------------------------------------------------------------------------- # 7. Core Title Extraction # --------------------------------------------------------------------------- # Brand/product-line prefixes to strip from title BRAND_PREFIXES = [ ("Compliments Naturally Simple ", "Compliments "), ("Compliments Balance ", "Compliments Balance "), ("Compliments Organic ", "Compliments Organic "), ("Compliments Green Care ", "Compliments Green "), ("Compliments Little Ones ", "Compliments Little Ones "), ("Compliments ", ""), ("Sensations ", ""), ] # Trailing size/weight/count patterns to strip SIZE_PATTERNS = [ r"\s+\d[\d,.]*\s*(?:g|kg|ml|l|oz|lb|count|ea|pack|piece|slice|cups?|litre|liters?|pound)s?\s*$", r"\s+\d+\s*x\s+\d+\s*(?:g|kg|ml|l)\s*$", r"\s+\d+\s*x\s*$", r"\s+\d+\s+per\s+pack\s*$", r"\s+\d+\s+count\s*$", ] # Bracket content to strip BRACKET_PATTERN = r"\s*\([^)]*\)\s*" def extract_core_title(title: str, product_line: str) -> str: """Extract core product title by stripping brand prefix and size info.""" t = title.strip() # Step 1: Strip brand/product-line prefix for prefix, replacement in BRAND_PREFIXES: if t.lower().startswith(prefix.lower()): t = replacement + t[len(prefix):] break # Step 2: Strip trailing size/weight/count patterns for pat in SIZE_PATTERNS: t = re.sub(pat, "", t, flags=re.IGNORECASE) # Step 3: Strip bracket content t = re.sub(BRACKET_PATTERN, " ", t) # Step 4: Clean up whitespace t = re.sub(r"\s+", " ", t).strip() return t # --------------------------------------------------------------------------- # 8. Identity Hash (for reference — grouping happens in Phase 3) # --------------------------------------------------------------------------- def build_identity_hash(row: dict) -> str: """Build deterministic identity hash from extracted attributes.""" parts = [] for col in ["is_organic", "is_gluten_free", "is_naturally_simple", "is_sugar_free", "is_unsalted", "is_lactose_free", "is_peanut_free", "is_plant_based", "is_reduced_sodium"]: parts.append("1" if row.get(col) else "0") fp = row.get("fat_percentage") if pd.notna(fp) and fp is not None: parts.append(f"fat{fp:.2g}" if fp else "fat0") else: parts.append("fat_none") parts.append(str(row.get("fat_level", "regular"))) parts.append(str(row.get("product_line", "Core"))) fl = row.get("flavour", []) parts.append(",".join(sorted(fl)) if fl else "") fm = row.get("formulation", []) parts.append(",".join(sorted(fm)) if fm else "") return "|".join(parts) # --------------------------------------------------------------------------- # Main Processing # --------------------------------------------------------------------------- def process_phase2(df: pd.DataFrame) -> pd.DataFrame: """Apply all Phase 2 transformations.""" log(f"Processing {len(df)} products ...") # CRITICAL: Extract identity from ORIGINAL title BEFORE any normalization log("Step 1: Extracting identity flags from original titles ...") identity_flags = df["title"].apply(extract_identity_flags) id_df = pd.DataFrame(identity_flags.tolist()) for col in id_df.columns: df[col] = id_df[col].values log("Step 2: Extracting fat info ...") fat_info = df["title"].apply(extract_fat_info) fat_df = pd.DataFrame(fat_info.tolist()) df["fat_level"] = fat_df["fat_level"].values df["fat_percentage"] = fat_df["fat_percentage"].values log("Step 3: Extracting flavours ...") df["flavour"] = df["title"].apply(extract_flavours) log("Step 4: Extracting formulations ...") df["formulation"] = df["title"].apply(extract_formulation) log("Step 5: Normalizing brands (using brand_clean from Phase 1) ...") brand_col = "brand_clean" if "brand_clean" in df.columns else "brand" brand_results = df[brand_col].apply(normalize_brand) df["brand_norm"] = brand_results.apply(lambda x: x[0]) df["product_line"] = brand_results.apply(lambda x: x[1]) log("Step 6: Parsing variant attributes ...") df["variant_attributes"] = df["size"].apply( lambda x: json.dumps(build_variant_attributes(x)) if pd.notna(x) else "{}" ) log("Step 7: Extracting core titles ...") df["core_title"] = df.apply( lambda row: extract_core_title(row["title"], row["product_line"]), axis=1 ) log("Step 8: Building identity hash ...") df["identity_hash"] = df.apply(lambda row: build_identity_hash(row.to_dict()), axis=1) log(f"Processing complete. Output shape: {df.shape}") return df # --------------------------------------------------------------------------- # Validation # --------------------------------------------------------------------------- def validate_phase2(df_in: pd.DataFrame, df_out: pd.DataFrame) -> dict: """Validate Phase 2 output.""" checks = {} # Row count preserved checks["row_count"] = { "input": len(df_in), "output": len(df_out), "pass": len(df_in) == len(df_out), } # All original columns preserved original_cols = set(df_in.columns) output_cols = set(df_out.columns) missing_original = original_cols - output_cols checks["original_columns_preserved"] = { "missing": list(missing_original), "pass": len(missing_original) == 0, } # New columns added new_cols = output_cols - original_cols expected_new = { "brand_norm", "product_line", "is_organic", "is_gluten_free", "is_naturally_simple", "is_sugar_free", "is_unsalted", "is_lactose_free", "is_peanut_free", "is_plant_based", "is_reduced_sodium", "fat_level", "fat_percentage", "flavour", "formulation", "variant_attributes", "core_title", "identity_hash", } missing_new = expected_new - new_cols checks["new_columns_added"] = { "expected": sorted(expected_new), "actual": sorted(new_cols), "missing": list(missing_new), "pass": len(missing_new) == 0, } # Brand normalization checks["brand_norm"] = { "unique_values": sorted(df_out["brand_norm"].unique().tolist()), "pass": True, } # Product line checks["product_line"] = { "unique_values": sorted(df_out["product_line"].unique().tolist()), "pass": True, } # Fat level values valid_fat_levels = {"fat_free", "reduced_fat", "regular"} actual_fat_levels = set(df_out["fat_level"].unique()) invalid_fat = actual_fat_levels - valid_fat_levels checks["fat_level"] = { "valid_values": sorted(valid_fat_levels), "actual_values": sorted(actual_fat_levels), "invalid": sorted(invalid_fat), "pass": len(invalid_fat) == 0, } # Identity flags are boolean flag_cols = ["is_organic", "is_gluten_free", "is_naturally_simple", "is_sugar_free", "is_unsalted", "is_lactose_free", "is_peanut_free", "is_plant_based", "is_reduced_sodium"] all_bool = all(df_out[col].dtype == bool for col in flag_cols) checks["identity_flags_boolean"] = {"pass": all_bool} # core_title not empty empty_core = (df_out["core_title"].str.strip() == "").sum() checks["core_title"] = { "empty_count": int(empty_core), "pass": empty_core == 0, } # identity_hash not empty empty_hash = (df_out["identity_hash"].str.strip() == "").sum() checks["identity_hash"] = { "empty_count": int(empty_hash), "pass": empty_hash == 0, } # No duplicate external_ids ext_dupes = df_out["external_id"].duplicated().sum() checks["external_id_uniqueness"] = { "duplicate_count": int(ext_dupes), "pass": ext_dupes == 0, } # Overall all_pass = all(c.get("pass", True) for c in checks.values()) checks["overall"] = {"result": "PASS" if all_pass else "FAIL"} return checks # --------------------------------------------------------------------------- # Statistics # --------------------------------------------------------------------------- def build_statistics(df_in: pd.DataFrame, df_out: pd.DataFrame) -> dict: """Build Phase 2 statistics.""" return { "version": VERSION, "timestamp": TIMESTAMP, "input": { "source": "phase1_output.parquet", "row_count": len(df_in), "column_count": len(df_in.columns), }, "output": { "row_count": len(df_out), "column_count": len(df_out.columns), "columns_added": sorted(set(df_out.columns) - set(df_in.columns)), }, "brand_normalization": { "input_brand_column": "brand_clean" if "brand_clean" in df_in.columns else "brand", "raw_brands": {str(k): int(v) for k, v in df_in.get("brand_clean", df_in["brand"]).value_counts().items()}, "normalized_brands": {str(k): int(v) for k, v in df_out["brand_norm"].value_counts().items()}, "product_lines": {str(k): int(v) for k, v in df_out["product_line"].value_counts().items()}, }, "identity_flags": { col: int(df_out[col].sum()) for col in [ "is_organic", "is_gluten_free", "is_naturally_simple", "is_sugar_free", "is_unsalted", "is_lactose_free", "is_peanut_free", "is_plant_based", "is_reduced_sodium", ] }, "fat_info": { "fat_level_distribution": {str(k): int(v) for k, v in df_out["fat_level"].value_counts().items()}, "fat_percentage_nulls": int(df_out["fat_percentage"].isna().sum()), "fat_percentage_non_null": int(df_out["fat_percentage"].notna().sum()), }, "flavour": { "products_with_flavour": int((df_out["flavour"].apply(len) > 0).sum()), "products_without_flavour": int((df_out["flavour"].apply(len) == 0).sum()), }, "formulation": { "products_with_formulation": int((df_out["formulation"].apply(len) > 0).sum()), "products_without_formulation": int((df_out["formulation"].apply(len) == 0).sum()), }, "core_title": { "unique_count": int(df_out["core_title"].nunique()), }, "identity_hash": { "unique_count": int(df_out["identity_hash"].nunique()), }, } # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main(): log("Starting Phase 2") # Ensure output dirs exist OUTPUT_DIR.mkdir(parents=True, exist_ok=True) VALIDATION_DIR.mkdir(parents=True, exist_ok=True) STATISTICS_DIR.mkdir(parents=True, exist_ok=True) # Load Phase 1 output log(f"Loading Phase 1 output from: {INPUT_PATH}") df_in = pd.read_parquet(INPUT_PATH) log(f"Loaded: {df_in.shape[0]} rows, {df_in.shape[1]} columns") # Process df_out = process_phase2(df_in.copy()) # Validate log("Validating output ...") validation = validate_phase2(df_in, df_out) # Statistics log("Building statistics ...") statistics = build_statistics(df_in, df_out) # Save outputs log("Saving outputs ...") df_out.to_parquet(OUTPUT_DIR / "phase2_output.parquet", index=False) log(f" Saved phase2_output.parquet ({df_out.shape[0]} rows, {df_out.shape[1]} cols)") with open(VALIDATION_DIR / "phase2_validation.json", "w") as f: json.dump(validation, f, indent=2, default=str) log(" Saved phase2_validation.json") with open(STATISTICS_DIR / "phase2_statistics.json", "w") as f: json.dump(statistics, f, indent=2, default=str) log(" Saved phase2_statistics.json") # Summary log("") log("=== PHASE 2 COMPLETE ===") log(f"Input: {df_in.shape[0]} rows, {df_in.shape[1]} columns") log(f"Output: {df_out.shape[0]} rows, {df_out.shape[1]} columns") log(f"Columns added: {sorted(set(df_out.columns) - set(df_in.columns))}") log(f"Validation: {validation['overall']['result']}") log("========================") return df_out, validation, statistics if __name__ == "__main__": main()