First Code Commit

Author: SunnyySachdeva

README.md

@@ -0,0 +1,5 @@
+# Spotify API Oauth Implementation without any 3rd party library
+- Set the API Key, and API Secret in spotify_apis.py
+- Now run the main.py
+- Go to the url, Accept the auth request and paste the url from url bar in terminal
+- The demo list of songs will be added to a playlist.

main.py

@@ -0,0 +1,24 @@
+from scrape import Scrape
+from spotify_apis import Spotify
+
+user_date = input("Which year do you want to travel to? Enter the date in this format YYYY-MM-DD: ")
+
+# Data from billboards
+scrape = Scrape(user_date)
+data = scrape.scrape_data()
+for i in range(len(data)):
+    print(f"Song {i+1}: {data[i]}")
+
+
+# Search songs from spotify
+spotify = Spotify()
+spotify.authenticate()
+user_id = spotify.get_profile()
+playlist_id = spotify.create_playlist(playlist_name=f'Billboard\'s hot songs - {user_date}')
+uris = []
+for song in data:
+    uri = spotify.search(song)
+    uris.append(f"{uri}")
+# uris = uris.split(', ')
+print(uris, len(uris))
+spotify.add_song_to_list(playlist_id, uris)

scrape.py

@@ -0,0 +1,14 @@
+from bs4 import BeautifulSoup
+import requests
+
+BILLBOARDS_BASE_URL = 'https://www.billboard.com/charts/hot-100/'
+
+class Scrape:
+    def __init__(self, date):
+        self.url = f"{BILLBOARDS_BASE_URL}{date}"
+
+    def scrape_data(self) -> list:
+        response = requests.get(url=self.url)
+        soup = BeautifulSoup(markup=response.text, features='html.parser')
+        songs = [song.getText().strip() for song in soup.select(selector='ul li h3')]
+        return songs[:100]

spotify_apis.py

@@ -0,0 +1,130 @@
+import requests
+import requests.auth
+import os
+from urllib.parse import parse_qs, urlparse
+import base64
+
+# Add SPOTIFY_CLIENT_ID, SPOTIFY_CLIENT_SECRET in environment variables
+# CLIENT_ID = os.environ['SPOTIFY_CLIENT_ID']
+# CLIENT_SECRET = os.environ['SPOTIFY_CLIENT_SECRET']
+# Or for testing add them here ID and SECRET. Change code below accordingly
+ID = 'ADD YOUR API ID HERE'
+SECRET = 'ADD API SECRET HERE'
+BASE_URL = 'https://accounts.spotify.com'
+AUTHORIZE_ENDPOINT = '/authorize'
+TOKEN_ENDPOINT = '/api/token'
+REDIRECT_URI = 'https://spotify.com/'
+API_URL = 'https://api.spotify.com/v1'
+
+class Spotify:
+    def __init__(self):
+        self.auth_request_parameter = {
+            'client_id': ID,
+            'response_type': 'code',
+            'redirect_uri': REDIRECT_URI,
+            'state': 'state',
+            'scope': 'playlist-modify-public playlist-modify-private'
+        }
+        self.token_request_headers = {
+            'Authorization': f"Basic {base64.b64encode(bytes(f'{ID}:{SECRET}', 'utf-8')).decode('utf-8')}",
+            'Content-Type': 'application/x-www-form-urlencoded'
+        }
+        self.access_token = None
+        self.id = None
+
+    def authenticate(self):
+        '''This method authennticates the user using cliend id in 3 steps. First it makes get request to get a code,
+        then it gets the code out of the url, and then it makes a post request using the code, client id and client
+        secret, and receives the token in current object\'s access_token attribute.'''
+
+        # STEP 1: get 'code' using ' /authorize ' endpoint
+        authorize_response = requests.get(url=f"{BASE_URL}{AUTHORIZE_ENDPOINT}", params=self.auth_request_parameter)
+        print(authorize_response.status_code)
+        authorize_response.raise_for_status()
+        print(f"Go here to authorise: {authorize_response.url}")
+
+        # STEP 2: parse the url to get 'code' and 'state'
+        authorized_url = input("Enter the url: ")
+        parsed_url = urlparse(authorized_url)
+        print(parsed_url)
+        code = parse_qs(parsed_url.query)['code'][0]
+        state = parse_qs(parsed_url.query)['state'][0]
+        print(code, state)
+
+        # STEP 3: Use the code with ' /api/token ' endpoint
+        token_request_body = {
+            'grant_type': 'authorization_code',
+            'code': code,
+            'redirect_uri': REDIRECT_URI,
+        }
+        token_response = requests.post(url=f"{BASE_URL}{TOKEN_ENDPOINT}",
+                                       headers=self.token_request_headers,
+                                       params=token_request_body)
+        print(token_response.text)
+        token_response.raise_for_status()
+        self.access_token = token_response.json()['access_token']
+        print(token_response.json()['access_token'])
+
+    def get_profile(self):
+        '''This method uses token to get user\'s profile, and gets user id out of the response received.'''
+        header = {
+            'Authorization': f'Bearer {self.access_token}'
+        }
+        endpoint = '/me'
+        response = requests.get(url=f"{API_URL}{endpoint}", headers=header)
+        print(response.text)
+        response.raise_for_status()
+        print(response.status_code)
+        self.id = response.json()['id']
+
+    def create_playlist(self, playlist_name):
+        endpoint = f'/users/{self.id}/playlists'
+        name = playlist_name
+        header = {
+            'Authorization': f'Bearer {self.access_token}'
+        }
+        body = {
+            'name': name,
+            'description': 'Created with code',
+            'public': 'True'
+        }
+        response = requests.post(url=f"{API_URL}{endpoint}", headers=header, json=body)
+        print(response.text)
+        response.raise_for_status()
+        return response.json()['id']
+
+    def search(self, song):
+        '''This method takes a string as input parameter, then searches the string using /search endpoint and
+        access token. Finally returns the song\'s uri.'''
+        endpoint = '/search'
+        header = {
+            'Authorization': f'Bearer {self.access_token}'
+        }
+        body = {
+            'q': song,
+            'type': 'track',
+            'limit': 10,
+            'market': 'IN'
+        }
+        response = requests.get(url=f"{API_URL}{endpoint}", headers=header, params=body)
+        # print(response.text)
+        response.raise_for_status()
+        print(f"uri for {song}: {response.json()['tracks']['items'][0]['uri']}")
+        return response.json()['tracks']['items'][0]['uri']
+
+    def add_song_to_list(self, id, songs):
+        endpoint = f"/playlists/{id}/tracks"
+        headers = {
+            'Content-Type': 'application/json',
+            'Authorization': f'Bearer {self.access_token}'
+        }
+        body = {
+            'uris': songs,
+            'position': 0
+        }
+        add_song_response = requests.post(url=f"{API_URL}{endpoint}", headers=headers, json=body)
+        print(add_song_response.text)
+        add_song_response.raise_for_status()
+        if add_song_response.status_code == 200:
+            print("Songs added to playlist")
+