106 lines
3.5 KiB
Python
106 lines
3.5 KiB
Python
import os
|
|
import re
|
|
import time
|
|
import ftfy
|
|
import pandas as pd
|
|
from curl_cffi import requests
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
|
|
DATA_PATH = "data/master_artists_intelligence_17k_final.csv"
|
|
OUTPUT_PATH = "data/master_artists_intelligence_17k_final.csv"
|
|
LASTFM_API_KEY = os.getenv("LASTFM_API_KEY")
|
|
|
|
|
|
def clean_artist_query(raw_name: str) -> str:
|
|
"""Fix mojibake encoding and clean handle artifacts."""
|
|
if not raw_name or pd.isna(raw_name):
|
|
return ""
|
|
|
|
# 1. Fix Mojibake encoding (ãl -> al, Amoré -> Amoré)
|
|
name = ftfy.fix_text(str(raw_name))
|
|
|
|
# 2. Strip leading numbers attached to handles (e.g. 1dessdior -> dess dior)
|
|
name = re.sub(r"^[0-9]+([a-zA-Z])", r"\1", name)
|
|
|
|
# 3. Strip trailing underscores or dots commonly found in IG handles
|
|
name = re.sub(r"[_\.]+$", "", name)
|
|
name = name.replace("_", " ").replace(".", " ")
|
|
|
|
# 4. Remove collaboration trails like & the Brand New S
|
|
if " & the " in name:
|
|
name = name.split(" & the ")[0]
|
|
|
|
return name.strip()
|
|
|
|
|
|
def query_lastfm_search(session: requests.Session, query: str) -> dict:
|
|
url = "https://ws.audioscrobbler.com/2.0/"
|
|
params = {
|
|
"method": "artist.search",
|
|
"artist": query,
|
|
"api_key": LASTFM_API_KEY,
|
|
"format": "json",
|
|
"limit": 3
|
|
}
|
|
headers = {"User-Agent": "HIFFIRecovery/1.0 (srilekhar86@gmail.com)"}
|
|
|
|
try:
|
|
resp = session.get(url, params=params, headers=headers, impersonate="chrome120", timeout=8)
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
matches = data.get("results", {}).get("artistmatches", {}).get("artist", [])
|
|
if isinstance(matches, dict):
|
|
matches = [matches]
|
|
if matches:
|
|
# Pick the top result with registered listeners
|
|
for match in matches:
|
|
listeners = int(match.get("listeners", 0))
|
|
if listeners > 0:
|
|
return {
|
|
"canonical_name": match.get("name"),
|
|
"listeners": listeners,
|
|
"url": match.get("url")
|
|
}
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
def run_recovery():
|
|
print(f"Loading master catalog: {DATA_PATH}")
|
|
df = pd.read_csv(DATA_PATH)
|
|
|
|
unv_mask = df["catalog_rank"].isna() | (df["catalog_intelligence_tier"] == "TIER_3_UNAUDITED_DIGITAL_CATALOG")
|
|
unv_indices = df[unv_mask].index.tolist()
|
|
print(f"Targeting {len(unv_indices)} unverified candidates for registry recovery...\n")
|
|
|
|
session = requests.Session()
|
|
recovered = 0
|
|
|
|
for idx in unv_indices:
|
|
raw_name = df.at[idx, "artist_name"]
|
|
cleaned_query = clean_artist_query(raw_name)
|
|
|
|
if not cleaned_query or len(cleaned_query) <= 1:
|
|
continue
|
|
|
|
res = query_lastfm_search(session, cleaned_query)
|
|
time.sleep(0.2) # Respect Last.fm rate limits
|
|
|
|
if res:
|
|
df.at[idx, "lastfm_listeners"] = res["listeners"]
|
|
df.at[idx, "lastfm_url"] = res["url"]
|
|
df.at[idx, "catalog_intelligence_tier"] = "TIER_3_LASTFM_VERIFIED_ENTITY"
|
|
recovered += 1
|
|
print(f"[{recovered}] Recovered: '{raw_name}' -> '{res['canonical_name']}' ({res['listeners']:,} listeners)")
|
|
|
|
print(f"\nRecovery pass complete: {recovered} / {len(unv_indices)} artists recovered.")
|
|
df.to_csv(OUTPUT_PATH, index=False)
|
|
print(f"Master file updated: {OUTPUT_PATH}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run_recovery()
|
|
|