feat(pipeline): finalize audited multi-platform dataset with clean provenance and entity resolution

pipeline/batch_500_runner.py

@@ -0,0 +1,234 @@
+"""
+pipeline/batch_500_runner.py
+Production Multi-Platform HIFFI Batch Runner.
+
+Governance Standards Enforced:
+- No synthetic proxies or fabricated estimates for Spotify streaming metrics.
+- Uncollected metrics are explicitly recorded as NULL (NaN) with provenance tags.
+- Spotify API status explicitly logged as BLOCKED_HTTP_403.
+- Separation of official Production YouTube HIFFI rank and Experimental HIFFI-XP score.
+- Checkpointing enabled with automated resume functionality.
+"""
+
+import sys
+import os
+import time
+import random
+from pathlib import Path
+from datetime import datetime, timezone
+
+PROJECT_ROOT = Path(__file__).resolve().parent.parent
+if str(PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(PROJECT_ROOT))
+
+import pandas as pd
+import numpy as np
+
+from collectors.wikidata_resolver import search_wikidata_qid, get_identifiers_for_qid
+from collectors.browser_collector import MultiPlatformSignalCollector
+from playwright.sync_api import sync_playwright
+
+CHECKPOINT_DIR = PROJECT_ROOT / "data" / "checkpoints"
+CHECKPOINT_DIR.mkdir(parents=True, exist_ok=True)
+CHECKPOINT_FILE = CHECKPOINT_DIR / "multiplatform_checkpoints.csv"
+FINAL_OUTPUT_FILE = PROJECT_ROOT / "data" / "hiffi_multiplatform_500.csv"
+
+
+def min_max_normalize(series: pd.Series) -> pd.Series:
+    """Scales numeric series strictly between 0 and 100 while preserving NaNs."""
+    valid = series.dropna()
+    if valid.empty or valid.max() == valid.min():
+        return pd.Series(np.nan, index=series.index)
+    return ((series - valid.min()) / (valid.max() - valid.min())) * 100.0
+
+
+class BatchMultiPlatformRunner:
+    def __init__(self, youtube_csv_path: str = "enriched_artists.csv"):
+        self.input_file = PROJECT_ROOT / youtube_csv_path
+        self.df_base = pd.read_csv(self.input_file)
+        self.df_base["name_clean"] = self.df_base["name"].astype(str).str.strip().str.lower()
+        self.collector = MultiPlatformSignalCollector(headless=True)
+
+    def load_checkpoint(self) -> pd.DataFrame:
+        """Loads existing checkpoint records to allow resuming interrupted runs."""
+        if CHECKPOINT_FILE.exists():
+            return pd.read_csv(CHECKPOINT_FILE)
+        return pd.DataFrame(columns=[
+            "name", "name_clean", "wikidata_qid", "spotify_id", 
+            "instagram_handle", "instagram_followers", "resolution_status", "ig_status"
+        ])
+
+    def save_checkpoint(self, df_checkpoint: pd.DataFrame):
+        """Persists current state to CSV."""
+        df_checkpoint.to_csv(CHECKPOINT_FILE, index=False)
+
+    def run_full_pipeline(self, batch_size: int = 25, delay_range: tuple = (0.5, 1.2)):
+        print("=" * 90)
+        print(f"HIFFI MULTI-PLATFORM RUNNER (STRICT DATA GOVERNANCE)")
+        print(f"Total Base Cohort: {len(self.df_base)} artists")
+        print("=" * 90)
+
+        df_ckpt = self.load_checkpoint()
+        processed_names = set(df_ckpt["name_clean"].dropna().tolist())
+        print(f"Existing Checkpoints Loaded: {len(processed_names)} artists already processed.\n")
+
+        artists_to_process = self.df_base[~self.df_base["name_clean"].isin(processed_names)].copy()
+        print(f"Artists Remaining to Process: {len(artists_to_process)}")
+
+        # Step 1: Collect Missing Social and Identifier Signals
+        if not artists_to_process.empty:
+            with sync_playwright() as p:
+                browser = p.chromium.launch(headless=True)
+                page = browser.new_page(
+                    viewport={"width": 1920, "height": 1080},
+                    user_agent=(
+                        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+                        "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
+                    )
+                )
+
+                new_rows = []
+                for idx, row in artists_to_process.iterrows():
+                    artist_name = str(row["name"]).strip()
+                    name_clean = artist_name.lower()
+                    current_count = len(df_ckpt) + len(new_rows) + 1
+                    print(f"[{current_count}/{len(self.df_base)}] Resolving: {artist_name}...")
+
+                    # 1. Resolve Wikidata Identifiers
+                    qid = search_wikidata_qid(artist_name)
+                    ids = get_identifiers_for_qid(qid) if qid else {"spotify_id": "N/A", "instagram_handle": "N/A"}
+
+                    handle = ids.get("instagram_handle", "N/A")
+                    ig_followers = None
+                    ig_status = "NO_HANDLE"
+
+                    # 2. Extract Instagram Follower Count if handle exists
+                    if handle and handle != "N/A":
+                        ig_url = f"https://www.instagram.com/{handle}/"
+                        res = self.collector.collect_artist({"name": artist_name, "ig_url": ig_url}, page)
+                        ig_followers = res.get("instagram_followers")
+                        ig_status = res.get("ig_status", "FAILED")
+                        f_display = f"{ig_followers:,}" if ig_followers is not None else "Failed"
+                        print(f"    -> QID: {qid} | Spotify URI: {ids.get('spotify_id')} | IG: @{handle} ({f_display})")
+                    else:
+                        print(f"    -> QID: {qid or 'Not Found'} | Spotify URI: {ids.get('spotify_id')} | IG: None")
+
+                    record = {
+                        "name": artist_name,
+                        "name_clean": name_clean,
+                        "wikidata_qid": qid or "N/A",
+                        "spotify_id": ids.get("spotify_id", "N/A"),
+                        "instagram_handle": handle,
+                        "instagram_followers": ig_followers,
+                        "resolution_status": "RESOLVED" if qid else "NOT_FOUND",
+                        "ig_status": ig_status
+                    }
+                    new_rows.append(record)
+
+                    # Periodic Checkpointing
+                    if len(new_rows) % batch_size == 0:
+                        df_ckpt = pd.concat([df_ckpt, pd.DataFrame(new_rows)], ignore_index=True)
+                        self.save_checkpoint(df_ckpt)
+                        new_rows = []
+                        print(f"\n>>> CHECKPOINT SAVED: {len(df_ckpt)} records committed. <<<\n")
+
+                    time.sleep(random.uniform(*delay_range))
+
+                if new_rows:
+                    df_ckpt = pd.concat([df_ckpt, pd.DataFrame(new_rows)], ignore_index=True)
+                    self.save_checkpoint(df_ckpt)
+
+                browser.close()
+
+        print("\n" + "=" * 90)
+        print("PHASE 2: MERGING SIGNALS & ENFORCING PROVENANCE SCHEMAS")
+        print("=" * 90)
+
+        # Merge check-pointed external data with YouTube baseline
+        merged = pd.merge(
+            self.df_base,
+            df_ckpt,
+            on="name_clean",
+            how="left",
+            suffixes=("", "_ext")
+        )
+
+        now_utc = datetime.now(timezone.utc).isoformat()
+        
+        # ---------------------------------------------------------
+        # Strict Spotify Ingestion Contract:
+        # NO synthetic math, NO fake proxies. Explicit NULLs & status flags.
+        # ---------------------------------------------------------
+        merged["spotify_monthly_listeners"] = np.nan
+        merged["spotify_popularity"] = np.nan
+        merged["spotify_status"] = "BLOCKED_HTTP_403_PREMIUM_REQUIRED"
+        merged["spotify_data_source"] = None
+        merged["spotify_collected_at"] = None
+        merged["is_spotify_verified"] = False
+
+        # Clean Instagram metrics & provenance
+        merged["instagram_followers"] = pd.to_numeric(merged["instagram_followers"], errors="coerce")
+        merged["instagram_collected_at"] = np.where(merged["instagram_followers"].notna(), now_utc, None)
+        merged["instagram_data_source"] = np.where(merged["instagram_followers"].notna(), "playwright_meta_scrape", None)
+        merged["is_instagram_handle_verified"] = False  # Flag for manual/future handle audit
+
+        # YouTube Production Metrics
+        merged["youtube_data_source"] = "youtube_data_api_v3"
+        merged["yt_baseline_score"] = pd.to_numeric(merged["hiffi_score"], errors="coerce")
+
+        # ---------------------------------------------------------
+        # Scoring Layer:
+        # 1. Official Production Rank (YouTube baseline ground truth)
+        # ---------------------------------------------------------
+        merged["production_hiffi_score"] = merged["yt_baseline_score"]
+        merged["production_hiffi_rank"] = merged["production_hiffi_score"].rank(ascending=False, method="min")
+
+        # ---------------------------------------------------------
+        # 2. Experimental Cross-Platform Sandbox (HIFFI-XP)
+        # Evaluated strictly on available signals. Does not impute Spotify.
+        # ---------------------------------------------------------
+        norm_ig = min_max_normalize(merged["instagram_followers"]).fillna(0)
+        norm_yt = min_max_normalize(merged["yt_baseline_score"]).fillna(0)
+
+        # Normalized two-pillar proxy (until authorized Spotify credentials are provided)
+        # Weights normalized across available active channels: YouTube 58% (35/60), Instagram 42% (25/60)
+        merged["hiffi_xp_score"] = ((0.58 * norm_yt) + (0.42 * norm_ig)).round(2)
+        merged["hiffi_xp_rank"] = merged["hiffi_xp_score"].rank(ascending=False, method="min").astype(int)
+
+        # Telemetry Completeness Flag
+        merged["has_complete_multiplatform_signals"] = (
+            merged["spotify_monthly_listeners"].notna() &
+            merged["instagram_followers"].notna() &
+            merged["yt_baseline_score"].notna()
+        )
+
+        # Export final dataset
+        FINAL_OUTPUT_FILE.parent.mkdir(parents=True, exist_ok=True)
+        merged.to_csv(FINAL_OUTPUT_FILE, index=False)
+        print(f"\nFinal Audited Multi-Platform Dataset Exported: {FINAL_OUTPUT_FILE}")
+
+        # Summary Inspection
+        print("\n" + "=" * 90)
+        print("PIPELINE DATA PROVENANCE AUDIT SUMMARY")
+        print("=" * 90)
+        print(f"Total Cohort Rows:                 {len(merged)}")
+        print(f"Spotify Candidate IDs Found:       {(merged['spotify_id'] != 'N/A').sum()} / {len(merged)}")
+        print(f"Spotify Live Metrics Populated:    {merged['spotify_monthly_listeners'].notna().sum()} (0.0% - Status: BLOCKED_HTTP_403)")
+        print(f"Instagram Live Follower Counts:    {merged['instagram_followers'].notna().sum()} / {len(merged)}")
+        print(f"Complete Multi-Platform Coverage:  {merged['has_complete_multiplatform_signals'].sum()} (Awaiting Spotify API key)")
+        print("-" * 90)
+
+        preview_cols = [
+            "name", "production_hiffi_rank", "hiffi_xp_rank", 
+            "hiffi_xp_score", "spotify_status", "spotify_monthly_listeners", 
+            "instagram_followers"
+        ]
+        available_cols = [c for c in preview_cols if c in merged.columns]
+        print("\nTOP 10 EXPERIMENTAL XP RANK PREVIEW (STRICT NULL PROVENANCE):")
+        print(merged[available_cols].sort_values("hiffi_xp_rank").head(10).to_string(index=False))
+        print("=" * 90)
+
+
+if __name__ == "__main__":
+    runner = BatchMultiPlatformRunner("enriched_artists.csv")
+    runner.run_full_pipeline(batch_size=15, delay_range=(0.5, 1.0))