from refresh import Refresh import difflib import os import json import requests import time import sqlite3 from alive_progress import alive_bar from sqlite3 import Error from secrets import spotify_user_id refreshCaller = Refresh() class Minidupe: def __init__(self): self.spotify_token = "" self.here_i_am = os.path.dirname(os.path.abspath(__file__)) self.database = self.here_i_am + r"\\MASTER.db" #self.database = self.here_i_am + r"\\Endless Feed.db" self.limit = 0.9 self.user_id = spotify_user_id self.bad_names_file_1 = "OLLIE UNWANTED ARTISTS.TXT" self.bad_names_file_2 = "FREDDY UNWANTED ARTISTS.TXT" self.bad_names = [] ########################################################################################################## ########## ENTER SOURCE PLAYLIST HERE #################################################################### self.source_playlist_id = "1U9VS14NXtySos8Su6oH6k" ########################################################################################################## self.target_playlist_id = "" #NOTE: LEAVE THIS BLANK def dupecheck(self): os.system('cls') print("\033[33mMinidupe v.1.1.0 Enter Source Playlist in Line 27...\033[0m") print("\033[33mPut this code and FREDDY UNWANTED ARTISTS.TXT (optional) in...\033[0m") print("\033[33m the same folder as your MASTER database...\033[0m") self.unwanted() self.get_source_info() self.database_config() self.get_source_tracks() self.check_tracks() self.create_target_playlist() self.add_tracks_to_target() print() self.clearup_db_tables() def get_source_info(self): try: query = "https://api.spotify.com/v1/playlists/{}".format(self.source_playlist_id) response = requests.request("get", url=query, headers={"Content-type": "application/json", "Authorization": "Bearer {}".format(self.spotify_token)}) json_data = json.loads(response.text) self.queue_total = (json_data["tracks"]["total"]) self.playlist_name = (json_data["name"]) return except: self.nopl = True return def database_config(self): conn = self.create_connection(self.database) cur = conn.cursor() try: with conn: cur.execute("CREATE TABLE IF NOT EXISTS SOURCE_DATA (URI text PRIMARY KEY, TRACK_NAME text, ARTIST_NAME text);") conn.commit() cur.execute("CREATE TABLE IF NOT EXISTS FOR_REVIEW (URI text PRIMARY KEY, TRACK_NAME text, ARTIST_NAME text);") conn.commit() except Error as e2: print("Database Config Error") print(e2) return def get_source_tracks(self): with alive_bar(self.queue_total, title="Getting Playlist Tracks") as bar: t = 0 while t < self.queue_total: try: if t == 1000: self.call_refresh() self.offset = t query = "https://api.spotify.com/v1/playlists/{}/tracks".format(self.source_playlist_id) response = requests.request("get", url=query, headers={"Content-type": "application/json", "Authorization": "Bearer {}".format(self.spotify_token)}, params={"limit": 1, "offset": self.offset}) json_data = json.loads(response.text) self.track_uri = json_data["items"][0]["track"]["uri"] self.track_name = json_data["items"][0]["track"]["name"] self.artist_name = json_data["items"][0]["track"]["artists"][0]["name"] self.configure_the_information() if response.status_code == 200: self.add_to_source_table() bar() t = t + 1 continue except: print("error" + str(t)) t = t - 2 continue return def add_to_source_table(self): conn = self.create_connection(self.database) cur = conn.cursor() try: tracks = (self.track_uri, self.track_name, self.artist_name) sql = ''' INSERT OR IGNORE INTO SOURCE_DATA (URI,TRACK_NAME,ARTIST_NAME) VALUES(?,?,?) ''' cur = conn.cursor() cur.execute(sql, tracks) conn.commit() return cur.lastrowid except Error as e3: print("Error adding new track to database") print(e3) return cur.lastrowid # NOTE return with the last rowid def check_tracks(self): t = 1 conn = self.create_connection(self.database) cur = conn.cursor() cur.execute("SELECT * FROM SOURCE_DATA") rows = cur.fetchall() for row in rows: print("Checking Track " + str(t) + " of " + str(self.queue_total), end='\r') self.artist_found = False self.new_uri = row[0] self.new_track = row[1] self.new_artist = row[2] self.bad_name_check() if self.artist_found: t += 1 continue cur.execute("SELECT URI FROM Track_list WHERE URI = '" + self.new_uri + "'") t+=1 wow = cur.fetchone() if wow is None: self.sort_out_the_track_name() self.track_similarity_check() if not self.found: cur6 = conn.cursor() track4 = (self.new_uri, self.new_track, self.new_artist) sql2 = " INSERT OR IGNORE INTO FOR_REVIEW (URI,TRACK_NAME,ARTIST_NAME) VALUES(?,?,?) " cur6.execute(sql2, track4) conn.commit() continue def track_similarity_check(self): conn = self.create_connection(self.database) cur = conn.cursor() cur.execute("SELECT * FROM Track_list") rows2 = cur.fetchall() for row in rows2: self.track_name = row[1] self.artist_name = row[2] self.found = False self.string_similarity(self.new_track, self.track_name) if self.result_ratio == 1: self.string_similarity(self.new_artist, self.artist_name) if self.result_ratio > self.limit: self.found = True return if self.result_ratio > self.limit: self.string_similarity(self.new_artist, self.artist_name) if self.result_ratio > self.limit: self.found = True return def string_similarity(self, str1, str2): result = difflib.SequenceMatcher(a=str1.lower(), b=str2.lower()) self.result_ratio = result.ratio() return self.result_ratio def create_target_playlist(self): query = "https://api.spotify.com/v1/users/{}/playlists".format(self.user_id) request_body = json.dumps({"name": self.playlist_name + " - Rebuilt", "description": "Tracks to Review from " + self.playlist_name + " Original playlist", "public": True}) response = requests.post(query, data=request_body, headers={"Content-Type": "application/json","Authorization": "Bearer {}".format(self.spotify_token)}) response_json = response.json() self.rebuilt_playlist_id = response_json["id"] return def add_tracks_to_target(self): print() t = 1 conn = self.create_connection(self.database) cur = conn.cursor() cur.execute("SELECT COUNT (*) FROM FOR_REVIEW") self.total_records = cur.fetchone() cur.execute("SELECT URI FROM FOR_REVIEW") rows = cur.fetchall() for row in rows: self.add_new_uri = row[0] print("Adding Track " + str(t) + " of " + str(self.total_records[0]) + " to Playlist", end='\r') self.new_track_for_target() t+=1 def new_track_for_target(self): query = "https://api.spotify.com/v1/playlists/{}/tracks?uris={}".format(self.rebuilt_playlist_id, self.add_new_uri) response = requests.request("post", url=query, headers={"Content-type": "application/json","Authorization":"Bearer {}".format(self.spotify_token)}) return def clearup_db_tables(self): conn = self.create_connection(self.database) cur = conn.cursor() cur.execute("DROP TABLE IF EXISTS FOR_REVIEW") conn.commit() cur.execute("DROP TABLE IF EXISTS SOURCE_DATA") conn.commit() cur.execute("VACUUM") conn.commit() def create_connection(self, db_file): conn = None try: conn = sqlite3.connect(db_file) return conn except Error as e1: print("Connection Error") print(e1) return conn def call_refresh(self): global refreshCaller while True: self.spotify_token = refreshCaller.refresh() if self.spotify_token == -1: print( "==============================\n[!] Error refreshing token, retrying in 30 Seconds\n==============================") time.sleep(30) else: break def unwanted(self): try: f = open(self.here_i_am + '\\' + self.bad_names_file_1, encoding='utf-8') for line in f: self.bad_names.append(line.strip()) return except FileNotFoundError: try: f = open(self.here_i_am + '\\' + self.bad_names_file_2, encoding='utf-8') for line in f: self.bad_names.append(line.strip()) return except FileNotFoundError: return def bad_name_check(self): for name in self.bad_names: if name in self.new_artist: self.artist_found = True return return def configure_the_information(self): self.track_name = self.track_name.replace("'", "") self.artist_name = self.artist_name.replace("'", "") self.hyphen = str(self.track_name).find(" - ") if self.hyphen != -1: self.track_name = self.track_name[:self.hyphen] self.sqbrack = str(self.track_name).find("[") if self.sqbrack != -1: self.track_name = self.track_name[:self.sqbrack] self.feat = str(self.track_name).find(" (feat. ") if self.feat != -1: self.track_name = self.track_name[:self.feat] self.remaster = str(self.track_name).find("(Remastered)") if self.remaster != -1: self.track_name = self.track_name[:self.remaster] self.rerecord = str(self.track_name).find("(Rerecorded)") if self.rerecord != -1: self.track_name = self.track_name[:self.rerecord] self.digrem = str(self.track_name).find("(Digitally") if self.digrem != -1: self.track_name = self.track_name[:self.digrem] self.dubstep = str(self.track_name).find("(Dubstep") if self.dubstep != -1: self.track_name = self.track_name[:self.dubstep] self.track_name = self.track_name.strip() self.artist_name = self.artist_name.strip() def sort_out_the_track_name(self): self.new_track = self.new_track.replace("'", "") self.new_artist = self.new_artist.replace("'", "") self.hyphen = str(self.new_track).find(" - ") if self.hyphen != -1: self.new_track = self.new_track[:self.hyphen] self.sqbrack = str(self.new_track).find("[") if self.sqbrack != -1: self.new_track = self.new_track[:self.sqbrack] self.feat = str(self.new_track).find(" (feat. ") if self.feat != -1: self.new_track = self.new_track[:self.feat] self.remaster = str(self.new_track).find("(Remastered)") if self.remaster != -1: self.new_track = self.new_track[:self.remaster] self.rerecord = str(self.new_track).find("(Rerecorded)") if self.rerecord != -1: self.new_track = self.new_track[:self.rerecord] self.digrem = str(self.new_track).find("(Digitally") if self.digrem != -1: self.new_track = self.new_track[:self.digrem] self.dubstep = str(self.new_track).find("(Dubstep") if self.dubstep != -1: self.new_track = self.new_track[:self.dubstep] self.new_track = self.new_track.strip() self.new_artist = self.new_artist.strip() a = Minidupe() a.call_refresh() a.dupecheck()