tmp

.gitignore

@@ -1,2 +1,4 @@
-prefetched
+prefetched*
 __pycache__
+.idea
+secret

common.py

@@ -1,18 +1,10 @@
-import requests
 import json
 import os
 
-# Directory for storing JSON files
 data_dir = 'prefetched'
-token = ''
 
-with open('token', 'r') as file:
-    token = file.readline().strip()
 
-print(f'Using auth token : {token}')
-
-def getData(name):
-    """Opens a JSON file in the 'prefetched' folder and returns its content."""
+def get_data(name):
     file_path = os.path.join(data_dir, f"{name}.json")
     try:
         with open(file_path, 'r') as file:
@@ -23,26 +15,11 @@ def getData(name):
     except json.JSONDecodeError:
         return f"Error decoding JSON from {name}.json."
 
-def saveData(data, name):
-    """Saves JSON data in the 'prefetched' folder with the given name."""
+
+def save_data(data, name):
+    os.makedirs(data_dir, exist_ok=True)
+
     file_path = os.path.join(data_dir, f"{name}.json")
     with open(file_path, 'w') as file:
         json.dump(data, file, indent=4)
     return f"Data saved to {name}.json."
-
-def fetch(endpoint, method='GET', body=None):
-    url = f'https://api.spotify.com/{endpoint}'
-    headers = {
-        'Authorization': f'Bearer {token}',
-    }
-    response = requests.request(method, url, headers=headers, json=body)
-    response.raise_for_status()
-    return response.json()
-
-def main():
-    data = fetch('v1/artists/4Z8W4fKeB5YxbusRsdQVPb')
-    data2 = fetch('v1/me/playlists?limit=1')['total']
-    print(data2)
-
-if __name__ == "__main__":
-    main()

loadArtwork.py

@@ -0,0 +1,45 @@
+import requests
+import os
+from PIL import Image
+from io import BytesIO
+
+from common import *
+
+
+def save_image_from_url(url, name, image_dir="playlist_covers"):
+    response = requests.get(url)
+
+    if response.status_code != 200:
+        raise "Cannot fetch the playlist cover"
+
+    image = Image.open(BytesIO(response.content))
+
+    directory = os.path.join(data_dir, image_dir)
+    os.makedirs(directory, exist_ok=True)
+    file_path = os.path.join(directory, f"{name}.jpg")
+    print(f"Image saved {file_path}")
+    image.save(file_path)
+
+
+def load_playlist_covers():
+    playlists = get_data('playlists')
+
+    for pl in playlists:
+        if len(pl['images']):
+            url = pl['images'][0]['url']
+            save_image_from_url(url, pl['name'])
+
+
+def load_user_cover():
+    user_data = get_data("user_data")
+    url = user_data['images'][1]['url']
+    save_image_from_url(url, "user", ".")
+
+
+def load():
+    load_user_cover()
+    load_playlist_covers()
+
+
+if __name__ == "__main__":
+    load()

main.py

@@ -0,0 +1,106 @@
+from common import *
+from spotifyFetch import fetch_data
+import cmd
+
+playlists = get_data('playlistTracks')
+top_artists = get_data('top_artists')
+top_tracks = get_data('top_tracks')
+user_tracks = get_data('tracks')
+
+
+def get_playlist_names():
+    return [name for name, items in playlists.items()]
+
+
+def get_artists_str(artists_pack):
+    track_artists = [artist['name'] for artist in artists_pack]
+    artists = " ".join(track_artists)
+    return artists
+
+
+def print_playlist_tracks(name):
+    print(name)
+    for track in playlists[name]:
+        track_data = track['track']
+        track_name = track_data['name']
+        artists = get_artists_str(track_data['artists'])
+        print(f"   '{track_name}' - '{artists}' ")
+
+
+def print_top_artists():
+    for artist in top_artists:
+        name = artist['name']
+        print(f"   '{name}'")
+
+
+def print_top_tracks():
+    for track in top_tracks:
+        track_name = track['name']
+        artists = get_artists_str(track['artists'])
+        print(f"   '{track_name}' - '{artists}' ")
+
+
+def print_stats():
+    print(f" tracks - {len(user_tracks)}")
+    print(f" playlists - {len(playlists)}")
+
+    for name, tracks in playlists.items():
+        print(f"    '{name}' - {len(tracks)}")
+
+    print(f" top tracks - {len(top_tracks)}")
+    print(f" top artists - {len(top_artists)}")
+
+
+def print_tracks():
+    for trackItem in user_tracks:
+        track = trackItem['track']
+        print(f"    '{track['name']}' - {get_artists_str(track['artists'])}")
+
+
+class Interpreter(cmd.Cmd):
+    intro = "Welcome to the interactive command loop. Type help or ? to list commands.\n"
+    prompt = "(spotify) "
+
+    def do_playlists(self, arg):
+        """prints user playlists"""
+        print("\n".join(get_playlist_names()))
+
+    def do_playlist_tracks(self, arg):
+        """prints playlist [name]"""
+        try:
+            name = arg.split()[0]
+            print_playlist_tracks(name)
+        except IndexError:
+            print("Invalid input")
+
+    def do_top_artists(self, arg):
+        """Top artists"""
+        print_top_artists()
+
+    def do_top_tracks(self, arg):
+        """Top tracks"""
+        print_top_tracks()
+
+    def do_stat(self, arg):
+        """Print stats"""
+        print_stats()
+
+    def do_tracks(self, arg):
+        """Print tracks"""
+        print_tracks()
+
+    def do_fetch(self, arg):
+        """Fetch all spotify data"""
+        try:
+            fetch_data()
+        except ValueError:
+            print("Can not fetch the data.")
+
+    def do_exit(self, arg):
+        """Exit the command loop: exit"""
+        print("Goodbye!")
+        return True
+
+
+if __name__ == '__main__':
+    Interpreter().cmdloop()

prefetched_1/user_data.json

@@ -0,0 +1,26 @@
+{
+    "display_name": "ilusha",
+    "external_urls": {
+        "spotify": "https://open.spotify.com/user/31uszk7zb6zdzrzst4q3rvkfrfz4"
+    },
+    "href": "https://api.spotify.com/v1/users/31uszk7zb6zdzrzst4q3rvkfrfz4",
+    "id": "31uszk7zb6zdzrzst4q3rvkfrfz4",
+    "images": [
+        {
+            "url": "https://i.scdn.co/image/ab67757000003b82b4a21a7f0c35258b3fa4b79d",
+            "height": 64,
+            "width": 64
+        },
+        {
+            "url": "https://i.scdn.co/image/ab6775700000ee85b4a21a7f0c35258b3fa4b79d",
+            "height": 300,
+            "width": 300
+        }
+    ],
+    "type": "user",
+    "uri": "spotify:user:31uszk7zb6zdzrzst4q3rvkfrfz4",
+    "followers": {
+        "href": null,
+        "total": 7
+    }
+}

prefetched_1/user_profile.json

@@ -0,0 +1,33 @@
+{
+    "display_name": "ilusha",
+    "external_urls": {
+        "spotify": "https://open.spotify.com/user/31uszk7zb6zdzrzst4q3rvkfrfz4"
+    },
+    "href": "https://api.spotify.com/v1/users/31uszk7zb6zdzrzst4q3rvkfrfz4",
+    "id": "31uszk7zb6zdzrzst4q3rvkfrfz4",
+    "images": [
+        {
+            "url": "https://i.scdn.co/image/ab67757000003b82b4a21a7f0c35258b3fa4b79d",
+            "height": 64,
+            "width": 64
+        },
+        {
+            "url": "https://i.scdn.co/image/ab6775700000ee85b4a21a7f0c35258b3fa4b79d",
+            "height": 300,
+            "width": 300
+        }
+    ],
+    "type": "user",
+    "uri": "spotify:user:31uszk7zb6zdzrzst4q3rvkfrfz4",
+    "followers": {
+        "href": null,
+        "total": 7
+    },
+    "country": "US",
+    "product": "free",
+    "explicit_content": {
+        "filter_enabled": false,
+        "filter_locked": false
+    },
+    "email": "ilusha.basic@gmail.com"
+}

spotifyAuth.py

@@ -1,88 +1,31 @@
-import requests
 import base64
-import re
+
+import requests
 import webbrowser
 from http.server import BaseHTTPRequestHandler, HTTPServer
 import urllib.parse
 import threading
 import time
-
-CODE = ''
-TOKEN = ''
-
-# Define the HTTP request handler class
-class RequestHandler(BaseHTTPRequestHandler):
-    
-    # Handle GET requests
-    def do_GET(self):
-        global CODE
-
-        # Parse the URL and query parameters
-        parsed_url = urllib.parse.urlparse(self.path)
-        query_params = urllib.parse.parse_qs(parsed_url.query)
-        
-        # Extract the authorization code from query parameters
-        if 'code' in query_params:
-            authorization_code = query_params['code'][0]
-            print(f"Authorization Code: {authorization_code}")
-            
-            # You can now proceed with exchanging this authorization code for an access token
-            # Example: Call a function to exchange code for access token
-            #with open('code', 'w') as file:
-            #    file.write(authorization_code)
-
-            # code = authorization_code
-            # print("Shutting down the callback server")
-            
-            # Send response back to the browser
-            self.send_response(200)
-            self.send_header('Content-type', 'text/html')
-            self.end_headers()
-            self.wfile.write(b'<html><body>Authorization code received. You can close this window.</body></html>')
-
-            print("\n\nCode recieved!!!\n")
-            CODE = authorization_code
-            # self.server.shutdown()
-
-        else:
-            # Handle cases where code is not found
-            self.send_response(400)
-            self.send_header('Content-type', 'text/html')
-            self.end_headers()
-            self.wfile.write(b'<html><body>Authorization code not found.</body></html>')
-
-
-# Set up and start the HTTP server
-def startLoginCallbackServer():
-    server_address = ('localhost', 8888)  # Use any free port you prefer
-    httpd = HTTPServer(server_address, RequestHandler)
-    print('Starting HTTP server on port 8080...')
-    httpd.serve_forever()
-
+import os.path
 
 client_id = '***REDACTED***'
-client_secret = '***REDACTED***'
+client_secret = None
+token = None
 
 
-def openLoginPage():
-    authorize_url = 'https://accounts.spotify.com/authorize'
-
-    params = {
-        'client_id': client_id,
-        'response_type': 'code',
-        'redirect_uri': 'http://localhost:8888',
-        'scope': 'user-read-private user-read-email',  # Add scopes based on your needs
-    }
-
-    auth_url = f"{authorize_url}?client_id={params['client_id']}&response_type={params['response_type']}&redirect_uri={params['redirect_uri']}&scope={params['scope']}"
-    webbrowser.open(auth_url)
+def resolve_client_secret():
+    global client_secret
+    if os.path.exists('secret'):
+        with open('secret', 'r') as file:
+            client_secret = file.readline().strip()
+    else:
+        client_secret = input("Enter the application secret: ")
+        with open('secret', 'w') as file:
+            file.write(client_secret)
 
 
-def authenticateWepAPIAfterLogin(code):
-    global TOKEN
-
-    client_creds = f"{client_id}:{client_secret}"
-    client_creds_b64 = base64.b64encode(client_creds.encode())
+def retrieve_web_api_token(authorization_code):
+    global token
 
     token_url = 'https://accounts.spotify.com/api/token'
 
@@ -90,56 +33,90 @@ def authenticateWepAPIAfterLogin(code):
         "grant_type": "authorization_code",
         "client_secret": client_secret,
         "client_id":     client_id,
-        "code": code,
+        "code": authorization_code,
         'redirect_uri': 'http://localhost:8888',
-        "scope": "user-read-private, playlist-read-private, playlist-read-collaborative",
-    }
-
-    token_headers = {
-#    "Authorization": f"Basic {client_creds_b64.decode()}"
-        "Content-Type" : "application/x-www-form-urlencoded"
     }
 
     print("\nAuthenticating WEB API\n")
 
-    token_response = requests.post(token_url, data=token_data)
+    auth_string = f'{client_id}:{client_secret}'
+    b64_auth_string = base64.b64encode(auth_string.encode()).decode()
+
+    # Define the headers
+    headers = {
+        'Content-Type': 'application/x-www-form-urlencoded',
+        'Authorization': f'Basic {b64_auth_string}'
+    }
+
+    token_response = requests.post(token_url, data=token_data, headers=headers)
 
     if token_response.status_code not in range(200, 299):
         print(f"Failed to retrieve access token: {token_response.status_code}")
         print(token_response.json())
+        raise ValueError
 
     token = token_response.json()['access_token']
-    token_type = token_response.json()['token_type']
-    token_time = token_response.json()['expires_in']
-
-    print(token_response.json())
-
-    TOKEN = token
-
-    print(f"\nGot the token! - {TOKEN} \n")
 
 
+class RequestHandler(BaseHTTPRequestHandler):
+    def do_GET(self):
 
-def main():
-    thread1 = threading.Thread(target=startLoginCallbackServer)
-    thread1.start()
+        parsed_url = urllib.parse.urlparse(self.path)
+        query_params = urllib.parse.parse_qs(parsed_url.query)
         
-    time.sleep(3)
+        if 'code' in query_params:
+            authorization_code = query_params['code'][0]
 
-    openLoginPage()
+            self.send_response(200)
+            self.send_header('Content-type', 'text/html')
+            self.end_headers()
+            self.wfile.write(b'<html><body>Authorization code received. You can close this window.</body></html>')
 
-    time.sleep(3)
+            retrieve_web_api_token(authorization_code)
 
-    print("Got the code ", CODE)
+            threading.Thread(target=self.server.shutdown).start()
 
-    authenticateWepAPIAfterLogin(CODE)
+        else:
+            self.send_response(400)
+            self.send_header('Content-type', 'text/html')
+            self.end_headers()
+            self.wfile.write(b'<html><body>Authorization code not found.</body></html>')
 
-    with open('token', 'w') as file:
-        file.write(TOKEN)
 
-    thread1.stop()
+def run_redirect_server():
+    httpd = HTTPServer(('localhost', 8888), RequestHandler)
+    print('Starting HTTP redirection server')
+    httpd.serve_forever()
+
+
+def open_authenticate_page():
+    authorize_url = 'https://accounts.spotify.com/authorize'
+
+    params = {
+        'client_id': client_id,
+        'response_type': 'code',
+        'redirect_uri': 'http://localhost:8888',
+        "scope": "user-top-read playlist-read-collaborative user-read-email user-library-read user-read-private playlist-read-private",
+    }
+
+    auth_url = f"{authorize_url}?client_id={params['client_id']}&response_type={params['response_type']}&redirect_uri={params['redirect_uri']}&scope={params['scope']}"
+    webbrowser.open(auth_url)
+
+
+def authenticate():
+    resolve_client_secret()
+
+    redirect_server = threading.Thread(target=run_redirect_server)
+    redirect_server.start()
+
+    time.sleep(1)
+    open_authenticate_page()
+
+    redirect_server.join()
+
+    return token
 
 
 if __name__ == "__main__":
-    main()
+    authenticate()
 

spotifyFetch.py

@@ -1,49 +1,147 @@
-import common as cm
-import os
+import requests
+from loadArtwork import load
+from spotifyAuth import authenticate
+from common import *
 
-os.makedirs(cm.data_dir, exist_ok=True)
+FETCH_STEP = 50
+token = ''
 
-fetch_step = 50
 
-playlists = []
-playlist_items = {}
+def fetch(endpoint, method='GET', body=None):
+    url = f'https://api.spotify.com/{endpoint}'
+    headers = {
+        'Authorization': f'Bearer {token}',
+    }
 
-def loadPLaylists(pl):
+    response = requests.request(method, url, headers=headers, json=body)
+    response.raise_for_status()
+    return response.json()
+
+
+def fetch_playlists(user_id):
+    print("Fetching playlists")
+    endpoint = f"v1/me/playlists"
+    pl_total = fetch(f'{endpoint}?limit=1')['total']
     pl_count = 0
-    pl_total = 0
-
-    pl_total = cm.fetch('v1/me/playlists?limit=1')['total']
+    playlists = []
 
     while pl_count < pl_total:
-        res = cm.fetch(f'v1/me/playlists?limit={fetch_step}&offset={pl_count}')
-        pl += res['items']
-        pl_count += fetch_step
+        res = fetch(f'{endpoint}?limit={FETCH_STEP}&offset={pl_count}')
+        playlists += res['items']
+        pl_count += FETCH_STEP
+
+    return playlists
 
 
-    cm.saveData(playlists, 'playlists')
-
-
-def loadPlaylistItems(playlistId, traks):
-    items = cm.fetch(f'v1/playlists/{playlistId}/tracks?fields=total')['total']
+def fetch_playlist_items(playlist_id):
+    endpoint = f"v1/playlists/{playlist_id}/tracks"
+    items = fetch(f'{endpoint}?fields=total')['total']
     loaded = 0
+    tracks = []
 
     while loaded < items:
-        res = cm.fetch(f'v1/playlists/{playlistId}/tracks?limit={fetch_step}&offset={items}')
-        traks += res['items']
-        loaded += fetch_step
+        res = fetch(f'{endpoint}?limit={FETCH_STEP}&offset={loaded}')
         
-    print(items)
+        tracks += res['items']
+        loaded += FETCH_STEP
 
-loadPLaylists(playlists)
-print(len(playlists))
+    return tracks
 
 
-for pl in playlists:
-    id = pl['id']
-    name = pl['name']
-    playlist_items[name] = []
+def fetch_tracks(user_id):
+    print("Fetching tracks")
+    endpoint = f"v1/me/tracks"
+    total = fetch(f"{endpoint}?limit=1&market=ES")['total']
+    current = 0
+    tracks = []
 
-    loadPlaylistItems(id, playlist_items[name])
+    while total > current:
+        res = fetch(f"{endpoint}?limit={FETCH_STEP}&offset={current}&market=ES")
+        tracks += res['items']
+        current += FETCH_STEP
 
-cm.saveData(playlist_items, "playlistTraks")
+    save_data(tracks, "tracks")
+    return tracks
 
+
+def fetch_tracks_top():
+    print("Fetching top tracks")
+    total = fetch(f'v1/me/top/tracks?fields=total')['total']
+    current = 0
+    tracks = []
+
+    while current < total:
+        res = fetch(f'v1/me/top/tracks?limit={FETCH_STEP}&offset={current}&time_range=long_term')
+        tracks += res['items']
+        current += FETCH_STEP
+
+    save_data(tracks, "top_tracks")
+    return tracks
+
+
+def fetch_artists_top():
+    print("Fetching top artists")
+
+    total = fetch(f'v1/me/top/artists?fields=total')['total']
+    current = 0
+    artists = []
+
+    while current < total:
+        res = fetch(f'v1/me/top/artists?limit={FETCH_STEP}&offset={current}&time_range=long_term')
+        artists += res['items']
+        current += FETCH_STEP
+
+    save_data(artists, "top_artists")
+    return artists
+
+
+def fetch_all_playlist_data(user_id):
+    playlists = fetch_playlists(user_id)
+
+    save_data(playlists, 'playlists')
+
+    print(len(playlists))
+
+    playlist_items = {}
+
+    print("Fetching playlists items")
+
+    index = 0
+    for pl in playlists:
+        pl_id = pl['id']
+        name = pl['name']
+        playlist_items[name] = fetch_playlist_items(pl_id)
+        print(f"{index} - {len(playlist_items[name])}")
+        index += 1
+
+    save_data(playlist_items, "playlistTracks")
+
+
+def get_user_data():
+    print("Fetching user data")
+
+    user_profile = fetch(f'v1/me')
+    user_id = user_profile['id']
+    user_data = fetch(f'v1/users/{user_id}')
+
+    save_data(user_profile, "user_profile")
+    save_data(user_data, "user_data")
+    return user_id
+
+
+def fetch_data():
+    global token
+    token = authenticate()
+
+    user_id = get_user_data()
+
+    fetch_tracks(user_id)
+    fetch_all_playlist_data(user_id)
+    fetch_tracks_top()
+    fetch_artists_top()
+
+    load()
+
+
+if __name__ == "__main__":
+    fetch_data()

stat.py

@@ -1,6 +0,0 @@
-import common as cm
-
-playlists = cm.getData('playlists')
-
-for pl in playlists:
-    print(pl['name'])

token

@@ -1 +0,0 @@
-BQA-OcaRz7YObEOVptzs_uE9hg8o3wbZozKn1JBgduwxIyq_WebpieN5b0bsea7ocHCE2-aTJ9bQxR4Wg9eq9FuiwTog25QIMR52yIzV2FZ66H_pjqoDRn3vtObnhTpkc4LWauwoWE-8WFZVVWbXwOhJs4TH9242cnZ-hJcS_v_LepDJHHZnKn84jb3inxRTFP0lKbYU98nTC4rCJvGaYQ