SpotifyRecorder.py

import json
import os.path
import time
from typing import List, Dict

import spotipy
from TheCodeLabs_BaseUtils.DefaultLogger import DefaultLogger
from spotipy import SpotifyOAuth, CacheFileHandler

from SpotifyAudioRecorder import SpotifyAudioRecorder

LOG_FORMAT = '[%(levelname)-7s] - %(asctime)s - %(message)s'
LOGGER = DefaultLogger().create_logger_if_not_exists('SpotifyRecorder', logFormat=LOG_FORMAT)


class SpotifyRecorder:
    _MAX_WAIT_TIME_TRACK_STARTING_IN_S = 10
    _WAIT_TIME_TRACK_PLAYING_SHORT_IN_S = 1
    _THRESHOLD_TRACK_END_IN_MS = 5000

    def __init__(self, clientID: str,
                 clientSecret: str,
                 redirectUrl: str,
                 openBrowser: bool,
                 cacheFilePath: str,
                 playlist: Dict[str, str],
                 spotifyDeviceName: str,
                 audioDeviceName: str,
                 destinationFolder: str,
                 startNumber: int = 0,
                 limit: int = -1):
        self._clientID = clientID
        self._clientSecret = clientSecret
        self._redirectUrl = redirectUrl
        self._openBrowser = openBrowser
        self._cacheFilePath = cacheFilePath
        self._playlist = playlist
        self._spotifyDeviceName = spotifyDeviceName
        self._audioDeviceName = audioDeviceName
        self._destinationFolder = destinationFolder
        self._startNumber = startNumber
        self._limit = limit

        os.makedirs(self._destinationFolder, exist_ok=True)

        self._spotify = self.login()

    def login(self) -> spotipy.Spotify:
        client_credentials_manager = SpotifyOAuth(client_id=self._clientID,
                                                  client_secret=self._clientSecret,
                                                  redirect_uri=self._redirectUrl,
                                                  scope='user-read-playback-state,user-modify-playback-state',
                                                  open_browser=self._openBrowser,
                                                  cache_handler=CacheFileHandler(cache_path=self._cacheFilePath))
        return spotipy.Spotify(client_credentials_manager=client_credentials_manager)

    def run(self):
        LOGGER.info(f'Fetching all tracks for playlist {self._playlist["name"]}...')
        playlist = self.__get_playlist(self._playlist['user'], self._playlist['id'])
        tracks = self.__get_tracks(playlist)

        self.__record_tracks(tracks)

    def __get_playlist(self, username: str, playlistID: str) -> Dict:
        LOGGER.info(f'Fetching playlist with ID: {playlistID} by {username}...')
        identifier = f'spotify:user:{username}:playlist:{playlistID}'
        playlist = self._spotify.playlist(identifier)
        LOGGER.info(f'Found playlist "{playlist["name"]}"')
        return playlist

    def __get_tracks(self, playlist) -> List[Dict]:
        tracks = playlist['tracks']
        results = tracks['items']

        while tracks['next']:
            tracks = self._spotify.next(tracks)
            results.extend(tracks['items'])

        LOGGER.info(f'Found {len(tracks)} tracks in playlist')
        return results

    def __extract_track_uris(self, tracks: List) -> List[str]:
        return [track['track']['uri'] for track in tracks]

    def __record_tracks(self, tracks: list):
        deviceId = self.__get_device_id_by_name(self._spotifyDeviceName)

        recordedTrackNumbers = []
        skippedTrackNumbers = []
        errorTrackNumbers = []

        if self._limit == -1:
            LOGGER.info(f'Recording track #{self._startNumber} to all')
            tracks = tracks[self._startNumber - 1:]
        else:
            LOGGER.info(f'Recording track #{self._startNumber} to (including) #{self._startNumber + self._limit - 1}')
            tracks = tracks[self._startNumber - 1:self._startNumber - 1 + self._limit]

        for index, track in enumerate(tracks):
            indexInPlaylist = self._startNumber + index

            if track['is_local']:
                # It's not possible to add a local track to a playlist using the web API.
                # https://github.com/plamere/spotipy/issues/793#issuecomment-1082421408
                LOGGER.info(f'Skipping local track "{track["track"]["name"]}"')
                skippedTrackNumbers.append(indexInPlaylist)
                continue

            LOGGER.info(
                f'>>> Recording track {index + 1}/{len(tracks)}: #{indexInPlaylist} "{track["track"]["name"]}"...')
            try:
                self.__stop_playback_if_playing(deviceId)

                filePath = self.__determine_file_path(indexInPlaylist + 1, track)
                recorder = SpotifyAudioRecorder(self._audioDeviceName, filePath)
                with recorder.record():
                    self.__play_track(deviceId, track['track']['uri'])
                    timeWaitedForPlaying = self.__wait_for_track_playing(track['track']['id'])
                    self.__wait_for_track_end(track, timeWaitedForPlaying)

                recordedTrackNumbers.append(indexInPlaylist)
            except Exception as e:
                LOGGER.error(f'An error occurred while recording track "{track["track"]["name"]}"', exc_info=e)
                errorTrackNumbers.append(indexInPlaylist)

        LOGGER.info('### DONE ###')

        LOGGER.info('>>> Skipped <<<')
        for number in skippedTrackNumbers:
            LOGGER.info(f'Skipped #{number}')

        LOGGER.info('>>> Errors <<<')
        for number in errorTrackNumbers:
            LOGGER.info(f'Error #{number}')

        LOGGER.info(
            f'### {len(tracks)} tracks, {len(recordedTrackNumbers)} recorded, {len(skippedTrackNumbers)} skipped, {len(errorTrackNumbers)} errors ###')

    def __determine_file_path(self, index: int, track) -> str:
        artists = track['track']['artists']
        artists = ' & '.join(artist['name'] for artist in artists)
        fileName = f'{index} - {artists} - {track["track"]["name"]}.wav'
        return os.path.join(self._destinationFolder, fileName)

    def __wait_for_track_end(self, track, timeWaitedForPlaying: float) -> None:
        trackDurationInMs = track['track']['duration_ms']
        trackDurationInSeconds = trackDurationInMs // 1000
        LOGGER.info(f'\tTrack duration: {self.__convert_seconds_to_duration(trackDurationInSeconds)}')

        startTime = time.time()

        while time.time() - startTime < trackDurationInSeconds:
            currentPlayback = self._spotify.current_playback()
            if currentPlayback['is_playing']:
                remainingTimeInMs = trackDurationInMs - currentPlayback['progress_ms']
                remainingTimeInSeconds = remainingTimeInMs // 1000

                sleepTime = self._WAIT_TIME_TRACK_PLAYING_SHORT_IN_S

                if remainingTimeInMs > self._THRESHOLD_TRACK_END_IN_MS:
                    sleepTime = remainingTimeInSeconds / 2

                LOGGER.debug(f'\t\tWaiting for track to end (remaining: '
                             f'{self.__convert_seconds_to_duration(remainingTimeInSeconds)}, '
                             f'sleep: {self.__convert_seconds_to_duration(int(sleepTime))}s)...')
                time.sleep(sleepTime)
            else:
                waitDuration = int(time.time() - startTime)
                waitDuration += timeWaitedForPlaying

                if waitDuration < trackDurationInSeconds - self._WAIT_TIME_TRACK_PLAYING_SHORT_IN_S:
                    raise RuntimeError(
                        f'Track finished too early (waited: {waitDuration}s, expected: {trackDurationInSeconds}s)')
                if waitDuration > trackDurationInSeconds + self._WAIT_TIME_TRACK_PLAYING_SHORT_IN_S * 3:
                    raise RuntimeError(
                        f'Track finished too late (waited: {waitDuration}s, expected: {trackDurationInSeconds})')

                LOGGER.info(f'Track finished. Waited {waitDuration}s, expected {trackDurationInSeconds}s, OK')
                break

    @staticmethod
    def __convert_seconds_to_duration(seconds: int):
        secs = seconds % 60
        minutes = int(seconds / 60) % 60
        hours = int(seconds / (60 * 60)) % 24
        return f'{str(hours).zfill(2)}:{str(minutes).zfill(2)}:{str(secs).zfill(2)}'

    def __get_device_id_by_name(self, deviceName: str) -> str:
        devices = self._spotify.devices()['devices']
        for device in devices:
            if device['name'] == deviceName:
                return device['id']

        raise RuntimeError('Device not found')

    def __play_track(self, deviceId: str, trackUri: str) -> None:
        self._spotify.start_playback(device_id=deviceId, uris=[trackUri])

    def __stop_playback_if_playing(self, deviceId: str) -> None:
        if self._spotify.current_playback()['is_playing']:
            self._spotify.pause_playback(device_id=deviceId)

    def __wait_for_track_playing(self, expectedTrackId: str) -> float:
        LOGGER.debug(f'\t\tWait for track to start playing...')
        startTime = time.time()
        duration = 0
        while time.time() - startTime < self._MAX_WAIT_TIME_TRACK_STARTING_IN_S:
            currentPlayback = self._spotify.current_playback()
            if currentPlayback['is_playing']:
                duration = time.time() - startTime
                if currentPlayback['item']['id'] == expectedTrackId:
                    LOGGER.debug(f'\t\tTrack started playing after {duration:.1f}s')
                    break
                else:
                    raise RuntimeError(f'Wrong track started playing (actual: {currentPlayback["item"]["id"]}, expected: {expectedTrackId})')

            time.sleep(1)

        return duration


if __name__ == '__main__':
    with open('config/settings-recorder.json', 'r', encoding='utf-8') as f:
        SETTINGS = json.load(f)

    spotifyBackup = SpotifyRecorder(SETTINGS['spotifyAPI']['clientID'],
                                    SETTINGS['spotifyAPI']['clientSecret'],
                                    SETTINGS['redirectUrl'],
                                    SETTINGS['openBrowser'],
                                    SETTINGS['cacheFilePath'],
                                    SETTINGS['playlist'],
                                    SETTINGS['spotifyDeviceName'],
                                    SETTINGS['audioDeviceName'],
                                    SETTINGS['destinationFolder'],
                                    SETTINGS['startNumber'],
                                    SETTINGS['limit'], )

    spotifyBackup.run()

    LOGGER.info('### DONE ###')