""" collectors/chartmasters_collector.py Public commercial record sales and CSPC telemetry collector. Indexes artist all-time equivalent album sales (EAS/CSPC), pure album sales, physical singles, digital singles, and streaming conversions from verified ChartMasters all-time ranking records. Adheres strictly to the HIFFI Zero-Imputation Policy: - Artists not cataloged on ChartMasters are stored as NOT_CATALOGED_YET with None/NaN metrics. - Uncataloged artists are NEVER imputed with 0. - Pure album sales are kept strictly distinct from reported Total EAS. - Computes cohort_sales_rank restricted solely to verified cataloged artists. """ from pathlib import Path import re from typing import Any, Dict, List, Optional import pandas as pd PROJECT_ROOT = Path(__file__).resolve().parent.parent DEFAULT_SALES_REF_PATH = PROJECT_ROOT / "data" / "chartmasters_sales_reference.csv" class ChartMastersSalesCollector: """ Ingests and normalizes commercial record sales metrics. Enforces strict typing, null governance, multi-pass alias resolution, and cohort ranking. """ STAGE_ALIASES: Dict[str, str] = { "a$ap rocky": "asap rocky", "asap rocky": "asap rocky", "a$ap ferg": "asap ferg", "asap ferg": "asap ferg", "a$ap mob": "asap mob", "joey bada$$": "joey badass", "6ix9ine": "sixnine", "curren$y": "currensy", "ty dolla $ign": "ty dolla sign", "kool g rap": "kool g rap", "t-pain": "tpain", "jay-z": "jay z", "jay z": "jay z", "e-40": "e40", "jeezy": "young jeezy", "young jeezy": "young jeezy", "the notorious big": "notorious big", "notorious big": "notorious big", "2pac": "tupac shakur", "tupac shakur": "tupac shakur", "tupac": "tupac shakur", "tyler the creator": "tyler the creator", "fivio foreign": "fivio foreign", "fivioforeign": "fivio foreign", "french montana": "french montana", "frenchmontana": "french montana", } def __init__(self, reference_path: Optional[Path] = None): self.reference_path = reference_path or DEFAULT_SALES_REF_PATH self.cached_metrics: Dict[str, Dict[str, Any]] = {} self.is_loaded = False @classmethod def normalize_artist_name(cls, name: str) -> str: """ Normalizes artist names across multiple passes: 1. Strips punctuation, quotes, periods, and currency symbols. 2. Resolves typographical ligature variations (e.g., $ -> s, & -> and). 3. Maps against canonical STAGE_ALIASES. """ if not name: return "" norm = str(name).strip().lower() # Pass 1: Common stylized replacements norm = norm.replace("$", "s").replace("&", "and").replace("+", "and") # Pass 2: Strip all non-alphanumeric except spaces norm = re.sub(r"[^\w\s]", "", norm) norm = re.sub(r"\s+", " ", norm).strip() # Pass 3: Alias dictionary lookup if norm in cls.STAGE_ALIASES: norm = cls.STAGE_ALIASES[norm] return norm @staticmethod def _parse_numeric(val: Any) -> Optional[int]: """Safely casts numeric values or strings to integers, returning None if invalid.""" if val is None or pd.isna(val): return None cleaned = re.sub(r"[^\d]", "", str(val).strip()) if not cleaned: return None try: return int(cleaned) except ValueError: return None def load_sales_table(self, html_content: Optional[str] = None) -> int: """ Loads and indexes record sales metrics. If html_content is provided (e.g. unit tests), parses the HTML table. Otherwise, loads verified records from the reference CSV. """ if html_content: from bs4 import BeautifulSoup soup = BeautifulSoup(html_content, "html.parser") table = soup.find("table") if not table: tables = soup.find_all("table") table = tables[0] if tables else None if not table: return 0 parsed_count = 0 for row in table.find_all("tr"): cells = row.find_all(["td", "th"]) if len(cells) < 7: continue rank = self._parse_numeric(cells[0].get_text(strip=True)) if rank is None: continue raw_name = cells[1].get_text(strip=True) norm_name = self.normalize_artist_name(raw_name) if not norm_name: continue self.cached_metrics[norm_name] = { "chartmasters_raw_name": raw_name, "chartmasters_sales_rank": rank, "pure_album_sales": self._parse_numeric(cells[2].get_text(strip=True)), "physical_singles_sales": self._parse_numeric(cells[3].get_text(strip=True)), "digital_singles_sales": self._parse_numeric(cells[4].get_text(strip=True)), "streaming_eas": self._parse_numeric(cells[5].get_text(strip=True)), "total_eas_cspc": self._parse_numeric(cells[6].get_text(strip=True)), "sales_source_url": "https://chartmasters.org/best-selling-artists-of-all-time/", "sales_collected_at": "2026-10-28T12:00:00Z", "sales_data_status": "VERIFIED_CSPC", } parsed_count += 1 self.is_loaded = True return parsed_count # Load from reference dataset if not self.reference_path.exists(): print(f"[ChartMastersCollector] Reference file not found at {self.reference_path}") return 0 ref_df = pd.read_csv(self.reference_path) parsed_count = 0 for _, row in ref_df.iterrows(): raw_name = str(row["artist_name"]) norm_name = self.normalize_artist_name(raw_name) if not norm_name: continue self.cached_metrics[norm_name] = { "chartmasters_raw_name": raw_name, "chartmasters_sales_rank": self._parse_numeric(row.get("chartmasters_global_rank")), "pure_album_sales": self._parse_numeric(row.get("pure_album_sales")), "physical_singles_sales": self._parse_numeric(row.get("physical_singles_sales")), "digital_singles_sales": self._parse_numeric(row.get("digital_singles_sales")), "streaming_eas": self._parse_numeric(row.get("streaming_eas")), "total_eas_cspc": self._parse_numeric(row.get("total_eas_cspc")), "sales_source_url": str(row.get("source_url", "https://chartmasters.org/best-selling-artists-of-all-time/")), "sales_collected_at": str(row.get("collected_at", "2026-10-28T12:00:00Z")), "sales_data_status": "VERIFIED_CSPC", } parsed_count += 1 self.is_loaded = True print(f"[ChartMastersCollector] Indexed {parsed_count} verified artist sales benchmark records.") return parsed_count def get_artist_sales_metrics(self, artist_name: str) -> Dict[str, Optional[Any]]: """ Retrieves record sales telemetry for a given artist name. Enforces strict null retention: unmatched artists return NOT_CATALOGED_YET with None values. """ if not self.is_loaded: self.load_sales_table() norm_name = self.normalize_artist_name(artist_name) if norm_name in self.cached_metrics: return self.cached_metrics[norm_name].copy() return { "chartmasters_raw_name": None, "chartmasters_sales_rank": None, "pure_album_sales": None, "physical_singles_sales": None, "digital_singles_sales": None, "streaming_eas": None, "total_eas_cspc": None, "sales_source_url": None, "sales_collected_at": None, "sales_data_status": "NOT_CATALOGED_YET", } def enrich_cohort(self, df: pd.DataFrame, name_column: str = "name") -> pd.DataFrame: """ Enriches an artist DataFrame with sales metrics, maintaining zero-imputation policy. Computes cohort_sales_rank solely across matched artists. """ if not self.is_loaded: self.load_sales_table() enriched = df.copy() sales_records: List[Dict[str, Any]] = [] for artist in enriched[name_column]: metrics = self.get_artist_sales_metrics(str(artist)) sales_records.append(metrics) sales_df = pd.DataFrame(sales_records, index=enriched.index) for col in sales_df.columns: enriched[col] = sales_df[col] # Calculate cohort-specific rank for cataloged artists enriched["cohort_sales_rank"] = pd.NA matched_mask = enriched["sales_data_status"] == "VERIFIED_CSPC" if matched_mask.any(): cohort_ranks = ( enriched.loc[matched_mask, "total_eas_cspc"] .rank(ascending=False, method="min") .astype(int) ) enriched.loc[matched_mask, "cohort_sales_rank"] = cohort_ranks return enriched if __name__ == "__main__": collector = ChartMastersSalesCollector() collector.load_sales_table() test_cohort = [ "Eminem", "Drake", "Kendrick Lamar", "Michael Jackson", "Jay-Z", "A$AP Rocky", "SZA", "Future", "The Notorious B.I.G.", "Tyler, The Creator", "Ice Cube", "DMX", "Unknown Indie Artist" ] print("\nValidating ChartMasters Benchmark Data & Strict Governance:") print(f"{'Artist':25} | {'Global #':>8} | {'Total EAS (CSPC)':>18} | {'Pure Albums':>14} | {'Status':>18}") print("-" * 91) for artist in test_cohort: m = collector.get_artist_sales_metrics(artist) rank_str = f"#{m['chartmasters_sales_rank']}" if m["chartmasters_sales_rank"] is not None else "N/A" eas_str = f"{m['total_eas_cspc']:,}" if m["total_eas_cspc"] is not None else "NaN" album_str = f"{m['pure_album_sales']:,}" if m["pure_album_sales"] is not None else "NaN" print(f"{artist:25} | {rank_str:>8} | {eas_str:>18} | {album_str:>14} | {m['sales_data_status']:>18}")