97 lines
3.7 KiB
Python
97 lines
3.7 KiB
Python
import os
|
|
import pandas as pd
|
|
|
|
DATA_DIR = "data"
|
|
INPUT_PATH = os.path.join(DATA_DIR, "master_artists_intelligence_17k_final.csv")
|
|
OUTPUT_PATH = os.path.join(DATA_DIR, "master_artists_intelligence_17k_final.csv")
|
|
|
|
|
|
def run_empirical_catalog_ranking():
|
|
if not os.path.exists(INPUT_PATH):
|
|
print(f"ERROR: Dataset not found at {INPUT_PATH}")
|
|
return
|
|
|
|
print(f"Loading master verified catalog: {INPUT_PATH}")
|
|
df = pd.read_csv(INPUT_PATH)
|
|
|
|
# 1. Define Strict Registry Tier Hierarchy
|
|
tier_priority = {
|
|
"TIER_1_GLOBAL_SUPERSTAR_CSPC": 1,
|
|
"TIER_2A_MAJOR_COMMERCIAL_CORE": 2,
|
|
"TIER_2B_CERTIFIED_CATALOG": 3,
|
|
"TIER_3A_VERIFIED_DIGITAL_CATALOG": 4,
|
|
"TIER_3B_ACTIVE_MONTHLY_STREAMERS": 5,
|
|
"TIER_3_LASTFM_VERIFIED_ENTITY": 6,
|
|
"TIER_3_UNAUDITED_DIGITAL_CATALOG": 7
|
|
}
|
|
|
|
df["tier_order"] = df["catalog_intelligence_tier"].map(tier_priority).fillna(99)
|
|
|
|
# 2. Extract Primary Audited Metric per Tier (No conversions, pure integers)
|
|
def extract_tier_metric(row):
|
|
order = row["tier_order"]
|
|
if order in [1, 2, 3]:
|
|
# CSPC EAS / RIAA Certified Units
|
|
val = row.get("audited_sales_units")
|
|
return float(val) if pd.notna(val) else 0.0
|
|
elif order == 4:
|
|
# Kworb Spotify Total Discography Streams
|
|
val = row.get("spotify_total_streams")
|
|
return float(val) if pd.notna(val) else 0.0
|
|
elif order == 5:
|
|
# Kworb Spotify Peak Monthly Listeners
|
|
val = row.get("spotify_monthly_listeners")
|
|
return float(val) if pd.notna(val) else 0.0
|
|
elif order == 6:
|
|
# Last.fm Global Verified Listeners
|
|
val = row.get("lastfm_listeners")
|
|
return float(val) if pd.notna(val) else 0.0
|
|
return 0.0
|
|
|
|
df["primary_empirical_metric"] = df.apply(extract_tier_metric, axis=1)
|
|
|
|
# 3. Sort strictly by Tier Order (1 -> 6), then by Primary Metric (Highest -> Lowest)
|
|
df.sort_values(
|
|
by=["tier_order", "primary_empirical_metric"],
|
|
ascending=[True, False],
|
|
inplace=True
|
|
)
|
|
df.reset_index(drop=True, inplace=True)
|
|
|
|
# 4. Generate Global Catalog Rank (Only for verified tiers 1 through 6)
|
|
verified_mask = df["tier_order"] <= 6
|
|
df["catalog_rank"] = None
|
|
df.loc[verified_mask, "catalog_rank"] = range(1, int(verified_mask.sum()) + 1)
|
|
|
|
# 5. Generate Peer Tier Rank
|
|
df["tier_rank"] = None
|
|
for tier_name in df["catalog_intelligence_tier"].unique():
|
|
if tier_name != "TIER_3_UNAUDITED_DIGITAL_CATALOG":
|
|
t_mask = df["catalog_intelligence_tier"] == tier_name
|
|
df.loc[t_mask, "tier_rank"] = range(1, int(t_mask.sum()) + 1)
|
|
|
|
# Clean up temporary helper columns
|
|
df.drop(columns=["tier_order", "primary_empirical_metric"], inplace=True)
|
|
|
|
# Save finalized ranked dataset
|
|
df.to_csv(OUTPUT_PATH, index=False)
|
|
print(f"\nEmpirical ranking complete! Saved to {OUTPUT_PATH}")
|
|
|
|
# Display Top 15 Overall
|
|
preview_cols = [
|
|
col for col in ["catalog_rank", "tier_rank", "artist_name", "catalog_intelligence_tier", "audited_sales_units", "lastfm_listeners"]
|
|
if col in df.columns
|
|
]
|
|
print("\n--- TOP 15 CATALOG ARTISTS ---")
|
|
print(df[preview_cols].head(15).to_string(index=False))
|
|
|
|
# Display Top 5 in Tier 3 Last.fm Verified Entities
|
|
lastfm_sample = df[df["catalog_intelligence_tier"] == "TIER_3_LASTFM_VERIFIED_ENTITY"]
|
|
if not lastfm_sample.empty:
|
|
print("\n--- TOP 5 TIER 3 (LAST.FM VERIFIED) ARTISTS ---")
|
|
lfm_cols = ["catalog_rank", "tier_rank", "artist_name", "lastfm_listeners", "lastfm_playcount"]
|
|
print(lastfm_sample[lfm_cols].head(5).to_string(index=False))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run_empirical_catalog_ranking() |