import requests import time import re import csv import os import json import multiprocessing from dotenv import load_dotenv import wikipedia # Import the wikipedia library # Load environment variables from .env file load_dotenv() # --- ANSI Color Codes --- class Colors: RESET = '\033[0m' RED = '\033[91m' GREEN = '\033[92m' YELLOW = '\033[93m' BLUE = '\033[94m' CYAN = '\033[96m' BOLD = '\033[1m' UNDERLINE = '\033[4m' def colored_print(message, color=Colors.RESET, bold=False): """Prints a message with specified ANSI color and boldness.""" prefix = color if bold: prefix += Colors.BOLD print(f"{prefix}{message}{Colors.RESET}") # --- Configuration for APIs --- MUSICBRAINZ_API_BASE_URL = "https://musicbrainz.org/ws/2" WIKIPEDIA_API_BASE_URL = "https://en.wikipedia.org/w/api.php" USER_AGENT = "SoFreshArtistFetcher/1.0 +https://github.com/yourusername/sofresh" # Unique User-Agent for this script MUSICBRAINZ_RATE_LIMIT_DELAY = 2.0 # MusicBrainz rate limit is 1 request per second, increased for robustness # --- Rate Limit Retry Settings --- MAX_RETRIES = 3 RETRY_DELAY_SECONDS = 5 # Initial delay for retries # --- Helper Functions --- def _clean_name(name): """Removes content within parentheses from a string.""" return re.sub(r'\s*\(.*\)\s*', '', name).strip() def get_musicbrainz_data(url, params=None, retries=0): """ Fetches data from the MusicBrainz API with appropriate headers, rate limiting, and retries. """ headers = { "User-Agent": USER_AGENT, "Accept": "application/json" # Request JSON response } try: response = requests.get(url, headers=headers, params=params) response.raise_for_status() # Raise an exception for HTTP errors (4xx or 5xx) time.sleep(MUSICBRAINZ_RATE_LIMIT_DELAY) # Adhere to MusicBrainz rate limit return response.json() except requests.exceptions.HTTPError as e: # Handle 429 (Too Many Requests) and 503 (Service Unavailable) if (e.response.status_code == 429 or e.response.status_code == 503) and retries < MAX_RETRIES: colored_print(f"DEBUG: MusicBrainz Rate limit/Service Unavailable hit ({e.response.status_code}). Retrying in {RETRY_DELAY_SECONDS * (2**retries)}s... (Attempt {retries + 1}/{MAX_RETRIES})", Colors.YELLOW) time.sleep(RETRY_DELAY_SECONDS * (2**retries)) # Exponential backoff return get_musicbrainz_data(url, params, retries + 1) colored_print(f"DEBUG: Error fetching MusicBrainz data from {url}: {e}", Colors.RED) if hasattr(e, 'response') and e.response is not None: colored_print(f"DEBUG: MusicBrainz Response content: {e.response.text}", Colors.RED) return None except requests.exceptions.RequestException as e: colored_print(f"DEBUG: Error fetching MusicBrainz data from {url}: {e}", Colors.RED) return None except json.JSONDecodeError: colored_print(f"DEBUG: Error decoding JSON from MusicBrainz response for {url}", Colors.RED) return None def search_wikipedia_page_title(query, retries=0): """ Searches Wikipedia for a given query and returns the title of the first article. This is used to get a precise title for the 'wikipedia' library. Includes retry logic for rate limits. """ params = { "action": "query", "list": "search", "srsearch": query, "format": "json", "srlimit": 1 # Get only the top result } try: response = requests.get(WIKIPEDIA_API_BASE_URL, params=params) response.raise_for_status() search_results = response.json() if search_results and search_results.get("query") and search_results["query"].get("search"): page_title = search_results["query"]["search"][0]["title"] return page_title, "Found" else: return None, "Not Found" except requests.exceptions.HTTPError as e: if (e.response.status_code == 429 or e.response.status_code == 503) and retries < MAX_RETRIES: colored_print(f"DEBUG: Wikipedia API rate limit/Service Unavailable hit for title search ({e.response.status_code}). Retrying in {RETRY_DELAY_SECONDS * (2**retries)}s... (Attempt {retries + 1}/{MAX_RETRIES})", Colors.YELLOW) time.sleep(RETRY_DELAY_SECONDS * (2**retries)) return search_wikipedia_page_title(query, retries + 1) colored_print(f"DEBUG: Network error searching Wikipedia page title for '{query}': {e}", Colors.RED) return None, "API Error" except requests.exceptions.RequestException as e: colored_print(f"DEBUG: Network error searching Wikipedia page title for '{query}': {e}", Colors.RED) return None, "Error" except Exception as e: colored_print(f"DEBUG: Unexpected error searching Wikipedia page title for '{query}': {e}", Colors.RED) return None, "Error" def get_wikipedia_summary(page_title, retries=0): """ Fetches the summary (first paragraph) of a Wikipedia page using the 'wikipedia' library. Includes retry logic for network/rate limit errors. """ try: summary = wikipedia.summary(page_title, sentences=2, auto_suggest=False, redirect=True) return summary, "Found" except wikipedia.exceptions.PageError: return None, "Page Not Found" except wikipedia.exceptions.DisambiguationError as e: if e.options and retries < MAX_RETRIES: # Try first option if disambiguation try: summary = wikipedia.summary(e.options[0], sentences=2, auto_suggest=False, redirect=True) return summary, "Found (Disambiguation Handled)" except requests.exceptions.HTTPError as http_e: if (http_e.response.status_code == 429 or http_e.response.status_code == 503): colored_print(f"DEBUG: Wikipedia API rate limit/Service Unavailable hit for disambiguation ({http_e.response.status_code}). Retrying in {RETRY_DELAY_SECONDS * (2**retries)}s... (Attempt {retries + 1}/{MAX_RETRIES})", Colors.YELLOW) time.sleep(RETRY_DELAY_SECONDS * (2**retries)) return get_wikipedia_summary(page_title, retries + 1) # Retry original page title return None, "Disambiguation HTTP Error" except Exception: return None, "Disambiguation Error" return None, "Disambiguation Error" except requests.exceptions.HTTPError as e: if (e.response.status_code == 429 or e.response.status_code == 503) and retries < MAX_RETRIES: colored_print(f"DEBUG: Wikipedia API rate limit/Service Unavailable hit for summary ({e.response.status_code}). Retrying in {RETRY_DELAY_SECONDS * (2**retries)}s... (Attempt {retries + 1}/{MAX_RETRIES})", Colors.YELLOW) time.sleep(RETRY_DELAY_SECONDS * (2**retries)) return get_wikipedia_summary(page_title, retries + 1) colored_print(f"DEBUG: Network error fetching Wikipedia summary for '{page_title}': {e}", Colors.RED) return None, "Network Error" except requests.exceptions.RequestException as e: # Catch other request errors colored_print(f"DEBUG: Network error fetching Wikipedia summary for '{page_title}': {e}", Colors.RED) return None, "Network Error" except Exception as e: colored_print(f"DEBUG: Unexpected error fetching Wikipedia summary for '{page_title}': {e}", Colors.RED) return None, "Unexpected Error" def process_artist_data(artist_name): """ Fetches detailed information for a single artist from MusicBrainz and Wikipedia. """ artist_data = { "Artist Name": artist_name, "Artist Bio": "N/A", "Artist Image": "N/A (Requires specific image API/scraping)", # Placeholder "Artist Website": "N/A", "Artist Facebook": "N/A", "Artist Instagram": "N/A", "Artist X (Twitter)": "N/A", "Artist Birth/Formation Date": "N/A", "Artist Location": "N/A", "Artist Wikipedia URL": "N/A" # New field for Wikipedia URL } bio_status = "N/A" website_status = "N/A" facebook_status = "N/A" instagram_status = "N/A" x_status = "N/A" birth_date_status = "N/A" location_status = "N/A" wikipedia_url_status = "N/A" # New status for Wikipedia URL colored_print(f"\nDEBUG: Processing artist: '{artist_name}'", Colors.BLUE) # --- Attempt 1: Prioritize Wikipedia direct search for Artist Bio (original name) --- colored_print(f"DEBUG: Attempting direct Wikipedia search for bio of '{artist_name}'", Colors.BLUE) wiki_title, wiki_status = search_wikipedia_page_title(f"{artist_name} musician") if wiki_title: bio_text, bio_status = get_wikipedia_summary(wiki_title) artist_data["Artist Bio"] = bio_text if bio_text else "N/A" if bio_text: artist_data["Artist Wikipedia URL"] = f"https://en.wikipedia.org/wiki/{wiki_title.replace(' ', '_')}" wikipedia_url_status = "Found (Direct)" colored_print(f"DEBUG: Direct Wikipedia bio for '{artist_name}': {bio_status}", Colors.GREEN if bio_text else Colors.YELLOW) else: bio_status = wiki_status colored_print(f"DEBUG: Direct Wikipedia bio search for '{artist_name}' failed: {bio_status}", Colors.YELLOW) # --- Now proceed with MusicBrainz for other data, and fallback for bio --- search_url = f"{MUSICBRAINZ_API_BASE_URL}/artist" params = {"query": artist_name, "fmt": "json", "limit": 5} # Get a few results to pick best score mb_search_results = get_musicbrainz_data(search_url, params) mbid = None best_score = -1 selected_mb_artist_name = "N/A" if mb_search_results and mb_search_results.get("artists"): colored_print(f"DEBUG: MusicBrainz search for '{artist_name}' returned {len(mb_search_results['artists'])} results.", Colors.BLUE) for artist_match in mb_search_results["artists"]: score = artist_match.get("score", 0) if score > best_score and artist_match.get("name", "").lower() == artist_name.lower(): best_score = score mbid = artist_match.get("id") selected_mb_artist_name = artist_match.get("name") colored_print(f"DEBUG: Found exact match '{selected_mb_artist_name}' with score {score}.", Colors.BLUE) break if not mbid and mb_search_results["artists"]: best_match = max(mb_search_results["artists"], key=lambda x: x.get("score", 0)) mbid = best_match.get("id") selected_mb_artist_name = best_match.get("name") colored_print(f"DEBUG: No exact match, selected best score '{selected_mb_artist_name}' with score {best_match.get('score',0)}.", Colors.BLUE) else: colored_print(f"DEBUG: MusicBrainz search for '{artist_name}' returned no artists or error.", Colors.YELLOW) if mbid: colored_print(f"DEBUG: Fetching details for MBID: {mbid} ('{selected_mb_artist_name}')", Colors.BLUE) artist_detail_url = f"{MUSICBRAINZ_API_BASE_URL}/artist/{mbid}" colored_print(f"DEBUG: Artist detail URL: {artist_detail_url}", Colors.BLUE) artist_detail_params = {"fmt": "json", "inc": "url-rels+aliases+tags"} mb_artist_details = get_musicbrainz_data(artist_detail_url, artist_detail_params) if mb_artist_details: # Fallback for Artist Bio if not found by direct Wikipedia search if artist_data["Artist Bio"] == "N/A": wikipedia_page_title = None if mb_artist_details.get("relations"): for relation in mb_artist_details["relations"]: if relation.get("type") == "wikipedia" and relation.get("url") and relation["url"].get("resource"): wiki_url_match = re.search(r'wikipedia\.org/wiki/(.*)', relation["url"]["resource"]) if wiki_url_match: wikipedia_page_title = requests.utils.unquote(wiki_url_match.group(1)) colored_print(f"DEBUG: Found Wikipedia relation in MB as fallback: '{wikipedia_page_title}'", Colors.BLUE) break if wikipedia_page_title: bio_text, bio_status = get_wikipedia_summary(wikipedia_page_title) artist_data["Artist Bio"] = bio_text if bio_text else "N/A" if bio_text: artist_data["Artist Wikipedia URL"] = f"https://en.wikipedia.org/wiki/{wikipedia_page_title.replace(' ', '_')}" wikipedia_url_status = "Found (MB Fallback)" colored_print(f"DEBUG: Wikipedia summary from MB relation (fallback): {bio_status}", Colors.GREEN if bio_text else Colors.YELLOW) else: colored_print(f"DEBUG: No direct Wikipedia relation in MB for '{selected_mb_artist_name}'.", Colors.YELLOW) # Artist Birth/Formation Date - Access directly from artist object life_span = mb_artist_details.get("life-span") # Access directly if life_span and life_span.get("begin"): artist_data["Artist Birth/Formation Date"] = life_span["begin"] birth_date_status = "Found" colored_print(f"DEBUG: Birth/Formation Date: {life_span['begin']}", Colors.BLUE) # Artist Location - Access directly from artist object area = mb_artist_details.get("area") # Access directly if area and area.get("name"): artist_data["Artist Location"] = area["name"] location_status = "Found" colored_print(f"DEBUG: Location: {area['name']}", Colors.BLUE) # Artist Website/Social Media if mb_artist_details.get("relations"): for relation in mb_artist_details["relations"]: url_resource = relation["url"].get("resource") if url_resource: if relation.get("type") == "official homepage" and artist_data["Artist Website"] == "N/A": artist_data["Artist Website"] = url_resource website_status = "Found" colored_print(f"DEBUG: Website: {url_resource}", Colors.BLUE) elif relation.get("type") == "social network": if "facebook.com" in url_resource and artist_data["Artist Facebook"] == "N/A": artist_data["Artist Facebook"] = url_resource facebook_status = "Found" colored_print(f"DEBUG: Facebook: {url_resource}", Colors.BLUE) elif "instagram.com" in url_resource and artist_data["Artist Instagram"] == "N/A": artist_data["Artist Instagram"] = url_resource instagram_status = "Found" colored_print(f"DEBUG: Instagram: {url_resource}", Colors.BLUE) elif ("twitter.com" in url_resource or "x.com" in url_resource) and artist_data["Artist X (Twitter)"] == "N/A": artist_data["Artist X (Twitter)"] = url_resource x_status = "Found" colored_print(f"DEBUG: X (Twitter): {url_resource}", Colors.BLUE) else: colored_print(f"DEBUG: Failed to get MusicBrainz details for MBID {mbid}.", Colors.YELLOW) else: colored_print(f"DEBUG: No MBID found for '{artist_name}' in initial search.", Colors.YELLOW) # --- Attempt 2: Retry with Cleaned Names if data is still "N/A" --- if artist_data["Artist Bio"] == "N/A" or artist_data["Artist Birth/Formation Date"] == "N/A" or artist_data["Artist Location"] == "N/A": cleaned_artist_name = _clean_name(artist_name) if cleaned_artist_name != artist_name: colored_print(f"DEBUG: Retrying with cleaned name: '{cleaned_artist_name}'", Colors.YELLOW) # Prioritize Wikipedia direct search for Artist Bio (cleaned name) if artist_data["Artist Bio"] == "N/A": colored_print(f"DEBUG: Attempting direct Wikipedia search for bio of cleaned '{cleaned_artist_name}'", Colors.BLUE) wiki_title_cleaned, wiki_status_cleaned = search_wikipedia_page_title(f"{cleaned_artist_name} musician") if wiki_title_cleaned: bio_text_cleaned, bio_status = get_wikipedia_summary(wiki_title_cleaned) artist_data["Artist Bio"] = bio_text_cleaned if bio_text_cleaned else "N/A" if bio_text_cleaned: artist_data["Artist Wikipedia URL"] = f"https://en.wikipedia.org/wiki/{wiki_title_cleaned.replace(' ', '_')}" wikipedia_url_status = "Found (Cleaned, Direct)" colored_print(f"DEBUG: Direct Wikipedia bio for cleaned '{cleaned_artist_name}': {bio_status}", Colors.GREEN if bio_text_cleaned else Colors.YELLOW) else: bio_status = wiki_status_cleaned colored_print(f"DEBUG: Direct Wikipedia bio search for cleaned '{cleaned_artist_name}' failed: {bio_status}", Colors.YELLOW) search_url_cleaned = f"{MUSICBRAINZ_API_BASE_URL}/artist" params_cleaned = {"query": cleaned_artist_name, "fmt": "json", "limit": 5} mb_search_results_cleaned = get_musicbrainz_data(search_url_cleaned, params_cleaned) mbid_cleaned = None best_score_cleaned = -1 selected_mb_artist_name_cleaned = "N/A" if mb_search_results_cleaned and mb_search_results_cleaned.get("artists"): colored_print(f"DEBUG: MusicBrainz search for cleaned name '{cleaned_artist_name}' returned {len(mb_search_results_cleaned['artists'])} results.", Colors.BLUE) for artist_match_cleaned in mb_search_results_cleaned["artists"]: score_cleaned = artist_match_cleaned.get("score", 0) if score_cleaned > best_score_cleaned and artist_match_cleaned.get("name", "").lower() == cleaned_artist_name.lower(): best_score_cleaned = score_cleaned mbid_cleaned = artist_match_cleaned.get("id") selected_mb_artist_name_cleaned = artist_match_cleaned.get("name") colored_print(f"DEBUG: Found exact match (cleaned) '{selected_mb_artist_name_cleaned}' with score {score_cleaned}.", Colors.BLUE) break if not mbid_cleaned and mb_search_results_cleaned["artists"]: best_match_cleaned = max(mb_search_results_cleaned["artists"], key=lambda x: x.get("score", 0)) mbid_cleaned = best_match_cleaned.get("id") selected_mb_artist_name_cleaned = best_match_cleaned.get("name") colored_print(f"DEBUG: No exact match (cleaned), selected best score '{selected_mb_artist_name_cleaned}' with score {best_match_cleaned.get('score',0)}.", Colors.BLUE) else: colored_print(f"DEBUG: MusicBrainz search for cleaned name '{cleaned_artist_name}' returned no artists or error.", Colors.YELLOW) if mbid_cleaned: colored_print(f"DEBUG: Fetching details for cleaned MBID: {mbid_cleaned} ('{selected_mb_artist_name_cleaned}')", Colors.BLUE) artist_detail_url_cleaned = f"{MUSICBRAINZ_API_BASE_URL}/artist/{mbid_cleaned}" colored_print(f"DEBUG: Artist detail URL (cleaned): {artist_detail_url_cleaned}", Colors.BLUE) artist_detail_params_cleaned = {"fmt": "json", "inc": "url-rels+aliases+tags"} mb_artist_details_cleaned = get_musicbrainz_data(artist_detail_url_cleaned, artist_detail_params_cleaned) if mb_artist_details_cleaned: # Fallback for Artist Bio if still N/A after direct Wikipedia search if artist_data["Artist Bio"] == "N/A": wikipedia_page_title_cleaned = None if mb_artist_details_cleaned.get("relations"): for relation in mb_artist_details_cleaned["relations"]: if relation.get("type") == "wikipedia" and relation.get("url") and relation["url"].get("resource"): wiki_url_match_cleaned = re.search(r'wikipedia\.org/wiki/(.*)', relation["url"]["resource"]) if wiki_url_match_cleaned: wikipedia_page_title_cleaned = requests.utils.unquote(wiki_url_match_cleaned.group(1)) colored_print(f"DEBUG: Found Wikipedia relation in MB (cleaned) as fallback: '{wikipedia_page_title_cleaned}'", Colors.BLUE) break if wikipedia_page_title_cleaned: bio_text_cleaned, bio_status = get_wikipedia_summary(wikipedia_page_title_cleaned) artist_data["Artist Bio"] = bio_text_cleaned if bio_text_cleaned else "N/A" if bio_text_cleaned: artist_data["Artist Wikipedia URL"] = f"https://en.wikipedia.org/wiki/{wikipedia_page_title_cleaned.replace(' ', '_')}" wikipedia_url_status = "Found (Cleaned, MB Fallback)" colored_print(f"DEBUG: Wikipedia summary from MB relation (cleaned, fallback): {bio_status}", Colors.GREEN if bio_text_cleaned else Colors.YELLOW) else: colored_print(f"DEBUG: No direct Wikipedia relation in MB for cleaned '{selected_mb_artist_name_cleaned}'.", Colors.YELLOW) if artist_data["Artist Birth/Formation Date"] == "N/A": life_span_cleaned = mb_artist_details_cleaned.get("life-span") if life_span_cleaned and life_span_cleaned.get("begin"): artist_data["Artist Birth/Formation Date"] = life_span_cleaned["begin"] birth_date_status = "Found (Cleaned)" colored_print(f"DEBUG: Birth/Formation Date (cleaned): {life_span_cleaned['begin']}", Colors.BLUE) if artist_data["Artist Location"] == "N/A": area_cleaned = mb_artist_details_cleaned.get("area") if area_cleaned and area_cleaned.get("name"): artist_data["Artist Location"] = area_cleaned["name"] location_status = "Found (Cleaned)" colored_print(f"DEBUG: Location (cleaned): {area_cleaned['name']}", Colors.BLUE) if mb_artist_details_cleaned.get("relations"): for relation in mb_artist_details_cleaned["relations"]: url_resource = relation["url"].get("resource") if url_resource: if relation.get("type") == "official homepage" and artist_data["Artist Website"] == "N/A": artist_data["Artist Website"] = url_resource website_status = "Found (Cleaned)" colored_print(f"DEBUG: Website (cleaned): {url_resource}", Colors.BLUE) elif relation.get("type") == "social network": if "facebook.com" in url_resource and artist_data["Artist Facebook"] == "N/A": artist_data["Artist Facebook"] = url_resource facebook_status = "Found (Cleaned)" colored_print(f"DEBUG: Facebook (cleaned): {url_resource}", Colors.BLUE) elif "instagram.com" in url_resource and artist_data["Artist Instagram"] == "N/A": artist_data["Artist Instagram"] = url_resource instagram_status = "Found (Cleaned)" colored_print(f"DEBUG: Instagram: {url_resource}", Colors.BLUE) elif ("twitter.com" in url_resource or "x.com" in url_resource) and artist_data["Artist X (Twitter)"] == "N/A": artist_data["Artist X (Twitter)"] = url_resource x_status = "Found (Cleaned)" colored_print(f"DEBUG: X (Twitter) (cleaned): {url_resource}", Colors.BLUE) else: colored_print(f"DEBUG: Failed to get MusicBrainz details for cleaned MBID {mbid_cleaned}.", Colors.YELLOW) else: colored_print(f"DEBUG: No MBID found for cleaned name '{cleaned_artist_name}'.", Colors.YELLOW) else: colored_print(f"DEBUG: Cleaned name is same as original, no retry needed for '{artist_name}'.", Colors.BLUE) # Construct final status message status_message = ( f"Processed '{artist_name}'. " f"Bio: {bio_status}, Wikipedia URL: {wikipedia_url_status}, Website: {website_status}, Facebook: {facebook_status}, " f"Instagram: {instagram_status}, X: {x_status}, " f"Birth/Formation: {birth_date_status}, Location: {location_status}" ) return artist_data, status_message # --- Main Script to Fetch and Save Artist Info --- def save_artist_data_to_csv(data, output_file, headers): """Saves a list of artist data dictionaries to a CSV file.""" try: with open(output_file, mode='w', newline='', encoding='utf-8') as outfile: writer = csv.DictWriter(outfile, fieldnames=headers) writer.writeheader() writer.writerows(data) colored_print(f"Saved {len(data)} artists to '{output_file}' (periodic save).", Colors.CYAN) except IOError as e: colored_print(f"Error saving artist data to CSV file: {e}", Colors.RED) def fetch_and_save_artist_info(input_csv_file, output_csv_file="so_fresh_artists.csv"): """ Reads the main data CSV, extracts unique artists, fetches their details, and saves them to a new CSV file. """ colored_print(f"\n--- Starting Artist Data Fetcher for '{input_csv_file}' ---", Colors.BOLD + Colors.CYAN) unique_artists = set() try: with open(input_csv_file, mode='r', newline='', encoding='utf-8') as infile: reader = csv.DictReader(infile) for row in reader: artist_entry = row.get("Artist", "").strip() if artist_entry: # Split by comma and process each individual artist individual_artists = [a.strip() for a in artist_entry.split(',')] for artist in individual_artists: if artist and artist.lower() != "various": # Exclude "Various" artists and empty strings unique_artists.add(artist) except FileNotFoundError: colored_print(f"Error: Input CSV file '{input_csv_file}' not found. Please ensure the combined data script has run.", Colors.RED) return except Exception as e: colored_print(f"An error occurred while reading the input CSV: {e}", Colors.RED) return if not unique_artists: colored_print("No unique artists found in the input CSV to process.", Colors.YELLOW) return artists_to_process = sorted(list(unique_artists)) # Sort for consistent processing order colored_print(f"Found {len(artists_to_process)} unique artists to process.", Colors.CYAN) # Set num_processes to 1 to strictly adhere to MusicBrainz 1 request/second global rate limit. # This will make the artist data fetching sequential and slower, but more reliable. num_processes = 1 colored_print(f"Using {num_processes} process(es) to fetch artist data (due to MusicBrainz API rate limits)...", Colors.YELLOW) all_artist_data = [] # Define CSV headers here, as they are needed for periodic saves csv_headers = [ "Artist Name", "Artist Bio", "Artist Image", "Artist Website", "Artist Facebook", "Artist Instagram", "Artist X (Twitter)", "Artist Birth/Formation Date", "Artist Location", "Artist Wikipedia URL" # Added new header ] # Initialize the CSV file with headers if it doesn't exist or is empty if not os.path.exists(output_csv_file) or os.path.getsize(output_csv_file) == 0: save_artist_data_to_csv([], output_csv_file, csv_headers) # Save just headers with multiprocessing.Pool(processes=num_processes) as pool: for i, (artist_info, status_message) in enumerate(pool.imap_unordered(process_artist_data, artists_to_process)): all_artist_data.append(artist_info) colored_print(f"Progress: {i + 1}/{len(artists_to_process)} artists processed. Last: {status_message}", Colors.BLUE) # Periodic save every N artists if (i + 1) % 5 == 0: # Save every 5 artists colored_print(f"Saving progress at {i + 1} artists...", Colors.YELLOW) save_artist_data_to_csv(all_artist_data, output_csv_file, csv_headers) colored_print(f"\n--- All Artist Data Fetched. Performing final save to '{output_csv_file}' ---", Colors.BOLD + Colors.CYAN) save_artist_data_to_csv(all_artist_data, output_csv_file, csv_headers) # Final save colored_print("\n--- Artist Data Fetcher Complete ---", Colors.BOLD + Colors.GREEN) # --- Main Execution --- if __name__ == "__main__": multiprocessing.freeze_support() input_csv = "so_fresh_full_data.csv" # Output from the combined data fetcher output_csv = "so_fresh_artists_info.csv" # New CSV for artist information fetch_and_save_artist_info(input_csv, output_csv)