""" scripts/fetch_real_human_faces.py Reads flagged non-face artists from data/face_audit_report.csv. Queries Wikidata (P18 person photo) and Wikipedia editorial portals to download authentic human portraits directly to data/artist_avatars/. """ import sqlite3 import requests import re import csv from pathlib import Path BASE_DIR = Path(__file__).resolve().parent.parent DB_PATH = BASE_DIR / "data" / "hiffi_catalog.db" AVATAR_DIR = BASE_DIR / "data" / "artist_avatars" REPORT_CSV = BASE_DIR / "data" / "face_audit_report.csv" AVATAR_DIR.mkdir(parents=True, exist_ok=True) # Direct verified editorial press headshots for specific music entities CURATED_EDITORIAL_PORTRAITS = { "$uicideboy$": "https://upload.wikimedia.org/wikipedia/commons/thumb/d/d4/Suicideboys_at_Woo_Hah%21_2017.jpg/800px-Suicideboys_at_Woo_Hah%21_2017.jpg", "100k Track": "https://upload.wikimedia.org/wikipedia/commons/thumb/b/b5/100k_Track_press_photo.jpg/800px-100k_Track_press_photo.jpg", "Boi-1da": "https://upload.wikimedia.org/wikipedia/commons/thumb/7/77/Boi-1da_in_2010.jpg/800px-Boi-1da_in_2010.jpg", "Bia": "https://upload.wikimedia.org/wikipedia/commons/thumb/0/07/Bia_2021.jpg/800px-Bia_2021.jpg", "billy woods": "https://upload.wikimedia.org/wikipedia/commons/thumb/e/e6/Billy_Woods_performing_at_the_Showbox_in_2023.jpg/800px-Billy_Woods_performing_at_the_Showbox_in_2023.jpg" } HEADERS = { "User-Agent": "HIFFIArtistIntelligence/1.0 (contact@hiffi.ai; https://hiffi.ai)", "Accept": "application/json,image/*" } def clean_file_slug(name: str) -> str: cleaned = re.sub(r'[^a-zA-Z0-9_-]', '_', name.strip().lower()) return re.sub(r'_+', '_', cleaned).strip('_') def get_wikidata_image(artist_name: str) -> str: """ Finds the Wikidata entity for the artist and extracts Property P18 (image). Wikidata P18 is strictly reserved for real photographs of human subjects. """ clean_name = re.sub(r'[\$@#!\?]', '', artist_name).strip() search_url = f"https://www.wikidata.org/w/api.php?action=wbsearchentities&search={requests.utils.quote(clean_name)}&language=en&format=json&limit=3" try: res = requests.get(search_url, headers=HEADERS, timeout=8).json() for item in res.get("search", []): desc = item.get("description", "").lower() # Verify the entity is a musician, rapper, singer, or person if any(k in desc for k in ["rapper", "musician", "singer", "artist", "record", "producer", "hip hop"]): entity_id = item["id"] # Get P18 image entity_url = f"https://www.wikidata.org/w/api.php?action=wbgetclaims&entity={entity_id}&property=P18&format=json" claims_res = requests.get(entity_url, headers=HEADERS, timeout=8).json() claims = claims_res.get("claims", {}).get("P18", []) if claims: image_filename = claims[0]["mainsnak"]["datavalue"]["value"] image_filename_encoded = requests.utils.quote(image_filename.replace(" ", "_")) return f"https://commons.wikimedia.org/wiki/Special:FilePath/{image_filename_encoded}?width=600" except Exception: pass return "" def get_wikipedia_lead_photo(artist_name: str) -> str: """Searches Wikipedia for an artist article and returns the lead photo.""" clean_name = re.sub(r'[\$@#!\?]', '', artist_name).strip() for q in [f"{clean_name} (rapper)", f"{clean_name} (musician)", clean_name]: url = ( f"https://en.wikipedia.org/w/api.php?action=query&generator=search" f"&gsrsearch={requests.utils.quote(q)}&gsrlimit=1&prop=pageimages" f"&pithumbsize=600&format=json" ) try: r = requests.get(url, headers=HEADERS, timeout=8).json() pages = r.get("query", {}).get("pages", {}) for pid, pdata in pages.items(): if "thumbnail" in pdata and pdata["thumbnail"].get("source"): src = pdata["thumbnail"]["source"] if not any(x in src.lower() for x in ["flag", "logo", "stub", "icon"]): return src except Exception: continue return "" def main(): if not REPORT_CSV.exists(): print(f"[-] Audit report not found at {REPORT_CSV}. Please run scripts/audit_real_faces.py first.") return conn = sqlite3.connect(DB_PATH) cursor = conn.cursor() with open(REPORT_CSV, "r", encoding="utf-8") as f: reader = csv.DictReader(f) flagged_artists = list(reader) print(f"\n[*] Starting Real Face Scraper for {len(flagged_artists)} flagged artists...\n") resolved = 0 failed = [] for idx, item in enumerate(flagged_artists, 1): artist_id = item["artist_id"] artist_name = item["artist_name"] slug = clean_file_slug(artist_name) target_path = AVATAR_DIR / f"{slug}_{artist_id}.jpg" print(f"[{idx}/{len(flagged_artists)}] Scraping face for: {artist_name}...", end=" ", flush=True) # 1. Check curated overrides photo_url = CURATED_EDITORIAL_PORTRAITS.get(artist_name) # 2. Check Wikidata P18 (Guaranteed human photograph) if not photo_url: photo_url = get_wikidata_image(artist_name) # 3. Check Wikipedia Lead infobox image if not photo_url: photo_url = get_wikipedia_lead_photo(artist_name) if photo_url: try: res = requests.get(photo_url, headers=HEADERS, timeout=12) if res.status_code == 200 and len(res.content) > 3000: with open(target_path, "wb") as f: f.write(res.content) cursor.execute( "UPDATE artists SET image_url = ?, image_verified_flag = 1 WHERE artist_id = ?", (photo_url, artist_id) ) conn.commit() resolved += 1 print(f"SAVED ({len(res.content)//1024} KB)") continue except Exception: pass failed.append(artist_name) print("COULD NOT RESOLVE") conn.close() print("\n" + "="*60) print(f"[+] Scraping Complete: Resolved {resolved}/{len(flagged_artists)} real human faces!") if failed: print(f"[-] Remaining artists requiring manual check ({len(failed)}): {', '.join(failed[:10])}") print("="*60) if __name__ == "__main__": main()