import requests import time import re import csv import os import subprocess import json import multiprocessing from dotenv import load_dotenv # Load environment variables from .env file load_dotenv() # --- Configuration for APIs --- BASE_URL = os.getenv("BASE_URL", "https://api.discogs.com") DISCOGS_CONSUMER_KEY = os.getenv("DISCOGS_CONSUMER_KEY") DISCOGS_CONSUMER_SECRET = os.getenv("DISCOGS_CONSUMER_SECRET") 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" LYRICS_API_BASE_URL = "https://api.lyrics.ovh/v1" WIKIPEDIA_API_BASE_URL = "https://en.wikipedia.org/w/api.php" USER_AGENT = "SoFreshCombinedFetcher/1.0 +https://github.com/yourusername/sofresh" # --- Global variable for Spotify access token (per process) --- spotify_access_token = None spotify_token_expiry = 0 # --- Helper Functions --- def _clean_name(name): """Removes content within parentheses from a string.""" return re.sub(r'\s*\(.*\)\s*', '', name).strip() def get_discogs_data(url, params=None): """Fetches data from the Discogs API with authentication and rate limiting.""" if not DISCOGS_CONSUMER_KEY or not DISCOGS_CONSUMER_SECRET: print("Error: Discogs API key or secret not found. Check .env.") return None headers = { "User-Agent": USER_AGENT, "Authorization": f"Discogs key={DISCOGS_CONSUMER_KEY}, secret={DISCOGS_CONSUMER_SECRET}" } try: response = requests.get(url, headers=headers, params=params) response.raise_for_status() time.sleep(1) # Be kind to the API return response.json() except requests.exceptions.RequestException as e: print(f"Error fetching Discogs data from {url}: {e}") return None def parse_so_fresh_title(title): """Parses 'So Fresh' compilation title to extract season and year.""" original_title_lower = title.lower() release_type = None year = None year_match = re.search(r'\b(19|20)\d{2}\b', original_title_lower) if year_match: year = int(year_match.group(0)) if "summer" in original_title_lower: release_type = "Summer" elif "autumn" in original_title_lower or "fall" in original_title_lower: release_type = "Autumn" elif "winter" in original_title_lower: release_type = "Winter" elif "spring" in original_title_lower: release_type = "Spring" if not release_type: if "christmas" in original_title_lower or "xmas" in original_title_lower or "holiday" in original_title_lower: release_type = "Christmas Release" elif "top" in original_title_lower and "hits" in original_title_lower: release_type = "Top Hits" elif "greatest hits" in original_title_lower or "best of" in original_title_lower or "years of" in original_title_lower: release_type = "Greatest Hits" elif "ultimate" in original_title_lower or "platinum" in original_title_lower or "classic" in original_title_lower: release_type = "Special Compilation" elif "the hits" in original_title_lower: release_type = "The Hits" elif "random" in original_title_lower: release_type = "Random" return release_type, year def get_spotify_access_token(): """Obtains and caches a Spotify API access token.""" global spotify_access_token, spotify_token_expiry 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. 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 return spotify_access_token except requests.exceptions.RequestException as e: # print(f"Error obtaining Spotify access token: {e}") # Suppress for multiprocessing 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}"} query = f"track:{track_name} artist:{artist_name}" params = {"q": query, "type": "track", "limit": 1} 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 = track_item["album"]["images"][0]["url"] if track_item.get("album") and track_item["album"].get("images") else None return spotify_link, cover_art_url, "Found" else: return None, None, "Not Found" except requests.exceptions.RequestException as e: return None, None, "Error" def search_youtube_video_yt_dlp(track_name, artist_name): """Searches for a music video on YouTube using yt-dlp and returns its link.""" 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) if process.returncode != 0: return None, "yt-dlp Error" video_info = json.loads(process.stdout) youtube_link = video_info.get("webpage_url") return youtube_link, "Found" if youtube_link else "Not Found" except FileNotFoundError: print("Error: yt-dlp not found. Please install it.") return None, "yt-dlp Not Installed" except (subprocess.CalledProcessError, json.JSONDecodeError, Exception) as e: return None, "Error" def get_lyrics(artist_name, track_name): """Fetches lyrics for a given artist and track name using Lyrics.ovh API.""" encoded_artist = requests.utils.quote(artist_name) encoded_track = requests.utils.quote(track_name) url = f"{LYRICS_API_BASE_URL}/{encoded_artist}/{encoded_track}" headers = {"User-Agent": USER_AGENT} try: response = requests.get(url, headers=headers) response.raise_for_status() lyrics_data = response.json() return lyrics_data["lyrics"], "Found" if lyrics_data and lyrics_data.get("lyrics") else "Not Found" except requests.exceptions.RequestException as e: return None, "API Error" except json.JSONDecodeError: return None, "JSON Error" def search_wikipedia_link(query): """Searches Wikipedia for a given query and returns the first article link.""" 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"] # Construct the Wikipedia URL from the title wikipedia_link = f"https://en.wikipedia.org/wiki/{requests.utils.quote(page_title)}" return wikipedia_link, "Found" else: return None, "Not Found" except requests.exceptions.RequestException as e: return None, "API Error" except Exception as e: return None, "Error" # --- Worker function for multiprocessing --- def enrich_track_data(row): """ Enriches a single track row with Spotify, YouTube, Lyrics, and Wikipedia links. Includes retry logic with cleaned names. This function will be run by each worker process. """ track_name = row.get("Track Name", "") artist_name = row.get("Artist", "") album_name = row.get("Album Name", "") # New field if not track_name or not artist_name: return row, f"Skipped: Missing Track Name or Artist in row: {row}" # Initialize statuses for the current 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") lyrics = row.get("Lyrics", "N/A") wikipedia_song_link = row.get("Wikipedia Song Link", "N/A") # New field wikipedia_album_link = row.get("Wikipedia Album Link", "N/A") # New field # --- Initial Searches --- if spotify_link == "N/A": spotify_link, spotify_cover_art_url, 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" else: spotify_status = "Already Found" if youtube_link == "N/A": youtube_link, youtube_status = search_youtube_video_yt_dlp(track_name, artist_name) row["YouTube Link"] = youtube_link if youtube_link else "N/A" else: youtube_status = "Already Found" if lyrics == "N/A": lyrics, lyrics_status = get_lyrics(artist_name, track_name) row["Lyrics"] = lyrics if lyrics else "N/A" else: lyrics_status = "Already Found" if wikipedia_song_link == "N/A": wikipedia_song_link, wiki_song_status = search_wikipedia_link(f"{track_name} {artist_name} song") row["Wikipedia Song Link"] = wikipedia_song_link if wikipedia_song_link else "N/A" else: wiki_song_status = "Already Found" if wikipedia_album_link == "N/A" and album_name and album_name != "N/A": wikipedia_album_link, wiki_album_status = search_wikipedia_link(f"{album_name} album") row["Wikipedia Album Link"] = wikipedia_album_link if wikipedia_album_link else "N/A" else: wiki_album_status = "Already Found" if wikipedia_album_link != "N/A" else "Skipped (No Album Name)" # --- Retry with Cleaned Names if links are still "N/A" --- cleaned_track_name = _clean_name(track_name) cleaned_artist_name = _clean_name(artist_name) cleaned_album_name = _clean_name(album_name) if cleaned_track_name != track_name or cleaned_artist_name != artist_name or cleaned_album_name != album_name: if spotify_link == "N/A": spotify_link_retry, spotify_cover_art_url_retry, spotify_status = search_spotify_track(cleaned_track_name, cleaned_artist_name) if spotify_link_retry: row["Spotify Link"] = spotify_link_retry row["Spotify Cover Art URL"] = spotify_cover_art_url_retry spotify_status = "Found (Cleaned)" elif spotify_status == "Not Found": # Only update status if it was genuinely not found before spotify_status = "Not Found (Cleaned)" if youtube_link == "N/A": youtube_link_retry, youtube_status = search_youtube_video_yt_dlp(cleaned_track_name, cleaned_artist_name) if youtube_link_retry: row["YouTube Link"] = youtube_link_retry youtube_status = "Found (Cleaned)" elif youtube_status == "Not Found": youtube_status = "Not Found (Cleaned)" if lyrics == "N/A": lyrics_retry, lyrics_status = get_lyrics(cleaned_artist_name, cleaned_track_name) if lyrics_retry: row["Lyrics"] = lyrics_retry lyrics_status = "Found (Cleaned)" elif lyrics_status == "Not Found": lyrics_status = "Not Found (Cleaned)" if wikipedia_song_link == "N/A": wikipedia_song_link_retry, wiki_song_status = search_wikipedia_link(f"{cleaned_track_name} {cleaned_artist_name} song") if wikipedia_song_link_retry: row["Wikipedia Song Link"] = wikipedia_song_link_retry wiki_song_status = "Found (Cleaned)" elif wiki_song_status == "Not Found": wiki_song_status = "Not Found (Cleaned)" if wikipedia_album_link == "N/A" and cleaned_album_name and cleaned_album_name != "N/A": wikipedia_album_link_retry, wiki_album_status = search_wikipedia_link(f"{cleaned_album_name} album") if wikipedia_album_link_retry: row["Wikipedia Album Link"] = wikipedia_album_link_retry wiki_album_status = "Found (Cleaned)" elif wiki_album_status == "Not Found": wiki_album_status = "Not Found (Cleaned)" status_message = ( f"Processed '{track_name}' by '{artist_name}'. " f"Spotify: {spotify_status}, YouTube: {youtube_status}, " f"Lyrics: {lyrics_status}, Wiki Song: {wiki_song_status}, Wiki Album: {wiki_album_status}" ) time.sleep(0.1) return row, status_message # --- Main Orchestration Script --- def fetch_and_enrich_so_fresh_data(label_url, output_csv_file="so_fresh_full_data.csv"): """ Fetches all track information from Discogs, then enriches it with Spotify, YouTube, Lyrics, and Wikipedia links using multiprocessing. """ print("\n--- Starting Combined So Fresh Data Fetcher ---") # --- Step 1: Fetch initial Discogs data --- print("\n--- Step 1: Fetching Discogs Releases and Tracklists ---") discogs_tracks_data = [] label_match = re.search(r'/label/(\d+)', label_url) if not label_match: print(f"Error: Invalid Discogs label URL: {label_url}") return label_id = label_match.group(1) page = 1 total_pages = 1 while page <= total_pages: label_releases_url = f"{BASE_URL}/labels/{label_id}/releases" params = {"page": page, "per_page": 50} print(f"Fetching releases for label ID {label_id}, page {page}...") releases_data = get_discogs_data(label_releases_url, params) if releases_data and releases_data.get("releases"): for release_summary in releases_data["releases"]: release_id = release_summary.get("id") release_title = release_summary.get("title", "") release_year = release_summary.get("year") if "so fresh" not in release_title.lower(): # print(f"Skipping non-'So Fresh' release: '{release_title}'.") continue if not release_id: # print(f"Skipping release with no ID: '{release_title}'.") continue print(f"Processing Discogs release: '{release_title}' (ID: {release_id})...") detected_release_type, parsed_year = parse_so_fresh_title(release_title) final_year = release_year if release_year else parsed_year display_release_type = detected_release_type if detected_release_type else "General Compilation" display_year = final_year if final_year else "N/A" if display_release_type == "Summer" and display_year and re.search(r'hits of \b' + str(display_year - 1) + r'\b', release_title.lower()): display_release_type = f"Summer + {display_year - 1} Hits" release_detail_data = get_discogs_data(f"{BASE_URL}/releases/{release_id}") if release_detail_data: main_genre = release_detail_data["genres"][0] if release_detail_data.get("genres") else "N/A" discogs_release_url = f"https://www.discogs.com/release/{release_id}" # New fields from Discogs date_released = release_detail_data.get("released", "N/A") album_name = release_title # Album name is the release title itself record_label = "N/A" if release_detail_data.get("labels"): record_label = release_detail_data["labels"][0].get("name", "N/A") isrc = "N/A" if release_detail_data.get("identifiers"): for identifier in release_detail_data["identifiers"]: if identifier.get("type", "").lower() == "isrc": isrc = identifier.get("value", "N/A") break # Awards - Placeholder, as direct API fetching is not feasible without specialized APIs awards = "N/A (Manual Check Required)" if release_detail_data.get("tracklist"): for track in release_detail_data["tracklist"]: track_number = track.get("position", "") track_name = track.get("title", "") track_duration = track.get("duration", "") track_artists = [] if track.get("artists"): track_artists = [a.get("name") for a in track["artists"] if a.get("name")] elif release_detail_data.get("artists"): track_artists = [a.get("name") for a in release_detail_data["artists"] if a.get("name")] artist_name = ", ".join(track_artists) if track_artists else "Various" discogs_tracks_data.append({ "Track Number": track_number, "Track Name": track_name, "Artist": artist_name, "Track Length (Min:Sec)": track_duration, "Season": display_release_type, "Year": display_year, "Main Genre": main_genre, "Discogs Release URL": discogs_release_url, "Date Released": date_released, # New "Album Name": album_name, # New "Record Label": record_label, # New "ISRC": isrc, # New "Awards": awards, # New - Placeholder "Spotify Link": "N/A", # Will be filled by enrichment "Spotify Cover Art URL": "N/A", # Will be filled by enrichment "YouTube Link": "N/A", # Will be filled by enrichment "Lyrics": "N/A", # Will be filled by enrichment "Wikipedia Song Link": "N/A", # Will be filled by enrichment "Wikipedia Album Link": "N/A" # Will be filled by enrichment }) else: print(f"No tracklist found for release '{release_title}' (ID: {release_id}).") else: print(f"Failed to fetch detailed data for release '{release_title}' (ID: {release_id}).") pagination = releases_data.get("pagination", {}) total_pages = pagination.get("pages", 1) page += 1 else: break print(f"\n--- Step 1 Complete: Found {len(discogs_tracks_data)} tracks from Discogs. ---") # --- Step 2: Enrich data with Spotify, YouTube, Lyrics, and Wikipedia using multiprocessing --- print("\n--- Step 2: Enriching Track Data with Media Links and Lyrics ---") if not discogs_tracks_data: print("No tracks to enrich. Exiting.") return num_processes = multiprocessing.cpu_count() print(f"Using {num_processes} processes for enrichment...") final_enriched_data = [] with multiprocessing.Pool(processes=num_processes) as pool: for i, (processed_row, status_message) in enumerate(pool.imap_unordered(enrich_track_data, discogs_tracks_data)): final_enriched_data.append(processed_row) if (i + 1) % 10 == 0 or (i + 1) == len(discogs_tracks_data): print(f"Progress: {i + 1}/{len(discogs_tracks_data)} tracks enriched. Last: {status_message}") print(f"\n--- Step 2 Complete: Enriched {len(final_enriched_data)} tracks. ---") # --- Step 3: Write all data to CSV --- print(f"\n--- Step 3: Writing all data to '{output_csv_file}' ---") csv_headers = [ "Track Number", "Track Name", "Artist", "Track Length (Min:Sec)", "Season", "Year", "Main Genre", "Discogs Release URL", "Date Released", "Album Name", "Record Label", "ISRC", "Awards", # New fields "Spotify Link", "Spotify Cover Art URL", "YouTube Link", "Lyrics", "Wikipedia Song Link", "Wikipedia Album Link" # New Wikipedia fields ] try: with open(output_csv_file, mode='w', newline='', encoding='utf-8') as csvfile: writer = csv.DictWriter(csvfile, fieldnames=csv_headers) writer.writeheader() writer.writerows(final_enriched_data) print(f"Successfully saved {len(final_enriched_data)} tracks to '{output_csv_file}'") except IOError as e: print(f"Error saving data to CSV file: {e}") print("\n--- Combined Data Fetcher Complete ---") # --- Main Execution --- if __name__ == "__main__": multiprocessing.freeze_support() so_fresh_label_url = "https://www.discogs.com/label/334733-So-Fresh" fetch_and_enrich_so_fresh_data(so_fresh_label_url, output_csv_file="so_fresh_full_data.csv")