From 597e442c1678cd733f525a9717ef306118c62e4d Mon Sep 17 00:00:00 2001 From: Robert Goldmann <deadlocker@gmx.de> Date: Wed, 9 Jun 2021 23:14:25 +0200 Subject: [PATCH] #1 - externalized code duplication --- BackupHandler.py | 63 +++++++++++++++++++++++++++++ SpotifyBackup.py | 70 +++++++++++---------------------- SpotifyFollowedArtistsBackup.py | 57 ++++++--------------------- 3 files changed, 99 insertions(+), 91 deletions(-) create mode 100644 BackupHandler.py diff --git a/BackupHandler.py b/BackupHandler.py new file mode 100644 index 0000000..37837b9 --- /dev/null +++ b/BackupHandler.py @@ -0,0 +1,63 @@ +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)) diff --git a/SpotifyBackup.py b/SpotifyBackup.py index 9b8b0f7..8e5a161 100644 --- a/SpotifyBackup.py +++ b/SpotifyBackup.py @@ -1,38 +1,40 @@ import csv import json -import os -from datetime import datetime, timedelta +from typing import Dict, List import spotipy from TheCodeLabs_BaseUtils.DefaultLogger import DefaultLogger from spotipy.oauth2 import SpotifyClientCredentials +from BackupHandler import BackupHandler + LOG_FORMAT = '[%(levelname)-7s] - %(asctime)s - %(message)s' LOGGER = DefaultLogger().create_logger_if_not_exists('SpotifyBackup', logFormat=LOG_FORMAT) -class SpotifyBackup: - DATE_FORMAT = '%Y-%m-%d_%H-%M-%S' - DATE_FORMAT_SHORT = '%Y-%m-%d' - INVALID_CHARS = [' ', '\t', '\n'] - REPLACE_CHAR = '_' +class SpotifyBackup(BackupHandler): + def __init__(self, clientID: str, clientSecret: str, exportFolder: str, playlists: List[Dict[str, str]]): + self._clientSecret = clientSecret + super().__init__(exportFolder, clientID) + self._playlists = playlists - def __init__(self, clientID: str, clientSecret: str, exportFolder: str): - self._exportFolder = exportFolder - if self._exportFolder: - os.makedirs(self._exportFolder, exist_ok=True) + def login(self) -> spotipy.Spotify: + client_credentials_manager = SpotifyClientCredentials(client_id=self._clientID, + client_secret=self._clientSecret) + return spotipy.Spotify(client_credentials_manager=client_credentials_manager) - client_credentials_manager = SpotifyClientCredentials(client_id=clientID, client_secret=clientSecret) - self._spotify = spotipy.Spotify(client_credentials_manager=client_credentials_manager) + def backup(self): + 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) 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}"...') - self.__export_csv(exportPath, tracks) + self._export_csv(exportPath, tracks) LOGGER.info('Exporting DONE') def __get_playlist(self, username: str, playlistID: str): @@ -52,19 +54,12 @@ class SpotifyBackup: LOGGER.info(f'Fetched {len(results)} tracks') return results - def __determine_export_path(self, playlistName): - 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): + def _export_csv(self, exportPath: str, items: List): with open(exportPath, 'w', newline='', encoding='utf-8') as f: writer = csv.writer(f, delimiter=';', quotechar='|', quoting=csv.QUOTE_MINIMAL) writer.writerow(('title', 'artists', 'album', 'addedBy', 'addedAt', 'url')) - for track in tracks: + for track in items: try: trackInfo = track['track'] title = trackInfo['name'] @@ -79,24 +74,6 @@ class SpotifyBackup: except Exception: 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__': with open('config/settings.json', 'r', encoding='utf-8') as f: @@ -104,11 +81,10 @@ if __name__ == '__main__': spotifyBackup = SpotifyBackup(SETTINGS['spotifyAPI']['clientID'], SETTINGS['spotifyAPI']['clientSecret'], - SETTINGS['exportFolder']) + SETTINGS['exportFolder'], + SETTINGS['playlists']) spotifyBackup.clean_old_exports(SETTINGS['daysToKeep']) - - for playlist in SETTINGS['playlists']: - spotifyBackup.backup_playlist(playlist['user'], playlist['id']) + spotifyBackup.backup() LOGGER.info('### DONE ###') diff --git a/SpotifyFollowedArtistsBackup.py b/SpotifyFollowedArtistsBackup.py index 842d8a8..81fa776 100644 --- a/SpotifyFollowedArtistsBackup.py +++ b/SpotifyFollowedArtistsBackup.py @@ -1,39 +1,33 @@ import csv import json -import os -from datetime import datetime, timedelta from typing import List import spotipy from TheCodeLabs_BaseUtils.DefaultLogger import DefaultLogger from spotipy.oauth2 import SpotifyPKCE +from BackupHandler import BackupHandler + LOG_FORMAT = '[%(levelname)-7s] - %(asctime)s - %(message)s' LOGGER = DefaultLogger().create_logger_if_not_exists('SpotifyFollowerBackup', logFormat=LOG_FORMAT) -class SpotifyFollowedArtistsBackup: - DATE_FORMAT = '%Y-%m-%d_%H-%M-%S' - DATE_FORMAT_SHORT = '%Y-%m-%d' - INVALID_CHARS = [' ', '\t', '\n'] - REPLACE_CHAR = '_' - +class SpotifyFollowedArtistsBackup(BackupHandler): def __init__(self, clientID: str, exportFolder: str): - self._exportFolder = exportFolder - if self._exportFolder: - os.makedirs(self._exportFolder, exist_ok=True) + super().__init__(exportFolder, clientID) - 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') - 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() - exportPath = self.__determine_export_path('followed_artists') + exportPath = self._determine_export_path('followed_artists') LOGGER.info(f'>>> Exporting followed artists to "{exportPath}"...') - self.__export_csv(exportPath, followedArtists) + self._export_csv(exportPath, followedArtists) LOGGER.info('Exporting DONE') def __get_followed_artists(self): @@ -47,19 +41,12 @@ class SpotifyFollowedArtistsBackup: LOGGER.info(f'Fetched {len(results)} followed artists') return results - def __determine_export_path(self, fileName): - 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): + def _export_csv(self, exportPath: str, items: List): with open(exportPath, 'w', newline='', encoding='utf-8') as f: writer = csv.writer(f, delimiter=';', quotechar='|', quoting=csv.QUOTE_MINIMAL) writer.writerow(('id', 'name', 'url', 'uri')) - for artist in artists: + for artist in items: try: id = artist['id'] name = artist['name'] @@ -70,24 +57,6 @@ class SpotifyFollowedArtistsBackup: except Exception: 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__': with open('config/settings.json', 'r', encoding='utf-8') as f: @@ -95,5 +64,5 @@ if __name__ == '__main__': spotifyBackup = SpotifyFollowedArtistsBackup(SETTINGS['spotifyAPI']['clientID'], SETTINGS['exportFolder']) spotifyBackup.clean_old_exports(SETTINGS['daysToKeep']) - spotifyBackup.backup_followed_artists() + spotifyBackup.backup() LOGGER.info('### DONE ###') -- GitLab