import logging
from . import scraping
from .spotify import SpotifyApi
from .db import Session, Song
from .exceptions import NoHistoryFound
[docs]class Updater(object):
def __init__(self):
self.spotify = SpotifyApi()
self.is_authenticated = False
# Add new scrapers here
self.scrapers = [
scraping.KSHEScraper(),
# scraping.EagleScraper(), # TODO: Eagle website not reachable in the EU
# scraping.Q1043Scrapper(),
# scraping.WMGKScrapper(),
# scraping.KLOScrapper()
]
[docs] def spotify_auth(self):
"""Authenticates using Authorization Code Flow.
Returns:
str: URL to redirect to
"""
url = self.spotify._authorization_code_flow_authentication()
return url
[docs] def spotify_callback(self, authorization_code):
"""Function called by Spotify with access token in the request
parameters.
Args:
authorization_code (str): Authorization code
"""
response = self.spotify._client_credentials_authentication(
authorization_code
)
logging.info(response)
self.spotify._access_token = response['access_token']
self.spotify._token_type = response['token_type']
self.spotify._token_expires_in = response['expires_in']
if response.get('access_token'):
# TODO: add authenticated until timestamp
self.is_authenticated = True
[docs] def sync_db_with_existing_songs(self, playlist_id):
"""If the playlist already exist, look for songs in it and stores them
in the local database so we don't add duplicates.
Args:
playlist_id (str): Playlist ID
"""
playlist_exists = self.spotify.check_playlist_exists(playlist_id)
if playlist_exists:
tracks = self.spotify.get_track_uris_from_playlist(playlist_id)
session = Session()
for uri in tracks:
song = Song(playlist_id=playlist_id, spotify_uri=uri)
session.add(song)
try:
session.commit()
except:
continue
session.close()
session = Session()
count = session.query(Song).count()
logging.info("Playlist has {0} songs after sync".format(count))
session.close()
[docs] def search_songs_in_spotify(self, radio_history):
"""Retrieve songs informations from title and artist using Spotify
Search API.
Args:
radio_history (list(dict)): list of dict with title and \
artist as keys
Returns:
list(dict): list of dict of spotify songs
"""
spotify_songs = [self.spotify.search_track(s['title'], s['artist'])
for s in radio_history if 'title' in s]
spotify_songs = [s for s in spotify_songs if 'spotify_uri' in s]
return spotify_songs
[docs] def filter_and_save_songs_to_db(self, spotify_songs,
scraper_name, playlist_id):
"""Filter out songs that have already been added and add the
remaining songs to the playlist.
Args:
spotify_songs (list(dict)): List of spotify songs as dict
scraper_name (str): Scraper class name
Returns:
list(dict): List of spotify songs that are not in the playlist yet
"""
# save tracks in DB
session = Session()
spotify_filtered_songs = []
for i, song in enumerate(spotify_songs):
# Don't save and upload song if it exists
match = session.query(Song)\
.filter_by(
spotify_uri=song['spotify_uri'],
playlist_id=playlist_id
)\
.first()
if match is None:
song["scraper_name"] = scraper_name
song["playlist_id"] = playlist_id
session.add(Song(**song))
spotify_filtered_songs.append(song)
else:
logging.warning('{0} already in playlist'
.format(song['song_name']))
session.commit()
return spotify_filtered_songs
[docs] def add_songs_to_playlist(self, spotify_songs, playlist_id):
"""Add spotify songs to a playlist, using songs URI.
Args:
spotify_songs (list(dict)): List of spotify songs
Returns:
json: Json response from the Spotify API
"""
logging.info('will add {0} tracks'.format(len(spotify_songs)))
response = self.spotify.add_tracks_to_playlist(
[s['spotify_uri'] for s in spotify_songs],
playlist_id
)
return response
[docs] def single_scraper_pipeline(self, scraper):
# get song history
try:
song_history = scraper.get_song_history()
except NoHistoryFound:
return {}
# spotify songs
spotify_songs = self.search_songs_in_spotify(song_history)
# filter out already present songs and sync database
spotify_filtered_songs = self.filter_and_save_songs_to_db(
spotify_songs,
scraper_name=scraper.name,
playlist_id=scraper.playlist_id
)
# upload the filtered out songs to the spotify playlist
_ = self.add_songs_to_playlist(
spotify_filtered_songs,
playlist_id=scraper.playlist_id
)
return {
"scraper": scraper.name,
"playlist_id": scraper.playlist_id,
"songs": spotify_filtered_songs
}
[docs] def scrap_and_update(self):
"""Run the whole pipeline for every scraper:
- Scrap the concerned website and get their song history
- Search for the songs in Spotify
- Filter the songs already in playlist and save them to DB
- Add the filtered songs to the playlist
Returns:
list(dict): Inserted songs
"""
inserted_songs = [self.single_scraper_pipeline(scraper)
for scraper in self.scrapers]
n_inserted_songs = sum([len(r["songs"]) for r in inserted_songs])
return inserted_songs, n_inserted_songs