added timer methods, moved chart to top level, added artist timer script

spotfm/engine/chart_source.py

@@ -5,7 +5,7 @@ from spotframework.model.uri import Uri
 from spotframework.engine.playlistengine import TrackSource, SourceParameter
 from spotframework.engine.processor.abstract import AbstractProcessor
 
-from spotfm.charts.chart import get_chart_of_spotify_tracks
+from spotfm.chart import get_chart_of_spotify_tracks
 
 from typing import List
 import logging

spotfm/timer.py

@@ -0,0 +1,79 @@
+import logging
+from datetime import datetime
+
+from spotframework.net.network import Network as SpotNet
+from spotframework.model.uri import Uri
+from spotframework.model.track import TrackFull
+from fmframework.net.network import Network as FmNet
+from fmframework.model import Track
+from fmframework.net.scrape import LibraryScraper
+
+logger = logging.getLogger(__name__)
+
+def time_artist(spotnet: SpotNet, artist: str, username: str, fmnet: FmNet = None,
+                from_date: datetime = None, to_date: datetime = None, date_preset: str = None) -> int:
+    logger.info(f'timing {artist} for {username}')
+
+    fmtracks = LibraryScraper.get_scrobbled_tracks(username=username, artist=artist, whole_track=False,
+                                                   from_date=from_date, to_date=to_date, date_preset=date_preset)
+
+    return time_track_collection(tracks=fmtracks, spotnet=spotnet, username=username, fmnet=fmnet)
+
+def time_album(spotnet: SpotNet, artist: str, album: str, username: str, fmnet: FmNet = None,
+               from_date: datetime = None, to_date: datetime = None, date_preset: str = None) -> int:
+    logger.info(f'timing {album} / {artist} for {username}')
+
+    fmtracks = LibraryScraper.get_albums_tracks(username=username, artist=artist, album=album, whole_track=False,
+                                                from_date=from_date, to_date=to_date, date_preset=date_preset)
+
+    return time_track_collection(tracks=fmtracks, spotnet=spotnet, username=username, fmnet=fmnet)
+
+def time_track(spotnet: SpotNet, artist: str, track: str, username: str, fmnet: FmNet = None,
+               from_date: datetime = None, to_date: datetime = None, date_preset: str = None) -> int:
+    logger.info(f'timing {track} / {artist} for {username}')
+
+    fmtracks = LibraryScraper.get_track_scrobbles(username=username, artist=artist, track=track, whole_track=False,
+                                                  from_date=from_date, to_date=to_date, date_preset=date_preset)
+
+    return time_track_collection(tracks=fmtracks, spotnet=spotnet, username=username, fmnet=fmnet)
+
+def time_track_collection(tracks, spotnet: SpotNet, username: str, fmnet:FmNet = None):
+    track_pairs = []
+    if tracks is not None:
+        for track in tracks:
+            spottrack = spotnet.search(query_types=[Uri.ObjectType.track],
+                                       track=track.name,
+                                       artist=track.artist.name,
+                                       response_limit=1).tracks
+
+            if len(spottrack) == 1:
+                track_pairs.append((track, spottrack[0]))
+            else:
+                if fmnet is not None:
+                    logger.error(f'no track returned for search {track.name} / {track.artist.name} / {username}'
+                                 f', pulling last.fm track')
+
+                    fmtrack = fmnet.get_track(name=track.name, artist=track.artist.name, username=username)
+
+                    if fmtrack is not None and fmtrack.duration is not None and fmtrack.duration > 0:
+                        track_pairs.append((track, fmtrack))
+                    else:
+                        logger.error(f'no duration found on last.fm for {track.name} / {track.artist.name} / {username}')
+                else:
+                    logger.error(f'no track returned for search {track.name} / {track.artist.name} / {username}'
+                                 f', no fmnet to use as fallback')
+
+        total_ms = 0
+
+        for track_pair in track_pairs:
+            if isinstance(track_pair[1], TrackFull):
+                duration = track_pair[1].duration_ms
+            elif isinstance(track_pair[1], Track):
+                duration = track_pair[1].duration
+            else:
+                logger.critical(f'invalid track type found {type(track_pair[1])}')
+                duration = 0
+
+            total_ms += duration * track_pair[0].user_scrobbles
+
+        return total_ms

time_top_artists.py

@@ -0,0 +1,63 @@
+from spotframework.net.network import Network as Spotnet, NetworkUser
+from fmframework.net.network import Network as Fmnet
+from spotfm.timer import time_artist
+
+import logging
+import os
+
+spotframework_logger = logging.getLogger('spotframework')
+fmframework_logger = logging.getLogger('fmframework')
+spotfm_logger = logging.getLogger('spotfm')
+
+log_format = '%(levelname)s %(name)s:%(funcName)s - %(message)s'
+formatter = logging.Formatter(log_format)
+
+stream_handler = logging.StreamHandler()
+stream_handler.setFormatter(formatter)
+
+spotframework_logger.addHandler(stream_handler)
+fmframework_logger.addHandler(stream_handler)
+spotfm_logger.addHandler(stream_handler)
+
+spot_client = os.environ.get('SPOT_CLIENT')
+spot_secret = os.environ.get('SPOT_SECRET')
+spot_access = os.environ.get('SPOT_ACCESS_TOKEN')
+spot_refresh = os.environ.get('SPOT_REFRESH_TOKEN')
+fmclient = os.environ.get('FM_CLIENT')
+fmuser = os.environ.get('FM_USER')
+
+if spot_access is None and spot_refresh is None:
+    print('no spotify credentials')
+    exit(0)
+
+if fmclient is None:
+    print('no last.fm credentials')
+    exit(0)
+
+spotnet = Spotnet(NetworkUser(client_id=spot_client,
+                              client_secret=spot_secret,
+                              access_token=spot_access,
+                              refresh_token=spot_refresh)).refresh_access_token()
+
+while len(fmuser) == 0:
+    fmuser = input('last.fm username >> ')
+
+fmnet = Fmnet(username=fmuser, api_key=fmclient)
+
+def convert(seconds):
+    seconds = seconds % (24 * 3600)
+    hour = seconds // 3600
+    seconds %= 3600
+    minutes = seconds // 60
+    seconds %= 60
+
+    return "%d:%02d:%02d" % (hour, minutes, seconds)
+
+top_artists = fmnet.get_top_artists(period=Fmnet.Range.OVERALL, limit=10)
+
+artist_counts = dict()
+for artist in top_artists:
+    artist_counts[artist.name] = time_artist(spotnet=spotnet, fmnet=fmnet, artist=artist.name, username=fmnet.username)
+
+for name, count in artist_counts.items():
+    print(name, f'{count}ms,', f'{count/1000}s,', convert(count/1000))