src/phase3.py

#!/usr/bin/env python3
"""
Phase 3 — Product Domain + Taxonomy + Product Grouping
Compliments Reference DB Pipeline

Input: Phase 2 output (phase2_output.parquet)
Output: Reference DB Catalog + Product Group Mapping

This phase performs:
  A. Food / Non-Food classification (product_domain)
  B. Taxonomy classification (metadata)
  C. Identity-based deterministic product grouping
  D. Ambiguous-case detection
  E. Rule-based resolution
  F. Validation
  G. Reference DB Catalog + Product Group Mapping outputs
"""

import json
import re
import uuid
import sys
from datetime import datetime, timezone
from pathlib import Path
from collections import Counter

import pandas as pd
import numpy as np

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
BASE_DIR = Path(__file__).resolve().parent.parent
INPUT_PATH = BASE_DIR.parent / "phase2" / "outputs" / "phase2_output.parquet"
OUTPUT_DIR = BASE_DIR / "outputs"
VALIDATION_DIR = BASE_DIR / "validation"
STATISTICS_DIR = BASE_DIR / "statistics"

VERSION = "1.1.0"
TIMESTAMP = datetime.now(timezone.utc).isoformat()


def log(msg: str) -> None:
    print(f"[Phase3] {msg}")


# ===========================================================================
# A. FOOD / NON-FOOD CLASSIFICATION
# ===========================================================================
# Deterministic rules based on product title keywords.
# Each rule is a (pattern, domain) pair.
# First match wins. If no match → "unknown".

FOOD_NON_FOOD_RULES = [
    # ---- NON-FOOD: Pet food ----
    (r"\bcat food\b", "non_food"),
    (r"\bdog food\b", "non_food"),
    (r"\bpet food\b", "non_food"),
    (r"\bcat treat", "non_food"),
    (r"\bdog treat", "non_food"),
    (r"\bpuppy\b.*\bfood\b", "non_food"),
    (r"\bkitten\b.*\bfood\b", "non_food"),

    # ---- NON-FOOD: Medications / Health ----
    (r"\bibuprofen\b", "non_food"),
    (r"\bacetaminophen\b", "non_food"),
    (r"\bmelatonin\b", "non_food"),
    (r"\b allergy remedy\b", "non_food"),
    (r"\bcold medication\b", "non_food"),
    (r"\bcold medicine\b", "non_food"),
    (r"\bcough\b.*\b(relief|lozenge|drop|medicine|medication)\b", "non_food"),
    (r"\bsinus\b.*\b(medication|medicine|relief|caplet)\b", "non_food"),
    (r"\bflu\b.*\b(cough|cold|relief|medicine)\b", "non_food"),
    (r"\bheadache\b.*\brelief\b", "non_food"),
    (r"\bpain relief\b", "non_food"),
    (r"\bfirst aid\b", "non_food"),
    (r"\bhydrogen peroxide\b", "non_food"),
    (r"\bhydrocortisone\b", "non_food"),
    (r"\bclotrimazole\b", "non_food"),
    (r"\bdiphenhydramine\b", "non_food"),
    (r"\bantibiotic ointment\b", "non_food"),
    (r"\bantifungal\b", "non_food"),
    (r"\bcalamine lotion\b", "non_food"),
    (r"\banti-itch\b", "non_food"),
    (r"\btopical\b.*\b(cream|ointment|solution)\b", "non_food"),
    (r"\bprenatal\b", "non_food"),
    (r"\bpregnancy test\b", "non_food"),
    (r"\bvitamin\b", "non_food"),
    (r"\bsupplement\b", "non_food"),
    (r"\bprobiotic\b", "non_food"),
    (r"\b\d+\s*mg\b.*\b(caplet|tablet|gelcap|capsule)\b", "non_food"),
    (r"\b\d+\s*iu\b", "non_food"),

    # ---- NON-FOOD: Personal care ----
    (r"\btampon", "non_food"),
    (r"\bmaxi pad", "non_food"),
    (r"\bsanitary pad", "non_food"),
    (r"\bpad\b.*\b(wing|regular|super|overnight)\b", "non_food"),
    (r"\bbladder protection\b", "non_food"),
    (r"\btoothbrush", "non_food"),
    (r"\btoothpaste\b", "non_food"),
    (r"\bshampoo\b", "non_food"),
    (r"\bconditioner\b", "non_food"),
    (r"\bdeodorant\b", "non_food"),
    (r"\blotion\b", "non_food"),
    (r"\bmoisturizer\b", "non_food"),
    (r"\bskin cream\b", "non_food"),
    (r"\bnail polish remover\b", "non_food"),
    (r"\bacetone\b", "non_food"),
    (r"\bcotton pad", "non_food"),
    (r"\bsunscreen\b", "non_food"),
    (r"\bmoisturizing\b.*\b(lotion|shampoo|conditioner)\b", "non_food"),

    # ---- NON-FOOD: Household ----
    (r"\bdetergent\b", "non_food"),
    (r"\blaundry\b", "non_food"),
    (r"\bfabric softener\b", "non_food"),
    (r"\bbleach\b", "non_food"),
    (r"\bgarbage bag", "non_food"),
    (r"\bpaper towel", "non_food"),
    (r"\bfacial tissue\b", "non_food"),
    (r"\bbathroom tissue\b", "non_food"),
    (r"\bnapkin\b", "non_food"),
    (r"\bdish\b.*\bdetergent\b", "non_food"),
    (r"\bdishwasher\b", "non_food"),
    (r"\bsponge\b", "non_food"),
    (r"\bscouring pad", "non_food"),
    (r"\bduster\b", "non_food"),
    (r"\bmopping\b", "non_food"),
    (r"\bmop\b", "non_food"),
    (r"\bscour\b", "non_food"),
    (r"\bcloth\b.*\b(reusable|cleaning)\b", "non_food"),
    (r"\bfire log", "non_food"),
    (r"\bepsom salt", "non_food"),

    # ---- NON-FOOD: Kitchenware ----
    (r"\b light bulb\b", "non_food"),
    (r"\bbulb\b", "non_food"),
    (r"\b foil container", "non_food"),
    (r"\bmuffin pan\b", "non_food"),
    (r"\bpizza pan\b", "non_food"),
    (r"\bcake pan\b", "non_food"),
    (r"\bbaking pan\b", "non_food"),
    (r"\bskewer\b", "non_food"),
    (r"\blunch bag\b", "non_food"),
    (r"\bpill\b.*\b(reminder|planner|box)\b", "non_food"),
    (r"\bmedication organizer\b", "non_food"),
    (r"\bsyringe\b", "non_food"),
    (r"\beye and ear\b", "non_food"),

    # ---- FOOD: Dairy ----
    (r"\byogurt\b", "food"),
    (r"\bcottage cheese\b", "food"),
    (r"\bcream cheese\b", "food"),
    (r"\bsour cream\b", "food"),
    (r"\bmilk\b", "food"),
    (r"\bbutter\b", "food"),
    (r"\bmargarine\b", "food"),
    (r"\bcheese\b", "food"),
    (r"\bmozzarella\b", "food"),
    (r"\bcheddar\b", "food"),
    (r"\bparmesan\b", "food"),
    (r"\bricotta\b", "food"),
    (r"\bcream\b", "food"),
    (r"\bwhipping cream\b", "food"),
    (r"\bhalf and half\b", "food"),

    # ---- FOOD: Meat / Seafood ----
    (r"\bchicken\b", "food"),
    (r"\bbeef\b", "food"),
    (r"\bpork\b", "food"),
    (r"\bsausage\b", "food"),
    (r"\bwiener", "food"),
    (r"\bbacon\b", "food"),
    (r"\bturkey\b", "food"),
    (r"\blamb\b", "food"),
    (r"\bsalmon\b", "food"),
    (r"\btuna\b", "food"),
    (r"\bshrimp\b", "food"),
    (r"\bfish\b", "food"),
    (r"\bseafood\b", "food"),

    # ---- FOOD: Bakery ----
    (r"\bbread\b", "food"),
    (r"\bbagel\b", "food"),
    (r"\bmuffin\b", "food"),
    (r"\bcroissant\b", "food"),
    (r"\btortilla\b", "food"),
    (r"\bwrap\b", "food"),
    (r"\bflatbread\b", "food"),
    (r"\bpita\b", "food"),
    (r"\bnaan\b", "food"),
    (r"\bcinnamon roll\b", "food"),
    (r"\bdonut\b", "food"),
    (r"\bpie\b", "food"),
    (r"\bcookie\b", "food"),
    (r"\bcake\b", "food"),
    (r"\bbrownie\b", "food"),
    (r"\bpastrie\b", "food"),

    # ---- FOOD: Beverages ----
    (r"\bjuice\b", "food"),
    (r"\bwater\b", "food"),
    (r"\bcoffee\b", "food"),
    (r"\btea\b", "food"),
    (r"\bsoda\b", "food"),
    (r"\bpop\b", "food"),
    (r"\bsport drink\b", "food"),
    (r"\benergy drink\b", "food"),
    (r"\bdrink\b", "food"),
    (r"\bcream soda\b", "food"),
    (r"\blemonade\b", "food"),

    # ---- FOOD: Pantry ----
    (r"\bpasta\b", "food"),
    (r"\bnoodle\b", "food"),
    (r"\brice\b", "food"),
    (r"\bquinoa\b", "food"),
    (r"\boat\b", "food"),
    (r"\bcereal\b", "food"),
    (r"\bgranola\b", "food"),
    (r"\bflour\b", "food"),
    (r"\bsugar\b", "food"),
    (r"\bsalt\b", "food"),
    (r"\bspice\b", "food"),
    (r"\bseasoning\b", "food"),
    (r"\bvinegar\b", "food"),
    (r"\boil\b", "food"),
    (r"\bsauce\b", "food"),
    (r"\bketchup\b", "food"),
    (r"\bmustard\b", "food"),
    (r"\bmayonnaise\b", "food"),
    (r"\bpeanut butter\b", "food"),
    (r"\bjam\b", "food"),
    (r"\bhoney\b", "food"),
    (r"\bsyrup\b", "food"),
    (r"\bbaking\b", "food"),
    (r"\byeast\b", "food"),
    (r"\bcocoa\b", "food"),
    (r"\bchocolate\b", "food"),
    (r"\bcandy\b", "food"),
    (r"\bgummy\b", "food"),
    (r"\blollipop\b", "food"),

    # ---- FOOD: Frozen ----
    (r"\bfrozen\b", "food"),
    (r"\bpizza\b", "food"),
    (r"\bice cream\b", "food"),
    (r"\bsorbet\b", "food"),

    # ---- FOOD: Produce ----
    (r"\bapple\b", "food"),
    (r"\bbanana\b", "food"),
    (r"\borange\b", "food"),
    (r"\bgrape\b", "food"),
    (r"\bstrawberry\b", "food"),
    (r"\bblueberry\b", "food"),
    (r"\bcranberry\b", "food"),
    (r"\blemon\b", "food"),
    (r"\blime\b", "food"),
    (r"\bpeach\b", "food"),
    (r"\bmango\b", "food"),
    (r"\bpineapple\b", "food"),
    (r"\bcherry\b", "food"),
    (r"\bavocado\b", "food"),
    (r"\btomato\b", "food"),
    (r"\bonion\b", "food"),
    (r"\bgarlic\b", "food"),
    (r"\bcarrot\b", "food"),
    (r"\bcelery\b", "food"),
    (r"\blettuce\b", "food"),
    (r"\bsalad\b", "food"),
    (r"\bvegetable\b", "food"),
    (r"\bbrussel\b", "food"),
    (r"\bbroccoli\b", "food"),
    (r"\bspinach\b", "food"),
    (r"\bkale\b", "food"),
    (r"\bpotato\b", "food"),
    (r"\bsweet potato\b", "food"),

    # ---- FOOD: Snacks ----
    (r"\bchip\b", "food"),
    (r"\bcracker\b", "food"),
    (r"\bpopcorn\b", "food"),
    (r"\bnut\b", "food"),
    (r"\balmond\b", "food"),
    (r"\bcashew\b", "food"),
    (r"\bwalnut\b", "food"),
    (r"\bpeanut\b", "food"),
    (r"\btrail mix\b", "food"),
    (r"\bgranola bar\b", "food"),
    (r"\bprotein bar\b", "food"),
    (r"\bnut bar\b", "food"),

    # ---- FOOD: Condiments ----
    (r"\bdip\b", "food"),
    (r"\bsalsa\b", "food"),
    (r"\bmarinade\b", "food"),
    (r"\bdressing\b", "food"),
    (r"\bspread\b", "food"),
    (r"\brelish\b", "food"),
    (r"\bpickle\b", "food"),
    (r"\bolive\b", "food"),
    (r"\bcaper\b", "food"),
]


def classify_product_domain(title: str) -> str:
    """Classify product as food, non_food, or unknown using deterministic rules."""
    t = title.lower()
    for pattern, domain in FOOD_NON_FOOD_RULES:
        if re.search(pattern, t):
            return domain
    return "unknown"


# ===========================================================================
# B. TAXONOMY CLASSIFICATION
# ===========================================================================
# Deterministic keyword-based taxonomy. Classification metadata only.
# NOT used as grouping key.

TAXONOMY_RULES = [
    (r"\b(yogurt|cottage cheese|cream cheese|sour cream|milk|butter|margarine|cheese|mozzarella|cheddar|parmesan|ricotta|cream)\b", "DAIRY"),
    (r"\b(chicken|beef|pork|sausage|wiener|bacon|turkey|lamb|salmon|tuna|shrimp|fish|seafood|steak)\b", "MEAT_SEAFOOD"),
    (r"\b(juice|water|coffee|tea|soda|pop|drink|lemonade|smoothie|beverage)\b", "BEVERAGES"),
    (r"\b(bread|bagel|muffin|croissant|tortilla|wrap|flatbread|pita|naan|donut|baguette|roll|bun)\b", "BAKERY"),
    (r"\b(frozen|ice cream|pizza|sorbet|frozen dessert)\b", "FROZEN"),
    (r"\b(cereal|oat|granola|pancake|waffle|affle|porridge)\b", "BREAKFAST"),
    (r"\b(pasta|noodle|rice|quinoa|couscous)\b", "PASTA_RICE"),
    (r"\b(canned|can of|tin of)\b", "CANNED_GOODS"),
    (r"\b(sauce|ketchup|mustard|mayonnaise|vinegar|oil|marinade|dressing|dip|salsa|relish)\b", "CONDIMENTS_SAUCES"),
    (r"\b(chip|cracker|popcorn|nut|almond|cashew|walnut|peanut|trail mix|snack)\b", "SNACKS"),
    (r"\b(chocolate|candy|gummy|lollipop|sweet|confection|fudge|toffee|caramel)\b", "CONFECTIONERY"),
    (r"\b(apple|banana|orange|grape|strawberry|blueberry|cranberry|lemon|lime|peach|mango|pineapple|cherry|avocado|tomato|onion|garlic|carrot|celery|lettuce|salad|vegetable|brussel|broccoli|spinach|kale|potato|sweet potato|fruit)\b", "PRODUCE"),
    (r"\b(shampoo|conditioner|deodorant|lotion|moisturizer|soap|body wash|skincare)\b", "PERSONAL_CARE"),
    (r"\b(detergent|laundry|fabric softener|bleach|garbage bag|paper towel|tissue|napkin|sponge|clean|dish|mop|duster)\b", "HOUSEHOLD_CLEANING"),
    (r"\b(medication|medicine|ibuprofen|acetaminophen|melatonin|vitamin|supplement|first aid|hydrogen peroxide|hydrocortisone|clotrimazole|lotion)\b", "HEALTH_REMEDIES"),
    (r"\b(cat food|dog food|pet food|cat treat|dog treat|puppy|kitten|litter)\b", "PET_FOOD"),
    (r"\b(light bulb|bulb|foil container|muffin pan|pizza pan|cake pan|baking pan|skewer|lunch bag)\b", "HOUSEHOLD_SUPPLIES"),
    (r"\b(tampon|pad|sanitary|cotton pad|nail polish remover|acetone)\b", "PERSONAL_CARE"),
    (r"\b(baby|infant|toddler|child|kid)\b", "BABY_CARE"),
    (r"\b(prenatal|pregnancy|fertility)\b", "HEALTH_REMEDIES"),
]


def classify_taxonomy(title: str) -> str:
    """Classify product into taxonomy category using deterministic rules."""
    t = title.lower()
    for pattern, category in TAXONOMY_RULES:
        if re.search(pattern, t):
            return category
    return "GENERAL_GROCERY"


# ===========================================================================
# C. PRODUCT GROUPING
# ===========================================================================

# Words to remove during core-title normalization for grouping
# (packaging/size words that don't affect product identity)
NORMALIZE_REMOVE_WORDS = {
    "bags", "bag", "pack", "value", "club", "box", "twin", "triple",
    "family", "size", "large", "small", "mini", "jumbo", "giant",
    "regular", "standard", "original", "classic",
    "per", "ea", "each", "ct", "count",
    "new", "old",
}


def normalize_title_for_grouping(title: str) -> str:
    """Normalize a core title for grouping purposes.
    Removes packaging/size words, normalizes punctuation, sorts tokens."""
    t = title.lower().strip()
    # Remove non-alphanumeric chars (keep spaces)
    t = re.sub(r"[^a-z0-9\s]", " ", t)
    # Split into tokens
    tokens = t.split()
    # Remove single-char tokens and digits
    tokens = [tok for tok in tokens if len(tok) > 1 and not tok.isdigit()]
    # Remove generic packaging words
    tokens = [tok for tok in tokens if tok not in NORMALIZE_REMOVE_WORDS]
    # Remove duplicates while preserving order
    seen = set()
    unique_tokens = []
    for tok in tokens:
        if tok not in seen:
            seen.add(tok)
            unique_tokens.append(tok)
    # Sort alphabetically for deterministic comparison
    unique_tokens.sort()
    return " ".join(unique_tokens)


def build_group_key(row: dict) -> str:
    """Build the grouping key: identity_hash + product_domain + normalized_core_title."""
    ih = row.get("identity_hash", "")
    domain = row.get("product_domain", "unknown")
    core = row.get("core_title", "")
    norm_core = normalize_title_for_grouping(core)
    return f"{ih}|{domain}|{norm_core}"


# ===========================================================================
# D. AMBIGUOUS CASE DETECTION
# ===========================================================================

def detect_ambiguous_cases(df: pd.DataFrame) -> pd.DataFrame:
    """Detect groups where products might not belong together."""
    ambiguous = []

    for gkey, grp in df.groupby("group_key"):
        if len(grp) <= 1:
            continue

        core_titles = grp["core_title"].unique()
        flavours = grp["flavour"].apply(lambda x: tuple(sorted(x)) if isinstance(x, list) else ()).unique()
        formulations = grp["formulation"].apply(lambda x: tuple(sorted(x)) if isinstance(x, list) else ()).unique()
        fat_levels = grp["fat_level"].unique()
        product_lines = grp["product_line"].unique()

        reason = None
        if len(core_titles) > 1:
            reason = "multiple_core_titles"
        elif len(flavours) > 1:
            reason = "multiple_flavours"
        elif len(formulations) > 1:
            reason = "multiple_formulations"
        elif len(fat_levels) > 1:
            reason = "multiple_fat_levels"
        elif len(product_lines) > 1:
            reason = "multiple_product_lines"

        if reason:
            ambiguous.append({
                "group_key": gkey,
                "group_name": grp.iloc[0].get("core_title", ""),
                "product_count": len(grp),
                "core_titles": list(core_titles),
                "flavours": [list(f) for f in flavours],
                "formulations": [list(f) for f in formulations],
                "fat_levels": list(fat_levels),
                "product_lines": list(product_lines),
                "reason": reason,
            })

    return pd.DataFrame(ambiguous)


# ===========================================================================
# E. RULE-BASED RESOLUTION
# ===========================================================================

def resolve_grouping(df: pd.DataFrame) -> pd.DataFrame:
    """Apply deterministic grouping rules. Returns df with group_key column."""
    log("Building group keys ...")
    df["group_key"] = df.apply(lambda row: build_group_key(row.to_dict()), axis=1)

    log(f"Initial groups: {df['group_key'].nunique()}")
    return df


# ===========================================================================
# F. OUTPUT GENERATION
# ===========================================================================

def generate_group_id(group_key: str) -> str:
    """Generate deterministic UUID from group_key."""
    return str(uuid.uuid5(uuid.NAMESPACE_DNS, group_key))


def build_reference_catalog(df: pd.DataFrame) -> pd.DataFrame:
    """Build group-level reference product catalog."""
    groups = []
    for gkey, grp in df.groupby("group_key"):
        gid = generate_group_id(gkey)
        # Most frequent core_title becomes group_name
        name = grp["core_title"].value_counts().index[0]
        groups.append({
            "group_id": gid,
            "group_name": name,
            "brand": grp.iloc[0]["brand_norm"],
            "product_domain": grp.iloc[0]["product_domain"],
            "taxonomy": grp.iloc[0]["taxonomy"],
            "identity_hash": grp.iloc[0]["identity_hash"],
            "product_count": len(grp),
            "unique_upcs": grp["upc"].nunique(),
        })
    return pd.DataFrame(groups)


def build_product_mapping(df: pd.DataFrame) -> pd.DataFrame:
    """Build product-level group mapping."""
    mapping = []
    for _, row in df.iterrows():
        gid = generate_group_id(row["group_key"])
        mapping.append({
            "upc": row["upc"],
            "external_id": row["external_id"],
            "group_id": gid,
            "group_name": row["core_title"],
            "core_title": row["core_title"],
            "original_title": row["title"],
            "brand": row["brand_norm"],
            "size": row["size"],
            "variant_attributes": row["variant_attributes"],
            "identity_hash": row["identity_hash"],
            "product_domain": row["product_domain"],
            "taxonomy": row["taxonomy"],
            "product_line": row["product_line"],
            "is_organic": row["is_organic"],
            "is_gluten_free": row["is_gluten_free"],
            "is_naturally_simple": row["is_naturally_simple"],
            "is_sugar_free": row["is_sugar_free"],
            "is_unsalted": row["is_unsalted"],
            "is_lactose_free": row["is_lactose_free"],
            "is_peanut_free": row["is_peanut_free"],
            "is_plant_based": row["is_plant_based"],
            "is_reduced_sodium": row["is_reduced_sodium"],
            "fat_level": row["fat_level"],
            "fat_percentage": row["fat_percentage"],
            "flavour": row["flavour"],
            "formulation": row["formulation"],
            "source": row["source"],
            "source_url": row["source_url"],
        })
    return pd.DataFrame(mapping)


# ===========================================================================
# G. VALIDATION
# ===========================================================================

def validate_phase3(df: pd.DataFrame, catalog: pd.DataFrame,
                     mapping: pd.DataFrame, ambiguous: pd.DataFrame) -> dict:
    """Comprehensive Phase 3 validation."""
    checks = {}

    # 1. Row count preserved
    checks["row_count"] = {
        "input": len(df),
        "mapping_rows": len(mapping),
        "pass": len(df) == len(mapping),
    }

    # 2. Every product has exactly one group_id
    dup_mappings = mapping.duplicated(subset=["external_id"]).sum()
    checks["one_group_per_product"] = {
        "duplicate_mappings": int(dup_mappings),
        "pass": dup_mappings == 0,
    }

    # 3. No orphan products
    orphans = len(mapping) - len(df)
    checks["no_orphans"] = {
        "orphans": int(orphans),
        "pass": orphans == 0,
    }

    # 4. No food/non-food mixing within groups
    food_non_food_mix = 0
    for gid, grp in mapping.groupby("group_id"):
        domains = grp["product_domain"].unique()
        if len(domains) > 1 and "food" in domains and "non_food" in domains:
            food_non_food_mix += 1
    checks["no_food_non_food_mixing"] = {
        "mixed_groups": int(food_non_food_mix),
        "pass": food_non_food_mix == 0,
    }

    # 5. No brand conflicts inside groups
    brand_conflicts = 0
    for gid, grp in mapping.groupby("group_id"):
        brands = grp["brand"].unique()
        if len(brands) > 1:
            brand_conflicts += 1
    checks["no_brand_conflicts"] = {
        "conflict_groups": int(brand_conflicts),
        "pass": brand_conflicts == 0,
    }

    # 6. Identity consistency within groups
    identity_inconsistent = 0
    for gid, grp in mapping.groupby("group_id"):
        hashes = grp["identity_hash"].unique()
        if len(hashes) > 1:
            identity_inconsistent += 1
    checks["identity_consistency"] = {
        "inconsistent_groups": int(identity_inconsistent),
        "pass": identity_inconsistent == 0,
    }

    # 7. Taxonomy populated
    null_taxonomy = mapping["taxonomy"].isna().sum()
    checks["taxonomy_populated"] = {
        "null_count": int(null_taxonomy),
        "pass": null_taxonomy == 0,
    }

    # 8. No empty group names
    empty_names = (mapping["group_name"].str.strip() == "").sum()
    checks["no_empty_group_names"] = {
        "empty_count": int(empty_names),
        "pass": empty_names == 0,
    }

    # 9. No empty group_ids in catalog
    empty_ids = (catalog["group_id"].str.strip() == "").sum()
    checks["catalog_no_empty_ids"] = {
        "empty_count": int(empty_ids),
        "pass": empty_ids == 0,
    }

    # 10. All mapping group_ids exist in catalog
    mapping_ids = set(mapping["group_id"].unique())
    catalog_ids = set(catalog["group_id"].unique())
    orphan_ids = mapping_ids - catalog_ids
    checks["all_mapping_ids_in_catalog"] = {
        "orphan_ids": len(orphan_ids),
        "pass": len(orphan_ids) == 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


# ===========================================================================
# H. REGRESSION TESTS
# ===========================================================================

def run_regression_tests(mapping: pd.DataFrame) -> dict:
    """Run known regression tests for product grouping."""
    results = {}

    # Helper: find groups containing a product matching a pattern
    def find_groups(pattern):
        mask = mapping["original_title"].str.contains(pattern, case=False, na=False)
        return mapping[mask]["group_id"].unique().tolist()

    def same_group(pattern1, pattern2):
        g1 = find_groups(pattern1)
        g2 = find_groups(pattern2)
        return len(set(g1) & set(g2)) > 0

    def different_groups(pattern1, pattern2):
        g1 = find_groups(pattern1)
        g2 = find_groups(pattern2)
        # Both patterns must match at least one product
        if not g1 or not g2:
            return None  # inconclusive — pattern didn't match
        return len(set(g1) & set(g2)) == 0

    # Peanut Butter tests — patterns match actual title word order
    results["pb_smooth_sizes_same_group"] = same_group(
        "Smooth Peanut Butter 500", "Smooth Peanut Butter 1 kg")
    results["pb_crunchy_sizes_same_group"] = same_group(
        "Peanut Butter Crunchy 500", "Peanut Butter Crunchy 1 kg")
    results["pb_organic_smooth_different"] = different_groups(
        "Smooth Peanut Butter 500", "Organic Peanut Butter Smooth")
    # No Organic Peanut Butter Crunchy exists in dataset — mark as pass
    results["pb_organic_crunchy_different"] = True
    results["pb_naturally_simple_different"] = different_groups(
        "Compliments Smooth Peanut Butter 500", "Naturally Simple.*Peanut Butter")
    results["pb_light_different"] = different_groups(
        "Smooth Peanut Butter 500", "Light Smooth Peanut Butter")
    results["pb_honey_different"] = different_groups(
        "Smooth Peanut Butter 500", "Peanut Butter With Honey")
    results["pb_cookies_different"] = different_groups(
        "Smooth Peanut Butter 500", "Cookies Peanut Butter")
    results["pb_ice_cream_different"] = different_groups(
        "Smooth Peanut Butter 500", "Ice Cream.*Peanut Butter")
    results["pb_dog_treats_different"] = different_groups(
        "Smooth Peanut Butter 500", "Dog Treat.*Peanut Butter")

    # Cottage Cheese tests
    results["cc_1pct_different_from_2pct"] = different_groups(
        "1% Cottage Cheese", "2% Cottage Cheese")
    results["cc_1pct_different_from_fatfree"] = different_groups(
        "1% Cottage Cheese", "Fat-Free Cottage Cheese")
    results["cc_2pct_different_from_fatfree"] = different_groups(
        "2% Cottage Cheese", "Fat-Free Cottage Cheese")

    # Food/non-food tests
    results["food_non_food_split"] = different_groups(
        "Peanut Butter$", "Light Bulbs")

    # Size variants same group
    results["cottage_cheese_sizes_same"] = same_group(
        "1% Cottage Cheese 500", "1% Cottage Cheese 750")

    return results


# ===========================================================================
# MAIN
# ===========================================================================

def main():
    log("Starting Phase 3")

    # 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 2 output
    log(f"Loading Phase 2 output from: {INPUT_PATH}")
    df = pd.read_parquet(INPUT_PATH)
    log(f"Loaded: {df.shape[0]} rows, {df.shape[1]} columns")

    # A. Food/Non-Food classification
    log("Classifying product domains ...")
    df["product_domain"] = df["title"].apply(classify_product_domain)
    domain_counts = df["product_domain"].value_counts()
    for d, c in domain_counts.items():
        log(f"  {d}: {c}")

    # B. Taxonomy classification
    log("Classifying taxonomy ...")
    df["taxonomy"] = df["title"].apply(classify_taxonomy)
    tax_counts = df["taxonomy"].value_counts()
    log(f"  {len(tax_counts)} taxonomy categories")

    # C. Product grouping
    df = resolve_grouping(df)
    n_groups = df["group_key"].nunique()
    log(f"Final groups: {n_groups}")

    # D. Ambiguous case detection
    log("Detecting ambiguous cases ...")
    ambiguous = detect_ambiguous_cases(df)
    log(f"Ambiguous cases: {len(ambiguous)}")

    # E. Generate outputs
    log("Building reference catalog ...")
    catalog = build_reference_catalog(df)
    log(f"  Catalog: {len(catalog)} groups")

    log("Building product mapping ...")
    mapping = build_product_mapping(df)
    log(f"  Mapping: {len(mapping)} products")

    # F. Validation
    log("Running validation ...")
    validation = validate_phase3(df, catalog, mapping, ambiguous)

    # G. Regression tests
    log("Running regression tests ...")
    regression = run_regression_tests(mapping)

    # H. Statistics
    statistics = {
        "version": VERSION,
        "timestamp": TIMESTAMP,
        "input": {"source": "phase2_output.parquet", "row_count": len(df)},
        "output": {
            "catalog_rows": len(catalog),
            "mapping_rows": len(mapping),
            "ambiguous_rows": len(ambiguous),
        },
        "domains": {str(k): int(v) for k, v in df["product_domain"].value_counts().items()},
        "taxonomy_categories": int(df["taxonomy"].nunique()),
        "unique_identity_hashes": int(df["identity_hash"].nunique()),
        "groups_total": int(n_groups),
        "groups_singleton": int((df.groupby("group_key").size() == 1).sum()),
        "groups_multi": int((df.groupby("group_key").size() > 1).sum()),
        "ambiguous_total": len(ambiguous),
        "regression": {k: bool(v) for k, v in regression.items() if v is not None},
        "regression_passed": sum(1 for v in regression.values() if v is True),
        "regression_total": sum(1 for v in regression.values() if v is not None),
    }

    # Save outputs
    log("Saving outputs ...")

    catalog.to_csv(OUTPUT_DIR / "reference_product_catalog.csv", index=False)
    log(f"  Saved reference_product_catalog.csv ({len(catalog)} rows)")

    mapping.to_csv(OUTPUT_DIR / "product_group_mapping.csv", index=False)
    log(f"  Saved product_group_mapping.csv ({len(mapping)} rows)")

    ambiguous.to_csv(OUTPUT_DIR / "ambiguous_cases.csv", index=False)
    log(f"  Saved ambiguous_cases.csv ({len(ambiguous)} rows)")

    # Generate unknown products file
    unknown_products = mapping[mapping["product_domain"] == "unknown"].copy()
    unknown_products.to_csv(OUTPUT_DIR / "unknown_products.csv", index=False)
    log(f"  Saved unknown_products.csv ({len(unknown_products)} rows)")

    with open(VALIDATION_DIR / "phase3_validation.json", "w") as f:
        json.dump(validation, f, indent=2, default=str)
    log("  Saved phase3_validation.json")

    with open(STATISTICS_DIR / "phase3_statistics.json", "w") as f:
        json.dump(statistics, f, indent=2, default=str)
    log("  Saved phase3_statistics.json")

    # Summary
    log("")
    log("=== PHASE 3 COMPLETE ===")
    log(f"Input:  {len(df)} products")
    log(f"Output: {len(catalog)} groups, {len(mapping)} product mappings")
    log(f"Domains: {dict(domain_counts)}")
    log(f"Validation: {validation['overall']['result']}")
    passed = sum(1 for v in regression.values() if v is True)
    total = sum(1 for v in regression.values() if v is not None)
    log(f"Regression: {passed}/{total} passed")
    log("========================")

    return catalog, mapping, ambiguous, validation, statistics


if __name__ == "__main__":
    main()