import time, sqlite3, datetime from alive_progress import alive_bar class Dupefinder: def __init__(self): ############ SET YOUR VALUES HERE ################# self.dbpath = 'j:\\Spotty\secrets_and_database' self.database_name = "Enormous Feed" ############ SET YOUR VALUES HERE ################# self.track1_uri_list = [] self.track1_name_list = [] self.track1_artist_list = [] self.newdupe_track_uri_list = [] self.newdupe_track_name_list = [] self.newdupe_track_artist_list = [] self.adddupe_uri_list = [] self.adddupe_name_list = [] self.adddupe_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.3.1.0 03/11/2023 \U00002694 Viking Utilities \U00002694") print("\033[91m########################### Latest Update ################################\033[0m") print("\033[33mv.1.0.0 Shows only NEW duplicates. SET Reviewed equal to 1 in New_Dupes_Table to mark as done.") print("Also set database path and name correctly at lines 7 & 8. \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.add_tracks_to_new_dupe() 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[2] 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]) tk_nme = row[1] tk_nme2 = tk_nme.title() #if tk_nme != tk_nme2: #print(tk_nme) #print(tk_nme2) #self.tk_tit_nme = self.tk_nme.title() self.track1_name_list.append(tk_nme2) self.track1_artist_list.append(row[2]) item1 = 0 while item1 < self.total_records: with alive_bar(self.total_records, title="Comparing Database Tracks") as bar: for item1 in range(self.total_records): check = False self.dupecount = 0 name_test = self.track1_name_list[item1] exist_count = self.track1_name_list.count(name_test) if exist_count > 1: namedupe = self.track1_name_list[item1] if not check: namecheck = namedupe check = True result = [] result.append(item1) for qb, self.track1_name_list[item1] in enumerate(self.track1_name_list): if self.track1_name_list[item1] == namecheck: result.append(qb) lengthdupeindex = len(result) while self.dupecount < lengthdupeindex: cur4 = self.conn.cursor() cur4.execute("SELECT * FROM Reviewed_Dupe_Tracks where URI = '" + self.track1_uri_list[result[self.dupecount]] + "'") row = cur4.fetchone() if row is None: if self.track1_uri_list[result[self.dupecount]] not in self.newdupe_track_uri_list: self.newdupe_track_uri_list.append(self.track1_uri_list[result[self.dupecount]]) self.newdupe_track_name_list.append(namecheck) self.newdupe_track_artist_list.append(self.track1_artist_list[result[self.dupecount]]) self.dupecount += 1 item1 += 1 bar() else: item1 += 1 bar() continue def add_tracks_to_new_dupe(self): check_item = 0 dupchk = len(self.adddupe_uri_list) while check_item < dupchk: with alive_bar(dupchk, title="Adding Tracks To New Dupes") as bar: for check_item in range(dupchk): track1 = (self.adddupe_uri_list[check_item], self.adddupe_name_list[check_item], self.adddupe_artist_list[check_item], 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() check_item += 1 bar() def check_dupe_artists(self): x = 0 y = 0 dupetot = len(self.newdupe_track_uri_list) while x < dupetot: while y < dupetot: if x == y: y +=1 continue if self.newdupe_track_name_list[x] == self.newdupe_track_name_list[y] and self.newdupe_track_artist_list[x] == self.newdupe_track_artist_list[y] and self.newdupe_track_uri_list[x] != self.newdupe_track_uri_list[y]: self.adddupe_uri_list.append(self.newdupe_track_uri_list[x]) self.adddupe_name_list.append(self.newdupe_track_name_list[x]) self.adddupe_artist_list.append(self.newdupe_track_artist_list[x]) y+=1 continue else: y+=1 continue x+=1 y=0 continue print() 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()