refactor initial not fully recovered

src/spotify/ParseSpotify.py

@@ -0,0 +1,139 @@
+from datetime import datetime
+from typing import Any, Iterable
+from pathlib import Path
+import json
+
+from src.Library import *
+
+
+def get_artists_str(artists_pack):
+    track_artists = [artist['name'] for artist in artists_pack]
+    artists = " ".join(track_artists)
+    return artists
+
+class Parser:
+    def __init__(self):
+        # state: Spotify track ID -> Track object
+        self.track_map: dict[str, Track] = {}
+
+    # -------------------------
+    # Low-level helpers
+    # -------------------------
+    @staticmethod
+    def _parse_spotify_datetime(value: str | None) -> datetime | None:
+        if not value:
+            return None
+        return datetime.fromisoformat(value.replace("Z", "+00:00"))
+
+    @staticmethod
+    def _parse_spotify_artist(a: dict[str, Any]) -> SpotifyArtist:
+        return SpotifyArtist(
+            id=a["id"],
+            name=a["name"],
+            uri=a["uri"],
+            url=a["external_urls"]["spotify"],
+        )
+
+    @staticmethod
+    def _parse_spotify_album(a: dict[str, Any]) -> SpotifyAlbum:
+        return SpotifyAlbum(
+            id=a["id"],
+            name=a["name"],
+            release_date=a.get("release_date"),
+            total_tracks=a.get("total_tracks", 0),
+            uri=a["uri"],
+            url=a["external_urls"]["spotify"],
+        )
+
+    def _parse_spotify_data(self, track_json: dict[str, Any]) -> SpotifyData:
+        return SpotifyData(
+            id=track_json["id"],
+            uri=track_json["uri"],
+            url=track_json["external_urls"]["spotify"],
+            href=track_json["href"],
+            popularity=track_json.get("popularity"),
+            is_playable=track_json.get("is_playable"),
+            is_local=track_json.get("is_local"),
+            external_ids=track_json.get("external_ids", {}),
+            artists=[self._parse_spotify_artist(a) for a in track_json.get("artists", [])],
+            album=self._parse_spotify_album(track_json["album"])
+            if track_json.get("album")
+            else None,
+        )
+
+    # -------------------------
+    # Track parsers
+    # -------------------------
+    def parse_track_item(self, item: dict[str, Any]) -> Track:
+        """
+        Parses a single track item and returns Track object.
+        Reuses existing Track from track_map if already parsed.
+        """
+        t = item["track"]
+        track_id = t["id"]
+
+        # return existing track if already parsed
+        if track_id in self.track_map:
+            return self.track_map[track_id]
+
+        track = Track(
+            title=t["name"],
+            artists=[a["name"] for a in t.get("artists", [])],
+            album=t["album"]["name"] if t.get("album") else None,
+            duration_ms=t.get("duration_ms"),
+            added_at=self._parse_spotify_datetime(item.get("added_at")),
+            spotify=self._parse_spotify_data(t),
+            id=self._parse_spotify_data(t).id
+        )
+
+        self.track_map[track_id] = track
+        return track
+
+    def parse_track_items(self, items: Iterable[dict[str, Any]]) -> list[Track]:
+        return [self.parse_track_item(i) for i in items if i.get("track")]
+
+    # -------------------------
+    # Playlist parsers
+    # -------------------------
+    def parse_playlist_content(self, playlist_name: str, items: list[dict[str, Any]]) -> Playlist:
+        tracks = self.parse_track_items(items)
+        return Playlist(name=playlist_name, tracks=tracks)
+
+    def parse_all_playlist_content(self, data: dict[str, list[dict[str, Any]]]) -> list[Playlist]:
+        playlists: list[Playlist] = []
+        for name, items in data.items():
+            playlists.append(self.parse_playlist_content(name, items))
+        return playlists
+
+    def parse_all_tracks(self, items: dict[str, list[dict[str, Any]]]) -> list[Track]:
+        return [self.parse_track_item(i) for i in items]
+
+    # -------------------------
+    # Library builder
+    # -------------------------
+    def build_library_from_playlist_content(
+        self, playlist_content_json: dict[str, list[dict[str, Any]]],
+            tracks_content_json: dict[str, list[dict[str, Any]]]
+    ) -> Library:
+        library = Library()
+
+        self.parse_all_tracks(tracks_content_json)
+
+        library.tracklists.append(TrackList(tracks=list((self.track_map.values()))))
+
+        library.playlists = self.parse_all_playlist_content(playlist_content_json)
+        library.tracks.extend(self.track_map.values())
+
+        return library
+
+    # -------------------------
+    # Convenience method
+    # -------------------------
+    def parse(
+        self,
+        playlist_content_file: Path,
+        tracks_file: Path,
+    ) -> Library:
+        playlist_content = json.loads(playlist_content_file.read_text())
+        tracks_content = json.loads(tracks_file.read_text())
+        return self.build_library_from_playlist_content(playlist_content, tracks_content)

src/spotify/SpotifyAuthenticator.py

@@ -0,0 +1,116 @@
+import base64
+
+import requests
+import webbrowser
+from http.server import BaseHTTPRequestHandler, HTTPServer
+import urllib.parse
+import threading
+import time
+import os.path
+
+client_id = None
+client_secret = None
+token = None
+
+
+def retrieve_web_api_token(authorization_code):
+    global token
+
+    token_url = 'https://accounts.spotify.com/api/token'
+
+    token_data = {
+        "grant_type": "authorization_code",
+        "client_secret": client_secret,
+        "client_id":     client_id,
+        "code": authorization_code,
+        'redirect_uri': 'http://127.0.0.1:8888',
+    }
+
+    print("\nAuthenticating WEB API\n")
+
+    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']
+
+
+class RequestHandler(BaseHTTPRequestHandler):
+    def do_GET(self):
+
+        parsed_url = urllib.parse.urlparse(self.path)
+        query_params = urllib.parse.parse_qs(parsed_url.query)
+        
+        if 'code' in query_params:
+            authorization_code = query_params['code'][0]
+
+            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>')
+
+            retrieve_web_api_token(authorization_code)
+
+            threading.Thread(target=self.server.shutdown).start()
+
+        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>')
+
+
+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://127.0.0.1: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(id, secret):
+
+    global client_secret
+    global client_id
+
+    client_secret=secret
+    client_id=id
+
+    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__":
+    authenticate()
+

src/spotify/SpotifyWebAPI.py

@@ -0,0 +1,196 @@
+import requests
+from PIL import Image
+from io import BytesIO
+
+from src.spotify.SpotifyAuthenticator import authenticate
+
+from src.helpers import *
+
+FETCH_STEP = 50
+token = ''
+
+
+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 find_song(song_pattern):
+    return fetch(f"v1/search/?type=track&track=1&q={song_pattern}")['tracks']['items']
+
+
+def fetch_playlists(user_id):
+    print("Fetching playlists")
+    endpoint = f"v1/me/playlists"
+    pl_total = fetch(f'{endpoint}?limit=1')['total']
+    pl_count = 0
+    playlists = []
+
+    while pl_count < pl_total:
+        res = fetch(f'{endpoint}?limit={FETCH_STEP}&offset={pl_count}')
+        playlists += res['items']
+        pl_count += FETCH_STEP
+
+    return playlists
+
+
+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 = fetch(f'{endpoint}?limit={FETCH_STEP}&offset={loaded}')
+        
+        tracks += res['items']
+        loaded += FETCH_STEP
+
+    return tracks
+
+
+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 = []
+
+    while total > current:
+        res = fetch(f"{endpoint}?limit={FETCH_STEP}&offset={current}&market=ES")
+        tracks += res['items']
+        current += FETCH_STEP
+
+    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 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(get_workdir(), 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_artworks():
+    load_user_cover()
+    load_playlist_covers()
+
+
+def update_access_token(id, secret):
+    global token
+    token = authenticate(id, secret)
+
+
+def fetch_data():
+    user_id = get_user_data()
+
+    fetch_tracks(user_id)
+    fetch_all_playlist_data(user_id)
+    fetch_tracks_top()
+    fetch_artists_top()
+
+    load_artworks()
+
+def fetch_all(id, secret, workdir):
+    set_workdir(workdir / "spotify")
+    update_access_token(id, secret)
+    fetch_data()
+
+if __name__ == "__main__":
+    update_access_token("", "")
+    fetch_data()