Major stats and reporting enhancements for Spotify data

.gitignore

@@ -132,3 +132,8 @@ dmypy.json
 *.zip
 MyData/
 /MyData/
+Spotify Extended Streaming History/
+*.pdf
+*.csv
+stats_report.html
+stats_report.md

filemgr/data.py

@@ -1,3 +1,4 @@
+import json
 from dataclasses import dataclass
 from datetime import timedelta
 from os import listdir
@@ -51,17 +52,36 @@ def my_library(self):
 
     # data loaders (from file)
     def _load_streaming_history(self):
-        files = list(filter(lambda x: x.startswith('StreamingHistory'), listdir(self._root)))
+        files = list(filter(lambda x: x.startswith('Streaming_History'), listdir(self._root)))
         files.sort()
 
         all_ = []
         for file in files:
             with open(self._root + file, encoding='UTF-8') as f:
-                all_ += eval(f.read())
+                data = json.load(f)
+                for item in data:
+                    # Filter out podcasts and audiobooks
+                    if (item.get('master_metadata_track_name') 
+                        and not item.get('episode_name') 
+                        and not item.get('audiobook_title')):
+
+                        all_.append({
+                            'endTime': item['ts'],
+                            'artistName': item['master_metadata_album_artist_name'],
+                            'albumName': item['master_metadata_album_album_name'],
+                            'trackName': item['master_metadata_track_name'],
+                            'msPlayed': item['ms_played'],
+                            'platform': item['platform'],
+                            'connCountry': item['conn_country'],
+                            'reasonStart': item['reason_start'],
+                            'reasonEnd': item['reason_end'],
+                            'shuffle': item['shuffle'],
+                            'skipped': item['skipped']
+                        })
 
         # Convert string timestamps to time objects
         for h in all_:
-            h['endTime'] = strptime(h['endTime'], '%Y-%m-%d %H:%M')
+            h['endTime'] = strptime(h['endTime'], '%Y-%m-%dT%H:%M:%SZ')
             h['msPlayed'] = timedelta(milliseconds=h['msPlayed'])
 
         return all_
@@ -84,7 +104,9 @@ def _load_playlists(self):
         return all_
 
     def _load_my_library(self):
-        with open(self._root + 'YourLibrary.json', encoding='UTF-8') as f:
-            all_ = eval(f.read())
-
-        return all_['tracks'], all_['albums'], all_['shows'], all_['episodes'], all_['artists']
+        try:
+            with open(self._root + 'YourLibrary.json', encoding='UTF-8') as f:
+                all_ = eval(f.read())
+            return all_['tracks'], all_['albums'], all_['shows'], all_['episodes'], all_['artists']
+        except FileNotFoundError:
+            return [], [], [], [], []

filemgr/functions.py

@@ -5,7 +5,7 @@
 
 
 def _extract_data(path: str) -> None:
-    if not os.path.exists('MyData'):
+    if not os.path.exists('Spotify Extended Streaming History'):
         with ZipFile(path) as zip_:
             # todo: check extracted location & modifiable root path
             zip_.extractall()
@@ -18,6 +18,4 @@ def load_zipped_data(path: str = 'my_spotify_data.zip') -> MyData:
     :param path: relative path to zip file ('my_spotify_data.zip' default)
     """
     _extract_data(path)
-
-    # todo: modifiable root path
-    return MyData()
+    return MyData(root_path='Spotify Extended Streaming History/')

filemgr/types.py

@@ -28,5 +28,12 @@ class PlayList(TypedDict):
 class History(TypedDict):
     endTime: struct_time  # str  # yyyy-mm-dd HH:MM
     artistName: str
+    albumName: str
     trackName: str
     msPlayed: timedelta  # int
+    platform: str
+    connCountry: str
+    reasonStart: str
+    reasonEnd: str
+    shuffle: bool
+    skipped: bool