import requests import time import re import csv import os import subprocess import json from dotenv import load_dotenv import multiprocessing # Import multiprocessing module # Load environment variables from .env file load_dotenv() # --- Configuration for Spotify API --- SPOTIFY_CLIENT_ID = os.getenv("SPOTIFY_CLIENT_ID") SPOTIFY_CLIENT_SECRET = os.getenv("SPOTIFY_CLIENT_SECRET") SPOTIFY_TOKEN_URL = "https://accounts.spotify.com/api/token" SPOTIFY_API_BASE_URL = "https://api.spotify.com/v1" USER_AGENT = "SoFreshMediaLinkFetcher/1.0 +https://github.com/yourusername/sofresh" # Updated User-Agent # --- Global variable to store Spotify access token and its expiry --- # Note: In multiprocessing, each process will have its own copy of global variables. # So, each worker process will obtain its own Spotify token. This is generally fine # but means multiple token requests will occur initially. spotify_access_token = None spotify_token_expiry = 0 # Unix timestamp # --- Helper Functions for Spotify API --- def get_spotify_access_token(): """ Obtains a Spotify API access token using the Client Credentials Flow. Caches the token and refreshes it if expired. """ global spotify_access_token, spotify_token_expiry # Check if the token is still valid if spotify_access_token and time.time() < spotify_token_expiry: return spotify_access_token if not SPOTIFY_CLIENT_ID or not SPOTIFY_CLIENT_SECRET: print("Error: Spotify CLIENT_ID or CLIENT_SECRET not found in environment variables. Cannot get Spotify token.") return None headers = { "Content-Type": "application/x-www-form-urlencoded" } data = { "grant_type": "client_credentials", "client_id": SPOTIFY_CLIENT_ID, "client_secret": SPOTIFY_CLIENT_SECRET } try: response = requests.post(SPOTIFY_TOKEN_URL, headers=headers, data=data) response.raise_for_status() token_info = response.json() spotify_access_token = token_info["access_token"] spotify_token_expiry = time.time() + token_info["expires_in"] - 60 # Refresh 60 seconds before actual expiry return spotify_access_token except requests.exceptions.RequestException as e: print(f"Error obtaining Spotify access token: {e}") if hasattr(e, 'response') and e.response is not None: print(f"Response content: {e.response.text}") return None def search_spotify_track(track_name, artist_name): """ Searches for a track on Spotify and returns its link and cover art URL. """ token = get_spotify_access_token() if not token: return None, None, None headers = { "Authorization": f"Bearer {token}" } # Construct a more precise query query = f"track:{track_name} artist:{artist_name}" params = { "q": query, "type": "track", "limit": 1 # We only need the top result } try: response = requests.get(f"{SPOTIFY_API_BASE_URL}/search", headers=headers, params=params) response.raise_for_status() search_results = response.json() if search_results and search_results.get("tracks") and search_results["tracks"].get("items"): track_item = search_results["tracks"]["items"][0] spotify_link = track_item["external_urls"]["spotify"] cover_art_url = None if track_item.get("album") and track_item["album"].get("images"): # Get the largest image available images = track_item["album"]["images"] if images: cover_art_url = images[0]["url"] # First image is usually the largest return spotify_link, cover_art_url, "Found" # Return "Found" status else: return None, None, "Not Found" # Indicate not found except requests.exceptions.RequestException as e: return None, None, "Error" # Indicate error # --- Helper Function for YouTube Search using yt-dlp --- def search_youtube_video_yt_dlp(track_name, artist_name): """ Searches for a music video on YouTube using yt-dlp and returns its link. """ # Construct the search query for yt-dlp query = f"ytsearch1:{track_name} {artist_name} official music video" try: command = ["yt-dlp", "--dump-json", "-q", query] process = subprocess.run( command, capture_output=True, text=True, check=False # Do not raise CalledProcessError automatically, check returncode manually ) if process.returncode != 0: return None, "yt-dlp Error" video_info = json.loads(process.stdout) youtube_link = video_info.get("webpage_url") if youtube_link: return youtube_link, "Found" else: return None, "Not Found (No webpage_url in yt-dlp output)" except FileNotFoundError: print("Error: yt-dlp not found. Please install it (e.g., pip install yt-dlp or via your package manager).") return None, "yt-dlp Not Installed" except json.JSONDecodeError: return None, "JSON Parse Error" except Exception as e: return None, "Unexpected Error" # --- Helper function to clean names by removing content in parentheses --- def _clean_name(name): """Removes content within parentheses from a string.""" return re.sub(r'\s*\(.*\)\s*', '', name).strip() # --- Worker function for multiprocessing --- def process_track_row(row): """ Processes a single track row to add Spotify and YouTube links, including a retry with cleaned names if initial search fails. This function will be run by each worker process. """ track_name = row.get("Track Name", "") artist_name = row.get("Artist", "") if not track_name or not artist_name: return row, f"Skipped: Missing Track Name or Artist in row: {row}" spotify_link = row.get("Spotify Link", "N/A") spotify_cover_art_url = row.get("Spotify Cover Art URL", "N/A") youtube_link = row.get("YouTube Link", "N/A") initial_spotify_status = "Already Found" if spotify_link != "N/A" else "Not Searched" initial_youtube_status = "Already Found" if youtube_link != "N/A" else "Not Searched" # --- Initial Search --- if spotify_link == "N/A": spotify_link, spotify_cover_art_url, initial_spotify_status = search_spotify_track(track_name, artist_name) row["Spotify Link"] = spotify_link if spotify_link else "N/A" row["Spotify Cover Art URL"] = spotify_cover_art_url if spotify_cover_art_url else "N/A" if youtube_link == "N/A": youtube_link, initial_youtube_status = search_youtube_video_yt_dlp(track_name, artist_name) row["YouTube Link"] = youtube_link if youtube_link else "N/A" # --- Retry with Cleaned Names if links are still "N/A" --- retry_status_spotify = "" retry_status_youtube = "" if spotify_link == "N/A": cleaned_track_name = _clean_name(track_name) cleaned_artist_name = _clean_name(artist_name) if cleaned_track_name != track_name or cleaned_artist_name != artist_name: spotify_link, spotify_cover_art_url, retry_status_spotify = search_spotify_track(cleaned_track_name, cleaned_artist_name) row["Spotify Link"] = spotify_link if spotify_link else "N/A" row["Spotify Cover Art URL"] = spotify_cover_art_url if spotify_cover_art_url else "N/A" if spotify_link != "N/A": retry_status_spotify = "Found (Cleaned)" else: retry_status_spotify = "Not Found (Cleaned)" if youtube_link == "N/A": cleaned_track_name = _clean_name(track_name) cleaned_artist_name = _clean_name(artist_name) if cleaned_track_name != track_name or cleaned_artist_name != artist_name: youtube_link, retry_status_youtube = search_youtube_video_yt_dlp(cleaned_track_name, cleaned_artist_name) row["YouTube Link"] = youtube_link if youtube_link else "N/A" if youtube_link != "N/A": retry_status_youtube = "Found (Cleaned)" else: retry_status_youtube = "Not Found (Cleaned)" # Construct the final status message final_spotify_status = "Found" if spotify_link != "N/A" else initial_spotify_status if initial_spotify_status != "Not Searched" else retry_status_spotify if retry_status_spotify else "Not Found" final_youtube_status = "Found" if youtube_link != "N/A" else initial_youtube_status if initial_youtube_status != "Not Searched" else retry_status_youtube if retry_status_youtube else "Not Found" status_message = ( f"Processed '{track_name}' by '{artist_name}'. " f"Spotify: {final_spotify_status}, YouTube: {final_youtube_status}" ) # Add a small delay to avoid hitting API rate limits for Spotify if many requests are made # This delay is per process, so overall rate will be higher with more processes. time.sleep(0.1) # Reduced sleep time for multiprocessing return row, status_message # --- Main Script to Process CSV and Add Media Links --- def add_media_links_to_csv(input_csv_file, output_csv_file): """ Reads an existing CSV, searches for Spotify and YouTube links for each track, and writes the updated data to a new CSV file using multiprocessing. """ print(f"\n--- Script: Adding Media Links to '{input_csv_file}' ---") tracks_to_process = [] original_headers = [] try: with open(input_csv_file, mode='r', newline='', encoding='utf-8') as infile: reader = csv.DictReader(infile) original_headers = reader.fieldnames for row in reader: tracks_to_process.append(row) except FileNotFoundError: print(f"Error: Input CSV file '{input_csv_file}' not found. Please run the first script first.") return except Exception as e: print(f"An error occurred while reading the input CSV: {e}") return if not tracks_to_process: print("No tracks found in the input CSV to process.") return # Define new headers including the media links if they don't already exist new_headers = original_headers[:] # Make a copy if "Spotify Link" not in new_headers: new_headers.append("Spotify Link") if "Spotify Cover Art URL" not in new_headers: new_headers.append("Spotify Cover Art URL") if "YouTube Link" not in new_headers: new_headers.append("YouTube Link") # Determine the number of processes to use # You can set this to a specific number, e.g., 4, 8, or multiprocessing.cpu_count() # Be mindful of API rate limits and your system's resources if increasing this significantly. num_processes = multiprocessing.cpu_count() # Uses all available CPU cores # num_processes = 8 # Example: set to a fixed number of processes print(f"Using {num_processes} processes to fetch media links...") tracks_with_media_data = [] # Use a multiprocessing Pool to distribute the work with multiprocessing.Pool(processes=num_processes) as pool: # Use imap_unordered for potentially faster results and to process them as they complete # This allows us to print progress as results come in. for i, (processed_row, status_message) in enumerate(pool.imap_unordered(process_track_row, tracks_to_process)): tracks_with_media_data.append(processed_row) if (i + 1) % 10 == 0 or (i + 1) == len(tracks_to_process): # Print every 10 tracks or at the end print(f"Progress: {i + 1}/{len(tracks_to_process)} tracks processed. Last: {status_message}") print(f"\n--- Writing updated data to '{output_csv_file}' ---") try: with open(output_csv_file, mode='w', newline='', encoding='utf-8') as outfile: writer = csv.DictWriter(outfile, fieldnames=new_headers) writer.writeheader() writer.writerows(tracks_with_media_data) print(f"Successfully saved {len(tracks_with_media_data)} tracks with media links to '{output_csv_file}'") except IOError as e: print(f"Error saving data to CSV file: {e}") print("-" * 50) # --- Main Execution --- if __name__ == "__main__": # This is important for multiprocessing on Windows. # It ensures that the child processes import the main module correctly. multiprocessing.freeze_support() input_csv = "so_fresh_compilation_tracks_with_media.csv" # This should be the output of your first script output_csv = "so_fresh_compilation_tracks_with_media_retry.csv" # New CSV with added info add_media_links_to_csv(input_csv, output_csv)