import time, sqlite3, datetime, os, difflib from alive_progress import alive_bar class Masterdupefinder: def __init__(self): self.here_i_am = os.path.dirname(os.path.abspath(__file__)) ############ SET YOUR VALUES HERE ################# self.master_database_name = "Endless Feed" self.boom_database_name = "plusothers" self.pl_count = 1 self.limit = 0.9 self.out_db = "tracks" ############ SET YOUR VALUES HERE ################# self.boom_table_list = [] self.table0 = self.out_db self.table1 = self.out_db + '_1960s' self.table2 = self.out_db + '_1970s' self.table3 = self.out_db + '_1980s' self.table4 = self.out_db + '_1990s' self.table5 = self.out_db + '_2000s' self.table6 = self.out_db + '_2010s' self.table7 = self.out_db + '_2020s' self.table8 = self.out_db + '_Others' def master_dupe_check(self): self.tic = time.perf_counter() now = datetime.datetime.now() print("Current Run Time:= " + now.strftime("%d-%m-%Y %H:%M:%S")) print() print(" BOOM Dupe Finder v.1.0.0 02/12/2022 \U00002694 Viking Utilities \U00002694") print("\033[91m########################### Latest Update ################################\033[0m") print("\033[33mv.1.0.0 INITIAL ATTEMPT") print("Currently set to " + str(self.limit) + " (" + str(self.limit*100) + "%) match... Set this value at line 10. ") print("Also set database names correctly at line 8 & 9.\033[0m") print("\033[91m##########################################################################\033[0m") self.master_database = self.here_i_am + r"\\" + self.master_database_name + ".db" self.boom_database = self.here_i_am + r"\\" + self.boom_database_name + ".db" self.boom_table_list.extend([self.table0, self.table1, self.table2, self.table3, self.table4, self.table5, self.table6, self.table7, self.table8]) self.config_results_tables() self.tracks_check() def config_results_tables(self): self.create_connection(self.master_database) cur = self.conn.cursor() cur.execute("ATTACH DATABASE '" + self.boom_database + "' AS '" + self.out_db + "'") self.conn.commit() cur.execute("DROP TABLE IF EXISTS '" + self.out_db + "'.URI_Matches") self.conn.commit() cur.execute( "CREATE TABLE IF NOT EXISTS '" + self.out_db +"'.URI_Matches (URI text PRIMARY KEY, TRACK_NAME text, ARTIST_NAME text, FROM_TABLE text);") self.conn.commit() cur.execute("DROP TABLE IF EXISTS '" + self.out_db +"'.Full_Matches") self.conn.commit() cur.execute( "CREATE TABLE IF NOT EXISTS '" + self.out_db +"'.Full_Matches (URI text PRIMARY KEY, TRACK_NAME text, ARTIST_NAME text, FROM_TABLE text);") self.conn.commit() cur.execute("DROP TABLE IF EXISTS '" + self.out_db +"'.Manual_Matches") self.conn.commit() cur.execute( "CREATE TABLE IF NOT EXISTS '" + self.out_db +"'.Manual_Matches (URI text PRIMARY KEY, TRACK_NAME text, ARTIST_NAME text, FROM_TABLE text);") self.conn.commit() cur.execute("DROP TABLE IF EXISTS '" + self.out_db +"'.No_Match") self.conn.commit() cur.execute( "CREATE TABLE IF NOT EXISTS '" + self.out_db +"'.No_Match (URI text PRIMARY KEY, TRACK_NAME text, ARTIST_NAME text, FROM_TABLE text);") self.conn.commit() def tracks_check(self): self.create_connection(self.master_database) cur = self.conn.cursor() cur.execute("ATTACH DATABASE '" + self.boom_database + "' AS '" + self.out_db +"'") self.conn.commit() self.tbid = 0 while self.tbid < self.pl_count: self.toc = time.perf_counter() ty_res = time.gmtime(self.toc - self.tic) res = time.strftime(" %H Hours : %M Minutes : %S Seconds", ty_res) print() print(f"Dupe Finder Took {res}") print() print("Working on playlist " + self.boom_table_list[self.tbid]) cur.execute("SELECT * FROM '" + self.out_db +"'." + self.boom_table_list[self.tbid]) rows=cur.fetchall() for row in rows: self.boom_uri = row[0] self.boom_track = row[1] self.boom_artist = row[2] cur.execute("SELECT URI FROM Track_list WHERE URI = '" + self.boom_uri + "'") wow = cur.fetchone() if wow is None: self.sort_out_the_track_name() self.track_similarity_check() if not self.found: cur6 = self.conn.cursor() track4 = (self.boom_uri, self.boom_track, self.boom_artist, self.boom_table_list[self.tbid]) sql2 = ' INSERT OR IGNORE INTO ' + self.out_db +'.No_Match (URI,TRACK_NAME,ARTIST_NAME,FROM_TABLE) VALUES(?,?,?,?) ' cur6.execute(sql2, track4) self.conn.commit() continue else: track = (self.boom_uri, self.boom_track, self.boom_artist, self.boom_table_list[self.tbid]) sql = ' INSERT OR IGNORE INTO ' + self.out_db +'.URI_Matches (URI,TRACK_NAME,ARTIST_NAME,FROM_TABLE) VALUES(?,?,?,?) ' cur.execute(sql, track) self.conn.commit() self.tbid+=1 def track_similarity_check(self): cur = self.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.boom_track, self.track_name) #100% or FULL Match of track name...... if self.result_ratio == 1: self.string_similarity(self.boom_artist, self.artist_name) if self.result_ratio > self.limit: cur4 = self.conn.cursor() track2 = (self.boom_uri, self.boom_track, self.boom_artist, self.boom_table_list[self.tbid]) sql = ' INSERT OR IGNORE INTO ' + self.out_db +'.Full_Matches (URI,TRACK_NAME,ARTIST_NAME,FROM_TABLE) VALUES(?,?,?,?) ' cur4.execute(sql, track2) self.conn.commit() self.found = True return #Manually set limit Match of track name...... if self.result_ratio > self.limit: self.string_similarity(self.boom_artist, self.artist_name) if self.result_ratio > self.limit: cur5 = self.conn.cursor() track3 = (self.boom_uri, self.boom_track, self.boom_artist, self.boom_table_list[self.tbid]) sql = ' INSERT OR IGNORE INTO ' + self.out_db +'.Manual_Matches (URI,TRACK_NAME,ARTIST_NAME,FROM_TABLE) VALUES(?,?,?,?) ' cur5.execute(sql, track3) self.conn.commit() self.found = True return # NOTE Format The Track Name for Duplicate Checks def sort_out_the_track_name(self): self.boom_track = self.boom_track.replace("'", "") self.boom_artist = self.boom_artist.replace("'", "") self.hyphen = str(self.boom_track).find(" - ") if self.hyphen != -1: self.boom_track = self.boom_track[:self.hyphen] self.sqbrack = str(self.boom_track).find("[") if self.sqbrack != -1: self.boom_track = self.boom_track[:self.sqbrack] self.feat = str(self.boom_track).find(" (feat. ") if self.feat != -1: self.boom_track = self.boom_track[:self.feat] self.remaster = str(self.boom_track).find("(Remastered)") if self.remaster != -1: self.boom_track = self.boom_track[:self.remaster] self.digrem = str(self.boom_track).find("(Digitally") if self.digrem != -1: self.boom_track = self.boom_track[:self.digrem] self.dubstep = str(self.boom_track).find("(Dubstep") if self.dubstep != -1: self.boom_track = self.boom_track[:self.dubstep] self.boom_track = self.boom_track.strip() self.boom_artist = self.boom_artist.strip() 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_connection(self, db_file): # NOTE Connect to the database self.conn = None self.conn = sqlite3.connect(db_file) return self.conn a = Masterdupefinder() a.master_dupe_check()