60 lines
2.3 KiB
Python
60 lines
2.3 KiB
Python
import re
|
|
import pandas as pd
|
|
from urllib.parse import urlparse
|
|
|
|
def clean_url(url: str) -> str:
|
|
"""Normalize URLs by stripping query params, trailing slashes, and spaces."""
|
|
if not isinstance(url, str) or not url.strip() or url.strip().lower() in ["nan", "none", ""]:
|
|
return ""
|
|
url = url.strip().rstrip("/")
|
|
if not re.match(r"^https?://", url, re.I):
|
|
url = "https://" + url
|
|
parsed = urlparse(url)
|
|
return f"{parsed.scheme}://{parsed.netloc}{parsed.path}".rstrip("/")
|
|
|
|
def extract_channel_key(yt_url: str) -> str:
|
|
"""Extract unique channel ID or handle to catch cross-name channel duplicates."""
|
|
clean = clean_url(yt_url).lower()
|
|
if not clean:
|
|
return ""
|
|
parts = [p for p in urlparse(clean).path.split("/") if p]
|
|
if not parts:
|
|
return ""
|
|
return parts[-1]
|
|
|
|
def deduplicate_inventory(input_path: str = "artist_inventory_artists.csv") -> pd.DataFrame:
|
|
print(f"Loading {input_path}...")
|
|
df = pd.read_csv(input_path)
|
|
initial_count = len(df)
|
|
print(f"Initial raw rows: {initial_count}")
|
|
|
|
# 1. Clean strings and URLs
|
|
for col in ["Name", "Location"]:
|
|
df[col] = df[col].fillna("").astype(str).str.strip()
|
|
|
|
for col in ["Instagram", "Youtube", "TikTok", "Facebook"]:
|
|
df[col] = df[col].apply(clean_url)
|
|
|
|
# 2. Generate normalized match keys
|
|
df["clean_name"] = df["Name"].str.lower().str.replace(r"[^\w\s]", "", regex=True).str.strip()
|
|
df["yt_key"] = df["Youtube"].apply(extract_channel_key)
|
|
|
|
# 3. Drop exact duplicates on Name + YouTube
|
|
df_dedup = df.drop_duplicates(subset=["clean_name", "yt_key"], keep="first").copy()
|
|
|
|
# 4. Check for duplicate YouTube channels across different names
|
|
yt_conflicts = df_dedup[df_dedup["yt_key"] != ""].duplicated(subset=["yt_key"], keep=False).sum()
|
|
print(f"Rows with shared YouTube channels across different names: {yt_conflicts}")
|
|
|
|
dropped = initial_count - len(df_dedup)
|
|
print(f"Deduplicated count: {len(df_dedup)} (Dropped {dropped} duplicate rows)")
|
|
|
|
# Clean temporary helper columns before returning
|
|
df_dedup.drop(columns=["clean_name", "yt_key"], inplace=True)
|
|
return df_dedup
|
|
|
|
if __name__ == "__main__":
|
|
cleaned_df = deduplicate_inventory()
|
|
cleaned_df.to_csv("cleaned_artists_inventory.csv", index=False)
|
|
print("Saved cleaned data to cleaned_artists_inventory.csv")
|