72 lines
2.9 KiB
Python
72 lines
2.9 KiB
Python
import os
|
|
import re
|
|
import pandas as pd
|
|
|
|
DATA_DIR = "data"
|
|
INPUT_PATH = os.path.join(DATA_DIR, "master_artists_intelligence_17k.csv")
|
|
FINAL_OUTPUT_PATH = os.path.join(DATA_DIR, "master_artists_intelligence_17k_final.csv")
|
|
|
|
|
|
def normalize_name(text: str) -> str:
|
|
if not text:
|
|
return ""
|
|
cleaned = re.sub(r"\(.*?\)|\[.*?\]", "", str(text))
|
|
cleaned = re.sub(r"[^\w\s]", "", cleaned)
|
|
return cleaned.strip().lower()
|
|
|
|
|
|
def run_final_qa():
|
|
print(f"Reading dataset: {INPUT_PATH}")
|
|
df = pd.read_csv(INPUT_PATH)
|
|
initial_count = len(df)
|
|
print(f"Initial total artists: {initial_count:,}")
|
|
|
|
# 1. Deduplication by normalized name and artist_id
|
|
df["norm_key"] = df["artist_name"].apply(normalize_name)
|
|
|
|
# Sort so verified records take priority over unverified during deduplication
|
|
tier_order = {
|
|
"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_UNAUDITED_DIGITAL_CATALOG": 6,
|
|
}
|
|
df["tier_rank"] = df["catalog_intelligence_tier"].map(tier_order).fillna(99)
|
|
df.sort_values(by=["tier_rank", "audited_sales_units", "spotify_total_streams", "spotify_monthly_listeners"],
|
|
ascending=[True, False, False, False], inplace=True)
|
|
|
|
# Drop duplicate normalized names, retaining highest-ranked tier
|
|
df.drop_duplicates(subset=["norm_key"], keep="first", inplace=True)
|
|
deduped_count = len(df)
|
|
print(f"Deduplicated count: {deduped_count:,} (Removed {initial_count - deduped_count:,} duplicate variants)")
|
|
|
|
# 2. Add Unified Sorting Metric Columns for Downstream UI
|
|
def get_primary_metric(row):
|
|
tier = row["catalog_intelligence_tier"]
|
|
if tier in ["TIER_1_GLOBAL_SUPERSTAR_CSPC", "TIER_2A_MAJOR_COMMERCIAL_CORE", "TIER_2B_CERTIFIED_CATALOG"]:
|
|
return pd.Series(["AUDITED_SALES_UNITS", row["audited_sales_units"]])
|
|
elif tier == "TIER_3A_VERIFIED_DIGITAL_CATALOG":
|
|
return pd.Series(["LIFETIME_STREAMS", row["spotify_total_streams"]])
|
|
elif tier == "TIER_3B_ACTIVE_MONTHLY_STREAMERS":
|
|
return pd.Series(["MONTHLY_LISTENERS", row["spotify_monthly_listeners"]])
|
|
return pd.Series(["UNAUDITED", None])
|
|
|
|
df[["primary_metric_type", "primary_metric_value"]] = df.apply(get_primary_metric, axis=1)
|
|
|
|
# Clean temporary helper columns
|
|
df.drop(columns=["norm_key", "tier_rank"], inplace=True)
|
|
|
|
# Sort final export: Tiered priority first, then primary metric descending
|
|
df.sort_values(by=["catalog_intelligence_tier", "primary_metric_value"],
|
|
ascending=[True, False], inplace=True)
|
|
|
|
df.to_csv(FINAL_OUTPUT_PATH, index=False)
|
|
print(f"\nFinal hardened intelligence catalog exported to: {FINAL_OUTPUT_PATH}")
|
|
print("\nFinal Tier Breakdown:")
|
|
print(df["catalog_intelligence_tier"].value_counts())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run_final_qa() |