import json import time import re from urllib.parse import quote_plus import requests from youtubesearchpython import VideosSearch from fuzzywuzzy import fuzz # Install required packages: # pip install youtube-search-python fuzzywuzzy python-Levenshtein requests def clean_text(text): """Remove special characters and normalize text for matching""" if not text: return "" # Handle lists (e.g., artist arrays) if isinstance(text, list): text = ', '.join(str(item) for item in text) # Convert to string if not already text = str(text) # Remove brackets and extra whitespace text = re.sub(r'[\[\]]', '', text) # Remove special characters but keep alphanumeric and spaces text = re.sub(r'[^\w\s-]', ' ', text) # Normalize whitespace text = ' '.join(text.split()) return text.lower() def extract_artists(artists_str): """Extract individual artist names from the artists field""" if not artists_str: return [] # Handle lists (already parsed) if isinstance(artists_str, list): return [str(a).strip() for a in artists_str if a] # Convert to string if not already artists_str = str(artists_str) # Remove brackets and split by common separators artists = re.sub(r'[\[\]]', '', artists_str) artists = re.split(r'[,/&]|\s+x\s+|\s+vs\.?\s+|\s+ft\.?\s+|\s+feat\.?\s+', artists) return [a.strip() for a in artists if a.strip()] def build_search_queries(song_data): """Generate multiple search query variations""" song_name = song_data.get('song_name', '') artists = song_data.get('artists', '') label = song_data.get('label', '') queries = [] # Primary query: song name + artists if song_name and artists: artist_clean = clean_text(artists) queries.append(f"{song_name} {artist_clean}") # Add label if it's a real label (not DistroKid/TuneCore) if label and label not in ['DistroKid', 'TuneCore', 'None']: queries.append(f"{song_name} {clean_text(artists)} {label}") # Try just song name if it's specific enough if song_name and len(song_name) > 15: queries.append(song_name) return queries def fuzzy_match_score(video_title, song_data): """Calculate fuzzy match score between video title and song data""" video_title_clean = clean_text(video_title) song_name_clean = clean_text(song_data.get('song_name', '')) artists_clean = clean_text(song_data.get('artists', '')) # Score based on song name match name_score = fuzz.partial_ratio(song_name_clean, video_title_clean) # Score based on artist match artist_score = 0 if artists_clean: artist_list = extract_artists(song_data.get('artists', '')) artist_scores = [fuzz.partial_ratio(clean_text(artist), video_title_clean) for artist in artist_list] artist_score = max(artist_scores) if artist_scores else 0 # Weighted combination (prioritize song name) combined_score = (name_score * 0.7) + (artist_score * 0.3) return combined_score def search_youtube(song_data, max_results=5): """Search YouTube and return best matching video""" queries = build_search_queries(song_data) best_match = None best_score = 0 for query in queries: try: search = VideosSearch(query, limit=max_results) results = search.result() if results and 'result' in results: for video in results['result']: score = fuzzy_match_score(video['title'], song_data) if score > best_score: best_score = score best_match = { 'url': video['link'], 'title': video['title'], 'channel': video['channel']['name'], 'duration': video.get('duration', 'N/A'), 'views': video.get('viewCount', {}).get('short', 'N/A'), 'match_score': round(score, 2), 'search_query': query } # Small delay between searches time.sleep(0.5) # If we found a good match (>70%), stop searching if best_score > 70: break except Exception as e: print(f"Error searching for '{query}': {e}") continue return best_match def process_songs(input_file, output_file, limit=20): """Process songs from JSONL file and find YouTube links""" results = [] with open(input_file, 'r', encoding='utf-8') as f: for i, line in enumerate(f): if i >= limit: break try: song_data = json.loads(line) print(f"\n[{i+1}/{limit}] Searching: {song_data.get('song_name', 'Unknown')}") print(f" Artist: {song_data.get('artists', 'Unknown')}") youtube_result = search_youtube(song_data) result = { 'original_data': song_data, 'youtube_match': youtube_result } if youtube_result: print(f" ✓ Found: {youtube_result['title']}") print(f" URL: {youtube_result['url']}") print(f" Match Score: {youtube_result['match_score']}%") else: print(f" ✗ No match found") results.append(result) # Rate limiting time.sleep(1) except json.JSONDecodeError as e: print(f"Error parsing line {i+1}: {e}") continue # Save results with open(output_file, 'w', encoding='utf-8') as f: json.dump(results, f, indent=2, ensure_ascii=False) print(f"\n✓ Results saved to {output_file}") # Print summary found = sum(1 for r in results if r['youtube_match']) print(f"\nSummary: Found YouTube links for {found}/{len(results)} songs") return results def export_to_csv(results, csv_file): """Export results to CSV for easy viewing""" import csv with open(csv_file, 'w', newline='', encoding='utf-8') as f: writer = csv.writer(f) writer.writerow([ 'Song Name', 'Artists', 'Label', 'Genre', 'BPM', 'YouTube URL', 'Video Title', 'Channel', 'Match Score' ]) for result in results: song = result['original_data'] yt = result['youtube_match'] writer.writerow([ song.get('song_name', ''), song.get('artists', ''), song.get('label', ''), song.get('genre', ''), song.get('bpm', ''), yt['url'] if yt else 'NOT FOUND', yt['title'] if yt else '', yt['channel'] if yt else '', yt['match_score'] if yt else '' ]) print(f"✓ CSV exported to {csv_file}") if __name__ == "__main__": # Configuration INPUT_FILE = "cleaned_mashup_data_wout_ws.jsonl" # Your input file OUTPUT_FILE = "youtube_results.json" CSV_FILE = "youtube_results.csv" SAMPLE_SIZE = 20 print("YouTube Mashup Search Script") print("=" * 50) print(f"Processing first {SAMPLE_SIZE} songs from {INPUT_FILE}\n") # Process songs results = process_songs(INPUT_FILE, OUTPUT_FILE, limit=SAMPLE_SIZE) # Export to CSV export_to_csv(results, CSV_FILE) print("\n✓ Done!")