Initial commit

.gitignore

@@ -0,0 +1,2 @@
+last_fm_credentials.json
+__pycache__/

last_fm.py

@@ -0,0 +1,29 @@
+import json
+from typing import List
+
+class LastFm:
+    def __init__(self, credentials_file):
+        with open(credentials_file, 'r') as f:
+            credentials = json.loads(f.read())
+
+            self.lastfm = pylast.LastFMNetwork(
+                api_key=credentials['api_kay'],
+                api_secret=credentials['api_secret']
+            )
+
+    def get_recent_plays(self, username):
+        self.lastfm.get_user(username).get_recent_tracks(limit=200)
+
+    def get_top_artists(self, username: str, min_scrobbles: int) -> List[pylast.TopItem]:
+        artists = []
+        page = 0
+        
+        while True:
+            artists_page = self.lastfm.get_user(username).get_top_artists()
+
+            for artist in artists_page:
+                if (artist.weight > min_scrobbles):
+                    artists.append(artist)
+                else:
+                    break
+            

main.py

@@ -0,0 +1,57 @@
+from sys import argv
+
+usage = """
+Usage:
+
+add_lastfm_user
+    recents
+    discography of artists over x plays
+    recommendations?
+    
+add_playlist <playlist_id>
+    spotify
+    tidal
+    lastfm?
+    last fm likes?
+    spotify likes
+    tidal likes?
+
+update existing data from mb
+auto import/data gather of downlads from bandcamp
+"""
+
+def sync():
+    pass
+
+def add_lastfm_user(username):
+    pass
+
+def add_playlist(playlist_id):
+    pass
+
+if __name__ == "__main__":
+    if len(argv) == 1:
+        print(usage)
+        exit()
+        
+    if argv[1] == "sync":
+        sync()
+
+    elif argv[1] == "add_lastfm_user":
+        if len(argv) < 3:
+            print("Please provide a lastfm username")
+            exit()
+            
+        username = argv[2]
+        add_lastfm_user(username)
+
+    elif argv[1] == "add_playlist":
+        if len(argv) < 3:
+            print("Please provide a playlist id")
+            exit()
+
+        playlist_id = argv[2]
+        add_playlist(playlist_id)
+
+    else:
+        print(f"Invalid argument '{argv[1]}'")

musicbrainz.py

@@ -0,0 +1,40 @@
+from datetime import datetime, timedelta, UTC
+from typing import Dict, List
+from requests import get
+from musicbrainz_entities import MbRecording, MbArtist
+from time import sleep         
+
+class MusicBrainz:
+    def __init__(self, application_name: str, version: str, contact: str, root_url: str = "https://musicbrainz.org/ws/2/"):
+        self._user_agent_string = f"{application_name}/{version} ( {contact} )"
+        self._last_request_time = datetime.now(UTC) + timedelta(seconds=-1)
+        self._root_url = root_url
+
+    def _request(self, method, endpoint:str, params: Dict[str, object] = dict()):
+        url = f"{self._root_url.rstrip('/')}/{endpoint.lstrip('/')}"
+        params['fmt'] = 'json'
+        # TODO: retries
+
+        if datetime.now(UTC) - self._last_request_time < timedelta(seconds=1):
+            sleep(1 - (datetime.now(UTC) - self._last_request_time).total_seconds())
+
+        response = method(url, params=params)
+        self._last_request_time = datetime.now(UTC)
+        return response
+
+    def isrc_lookup(self, isrc: str) -> List[MbRecording]:
+        result = self._request(get, f"/isrc/{isrc}", { 'inc': 'artists+isrcs+releases+url-rels' }) 
+        return [MbRecording(recording) for recording in result.json()['recordings']]
+
+    def get_artist_by_id(self, id: str) -> MbArtist:
+        result = self._request(get, f'/artist/{id}', { 'inc': 'genres' })
+        return MbArtist(result.json())
+
+
+if __name__ == "__main__":
+    mb = MusicBrainz("pob_tag_test", "0.1", "musicbrainz@pobnellion.com")
+
+    res = mb.isrc_lookup('JPPC09428330')
+    print(res[0].title)
+    artist = mb.get_artist_by_id(res[0].artist_credit[0].id)
+    print(artist.name)

musicbrainz_entities.py

@@ -0,0 +1,34 @@
+from typing import List
+from datetime import datetime, timedelta
+
+
+class MbArtistCredit:
+    def __init__(self, mb_json):
+        self.id: str = mb_json['artist']['id']
+        self.name: str = mb_json['artist']['name']
+        self.type: str = mb_json['artist']['type']
+        self.country: str = mb_json['artist']['country']
+        self.disambiguation: str = mb_json['artist']['disambiguation']
+        self.join_phrase: str = mb_json['joinphrase']
+
+
+class MbRecording:
+    def __init__(self, mb_json):
+        self.id: str = mb_json['id']
+        self.title: str = mb_json['title']
+        self.is_video: bool = mb_json['video']
+        self.disambiguation: str = mb_json['disambiguation']
+        self.isrcs: List[str] = mb_json['isrcs']
+        self.first_release_date: datetime = datetime.strptime(mb_json['first-release-date'], '%Y-%m-%d')
+        self.length: timedelta = timedelta(seconds = int(mb_json['length']))
+        self.artist_credit: List[MbArtistCredit] = [MbArtistCredit(artist_credit) for artist_credit in mb_json['artist-credit']]
+
+       
+class MbArtist:
+    def __init__(self, mb_json):
+        self.id: str = mb_json['id']
+        self.name: str = mb_json['name']
+        self.disambiguation: str = mb_json['disambiguation']
+        self.country: str = mb_json['country']
+        # TODO: more
+ 

track_cache_db.py

@@ -0,0 +1,5 @@
+import sqlite3
+
+class TrackCacheDb:
+    def __init__(self):
+        pass