spotify script

import os
import re
import subprocess
import spotipy
import music_tag
import ctypes
import time
from spotipy.oauth2 import SpotifyClientCredentials
from dotenv import load_dotenv
from collections import Counter
from mutagen.mp3 import MP3
from datetime import datetime, timedelta
from colorama import init, Fore, Back, Style
from send2trash import send2trash

class Container:
  def __init__(self):
    self.local_library_folder = "D:\Music"
    self.local_playlist_folder = ''
    self.local_temp_download_folder = "D:\Music\[Other]\DownOnSpot"
    self.allowed_extensions = ['.mp3']

    self.playlist_url = ''
    self.playlist_id = ''
    self.playlist_name = ''

    self.spotify_tracks_raw = []
    self.spotify_tracks_already_downloaded = []
    self.spotify_tracks_incomplete = []
    self.spotify_tracks_uploaded = []
    self.spotify_tracks_unavailable = []
    self.spotify_tracks_deleted = []
    self.spotify_tracks_to_download = []

    self.local_tracks_raw = []
    self.local_tracks_already_downloaded = []
    self.local_tracks_duplicate = []
    self.local_tracks_extra = []
    self.local_tracks_uploaded = []

    self.currently_downloading_track = {}
    self.newly_downloaded_track = {}
    self.completed_index = 0
    self.progress_bar_text = ''

    self.script_owner = 'HammadXP'
    self.region = 'PK'

    self.playlists = {
        'bollywood': 'https://open.spotify.com/playlist/2V6z0Yv2GSxYKhTlevre0J?si=b99b1fec8cf0492d',
        # 'bollywood_classic': 'https://open.spotify.com/playlist/0VUM3gcOCWTONPDLFKOqcT?si=6e8ec6e44d6741c4',
        # 'bollywood_more_previous': 'https://open.spotify.com/playlist/28Ftlt3xLfZAkIxxVpZgLL?si=4ef871e20bcf4936',
        'bollywood_previous': 'https://open.spotify.com/playlist/2vLAwXfPJJLT9zCZiBHLd7?si=a709e13fe5e1433f',
        'indian_hip_hop': 'https://open.spotify.com/playlist/71eOYF6sSARCrtmihyQEUP?si=7b72c2366ceb438b',
        'indian_pop': 'https://open.spotify.com/playlist/1MiOoq94f0YmJi5MYa9SwR?si=35f1bc8f275d44bf',
        # 'just_for_me': 'https://open.spotify.com/playlist/1J1ZhkZMKLttC1WhQegFoK?si=4739c9d0dc9944bf',
        'just_music': 'https://open.spotify.com/playlist/6X4aonsxqtHRFPIc37nln5?si=be7787a1028a4a61',
        'movie_soundtracks': 'https://open.spotify.com/playlist/4NSSc59unC5sBiByLDJb1D?si=44b8e3265e944d4c',
        'movie_tracks': 'https://open.spotify.com/playlist/78fHA86h7O84wBL4cXXKsN?si=d4a7f97eaed747ff',
        'not_bad': 'https://open.spotify.com/playlist/3sq8eo9pnCvPcxjYIQo0VN?si=d61c4dae712141ef',
        # 'ost': 'https://open.spotify.com/playlist/57rWznXfKMWOCxPRrPtuHN?si=7ef95cbfff71410d',
        'pakistani_classic': 'https://open.spotify.com/playlist/2O73qJgmZfB57rKnpo7Jzn?si=77800e4812a74c5e',
        'pakistani_pop': 'https://open.spotify.com/playlist/0XlE4fNzKuesDnSVmBb9bh?si=b18b659227204df9',
        'pop_music': 'https://open.spotify.com/playlist/2BOn4rcEwKZ4HxaHGOMpIf?si=d681376f9e6b42f5',
        'punjabi': 'https://open.spotify.com/playlist/0GWg1KQSiIAjquU9LANpYD?si=4bb8c09f9a0244ba',
        'shorts_less': 'https://open.spotify.com/playlist/0jLk6UGikPGpbbApFg7NvU?si=1bd4aece3d6240aa',
        'shorts_music': 'https://open.spotify.com/playlist/5rCPO6bFNZfwnZSC857LeS?si=823bb76b528a4e26',
        'solo_music': 'https://open.spotify.com/playlist/2NYdWNXYaOhuOLLWO2aY1X?si=673fab0500504229',
        'soundtracks': 'https://open.spotify.com/playlist/5wn5aufTRaLoYRykTl57u4?si=861d3c6683a943a2',
        'soundtracks_codm': 'https://open.spotify.com/playlist/0yVjrLYrbraAWGfr7xWAZt?si=4030e456b6e748ad',
        # 'superhero_music': 'https://open.spotify.com/playlist/2JilFHyef9EhsER3R6k6AR?si=93efd1985e3a4312',
        'uplifting_music': 'https://open.spotify.com/playlist/78u4RkP3zK9malm09IM9Ku?si=fbf691d3c4974d37',
    }

    self.playlists_test = {
      'DownOnSpot (Test)': 'https://open.spotify.com/playlist/4K1lpHuBT9j3TUbeHw51zt?si=570766aa646f440b',
    }

    self.playlists_2 = {
      'indian_pop': 'https://open.spotify.com/playlist/1MiOoq94f0YmJi5MYa9SwR?si=35f1bc8f275d44bf'
      # 'pakistani_pop': 'https://open.spotify.com/playlist/0XlE4fNzKuesDnSVmBb9bh?si=b18b659227204df9',
      # 'shorts_less': 'https://open.spotify.com/playlist/0jLk6UGikPGpbbApFg7NvU?si=1bd4aece3d6240aa',
    }

  ######################################################

  def create_session(self):
    load_dotenv()

    client_id = os.getenv("CLIENT_ID", "")
    client_secret = os.getenv("CLIENT_SECRET", "")

    client_credentials_manager = SpotifyClientCredentials(client_id=client_id, client_secret=client_secret)
    self.session = spotipy.Spotify(client_credentials_manager=client_credentials_manager)

  def update_window_title(self, text):
    ctypes.windll.kernel32.SetConsoleTitleW(f"Spotify Script by {self.script_owner} | {text}")

  ######################################################

  def get_playlist_id(self):
    if match := re.match(r"https://open.spotify.com/playlist/(.*)\?", self.playlist_url):
      self.playlist_id = match.groups()[0]
    else:
      raise ValueError("Bad URL format !!!")

  def get_playlist_name(self):
    self.playlist_name = self.session.user_playlist(user=None, playlist_id=self.playlist_id, fields="name")['name']

  def get_spotify_tracks_raw(self):
    self.spotify_tracks_raw = []

    response = self.session.playlist_tracks(self.playlist_id, market=self.region)
    tracks_fetched = response['items']

    while response['next']:
      response = self.session.next(response)
      tracks_fetched.extend(response['items'])

    for track in tracks_fetched:
      title = track['track']['name']
      artist = ", ".join([artist['name'] for artist in track['track']['artists']])
      album = track['track']['album']['name']
      duration = track['track']['duration_ms']
      release_date = track['track']['album']['release_date']
      added_at = track['added_at']
      track_url = track['track']['external_urls'].get('spotify', False)
      track_id = track['track']['id'] # For local -> 'None'
      is_local = track['track']['is_local']
      is_playable = track['track'].get('is_playable', True)

      self.spotify_tracks_raw.append({'title': title, 'artist': artist, 'album': album, 'duration': duration, 'release_date': release_date, 'added_at': added_at, 'track_url': track_url, 'track_id': track_id, 'is_local': is_local, 'is_playable': is_playable})

    self.spotify_tracks_raw.sort(key=lambda a: (a['title'].lower(), a['artist'].lower()))

  def get_local_tracks_raw(self):
    self.local_tracks_raw = []

    for folder, subfolders, files in os.walk(self.local_playlist_folder):
      for file in files:
        file_path = os.path.join(folder, file)
        file_name = os.path.splitext(file)[0]
        file_extension = os.path.splitext(file)[1]

        if file_extension in self.allowed_extensions:
          music_file = music_tag.load_file(file_path)
          music_file_mutagen = MP3(file_path)

          title = str(music_file['title'])
          artist = str(music_file['artist'])
          album = str(music_file['album'])
          duration = music_file_mutagen.info.length * 1000
          comment_raw = str(music_file['comment'])
          comment = comment_raw.split(',')[0]

          self.local_tracks_raw.append({"title": title, "artist": artist, "album": album, "duration": duration, "comment": comment, "file_name": file_name, "file_path": file_path})

    self.local_tracks_raw.sort(key=lambda a: (a['title'].lower(), a['artist'].lower()))

  def create_lists(self):
    for spotify_track in self.spotify_tracks_raw:
      for local_track in self.local_tracks_raw:
        if spotify_track['title'] == local_track['title'] and spotify_track['artist'] == local_track['artist'] and spotify_track['album'] == local_track['album']:

          # get list of tracks that have already been downloaded
          if spotify_track['track_id'] == local_track['comment']:
            self.spotify_tracks_already_downloaded.append(spotify_track) if spotify_track not in self.spotify_tracks_already_downloaded else None
            self.local_tracks_already_downloaded.append(local_track) if local_track not in self.local_tracks_already_downloaded else None

          # get list of incomplete tracks for re-download
          if spotify_track['duration'] != local_track['duration']:
            if (spotify_track['duration'] - local_track['duration']) > 1000: # duration difference in milliseconds, so 1000 = 1s
              self.spotify_tracks_incomplete.append(spotify_track) if spotify_track not in self.spotify_tracks_incomplete else None

          # get list of uploaded tracks
          if spotify_track['is_local']:
            self.spotify_tracks_uploaded.append(spotify_track) if spotify_track not in self.spotify_tracks_uploaded else None
            self.local_tracks_uploaded.append(local_track) if local_track not in self.local_tracks_uploaded else None

      # get list of tracks that have been deleted from spotify
      if spotify_track['title'] == '' and spotify_track['artist'] == '' and spotify_track['album'] == '':
        self.spotify_tracks_deleted.append(spotify_track) if spotify_track not in self.spotify_tracks_deleted else None

      # get list of unavailable tracks
      if not spotify_track['is_playable']:
        self.spotify_tracks_unavailable.append(spotify_track) if spotify_track not in self.spotify_tracks_unavailable else None

    # get list of extra local tracks that should be removed locally
    self.local_tracks_extra = [item for item in self.local_tracks_raw if item not in self.local_tracks_already_downloaded]

    # prevent uploaded tracks from getting added to extra local tracks list
    self.local_tracks_extra = [item for item in self.local_tracks_extra if item not in self.local_tracks_uploaded]

    # get list of tracks that have been downloaded twice with same ids but different filenames
    # checked_ids = set()

    # for local_track in self.local_tracks_raw:
    #   if local_track['comment'] != '' and local_track['comment'] in checked_ids:
    #     self.local_tracks_duplicate.append(local_track)
    #     self.local_tracks_duplicate.append(next(item for item in self.local_tracks_duplicate if item['comment'] == local_track['comment']))

    #   elif local_track['comment'] != '':
    #     checked_ids.add(local_track['comment'])

    # get list of tracks to download
    self.spotify_tracks_to_download = self.spotify_tracks_raw + self.spotify_tracks_incomplete

    for item in self.spotify_tracks_already_downloaded + self.spotify_tracks_uploaded + self.spotify_tracks_deleted + self.spotify_tracks_unavailable:
      if item in self.spotify_tracks_to_download:
        self.spotify_tracks_to_download.remove(item)

  def duplicate_proof_filename(self):
    for spotify_track in self.spotify_tracks_to_download:
      spotify_track['file_name_final'] = spotify_track['title']

      for local_track in self.local_tracks_raw:
        if spotify_track['title'].lower() == local_track['title'].lower():
          spotify_track['file_name_final'] = f"{spotify_track['title']} (by {spotify_track['artist']})"

        if spotify_track['title'].lower() == local_track['title'].lower() and spotify_track['artist'].lower() == local_track['artist'].lower():
          spotify_track['file_name_final'] = f"{spotify_track['title']} (by {spotify_track['artist']}) ({spotify_track['album']})" # for infinity war

  def remove_local_tracks_duplicate(self):
    for local_track in self.local_tracks_duplicate:
      file_path = local_track['file_path']
      does_exist = os.path.exists(file_path)

      if does_exist:
        send2trash(file_path)

  def remove_local_tracks_extra(self):
    for local_track in self.local_tracks_extra:
      file_path = local_track['file_path']
      does_exist = os.path.exists(file_path)

      if does_exist:
        send2trash(file_path)

  def empty_local_temporary_download_folder(self):
    for folder, subfolders, files in os.walk(self.local_temp_download_folder):
      for file in files:
        file_path = os.path.join(folder, file)
        send2trash(file_path)

  ######################################################

  def downloader(self):
    self.update_window_title(self.progress_bar_text)
    os.system(f"down_on_spot \"{self.currently_downloading_track['track_url']}\"")

  def get_newly_downloaded_track(self):
    for folder, subfolders, files in os.walk(self.local_temp_download_folder):
      for file in files:
        file_path = os.path.join(folder, file)
        file_name = os.path.splitext(file)[0]
        file_extension = os.path.splitext(file)[1]
        file_name_final = self.currently_downloading_track['file_name_final']

        if file_extension in self.allowed_extensions:
          music_file = music_tag.load_file(file_path)

          title = str(music_file['title'])
          artist = str(music_file['artist'])
          album = str(music_file['album'])
          comment_raw = str(music_file['comment'])
          comment = comment_raw.split(',')[0]

          self.newly_downloaded_track = {"title": title, "artist": artist, "album": album, "comment": comment, "file_name": file_name, "file_path": file_path, "file_extension": file_extension, "file_name_final": file_name_final}

  def edit_metadata(self):
    music_file = music_tag.load_file(self.newly_downloaded_track['file_path'])

    music_file['title'] = self.currently_downloading_track['title'] # forgot why i needed to edit them
    music_file['artist'] = self.currently_downloading_track['artist']
    music_file['album'] = self.currently_downloading_track['album']

    music_file['comment'] = self.currently_downloading_track['track_id']
    music_file.save()

  def edit_modification_date(self):
    timestamp_raw = self.currently_downloading_track['added_at']
    timestamp = datetime.strptime(timestamp_raw, "%Y-%m-%dT%H:%M:%SZ")
    timestamp_adjusted = timestamp + timedelta(hours=5)
    timestamp_epoch = timestamp_adjusted.timestamp()
    mdate_P1 = timestamp_adjusted.strftime("%Y-%m-%d")
    mdate_P2 = timestamp_adjusted.strftime("%Y-%m-%d %H:%M:%S")

    os.utime(self.newly_downloaded_track['file_path'], (timestamp_epoch, timestamp_epoch))

  def rename_move_downloaded_track(self):
    file_name_new = re.sub(r'[\\/*?:"<>|]', "", self.newly_downloaded_track['file_name_final'])

    file_path_old = self.newly_downloaded_track['file_path']
    file_path_new = os.path.join(self.local_playlist_folder, file_name_new + self.newly_downloaded_track['file_extension'])

    if not os.path.exists(self.local_playlist_folder):
      os.makedirs(self.local_playlist_folder)

    os.replace(file_path_old, file_path_new)

  def download_handler(self):
    for spotify_track in self.spotify_tracks_to_download:
      self.currently_downloading_track = spotify_track

      self.progress_bar_text = f"\nPlaylist: {self.playlist_name} (Total Songs: {len(self.spotify_tracks_raw)} | To Download: {len(self.spotify_tracks_to_download) - self.completed_index}) (Unavailable: {len(self.spotify_tracks_unavailable)}) | Downloading: {spotify_track['title']}\n"

      self.downloader()
      self.get_newly_downloaded_track()
      self.edit_metadata()
      self.edit_modification_date()
      self.rename_move_downloaded_track()

      self.completed_index += 1

    if not self.spotify_tracks_to_download:
      print(f"{self.playlist_name.ljust(30)} {Fore.GREEN + 'DOWNLOADED' + Style.RESET_ALL}")

    if self.spotify_tracks_to_download == self.completed_index and len(self.spotify_tracks_to_download) > 0:
      print(f"{self.playlist_name.ljust(30)} {Fore.GREEN + 'DOWNLOADED' + Fore.YELLOW + ' (Unavailable skipped)' + Style.RESET_ALL}")

  def reset_lists(self):
    self.spotify_tracks_raw = []
    self.spotify_tracks_already_downloaded = []
    self.spotify_tracks_incomplete = []
    self.spotify_tracks_uploaded = []
    self.spotify_tracks_unavailable = []
    self.spotify_tracks_deleted = []
    self.spotify_tracks_to_download = []

    self.local_tracks_raw = []
    self.local_tracks_already_downloaded = []
    self.local_tracks_duplicate = []
    self.local_tracks_extra = []
    self.local_tracks_uploaded = []

    self.completed_index = 0

  ######################################################

  def app_handler(self):
    self.__init__()
    self.create_session()
    init() # For colorama

    for _, playlist_url in self.playlists.items():
      self.playlist_url = playlist_url;
      self.get_playlist_id();
      self.get_playlist_name();
      self.local_playlist_folder = os.path.join(self.local_library_folder, self.playlist_name)

      self.update_window_title(f"Processing: {self.playlist_name}")
      self.get_spotify_tracks_raw()
      self.get_local_tracks_raw()
      self.create_lists()
      # return
      self.duplicate_proof_filename()

      # self.remove_local_tracks_duplicate()
      self.remove_local_tracks_extra()
      self.empty_local_temporary_download_folder()

      self.download_handler()
      self.reset_lists()

    self.update_window_title('Finished.')

# START

app = Container()
app.app_handler()

input('\nFinished.')

########################################################

# Notes:
# is_playable is 'true' on normal tracks, 'false' on unavailable tracks, and is not present on local tracks
# index was added to keep the compare process in sync, but not needed anymore
# sync might also be the reason for lambda sorting
# music_tag showed comment tag as twice
# append method in 'get_spotify_tracks_to_download' was causing infinte loop

########################################################