import sqlite3, datetime, os, difflib class Masterdupefinder: def __init__(self): self.here_i_am = os.path.dirname(os.path.abspath(__file__)) ############ SET YOUR VALUES HERE ################# self.boom_path = "C:\\Users\AlSmi\Dropbox\MUSIC-TIN\BOOM-Sorter" self.master_database_name = "Endless Feed" self.boom_database_name = "Boom Tracks Database" self.limit = 0.9 ############ SET YOUR VALUES HERE ################# self.boom_table_list = [] self.table0 = 'BOOM_1950s' self.table1 = 'BOOM_1960s' self.table2 = 'BOOM_1970s' self.table3 = 'BOOM_1980s' self.table4 = 'BOOM_1990s' self.table5 = 'BOOM_2000s' self.table6 = 'BOOM_2010s' self.table7 = 'BOOM_2020s' self.table8 = 'BOOM_Others' def master_dupe_check(self): now = datetime.datetime.now() print("Current Run Time:= " + now.strftime("%d-%m-%Y %H:%M:%S")) print() print(" BOOM Dupe Finder v.1.1.0 02/12/2022 \U00002694 Viking Utilities \U00002694") print("\033[91m########################### Latest Update ################################\033[0m") print("\033[33mv.1.1.0") print("Currently set to " + str(self.limit) + " (" + str(self.limit*100) + "%) match... Set this value at line 10. ") print("Also set database names and path correctly at line 7,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.boom_path + 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 boom") self.conn.commit() cur.execute("DROP TABLE IF EXISTS boom.URI_Matches") self.conn.commit() cur.execute( "CREATE TABLE IF NOT EXISTS boom.URI_Matches (URI text PRIMARY KEY, TRACK_NAME text, ARTIST_NAME text, FROM_TABLE text);") self.conn.commit() cur.execute("DROP TABLE IF EXISTS boom.Full_Matches") self.conn.commit() cur.execute( "CREATE TABLE IF NOT EXISTS boom.Full_Matches (URI text PRIMARY KEY, TRACK_NAME text, ARTIST_NAME text, FROM_TABLE text);") self.conn.commit() cur.execute("DROP TABLE IF EXISTS boom.Manual_Matches") self.conn.commit() cur.execute( "CREATE TABLE IF NOT EXISTS boom.Manual_Matches (URI text PRIMARY KEY, TRACK_NAME text, ARTIST_NAME text, FROM_TABLE text);") self.conn.commit() cur.execute("DROP TABLE IF EXISTS boom.No_Match") self.conn.commit() cur.execute( "CREATE TABLE IF NOT EXISTS boom.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 boom") self.conn.commit() self.tbid = 0 while self.tbid < 9: print("Working on playlist " + self.boom_table_list[self.tbid]) cur.execute("SELECT * FROM boom.'" + self.boom_table_list[self.tbid] + "' WHERE HISTORIC = FALSE") 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 boom.No_Match (URI,TRACK_NAME,ARTIST_NAME,FROM_TABLE) VALUES(?,?,?,?) ''' cur6.execute(sql2, track4) self.conn.commit() cur.execute("UPDATE boom.'" + self.boom_table_list[self.tbid] + "' SET HISTORIC = TRUE WHERE URI = '" + self.boom_uri + "'") 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 boom.URI_Matches (URI,TRACK_NAME,ARTIST_NAME,FROM_TABLE) VALUES(?,?,?,?) ''' cur.execute(sql, track) self.conn.commit() cur.execute("UPDATE boom.'" + self.boom_table_list[self.tbid] + "' SET HISTORIC = TRUE WHERE URI = '" + self.boom_uri + "'") 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) 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 boom.Full_Matches (URI,TRACK_NAME,ARTIST_NAME,FROM_TABLE) VALUES(?,?,?,?) ''' cur4.execute(sql, track2) self.conn.commit() cur.execute("UPDATE boom.'" + self.boom_table_list[self.tbid] + "' SET HISTORIC = TRUE WHERE URI = '" + self.boom_uri + "'") self.conn.commit() self.found = True return 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 boom.Manual_Matches (URI,TRACK_NAME,ARTIST_NAME,FROM_TABLE) VALUES(?,?,?,?) ''' cur5.execute(sql, track3) self.conn.commit() cur.execute("UPDATE boom.'" + self.boom_table_list[self.tbid] + "' SET HISTORIC = TRUE WHERE URI = '" + self.boom_uri + "'") self.found = True return 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()