hiffi-artist-intelligence/scripts/audit_platform_gaps.py

"""
scripts/audit_platform_gaps.py
Audits the exact volume and names of artists with unverified or missing
metrics across YouTube, Spotify, Instagram, and Commercial Sales.
"""

import sys
from pathlib import Path
import pandas as pd

ROOT_DIR = Path(__file__).resolve().parent.parent
if str(ROOT_DIR) not in sys.path:
    sys.path.insert(0, str(ROOT_DIR))

from core.database import get_connection

df500 = pd.read_csv(ROOT_DIR / "data" / "hiffi_multiplatform_500.csv")
cohort_names = [n.strip() for n in df500["name"]]

conn = get_connection()
cur = conn.cursor()

# 1. Enriched YouTube Check
enr_file = ROOT_DIR / "enriched_artists.csv"
if enr_file.exists():
    enr = pd.read_csv(enr_file)
    yt_verified = set(
        enr[enr["subscriber_count"].notna() & (enr["subscriber_count"] > 0)]["name"]
        .str.strip()
        .str.lower()
    )
else:
    yt_verified = set()

# 2. Instagram Checkpoint Check
chk_ig_file = ROOT_DIR / "data" / "checkpoints" / "multiplatform_checkpoints.csv"
if chk_ig_file.exists():
    chk_ig = pd.read_csv(chk_ig_file)
    ig_verified = set(
        chk_ig[chk_ig["instagram_followers"] > 0]["name"]
        .str.strip()
        .str.lower()
    )
else:
    ig_verified = set()

# 3. Spotify Monthly Listeners Check
sp_col = "spotify_monthly_listeners" if "spotify_monthly_listeners" in df500.columns else None
if sp_col:
    sp_verified = set(
        df500[df500[sp_col] > 0]["name"].str.strip().str.lower()
    )
else:
    sp_verified = set()

# 4. Commercial Sales Check from ChartMasters / SQLite
# Inspect columns of commercial_sales table
cur.execute("PRAGMA table_info(commercial_sales)")
cs_cols = [c[1] for c in cur.fetchall()]

# Sales in df500
sales_col = None
for col_candidate in ["audited_sales_units", "sales", "eas", "commercial_sales", "chartmasters_sales"]:
    if col_candidate in df500.columns:
        sales_col = col_candidate
        break

if sales_col:
    sales_verified = set(df500[df500[sales_col] > 0]["name"].str.strip().str.lower())
else:
    cur.execute("""
        SELECT a.artist_name 
        FROM artists a 
        JOIN commercial_sales c ON a.artist_id = c.artist_id 
        WHERE c.audited_sales_units > 0
    """)
    sales_verified = set(row[0].strip().lower() for row in cur.fetchall())

conn.close()

# Compute missing
yt_missing = [n for n in cohort_names if n.lower() not in yt_verified]
sp_missing = [n for n in cohort_names if n.lower() not in sp_verified]
ig_missing = [n for n in cohort_names if n.lower() not in ig_verified]
sales_missing = [n for n in cohort_names if n.lower() not in sales_verified]

print("=" * 70)
print("MULTI-PLATFORM LIVE DATA GAP AUDIT (500 ARTISTS)")
print("=" * 70)
print(f"YouTube Channels lacking original scrape : {len(yt_missing):>3} / 500")
print(f"Spotify Listeners lacking direct scrape  : {len(sp_missing):>3} / 500")
print(f"Instagram Handles lacking direct scrape  : {len(ig_missing):>3} / 500")
print(f"Commercial Sales lacking audited source  : {len(sales_missing):>3} / 500")
print("=" * 70)

print(f"\n[SAMPLE] YouTube Missing ({len(yt_missing)} total):")
print(", ".join(yt_missing[:20]))

print(f"\n[SAMPLE] Spotify Missing ({len(sp_missing)} total):")
print(", ".join(sp_missing[:20]))

print(f"\n[SAMPLE] Instagram Missing ({len(ig_missing)} total):")
print(", ".join(ig_missing[:20]))

print(f"\n[SAMPLE] Commercial Sales Missing ({len(sales_missing)} total):")
print(", ".join(sales_missing[:20]))