hiffi-artist-intelligence/scripts/verify_production_cohort.py

import sqlite3
import re
from pathlib import Path
import pandas as pd
import numpy as np

PROJECT_ROOT = Path(__file__).resolve().parent.parent
DB_PATH = PROJECT_ROOT / "data" / "hiffi_catalog.db"

conn = sqlite3.connect(DB_PATH)

query = """
    SELECT 
        a.artist_id,
        a.artist_name,
        a.spotify_url,
        t.spotify_monthly_listeners,
        a.youtube_url,
        t.subscribers_count AS yt_subscribers,
        a.instagram_url,
        t.instagram_followers,
        t.total_views AS yt_total_views
    FROM artists a
    LEFT JOIN telemetry_metrics t ON a.artist_id = t.artist_id
    WHERE a.image_verified_flag = 1
    ORDER BY t.spotify_monthly_listeners DESC
"""

df = pd.read_sql_query(query, conn)
conn.close()

# 1. Strict Canonical Regex Validation
SP_PATTERN = re.compile(r"^https://open\.spotify\.com/artist/[0-9A-Za-z]{22}$")
YT_PATTERN = re.compile(r"^https://www\.youtube\.com/channel/UC[0-9A-Za-z_-]{22}$")
IG_PATTERN = re.compile(r"^https://www\.instagram\.com/[A-Za-z0-9_.]+/?$")

df["sp_url_valid"] = df["spotify_url"].astype(str).apply(lambda x: bool(SP_PATTERN.match(x.strip())))
df["yt_url_valid"] = df["youtube_url"].astype(str).apply(lambda x: bool(YT_PATTERN.match(x.strip())))
df["ig_url_valid"] = df["instagram_url"].astype(str).apply(lambda x: bool(IG_PATTERN.match(x.strip())))

# 2. Suspicious Low-Count Artifact Flags (< 1000 followers/listeners for verified artists)
df["suspicious_ig"] = df["instagram_followers"].fillna(0).astype(int) < 1000
df["suspicious_yt"] = df["yt_subscribers"].fillna(0).astype(int) < 1000
df["suspicious_sp"] = df["spotify_monthly_listeners"].fillna(0).astype(int) < 1000

# 3. Ratio Anomaly Test: Large streaming volume vs trivial social counts
# If an artist has > 1,000,000 monthly listeners, IG < 10k or YT < 10k is flagged for review
df["ratio_anomaly"] = (
    (df["spotify_monthly_listeners"] > 1_000_000) &
    ((df["instagram_followers"] < 10_000) | (df["yt_subscribers"] < 10_000))
)

# 4. Overall Verification Status
df["audit_pass"] = (
    df["sp_url_valid"] &
    df["yt_url_valid"] &
    df["ig_url_valid"] &
    (~df["suspicious_ig"]) &
    (~df["suspicious_yt"]) &
    (~df["suspicious_sp"]) &
    (~df["ratio_anomaly"])
)

passed_count = df["audit_pass"].sum()
total_count = len(df)

print("=" * 70)
print("       HIFFI COHORT PRODUCTION TELEMETRY & IDENTITY AUDIT")
print("=" * 70)
print(f"Total Cohort Artists Evaluated:               {total_count}")
print(f"Spotify Canonical 22-char URLs:               {df['sp_url_valid'].sum()} / {total_count}")
print(f"YouTube Canonical Channel IDs (UC...):        {df['yt_url_valid'].sum()} / {total_count}")
print(f"Instagram Clean Canonical URLs:               {df['ig_url_valid'].sum()} / {total_count}")
print(f"Artists Passing 100% Zero-Defect Standards:   {passed_count} / {total_count}")
print("-" * 70)

# Flag any anomalies
anomalies = df[~df["audit_pass"]]
if not anomalies.empty:
    print(f"\n[ATTENTION] Found {len(anomalies)} artist(s) flagged for manual verification:")
    for _, row in anomalies.iterrows():
        reasons = []
        if not row["sp_url_valid"]: reasons.append("Malformed Spotify URL")
        if not row["yt_url_valid"]: reasons.append("Non-canonical YouTube URL")
        if not row["ig_url_valid"]: reasons.append("Malformed Instagram URL")
        if row["suspicious_sp"]: reasons.append(f"Low Spotify ({row['spotify_monthly_listeners']:,})")
        if row["suspicious_yt"]: reasons.append(f"Low YT Subs ({row['yt_subscribers']:,})")
        if row["suspicious_ig"]: reasons.append(f"Low IG Followers ({row['instagram_followers']:,})")
        if row["ratio_anomaly"]: reasons.append("Cross-platform ratio mismatch")
        
        print(f"\n* Artist: {row['artist_name']} (ID: {row['artist_id']})")
        print(f"  Flagged issues: {', '.join(reasons)}")
        print(f"  Spotify:   {row['spotify_url']}")
        print(f"  YouTube:   {row['youtube_url']}")
        print(f"  Instagram: {row['instagram_url']}")
else:
    print("\nSUCCESS: All 500 artists passed identity format and ratio verification.")

# Export full validation ledger
proof_export_path = PROJECT_ROOT / "data" / "production_verification_audit.csv"
df.to_csv(proof_export_path, index=False, encoding="utf-8")
print(f"\nDetailed per-artist audit saved to:\n{proof_export_path}")