hiffi-artist-intelligence/data/roster_loader.py

"""
data/roster_loader.py
Loads, cleans, and validates artist rosters from CSV files.
"""

import os
from typing import List, Dict, Any
import pandas as pd


class RosterLoader:
    REQUIRED_COLUMNS = ["name", "identifier"]

    def __init__(self, file_path: str = "data/seed_artists_500.csv"):
        self.file_path = file_path

    def load_roster(self) -> List[Dict[str, Any]]:
        """
        Loads the CSV roster, validates columns, drops empty/duplicate rows,
        and returns a clean list of artist dictionaries.
        """
        if not os.path.exists(self.file_path):
            raise FileNotFoundError(
                f"Seed file '{self.file_path}' not found. Please verify the path."
            )

        df = pd.read_csv(self.file_path)

        # Check for required schema columns
        missing_cols = [c for c in self.REQUIRED_COLUMNS if c not in df.columns]
        if missing_cols:
            raise ValueError(
                f"Seed CSV is missing required columns: {missing_cols}. "
                f"Expected columns: {self.REQUIRED_COLUMNS}"
            )

        # Clean string whitespace and drop missing entries
        df["name"] = df["name"].astype(str).str.strip()
        df["identifier"] = df["identifier"].astype(str).str.strip()

        df = df[
            (df["name"] != "")
            & (df["name"] != "nan")
            & (df["identifier"] != "")
            & (df["identifier"] != "nan")
        ]

        # Drop duplicates by artist name (keep first occurrence)
        initial_len = len(df)
        df = df.drop_duplicates(subset=["name"], keep="first")
        dedup_len = len(df)

        if initial_len != dedup_len:
            print(f"ℹ️ Removed {initial_len - dedup_len} duplicate artist entries from roster.")

        print(f"📋 Successfully loaded {len(df)} validated artists from '{self.file_path}'.")
        return df.to_dict(orient="records")