mchockal/TaylorSwiftDJ
1
1import json2import re3import numpy as np4from langchain.vectorstores import DeepLake5 6# Used to clean the inconsistencies in the format in which ChatGPT generated output.7# Also convert all characters to lower case8# Usage: clean_emotions_json("../data/spotify_song_url_emotions.json")9def clean_emotions_json(filename:str) -> None:10 with open(filename, "r") as f:11 input_data = json.load(f)12 13 output_data = []14 15 # Clean emotions data - Use only lower case letters and remove any ordered listing16 for song in input_data:17 emotions = song['emotions']18 cleaned_emotions = re.sub(r'\d+\.\s+', '', emotions.lower().replace('\n', ', '))19 output_data.append(20 {21 "song_name": song["song_name"],22 "iframe": song["iframe"],23 "emotions": cleaned_emotions24 })25 print(emotions, "\n", cleaned_emotions)26 27 # Write to output file which will be used to store the song emotions as embeddings28 with open(filename, "w") as f:29 json.dump(output_data, f, indent=4)30 31 print(f"Spotify song, url and song emotions saved to {filename}")32 33 34# Does np.random.choice and ensures we don't have duplicates in the final result35def weighted_random_sample(items: np.array, weights: np.array, n: int) -> np.array:36 37 indices = np.arange(len(items))38 out_indices = []39 40 for _ in range(n):41 chosen_index = np.random.choice(indices, p=weights)42 out_indices.append(chosen_index)43 44 mask = indices != chosen_index45 indices = indices[mask]46 weights = weights[mask]47 48 if weights.sum() != 0:49 weights = weights / weights.sum()50 51 return items[out_indices]52 53# Load DeepLake db54def load_db(dataset_path: str, *args, **kwargs) -> DeepLake:55 db = DeepLake(dataset_path, *args, **kwargs)56 return db