diff --git a/maloja/proccontrol/tasks/__init__.py b/maloja/proccontrol/tasks/__init__.py index 744534a..b86008f 100644 --- a/maloja/proccontrol/tasks/__init__.py +++ b/maloja/proccontrol/tasks/__init__.py @@ -12,14 +12,22 @@ def loadexternal(filename): return from .importer import import_scrobbles - imported,warning,skipped,failed = import_scrobbles(filename) - print("Successfully imported",imported,"scrobbles!") - if warning > 0: - print(col['orange'](f"{warning} Warning{'s' if warning != 1 else ''}!")) - if skipped > 0: - print(col['orange'](f"{skipped} Skipped!")) - if failed > 0: - print(col['red'](f"{failed} Error{'s' if failed != 1 else ''}!")) + result = import_scrobbles(filename) + + msg = f"Successfully imported {result['CONFIDENT_IMPORT'] + result['UNCERTAIN_IMPORT']} scrobbles" + if result['UNCERTAIN_IMPORT'] > 0: + warningmsg = col['orange'](f"{result['UNCERTAIN_IMPORT']} Warning{'s' if result['UNCERTAIN_IMPORT'] != 1 else ''}!") + msg += f" ({warningmsg})" + print(msg) + + msg = f"Skipped {result['CONFIDENT_SKIP'] + result['UNCERTAIN_SKIP']} scrobbles" + if result['UNCERTAIN_SKIP'] > 0: + warningmsg = col['orange'](f"{result['UNCERTAIN_SKIP']} Warning{'s' if result['UNCERTAIN_SKIP'] != 1 else ''}!") + msg += f" ({warningmsg})" + print(msg) + + if result['FAIL'] > 0: + print(col['red'](f"{result['FAIL']} Error{'s' if result['FAIL'] != 1 else ''}!")) def backuphere(): from .backup import backup diff --git a/maloja/proccontrol/tasks/importer.py b/maloja/proccontrol/tasks/importer.py index 774932a..69e3d3d 100644 --- a/maloja/proccontrol/tasks/importer.py +++ b/maloja/proccontrol/tasks/importer.py @@ -12,12 +12,22 @@ c = CleanerAgent() def warn(msg): print(col['orange'](msg)) +def skip(msg): + print(col['#ffcba4'](msg)) def err(msg): print(col['red'](msg)) def import_scrobbles(inputf): + result = { + "CONFIDENT_IMPORT": 0, + "UNCERTAIN_IMPORT": 0, + "CONFIDENT_SKIP": 0, + "UNCERTAIN_SKIP": 0, + "FAIL": 0 + } + filename = os.path.basename(inputf) if re.match(".*\.csv",filename): @@ -37,7 +47,7 @@ def import_scrobbles(inputf): else: print("File",inputf,"could not be identified as a valid import source.") - return 0,0,0,0 + return result print(f"Parsing {col['yellow'](inputf)} as {col['cyan'](type)} export") @@ -47,7 +57,7 @@ def import_scrobbles(inputf): while True: action = prompt(f"Already imported {type} data. [O]verwrite, [A]ppend or [C]ancel?",default='c').lower()[0] if action == 'c': - return 0,0,0,0 + return result elif action == 'a': mode = 'a' break @@ -61,18 +71,12 @@ def import_scrobbles(inputf): with open(outputf,mode) as outputfd: - success, warning, skipped, failed = 0, 0, 0, 0 + timestamps = set() for status,scrobble in importfunc(inputf): - if status == 'FAIL': - failed += 1 - elif status == 'SKIP': - skipped += 1 - else: - success += 1 - if status == 'WARN': - warning += 1 + result[status] += 1 + if status in ['CONFIDENT_IMPORT','UNCERTAIN_IMPORT']: while scrobble['timestamp'] in timestamps: scrobble['timestamp'] += 1 @@ -93,10 +97,10 @@ def import_scrobbles(inputf): ]) outputfd.write(outputline + '\n') - if success % 100 == 0: - print(f"Imported {success} scrobbles...") + if (result['CONFIDENT_IMPORT'] + result['UNCERTAIN_IMPORT']) % 100 == 0: + print(f"Imported {result['CONFIDENT_IMPORT'] + result['UNCERTAIN_IMPORT']} scrobbles...") - return success, warning, skipped, failed + return result def parse_spotify_lite(inputf): inputfolder = os.path.dirname(inputf) @@ -145,27 +149,29 @@ def parse_spotify_full(inputf): if title is None: - warn(f"{entry} has no title, skipping...") - yield ('SKIP',None) + skip(f"{entry} has no title, skipping...") + yield ('CONFIDENT_SKIP',None) continue if artist is None: - warn(f"{entry} has no artist, skipping...") - yield ('SKIP',None) + skip(f"{entry} has no artist, skipping...") + yield ('CONFIDENT_SKIP',None) continue if played < 30: - warn(f"{entry} is shorter than 30 seconds, skipping...") - yield ('SKIP',None) + skip(f"{entry} is shorter than 30 seconds, skipping...") + yield ('CONFIDENT_SKIP',None) continue # if offline_timestamp is a proper number, we treat it as # accurate and check duplicates by that exact timestamp if timestamp != 0: - status = 'SUCCESS' + if timestamp in timestamps and (artist,title) in timestamps[timestamp]: - warn(f"{entry} seems to be a duplicate, skipping...") - yield ('SKIP',None) + skip(f"{entry} seems to be a duplicate, skipping...") + yield ('CONFIDENT_SKIP',None) continue - timestamps.setdefault(timestamp,[]).append((artist,title)) + else: + status = 'CONFIDENT_IMPORT' + timestamps.setdefault(timestamp,[]).append((artist,title)) # if it's 0, we use ts instead, but identify duplicates differently # (cause the ts is not accurate) @@ -188,13 +194,13 @@ def parse_spotify_full(inputf): # - exact same track uri # - exact same ms_played if (abs(scr[0] - timestamp) < 30) and scr[1:] == scrobble_describe[1:]: - warn(f"{entry} has been identified as potential duplicate, skipping...") - yield ('SKIP',None) + warn(f"{entry} might be a duplicate, skipping...") + yield ('UNCERTAIN_SKIP',None) found_similar = True break else: # no duplicates, assume proper scrobble but warn - status = 'WARN' + status = 'UNCERTAIN_IMPORT' warn(f"{entry} might have an inaccurate timestamp.") inaccurate_timestamps.setdefault(ts_group,[]).append(scrobble_describe) @@ -230,7 +236,7 @@ def parse_lastfm(inputf): continue try: - yield ('SUCCESS',{ + yield ('CONFIDENT_IMPORT',{ 'title': title, 'artiststr': artist, 'album': album,