import time, sqlite3, datetime from alive_progress import alive_bar class Dupefinder: def __init__(self): self.dbpath = 'j:\\Spotty\secrets_and_database' ############ SET YOUR VALUES HERE ################# self.database_name = "Endless Feed" #self.database_name = "MEGA MASTER BACKUP" ############ SET YOUR VALUES HERE ################# self.track1_uri_list = [] self.track1_name_list = [] self.track1_artist_list = [] self.track2_uri_list = [] self.track2_name_list = [] self.track2_artist_list = [] def find_the_dupes(self): tic = time.perf_counter() now = datetime.datetime.now() print("Current Run Time:= " + now.strftime("%d-%m-%Y %H:%M:%S")) print() print("Dupe Finder v.1.1.2 01/07/2023 \U00002694 Viking Utilities \U00002694") print("\033[91m########################### Latest Update ################################\033[0m") print("\033[33mv.1.0.0 Now uses 'lists' to speed things up. SET Reviewed equal to 1 in New_Dupes_Table to mark as done.") print("Also set database name correctly at line 8. WORK IN PROGRESS BE CAREFUL WITH USEAGE\033[0m") print("\033[91m##########################################################################\033[0m") self.database = self.dbpath + r"\\" + self.database_name + ".db" self.create_connection(self.database) self.check_dupe_tables() self.xfer_new_reviewed() self.flex_new_dupe_table() self.compare_tracks() self.check_dupe_artists() self.hoover() toc = time.perf_counter() ty_res = time.gmtime(toc - tic) res = time.strftime(" %H Hours : %M Minutes : %S Seconds", ty_res) print() print(f"Dupe Finder Took {res}") print() def create_connection(self, db_file): # NOTE Connect to the database self.conn = None self.conn = sqlite3.connect(db_file) return self.conn def xfer_new_reviewed(self): cur = self.conn.cursor() cur.execute("SELECT * FROM New_Dupe_Tracks WHERE REVIEWED = 1") rows = cur.fetchall() for row in rows: revuri = row[0] revname = row[1] revartist = row[1] revrev = row[3] revtrack = (revuri, revname, revartist, revrev) sql = ''' INSERT OR IGNORE INTO Reviewed_Dupe_Tracks (URI,TRACK_NAME,ARTIST_NAME,REVIEWED) VALUES(?,?,?,?) ''' cur.execute(sql, revtrack) self.conn.commit() def flex_new_dupe_table(self): cur = self.conn.cursor() cur.execute("DROP TABLE IF EXISTS New_Dupe_Tracks") self.conn.commit() cur.execute("CREATE TABLE IF NOT EXISTS New_Dupe_Tracks (URI text PRIMARY KEY, TRACK_NAME text, ARTIST_NAME text, REVIEWED integer);") self.conn.commit() def compare_tracks(self): cur = self.conn.cursor() cur.execute("SELECT * FROM Track_List ORDER BY TRACK_NAME ASC") rows = cur.fetchall() for row in rows: self.track1_uri_list.append(row[0]) self.track1_name_list.append(row[1]) self.track1_artist_list.append(row[2]) self.track2_uri_list.append(row[0]) self.track2_name_list.append(row[1]) self.track2_artist_list.append(row[2]) i1 = 0 while i1 < self.total_records: with alive_bar(self.total_records, title="Comparing Database Tracks") as bar: for i1 in range(self.total_records): self.dc = 0 name_test = self.track1_name_list[i1] exist_count = self.track1_name_list.count(name_test) if exist_count > 1: dupes_indices = [] namedupe = self.track1_name_list[i1] nameslist = self.track1_name_list for namenumber in range(len(nameslist)): if nameslist[namenumber] == str(namedupe): dupes_indices.append(namenumber) ldi = len(dupes_indices) while self.dc < ldi: cur4 = self.conn.cursor() cur4.execute("SELECT * FROM Reviewed_Dupe_Tracks where URI = '" + self.track1_uri_list[dupes_indices[self.dc]] + "'") row = cur4.fetchone() if row is None: track1 = (self.track1_uri_list[dupes_indices[self.dc]], self.track1_name_list[ dupes_indices[self.dc]], self.track1_artist_list[dupes_indices[self.dc]], 0) sql = ''' INSERT OR IGNORE INTO New_Dupe_Tracks (URI,TRACK_NAME,ARTIST_NAME,REVIEWED) VALUES(?,?,?,?) ''' cur3 = self.conn.cursor() cur3.execute(sql, track1) self.conn.commit() self.dc += 1 i1 += 1 bar() else: i1 += 1 bar() continue def check_dupe_artists(self): cur = self.conn.cursor() cur2 = self.conn.cursor() cur.execute("SELECT * FROM New_Dupe_Tracks ORDER BY TRACK_NAME ASC") rows = cur.fetchall() trw = len(rows) rw = 1 for row in rows: print("Working on Item " + str(rw) + " of " + str(trw), end='\r') r1uri = row[0] r1track = row[1] r1track = r1track.replace("'", "\'") r1artist = row[2] r1artist = r1artist.replace("'", "\'") cur2.execute("SELECT * FROM New_Dupe_Tracks WHERE TRACK_NAME = '" + r1track + "'") rows2 = cur2.fetchall() tkcount = len(rows2) for row2 in rows2: r2uri = row2[0] r2artist = row2[2] r2artist = r2artist.replace("'", "\'") if r1artist == r2artist: rw+=1 continue else: cur2.execute("DELETE FROM New_Dupe_Tracks WHERE URI = '" + r2uri + "'") self.conn.commit() tkcount =-1 if tkcount == 1: cur2.execute("DELETE FROM New_Dupe_Tracks WHERE URI = '" + r1uri + "'") cur2.execute("DELETE FROM New_Dupe_Tracks WHERE URI = '" + r2uri + "'") self.conn.commit() rw+=1 continue def check_dupe_tables(self): cur = self.conn.cursor() cur.execute("SELECT COUNT (*) FROM Track_List") self.total_records = int(cur.fetchone()[0]) cur.execute("CREATE TABLE IF NOT EXISTS New_Dupe_Tracks (URI text PRIMARY KEY, TRACK_NAME text, ARTIST_NAME text, REVIEWED integer);") cur.execute("CREATE TABLE IF NOT EXISTS Reviewed_Dupe_Tracks (URI text PRIMARY KEY, TRACK_NAME text, ARTIST_NAME text, REVIEWED integer);") self.conn.commit() def hoover(self): cur = self.conn.cursor() cur.execute("VACUUM") self.conn.commit() a = Dupefinder() a.find_the_dupes()