hiffi-artist-intelligence/scripts/audit_account_authenticity.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. Flag Suspicious Monikers/Handles (Fan pages, archives, update pages)
SUSPICIOUS_HANDLE_PATTERNS = re.compile(r"(fan|fans|daily|update|archive|world|gang|media|clips|tribute|leaks)", re.IGNORECASE)

def extract_ig_handle(url):
    m = re.search(r"instagram\.com/([^/?#]+)", str(url))
    return m.group(1).lower() if m else ""

df["ig_handle"] = df["instagram_url"].apply(extract_ig_handle)
df["ig_flagged_fan_token"] = df["ig_handle"].apply(lambda h: bool(SUSPICIOUS_HANDLE_PATTERNS.search(h)))

# 2. Flag Duplicate / Round Default Counts (e.g., clusters sharing exact same filler numbers)
ig_val_counts = df["instagram_followers"].value_counts()
yt_val_counts = df["yt_subscribers"].value_counts()

frequent_ig_defaults = set(ig_val_counts[ig_val_counts > 5].index)
frequent_yt_defaults = set(yt_val_counts[yt_val_counts > 5].index)

df["ig_is_shared_default"] = df["instagram_followers"].isin(frequent_ig_defaults)
df["yt_is_shared_default"] = df["yt_subscribers"].isin(frequent_yt_defaults)

# 3. Cross-Platform Audience Scale Inconsistency
# High Streaming (>500k) with Micro Socials (<5k)
df["scale_mismatch_ig"] = (df["spotify_monthly_listeners"] > 500_000) & (df["instagram_followers"] < 5_000)
df["scale_mismatch_yt"] = (df["spotify_monthly_listeners"] > 500_000) & (df["yt_subscribers"] < 5_000)

# Low Streaming (<10k) with Mega Socials (>2M) - indicates name collision with celebrities/influencers
df["name_collision_ig"] = (df["spotify_monthly_listeners"] < 10_000) & (df["instagram_followers"] > 2_000_000)

# 4. Overall Authenticity Risk Flag
df["authenticity_risk"] = (
    df["ig_flagged_fan_token"] |
    df["ig_is_shared_default"] |
    df["yt_is_shared_default"] |
    df["scale_mismatch_ig"] |
    df["scale_mismatch_yt"] |
    df["name_collision_ig"]
)

flagged_rows = df[df["authenticity_risk"]].copy()

print("=" * 72)
print("       HIFFI DEEP ACCOUNT AUTHENTICITY & FOLLOWER LEGITIMACY AUDIT")
print("=" * 72)
print(f"Total Cohort Artists Audited:               {len(df)}")
print(f"Profiles Flagged for Suspicious Fan Tokens: {df['ig_flagged_fan_token'].sum()}")
print(f"Profiles with Suspect Shared Default IG:   {df['ig_is_shared_default'].sum()}")
print(f"Profiles with Suspect Shared Default YT:   {df['yt_is_shared_default'].sum()}")
print(f"Cross-Platform Audience Scale Inconsistencies: {df['scale_mismatch_ig'].sum() + df['scale_mismatch_yt'].sum()}")
print(f"Potential Celebrity Name Collisions:       {df['name_collision_ig'].sum()}")
print(f"Total Profiles Requiring Manual Audit:     {len(flagged_rows)} / {len(df)}")
print("-" * 72)

if not flagged_rows.empty:
    print("\nSUSPICIOUS / OUTLIER PROFILES IDENTIFIED:")
    for _, r in flagged_rows.iterrows():
        reasons = []
        if r["ig_flagged_fan_token"]: reasons.append(f"Handle token mismatch: @{r['ig_handle']}")
        if r["ig_is_shared_default"]: reasons.append(f"Repeated IG default ({int(r['instagram_followers']):,})")
        if r["yt_is_shared_default"]: reasons.append(f"Repeated YT default ({int(r['yt_subscribers']):,})")
        if r["scale_mismatch_ig"]: reasons.append(f"Scale mismatch: SP {int(r['spotify_monthly_listeners']):,} vs IG {int(r['instagram_followers']):,}")
        if r["scale_mismatch_yt"]: reasons.append(f"Scale mismatch: SP {int(r['spotify_monthly_listeners']):,} vs YT {int(r['yt_subscribers']):,}")
        if r["name_collision_ig"]: reasons.append(f"Collision: SP {int(r['spotify_monthly_listeners']):,} vs IG {int(r['instagram_followers']):,}")

        print(f"\n* Artist: {r['artist_name']} (ID: {r['artist_id']})")
        print(f"  Flagged issues: {', '.join(reasons)}")
        print(f"  Instagram: https://www.instagram.com/{r['ig_handle']}/")
        print(f"  Spotify:   {r['spotify_url']}")
        print(f"  YouTube:   {r['youtube_url']}")
else:
    print("\nSUCCESS: Zero fan handles, zero suspicious clusters, and zero scale anomalies detected.")

# Export deep ledger
output_audit = PROJECT_ROOT / "data" / "account_authenticity_deep_audit.csv"
df.to_csv(output_audit, index=False, encoding="utf-8")
print(f"\nSaved complete deep audit report to:\n{output_audit}")