Improved feedback of import

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)
+	result = import_scrobbles(filename)
-	print("Successfully imported",imported,"scrobbles!")
+
-	if warning > 0:
+	msg = f"Successfully imported {result['CONFIDENT_IMPORT'] + result['UNCERTAIN_IMPORT']} scrobbles"
-		print(col['orange'](f"{warning} Warning{'s' if warning != 1 else ''}!"))
+	if result['UNCERTAIN_IMPORT'] > 0:
-	if skipped > 0:
+		warningmsg = col['orange'](f"{result['UNCERTAIN_IMPORT']} Warning{'s' if result['UNCERTAIN_IMPORT'] != 1 else ''}!")
-		print(col['orange'](f"{skipped} Skipped!"))
+		msg += f" ({warningmsg})"
-	if failed > 0:
+	print(msg)
-		print(col['red'](f"{failed} Error{'s' if failed != 1 else ''}!"))
+
+	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

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':
+			result[status] += 1
-				failed += 1
+			if status in ['CONFIDENT_IMPORT','UNCERTAIN_IMPORT']:
-			elif status == 'SKIP':
-				skipped += 1
-			else:
-				success += 1
-				if status == 'WARN':
-					warning += 1
 
 				while scrobble['timestamp'] in timestamps:
 					scrobble['timestamp'] += 1
@@ -93,10 +97,10 @@ def import_scrobbles(inputf):
 				])
 				outputfd.write(outputline + '\n')
 
-				if success % 100 == 0:
+				if (result['CONFIDENT_IMPORT'] + result['UNCERTAIN_IMPORT']) % 100 == 0:
-					print(f"Imported {success} scrobbles...")
+					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,26 +149,28 @@ def parse_spotify_full(inputf):
 
 
 				if title is None:
-					warn(f"{entry} has no title, skipping...")
+					skip(f"{entry} has no title, skipping...")
-					yield ('SKIP',None)
+					yield ('CONFIDENT_SKIP',None)
 					continue
 				if artist is None:
-					warn(f"{entry} has no artist, skipping...")
+					skip(f"{entry} has no artist, skipping...")
-					yield ('SKIP',None)
+					yield ('CONFIDENT_SKIP',None)
 					continue
 				if played < 30:
-					warn(f"{entry} is shorter than 30 seconds, skipping...")
+					skip(f"{entry} is shorter than 30 seconds, skipping...")
-					yield ('SKIP',None)
+					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...")
+						skip(f"{entry} seems to be a duplicate, skipping...")
-						yield ('SKIP',None)
+						yield ('CONFIDENT_SKIP',None)
 						continue
+					else:
+						status = 'CONFIDENT_IMPORT'
 						timestamps.setdefault(timestamp,[]).append((artist,title))
 
 				# if it's 0, we use ts instead, but identify duplicates differently
@@ -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...")
+							warn(f"{entry} might be a duplicate, skipping...")
-							yield ('SKIP',None)
+							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,