Upload src/phase1.py with huggingface_hub

src/phase1.py

@@ -0,0 +1,424 @@
+#!/usr/bin/env python3
+"""
+Phase 1 — Load / Validate / Provenance
+Compliments Reference DB Pipeline
+
+Authoritative Input:
+  https://huggingface.co/datasets/saraNour/compliments-brand/blob/main/source_of_truth/products.parquet
+
+This phase:
+  1. Downloads the authoritative products.parquet from HuggingFace
+  2. Validates schema, row count, nulls, duplicates
+  3. Analyzes UPC patterns, brand values, size fields
+  4. Documents provenance of every column
+  5. Drops 100% null columns with explicit documentation
+  6. Produces clean Phase 1 output + validation + statistics
+"""
+
+import json
+import os
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+import pandas as pd
+from huggingface_hub import hf_hub_download
+
+# ---------------------------------------------------------------------------
+# Configuration
+# ---------------------------------------------------------------------------
+HF_REPO = "saraNour/compliments-brand"
+HF_FILE = "source_of_truth/products.parquet"
+HF_REPO_TYPE = "dataset"
+
+OUTPUT_DIR = Path(__file__).resolve().parent.parent / "outputs"
+VALIDATION_DIR = Path(__file__).resolve().parent.parent / "validation"
+STATISTICS_DIR = Path(__file__).resolve().parent.parent / "statistics"
+
+VERSION = "1.0.0"
+TIMESTAMP = datetime.now(timezone.utc).isoformat()
+
+
+def log(msg: str) -> None:
+    print(f"[Phase1] {msg}")
+
+
+# ---------------------------------------------------------------------------
+# 1. Load
+# ---------------------------------------------------------------------------
+def load_dataset() -> pd.DataFrame:
+    log(f"Downloading {HF_REPO}/{HF_FILE} ...")
+    path = hf_hub_download(HF_REPO, HF_FILE, repo_type=HF_REPO_TYPE)
+    log(f"Downloaded to: {path}")
+    df = pd.read_parquet(path)
+    log(f"Loaded shape: {df.shape}")
+    return df
+
+
+# ---------------------------------------------------------------------------
+# 2. Validate schema
+# ---------------------------------------------------------------------------
+EXPECTED_COLUMNS = [
+    "upc", "external_id", "brand", "title", "price", "price_currency",
+    "size", "size_amount", "size_unit", "size_qty", "size_per_unit",
+    "size_unit_norm", "size_total", "image_url", "source", "source_url",
+]
+
+EXPECTED_DTYPES = {
+    "upc": "str",
+    "external_id": "str",
+    "brand": "str",
+    "title": "str",
+    "price": "float64",
+    "price_currency": "str",
+    "size": "str",
+    "size_amount": "float64",
+    "size_unit": "str",
+    "size_qty": "int64",
+    "size_per_unit": "object",
+    "size_unit_norm": "str",
+    "size_total": "object",
+    "image_url": "str",
+    "source": "str",
+    "source_url": "str",
+}
+
+EXPECTED_ROW_COUNT = 4440
+
+
+def validate_schema(df: pd.DataFrame) -> dict:
+    checks = {}
+
+    # Row count
+    checks["row_count"] = {
+        "expected": EXPECTED_ROW_COUNT,
+        "actual": len(df),
+        "pass": len(df) == EXPECTED_ROW_COUNT,
+    }
+
+    # Column presence
+    missing = [c for c in EXPECTED_COLUMNS if c not in df.columns]
+    extra = [c for c in df.columns if c not in EXPECTED_COLUMNS]
+    checks["columns"] = {
+        "expected_count": len(EXPECTED_COLUMNS),
+        "actual_count": len(df.columns),
+        "missing": missing,
+        "extra": extra,
+        "pass": len(missing) == 0,
+    }
+
+    # Column names match exactly
+    checks["column_order"] = {
+        "expected": EXPECTED_COLUMNS,
+        "actual": list(df.columns),
+        "pass": list(df.columns) == EXPECTED_COLUMNS,
+    }
+
+    return checks
+
+
+# ---------------------------------------------------------------------------
+# 3. Inspect nulls
+# ---------------------------------------------------------------------------
+def inspect_nulls(df: pd.DataFrame) -> dict:
+    null_info = {}
+    for col in df.columns:
+        n = int(df[col].isna().sum())
+        null_info[col] = {
+            "null_count": n,
+            "null_pct": round(n / len(df) * 100, 2),
+            "is_100pct_null": n == len(df),
+        }
+    return null_info
+
+
+# ---------------------------------------------------------------------------
+# 4. Inspect duplicates
+# ---------------------------------------------------------------------------
+def inspect_duplicates(df: pd.DataFrame) -> dict:
+    full_dupes = int(df.duplicated().sum())
+    upc_dupes = int(df["upc"].duplicated().sum()) if "upc" in df.columns else 0
+    ext_dupes = int(df["external_id"].duplicated().sum()) if "external_id" in df.columns else 0
+
+    return {
+        "full_row_duplicates": full_dupes,
+        "upc_duplicates": upc_dupes,
+        "external_id_duplicates": ext_dupes,
+        "pass": full_dupes == 0 and ext_dupes == 0,
+    }
+
+
+# ---------------------------------------------------------------------------
+# 5. UPC analysis
+# ---------------------------------------------------------------------------
+def analyze_upc(df: pd.DataFrame) -> dict:
+    upc = df["upc"]
+    total = len(upc)
+    nulls = int(upc.isna().sum())
+    unique = int(upc.nunique())
+
+    # Reused UPCs (appear more than once)
+    counts = upc.value_counts()
+    reused = counts[counts > 1]
+    reused_upcs = reused.to_dict()
+
+    return {
+        "total_rows": total,
+        "null_count": nulls,
+        "unique_count": unique,
+        "reused_upc_count": int(len(reused)),
+        "reused_upc_examples": {str(k): int(v) for k, v in list(reused_upcs.items())[:10]},
+        "null_upc_rows": df[upc.isna()][["external_id", "title", "brand"]].to_dict("records"),
+    }
+
+
+# ---------------------------------------------------------------------------
+# 6. Brand analysis
+# ---------------------------------------------------------------------------
+def analyze_brand(df: pd.DataFrame) -> dict:
+    brand_counts = df["brand"].value_counts()
+    return {
+        "unique_count": int(brand_counts.shape[0]),
+        "distribution": {str(k): int(v) for k, v in brand_counts.items()},
+    }
+
+
+# ---------------------------------------------------------------------------
+# 7. Size analysis
+# ---------------------------------------------------------------------------
+def analyze_size(df: pd.DataFrame) -> dict:
+    size_col = df["size"]
+    size_amount = df["size_amount"]
+    size_unit = df["size_unit"]
+
+    return {
+        "size_string": {
+            "unique_count": int(size_col.nunique()),
+            "null_count": int(size_col.isna().sum()),
+            "top_20": {str(k): int(v) for k, v in size_col.value_counts().head(20).items()},
+        },
+        "size_amount": {
+            "null_count": int(size_amount.isna().sum()),
+            "null_pct": round(size_amount.isna().sum() / len(df) * 100, 2),
+            "min": float(size_amount.min()) if size_amount.notna().any() else None,
+            "max": float(size_amount.max()) if size_amount.notna().any() else None,
+            "mean": round(float(size_amount.mean()), 2) if size_amount.notna().any() else None,
+            "median": round(float(size_amount.median()), 2) if size_amount.notna().any() else None,
+        },
+        "size_unit": {
+            "null_count": int(size_unit.isna().sum()),
+            "null_pct": round(size_unit.isna().sum() / len(df) * 100, 2),
+            "distribution": {str(k): int(v) for k, v in size_unit.value_counts().items()},
+        },
+    }
+
+
+# ---------------------------------------------------------------------------
+# 8. Provenance documentation
+# ---------------------------------------------------------------------------
+def document_provenance() -> dict:
+    return {
+        "source_dataset": f"{HF_REPO}/{HF_FILE}",
+        "source_url": f"https://huggingface.co/datasets/{HF_REPO}/blob/main/{HF_FILE}",
+        "source_type": "HuggingFace dataset (private)",
+        "access_method": "huggingface_hub.hf_hub_download",
+        "original_source": "Voila.ca (Loblaw) Compliments private-label products",
+        "columns": {
+            "upc": "Universal Product Code. 1 null. 3,271 unique. Some reused across variants.",
+            "external_id": "Voila retailer product ID. 4,440 unique. No nulls.",
+            "brand": "Product brand. 11 variants of Compliments/Sensations.",
+            "title": "Raw product title from Voila. 4,375 unique.",
+            "price": "Price in CAD. 247 unique values.",
+            "price_currency": "Always 'CAD'.",
+            "size": "Raw size string from Voila. 631 unique values.",
+            "size_amount": "Parsed numeric size amount. 307 nulls (6.9%).",
+            "size_unit": "Parsed size unit. 307 nulls (6.9%).",
+            "size_qty": "Size quantity multiplier. Usually 1.",
+            "size_per_unit": "100% null. All values are NaN. DROPPED.",
+            "size_unit_norm": "Normalized size unit. 307 nulls (6.9%).",
+            "size_total": "100% null. All values are NaN. DROPPED.",
+            "image_url": "Product image URL from Voila. 4,440 unique.",
+            "source": "Always 'voila'.",
+            "source_url": "Product page URL on Voila. 4,440 unique.",
+        },
+        "dropped_columns": [
+            {
+                "column": "size_per_unit",
+                "reason": "100% null (4,440/4,440 values are NaN). No usable data.",
+            },
+            {
+                "column": "size_total",
+                "reason": "100% null (4,440/4,440 values are NaN). No usable data.",
+            },
+        ],
+        "preserved_columns": [
+            "upc", "external_id", "brand", "title", "price", "price_currency",
+            "size", "size_amount", "size_unit", "size_qty", "size_unit_norm",
+            "image_url", "source", "source_url",
+        ],
+    }
+
+
+# ---------------------------------------------------------------------------
+# 9. Build statistics
+# ---------------------------------------------------------------------------
+def build_statistics(df: pd.DataFrame, null_info: dict, dup_info: dict,
+                     upc_info: dict, brand_info: dict, size_info: dict) -> dict:
+    cols_100pct_null = [c for c, v in null_info.items() if v["is_100pct_null"]]
+    return {
+        "version": VERSION,
+        "timestamp": TIMESTAMP,
+        "input": {
+            "source": f"{HF_REPO}/{HF_FILE}",
+            "row_count": len(df),
+            "column_count": len(df.columns),
+        },
+        "output": {
+            "row_count": len(df),
+            "column_count": len(df.columns) - len(cols_100pct_null),
+            "columns_dropped": cols_100pct_null,
+        },
+        "nulls": {
+            "columns_with_nulls": {c: v for c, v in null_info.items() if v["null_count"] > 0},
+            "columns_100pct_null": cols_100pct_null,
+        },
+        "duplicates": dup_info,
+        "upc": upc_info,
+        "brand": brand_info,
+        "size": size_info,
+    }
+
+
+# ---------------------------------------------------------------------------
+# 10. Build validation report
+# ---------------------------------------------------------------------------
+def build_validation(schema_checks: dict, null_info: dict, dup_info: dict) -> dict:
+    all_pass = True
+    failures = []
+
+    # Schema checks
+    for key, check in schema_checks.items():
+        if not check.get("pass", True):
+            all_pass = False
+            failures.append(f"schema.{key}")
+
+    # Null checks
+    non_trivial_nulls = {
+        c: v for c, v in null_info.items()
+        if v["null_count"] > 0 and not v["is_100pct_null"]
+    }
+    # 100% null columns are expected (size_per_unit, size_total)
+    expected_100pct = {"size_per_unit", "size_total"}
+    unexpected_100pct = [c for c, v in null_info.items()
+                         if v["is_100pct_null"] and c not in expected_100pct]
+    if unexpected_100pct:
+        all_pass = False
+        failures.append(f"unexpected_100pct_null_columns: {unexpected_100pct}")
+
+    # Duplicate checks
+    if not dup_info["pass"]:
+        all_pass = False
+        failures.append("duplicates")
+
+    return {
+        "version": VERSION,
+        "timestamp": TIMESTAMP,
+        "result": "PASS" if all_pass else "FAIL",
+        "failures": failures,
+        "checks": {
+            "schema": schema_checks,
+            "duplicates": dup_info,
+            "non_trivial_nulls": non_trivial_nulls,
+            "unexpected_100pct_null_columns": unexpected_100pct,
+        },
+    }
+
+
+# ---------------------------------------------------------------------------
+# Main
+# ---------------------------------------------------------------------------
+def main():
+    log("Starting Phase 1")
+
+    # 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)
+
+    # 1. Load
+    df = load_dataset()
+
+    # 2. Validate schema
+    log("Validating schema ...")
+    schema_checks = validate_schema(df)
+
+    # 3. Nulls
+    log("Inspecting nulls ...")
+    null_info = inspect_nulls(df)
+
+    # 4. Duplicates
+    log("Inspecting duplicates ...")
+    dup_info = inspect_duplicates(df)
+
+    # 5. UPC analysis
+    log("Analyzing UPCs ...")
+    upc_info = analyze_upc(df)
+
+    # 6. Brand analysis
+    log("Analyzing brands ...")
+    brand_info = analyze_brand(df)
+
+    # 7. Size analysis
+    log("Analyzing sizes ...")
+    size_info = analyze_size(df)
+
+    # 8. Provenance
+    log("Documenting provenance ...")
+    provenance = document_provenance()
+
+    # 9. Statistics
+    log("Building statistics ...")
+    statistics = build_statistics(df, null_info, dup_info, upc_info, brand_info, size_info)
+
+    # 10. Validation
+    log("Building validation report ...")
+    validation = build_validation(schema_checks, null_info, dup_info)
+
+    # 11. Drop 100% null columns
+    cols_to_drop = [c for c, v in null_info.items() if v["is_100pct_null"]]
+    log(f"Dropping 100% null columns: {cols_to_drop}")
+    df_out = df.drop(columns=cols_to_drop)
+
+    # 12. Save outputs
+    log("Saving outputs ...")
+    df_out.to_parquet(OUTPUT_DIR / "phase1_output.parquet", index=False)
+    log(f"  Saved phase1_output.parquet ({df_out.shape[0]} rows, {df_out.shape[1]} cols)")
+
+    with open(VALIDATION_DIR / "phase1_validation.json", "w") as f:
+        json.dump(validation, f, indent=2, default=str)
+    log("  Saved phase1_validation.json")
+
+    with open(STATISTICS_DIR / "phase1_statistics.json", "w") as f:
+        json.dump(statistics, f, indent=2, default=str)
+    log("  Saved phase1_statistics.json")
+
+    with open(OUTPUT_DIR / "phase1_provenance.json", "w") as f:
+        json.dump(provenance, f, indent=2, default=str)
+    log("  Saved phase1_provenance.json")
+
+    # Summary
+    log("")
+    log("=== PHASE 1 COMPLETE ===")
+    log(f"Input:  {df.shape[0]} rows, {df.shape[1]} columns")
+    log(f"Output: {df_out.shape[0]} rows, {df_out.shape[1]} columns")
+    log(f"Dropped columns: {cols_to_drop}")
+    log(f"Validation: {validation['result']}")
+    if validation["failures"]:
+        log(f"Failures: {validation['failures']}")
+    log("========================")
+
+    return df_out, validation, statistics
+
+
+if __name__ == "__main__":
+    main()