import 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.boom_path = "C:\\Users\AlSmi\Dropbox\MUSIC-TIN\BOOM-Sorter" self.master_database_name_1 = "Endless Feed" self.master_database_name_2 = "MASTER" self.boom_database_name = "Boom Tracks Database" self.limit = 0.9 ############ SET YOUR VALUES HERE ################# self.boom_table_list = [] self.user_db_list = [] self.user_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' self.user_id1 = 'Ollie' self.user_id2 = 'Freddie' self.user = 0 def master_dupe_check(self): now = datetime.datetime.now() print("Current Run Time:= " + now.strftime("%d-%m-%Y %H:%M:%S")) print() self.boom_database = self.here_i_am + r"\\" + self.boom_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.user_db_list.extend([self.master_database_name_1, self.master_database_name_2]) self.user_list.extend([self.user_id1, self.user_id2]) while self.user < 2: self.master_database = self.here_i_am + r"\\" + self.user_db_list[self.user] + ".db" self.config_results_tables() self.tracks_check() self.user+=1 self.user = 0 while self.user < 2: self.master_database = self.here_i_am + r"\\" + self.user_db_list[self.user] + ".db" self.create_connection(self.master_database) cur = self.conn.cursor() cur.execute("ATTACH DATABASE '" + self.boom_database + "' AS boom") self.conn.commit() self.count_no_match_table() self.compare_tracks() self.user+=1 return 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." + self.user_list[self.user] + "_URI_Matches") self.conn.commit() cur.execute( "CREATE TABLE IF NOT EXISTS boom." + self.user_list[self.user] + "_URI_Matches (URI text PRIMARY KEY, TRACK_NAME text, ARTIST_NAME text, SOURCE text, FROM_TABLE text);") self.conn.commit() cur.execute("DROP TABLE IF EXISTS boom." + self.user_list[self.user] + "_Full_Matches") self.conn.commit() cur.execute( "CREATE TABLE IF NOT EXISTS boom." + self.user_list[self.user] + "_Full_Matches (URI text PRIMARY KEY, TRACK_NAME text, ARTIST_NAME text, SOURCE text, FROM_TABLE text);") self.conn.commit() cur.execute("DROP TABLE IF EXISTS boom." + self.user_list[self.user] + "_Manual_Matches") self.conn.commit() cur.execute( "CREATE TABLE IF NOT EXISTS boom." + self.user_list[self.user] + "_Manual_Matches (URI text PRIMARY KEY, TRACK_NAME text, ARTIST_NAME text, SOURCE text, FROM_TABLE text);") self.conn.commit() cur.execute("DROP TABLE IF EXISTS boom." + self.user_list[self.user] + "_No_Match") self.conn.commit() cur.execute( "CREATE TABLE IF NOT EXISTS boom." + self.user_list[self.user] + "_No_Match (URI text PRIMARY KEY, TRACK_NAME text, ARTIST_NAME text, SOURCE text, FROM_TABLE text, REVIEW integer);") self.conn.commit() cur.execute("DROP TABLE IF EXISTS boom." + self.user_list[self.user] + "_To_Review") self.conn.commit() cur.execute( "CREATE TABLE IF NOT EXISTS boom." + self.user_list[self.user] + "_To_Review (URI text PRIMARY KEY, TRACK_NAME text, ARTIST_NAME text, SOURCE 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: now = datetime.datetime.now() print("Working on playlist " + self.boom_table_list[self.tbid] + " Started at " + now.strftime("%H:%M:%S")) 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] self.boom_source = row[7] self.review = 0 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_source, self.boom_table_list[self.tbid], self.review) sql2 = " INSERT OR IGNORE INTO boom." + self.user_list[self.user] + "_No_Match (URI,TRACK_NAME,ARTIST_NAME,SOURCE,FROM_TABLE,REVIEW) VALUES(?,?,?,?,?,?) " cur6.execute(sql2, track4) self.conn.commit() if self.user == 1: 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_source, self.boom_table_list[self.tbid]) sql = " INSERT OR IGNORE INTO boom." + self.user_list[self.user] + "_URI_Matches(URI, TRACK_NAME, ARTIST_NAME, SOURCE, FROM_TABLE) VALUES(?,?,?,?,?) " cur.execute(sql, track) self.conn.commit() if self.user == 1: 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_source, self.boom_table_list[self.tbid]) sql = " INSERT OR IGNORE INTO boom." + self.user_list[self.user] + "_Full_Matches (URI,TRACK_NAME,ARTIST_NAME,SOURCE,FROM_TABLE) VALUES(?,?,?,?,?) " cur4.execute(sql, track2) self.conn.commit() if self.user == 1: 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_source, self.boom_table_list[self.tbid]) sql = " INSERT OR IGNORE INTO boom." + self.user_list[self.user] + "_Manual_Matches (URI,TRACK_NAME,ARTIST_NAME,SOURCE,FROM_TABLE) VALUES(?,?,?,?,?) " cur5.execute(sql, track3) self.conn.commit() if self.user == 1: 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 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.rerecord = str(self.boom_track).find("(Rerecorded)") if self.rerecord != -1: self.boom_track = self.boom_track[:self.rerecord] 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 def count_no_match_table(self): cur = self.conn.cursor() cur.execute("SELECT COUNT (*) FROM boom." + self.user_list[self.user] + "_No_Match") self.total_records = int(cur.fetchone()[0]) self.conn.commit() def compare_tracks(self): cur = self.conn.cursor() cur.execute("SELECT * FROM boom." + self.user_list[self.user] + "_No_Match ORDER BY TRACK_NAME ASC") rows = cur.fetchall() i = 0 while i < self.total_records: with alive_bar(self.total_records, title="Comparing " + self.user_list[self.user] + "_No_Match Tracks") as bar: for row in rows: self.track1uri = row[0] self.track1name = row[1] self.track1artist = row[2] self.track1source = row[3] self.track1from = row[4] self.review = row[5] cur.execute("SELECT * FROM boom." + self.user_list[self.user] + "_To_Review WHERE TRACK_NAME = '" + self.track1name + "' AND ARTIST_NAME = '" + self.track1artist + "'") rows2 =cur.fetchone() if rows2 is None: cur2 = self.conn.cursor() cur2.execute( "SELECT * FROM boom." + self.user_list[self.user] + "_No_Match WHERE TRACK_NAME = '" + self.track1name + "' AND ARTIST_NAME = '" + self.track1artist + "' AND REVIEW = FALSE") rows2 = cur2.fetchall() for row2 in rows2: cur.execute("UPDATE boom." + self.user_list[self.user] + "_No_Match SET REVIEW = TRUE WHERE TRACK_NAME = '" + self.track1name + "' AND ARTIST_NAME = '" + self.track1artist + "'") self.conn.commit() no_match_track = (self.track1uri, self.track1name, self.track1artist, self.track1source, self.track1from) sql = " INSERT OR IGNORE INTO boom." + self.user_list[self.user] + "_To_Review (URI,TRACK_NAME,ARTIST_NAME,SOURCE,FROM_TABLE) VALUES(?,?,?,?,?)" cur3 = self.conn.cursor() cur3.execute(sql, no_match_track) self.conn.commit() i += 1 bar() return a = Masterdupefinder() a.master_dupe_check()