Skip to content
Snippets Groups Projects
Commit 597e442c authored by Robert Goldmann's avatar Robert Goldmann
Browse files

#1 - externalized code duplication

parent 5c142398
No related branches found
No related tags found
No related merge requests found
import abc
import os
from datetime import datetime, timedelta
from typing import List
import spotipy
from TheCodeLabs_BaseUtils.DefaultLogger import DefaultLogger
LOG_FORMAT = '[%(levelname)-7s] - %(asctime)s - %(message)s'
LOGGER = DefaultLogger().create_logger_if_not_exists('SpotifyBackup', logFormat=LOG_FORMAT)
class BackupHandler(abc.ABC):
DATE_FORMAT = '%Y-%m-%d_%H-%M-%S'
DATE_FORMAT_SHORT = '%Y-%m-%d'
INVALID_CHARS = [' ', '\t', '\n']
REPLACE_CHAR = '_'
def __init__(self, exportFolder: str, clientID: str):
self._clientID = clientID
self._exportFolder = exportFolder
if self._exportFolder:
os.makedirs(self._exportFolder, exist_ok=True)
self._spotify = self.login()
@abc.abstractmethod
def login(self) -> spotipy.Spotify:
pass
@abc.abstractmethod
def backup(self):
pass
@abc.abstractmethod
def _export_csv(self, exportPath: str, items: List):
pass
def _determine_export_path(self, fileName: str) -> str:
for item in self.INVALID_CHARS:
fileName = fileName.replace(item, self.REPLACE_CHAR)
completeFileName = f'{datetime.strftime(datetime.now(), self.DATE_FORMAT)}_{fileName}.csv'
exportPath = os.path.join(self._exportFolder, completeFileName)
return exportPath
def clean_old_exports(self, daysToKeep):
if not daysToKeep:
LOGGER.info('Skipping deletion of old files')
return
minimumDate = datetime.today() - timedelta(days=int(daysToKeep))
LOGGER.info(f'>>> Deleting files older than {minimumDate} ({daysToKeep} days)')
files = [file for file in sorted(os.listdir(self._exportFolder)) if
os.path.isfile(os.path.join(self._exportFolder, file))]
for file in files:
parts = file.split('_')
creationDate = datetime.strptime(parts[0], self.DATE_FORMAT_SHORT)
if creationDate < minimumDate:
LOGGER.info(f'Removing old file "{file}"')
os.remove(os.path.join(self._exportFolder, file))
import csv import csv
import json import json
import os from typing import Dict, List
from datetime import datetime, timedelta
import spotipy import spotipy
from TheCodeLabs_BaseUtils.DefaultLogger import DefaultLogger from TheCodeLabs_BaseUtils.DefaultLogger import DefaultLogger
from spotipy.oauth2 import SpotifyClientCredentials from spotipy.oauth2 import SpotifyClientCredentials
from BackupHandler import BackupHandler
LOG_FORMAT = '[%(levelname)-7s] - %(asctime)s - %(message)s' LOG_FORMAT = '[%(levelname)-7s] - %(asctime)s - %(message)s'
LOGGER = DefaultLogger().create_logger_if_not_exists('SpotifyBackup', logFormat=LOG_FORMAT) LOGGER = DefaultLogger().create_logger_if_not_exists('SpotifyBackup', logFormat=LOG_FORMAT)
class SpotifyBackup: class SpotifyBackup(BackupHandler):
DATE_FORMAT = '%Y-%m-%d_%H-%M-%S' def __init__(self, clientID: str, clientSecret: str, exportFolder: str, playlists: List[Dict[str, str]]):
DATE_FORMAT_SHORT = '%Y-%m-%d' self._clientSecret = clientSecret
INVALID_CHARS = [' ', '\t', '\n'] super().__init__(exportFolder, clientID)
REPLACE_CHAR = '_' self._playlists = playlists
def __init__(self, clientID: str, clientSecret: str, exportFolder: str): def login(self) -> spotipy.Spotify:
self._exportFolder = exportFolder client_credentials_manager = SpotifyClientCredentials(client_id=self._clientID,
if self._exportFolder: client_secret=self._clientSecret)
os.makedirs(self._exportFolder, exist_ok=True) return spotipy.Spotify(client_credentials_manager=client_credentials_manager)
client_credentials_manager = SpotifyClientCredentials(client_id=clientID, client_secret=clientSecret) def backup(self):
self._spotify = spotipy.Spotify(client_credentials_manager=client_credentials_manager) for playlist in SETTINGS['playlists']:
self.__backup_playlist(playlist['user'], playlist['id'])
def backup_playlist(self, username: str, playlistID: str): def __backup_playlist(self, username: str, playlistID: str):
playlist = self.__get_playlist(username, playlistID) playlist = self.__get_playlist(username, playlistID)
tracks = self.__get_tracks(playlist) tracks = self.__get_tracks(playlist)
exportPath = self.__determine_export_path(playlist['name']) exportPath = self._determine_export_path(playlist['name'])
LOGGER.info(f'>>> Exporting tracks to "{exportPath}"...') LOGGER.info(f'>>> Exporting tracks to "{exportPath}"...')
self.__export_csv(exportPath, tracks) self._export_csv(exportPath, tracks)
LOGGER.info('Exporting DONE') LOGGER.info('Exporting DONE')
def __get_playlist(self, username: str, playlistID: str): def __get_playlist(self, username: str, playlistID: str):
...@@ -52,19 +54,12 @@ class SpotifyBackup: ...@@ -52,19 +54,12 @@ class SpotifyBackup:
LOGGER.info(f'Fetched {len(results)} tracks') LOGGER.info(f'Fetched {len(results)} tracks')
return results return results
def __determine_export_path(self, playlistName): def _export_csv(self, exportPath: str, items: List):
for item in self.INVALID_CHARS:
playlistName = playlistName.replace(item, self.REPLACE_CHAR)
fileName = f'{datetime.strftime(datetime.now(), self.DATE_FORMAT)}_{playlistName}.csv'
exportPath = os.path.join(self._exportFolder, fileName)
return exportPath
def __export_csv(self, exportPath, tracks):
with open(exportPath, 'w', newline='', encoding='utf-8') as f: with open(exportPath, 'w', newline='', encoding='utf-8') as f:
writer = csv.writer(f, delimiter=';', quotechar='|', quoting=csv.QUOTE_MINIMAL) writer = csv.writer(f, delimiter=';', quotechar='|', quoting=csv.QUOTE_MINIMAL)
writer.writerow(('title', 'artists', 'album', 'addedBy', 'addedAt', 'url')) writer.writerow(('title', 'artists', 'album', 'addedBy', 'addedAt', 'url'))
for track in tracks: for track in items:
try: try:
trackInfo = track['track'] trackInfo = track['track']
title = trackInfo['name'] title = trackInfo['name']
...@@ -79,24 +74,6 @@ class SpotifyBackup: ...@@ -79,24 +74,6 @@ class SpotifyBackup:
except Exception: except Exception:
LOGGER.exception(f'Error while exporting track "{track}"') LOGGER.exception(f'Error while exporting track "{track}"')
def clean_old_exports(self, daysToKeep):
if not daysToKeep:
LOGGER.info('Skipping deletion of old files')
return
minimumDate = datetime.today() - timedelta(days=int(daysToKeep))
LOGGER.info(f'>>> Deleting files older than {minimumDate} ({daysToKeep} days)')
files = [file for file in sorted(os.listdir(self._exportFolder)) if
os.path.isfile(os.path.join(self._exportFolder, file))]
for file in files:
parts = file.split('_')
creationDate = datetime.strptime(parts[0], self.DATE_FORMAT_SHORT)
if creationDate < minimumDate:
LOGGER.info(f'Removing old file "{file}"')
os.remove(os.path.join(self._exportFolder, file))
if __name__ == '__main__': if __name__ == '__main__':
with open('config/settings.json', 'r', encoding='utf-8') as f: with open('config/settings.json', 'r', encoding='utf-8') as f:
...@@ -104,11 +81,10 @@ if __name__ == '__main__': ...@@ -104,11 +81,10 @@ if __name__ == '__main__':
spotifyBackup = SpotifyBackup(SETTINGS['spotifyAPI']['clientID'], spotifyBackup = SpotifyBackup(SETTINGS['spotifyAPI']['clientID'],
SETTINGS['spotifyAPI']['clientSecret'], SETTINGS['spotifyAPI']['clientSecret'],
SETTINGS['exportFolder']) SETTINGS['exportFolder'],
SETTINGS['playlists'])
spotifyBackup.clean_old_exports(SETTINGS['daysToKeep']) spotifyBackup.clean_old_exports(SETTINGS['daysToKeep'])
spotifyBackup.backup()
for playlist in SETTINGS['playlists']:
spotifyBackup.backup_playlist(playlist['user'], playlist['id'])
LOGGER.info('### DONE ###') LOGGER.info('### DONE ###')
import csv import csv
import json import json
import os
from datetime import datetime, timedelta
from typing import List from typing import List
import spotipy import spotipy
from TheCodeLabs_BaseUtils.DefaultLogger import DefaultLogger from TheCodeLabs_BaseUtils.DefaultLogger import DefaultLogger
from spotipy.oauth2 import SpotifyPKCE from spotipy.oauth2 import SpotifyPKCE
from BackupHandler import BackupHandler
LOG_FORMAT = '[%(levelname)-7s] - %(asctime)s - %(message)s' LOG_FORMAT = '[%(levelname)-7s] - %(asctime)s - %(message)s'
LOGGER = DefaultLogger().create_logger_if_not_exists('SpotifyFollowerBackup', logFormat=LOG_FORMAT) LOGGER = DefaultLogger().create_logger_if_not_exists('SpotifyFollowerBackup', logFormat=LOG_FORMAT)
class SpotifyFollowedArtistsBackup: class SpotifyFollowedArtistsBackup(BackupHandler):
DATE_FORMAT = '%Y-%m-%d_%H-%M-%S'
DATE_FORMAT_SHORT = '%Y-%m-%d'
INVALID_CHARS = [' ', '\t', '\n']
REPLACE_CHAR = '_'
def __init__(self, clientID: str, exportFolder: str): def __init__(self, clientID: str, exportFolder: str):
self._exportFolder = exportFolder super().__init__(exportFolder, clientID)
if self._exportFolder:
os.makedirs(self._exportFolder, exist_ok=True)
client_credentials_manager = SpotifyPKCE(client_id=clientID, redirect_uri='http://localhost:8080', def login(self) -> spotipy.Spotify:
client_credentials_manager = SpotifyPKCE(client_id=self._clientID, redirect_uri='http://localhost:8080',
scope='user-follow-read') scope='user-follow-read')
self._spotify = spotipy.Spotify(client_credentials_manager=client_credentials_manager) return spotipy.Spotify(client_credentials_manager=client_credentials_manager)
def backup_followed_artists(self): def backup(self):
followedArtists = self.__get_followed_artists() followedArtists = self.__get_followed_artists()
exportPath = self.__determine_export_path('followed_artists') exportPath = self._determine_export_path('followed_artists')
LOGGER.info(f'>>> Exporting followed artists to "{exportPath}"...') LOGGER.info(f'>>> Exporting followed artists to "{exportPath}"...')
self.__export_csv(exportPath, followedArtists) self._export_csv(exportPath, followedArtists)
LOGGER.info('Exporting DONE') LOGGER.info('Exporting DONE')
def __get_followed_artists(self): def __get_followed_artists(self):
...@@ -47,19 +41,12 @@ class SpotifyFollowedArtistsBackup: ...@@ -47,19 +41,12 @@ class SpotifyFollowedArtistsBackup:
LOGGER.info(f'Fetched {len(results)} followed artists') LOGGER.info(f'Fetched {len(results)} followed artists')
return results return results
def __determine_export_path(self, fileName): def _export_csv(self, exportPath: str, items: List):
for item in self.INVALID_CHARS:
fileName = fileName.replace(item, self.REPLACE_CHAR)
completeFileName = f'{datetime.strftime(datetime.now(), self.DATE_FORMAT)}_{fileName}.csv'
exportPath = os.path.join(self._exportFolder, completeFileName)
return exportPath
def __export_csv(self, exportPath: str, artists: List):
with open(exportPath, 'w', newline='', encoding='utf-8') as f: with open(exportPath, 'w', newline='', encoding='utf-8') as f:
writer = csv.writer(f, delimiter=';', quotechar='|', quoting=csv.QUOTE_MINIMAL) writer = csv.writer(f, delimiter=';', quotechar='|', quoting=csv.QUOTE_MINIMAL)
writer.writerow(('id', 'name', 'url', 'uri')) writer.writerow(('id', 'name', 'url', 'uri'))
for artist in artists: for artist in items:
try: try:
id = artist['id'] id = artist['id']
name = artist['name'] name = artist['name']
...@@ -70,24 +57,6 @@ class SpotifyFollowedArtistsBackup: ...@@ -70,24 +57,6 @@ class SpotifyFollowedArtistsBackup:
except Exception: except Exception:
LOGGER.exception(f'Error while exporting artist "{artist}"') LOGGER.exception(f'Error while exporting artist "{artist}"')
def clean_old_exports(self, daysToKeep):
if not daysToKeep:
LOGGER.info('Skipping deletion of old files')
return
minimumDate = datetime.today() - timedelta(days=int(daysToKeep))
LOGGER.info(f'>>> Deleting files older than {minimumDate} ({daysToKeep} days)')
files = [file for file in sorted(os.listdir(self._exportFolder)) if
os.path.isfile(os.path.join(self._exportFolder, file))]
for file in files:
parts = file.split('_')
creationDate = datetime.strptime(parts[0], self.DATE_FORMAT_SHORT)
if creationDate < minimumDate:
LOGGER.info(f'Removing old file "{file}"')
os.remove(os.path.join(self._exportFolder, file))
if __name__ == '__main__': if __name__ == '__main__':
with open('config/settings.json', 'r', encoding='utf-8') as f: with open('config/settings.json', 'r', encoding='utf-8') as f:
...@@ -95,5 +64,5 @@ if __name__ == '__main__': ...@@ -95,5 +64,5 @@ if __name__ == '__main__':
spotifyBackup = SpotifyFollowedArtistsBackup(SETTINGS['spotifyAPI']['clientID'], SETTINGS['exportFolder']) spotifyBackup = SpotifyFollowedArtistsBackup(SETTINGS['spotifyAPI']['clientID'], SETTINGS['exportFolder'])
spotifyBackup.clean_old_exports(SETTINGS['daysToKeep']) spotifyBackup.clean_old_exports(SETTINGS['daysToKeep'])
spotifyBackup.backup_followed_artists() spotifyBackup.backup()
LOGGER.info('### DONE ###') LOGGER.info('### DONE ###')
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment