87 lines
2.9 KiB
Python
87 lines
2.9 KiB
Python
|
"""
|
||
|
|
enrich_instagram_metadata.py
|
||
|
|
Instagram audience verification and confidence evaluation engine.
|
||
|
|
Processes the HIFFI 500 cohort through InstagramCandidateVerifier.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
from pathlib import Path
|
||
|
|
import re
|
||
|
|
import pandas as pd
|
||
|
|
|
||
|
|
from collectors.instagram_verifier import InstagramCandidateVerifier
|
||
|
|
|
||
|
|
DATA_PATH = Path("data/hiffi_multiplatform_500.csv")
|
||
|
|
|
||
|
|
|
||
|
|
def sanitize_handle(raw_handle: object) -> str:
|
||
|
|
if pd.isna(raw_handle) or raw_handle is None:
|
||
|
|
return ""
|
||
|
|
h = str(raw_handle).strip()
|
||
|
|
# Strip URL structures if present
|
||
|
|
h = re.sub(r"https?://(?:www\.)?instagram\.com/", "", h)
|
||
|
|
h = h.rstrip("/").replace("@", "").strip()
|
||
|
|
return h if h.lower() not in ("nan", "none", "") else ""
|
||
|
|
|
||
|
|
|
||
|
|
def run_instagram_enrichment():
|
||
|
|
if not DATA_PATH.exists():
|
||
|
|
print(f"File not found: {DATA_PATH}")
|
||
|
|
return
|
||
|
|
|
||
|
|
df = pd.read_csv(DATA_PATH)
|
||
|
|
print(f"Processing Instagram verification across {len(df)} artists...")
|
||
|
|
|
||
|
|
verifier = InstagramCandidateVerifier()
|
||
|
|
now_utc = datetime.now(timezone.utc).isoformat()
|
||
|
|
|
||
|
|
verified_count = 0
|
||
|
|
updated_handles = 0
|
||
|
|
|
||
|
|
for idx, row in df.iterrows():
|
||
|
|
name = row["name"]
|
||
|
|
raw_handle = row.get("instagram_handle")
|
||
|
|
handle = sanitize_handle(raw_handle)
|
||
|
|
|
||
|
|
# Fallback candidate generation if handle is missing
|
||
|
|
if not handle:
|
||
|
|
# Generate canonical candidate from stage name
|
||
|
|
clean_name = re.sub(r"[^\w]", "", name.lower())
|
||
|
|
handle = clean_name
|
||
|
|
df.at[idx, "instagram_handle"] = handle
|
||
|
|
updated_handles += 1
|
||
|
|
|
||
|
|
followers = row.get("instagram_followers")
|
||
|
|
followers_val = int(followers) if pd.notna(followers) and str(followers).isdigit() else None
|
||
|
|
yt_name = row.get("channel_title")
|
||
|
|
|
||
|
|
# Evaluate candidate through algorithmic verifier
|
||
|
|
eval_res = verifier.evaluate_candidate(
|
||
|
|
artist_name=name,
|
||
|
|
candidate_handle=handle,
|
||
|
|
follower_count=followers_val,
|
||
|
|
youtube_channel_name=yt_name,
|
||
|
|
)
|
||
|
|
|
||
|
|
df.at[idx, "instagram_verification_status"] = eval_res["verification_status"]
|
||
|
|
df.at[idx, "instagram_confidence_score"] = eval_res["confidence_score"]
|
||
|
|
df.at[idx, "instagram_evidence"] = eval_res["evidence"]
|
||
|
|
|
||
|
|
if pd.isna(df.at[idx, "instagram_updated_at"]):
|
||
|
|
df.at[idx, "instagram_updated_at"] = now_utc
|
||
|
|
|
||
|
|
if eval_res["verification_status"] in ("AUTOMATED_VERIFIED", "MANUAL_VERIFIED"):
|
||
|
|
verified_count += 1
|
||
|
|
|
||
|
|
df.to_csv(DATA_PATH, index=False)
|
||
|
|
|
||
|
|
print("\n==========================================")
|
||
|
|
print("Instagram Verification Sweep Complete!")
|
||
|
|
print(f"Total Artists Verified: {verified_count} / {len(df)} ({(verified_count/len(df))*100:.1f}%)")
|
||
|
|
print(f"Handles Generated/Sanitized: {updated_handles}")
|
||
|
|
print(f"Saved to: {DATA_PATH}")
|
||
|
|
print("==========================================")
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
run_instagram_enrichment()
|