Added sorting

[djmango]

Updated the script to use sorting and DictReader. Not clean code but works. Was considering making async too for speed, this has been running for 3 days lol. But i have like 10k songs so yk. This sorts by date, oldest first added in playlist.

from sys import argv
import csv
import urllib.parse, urllib.request
import json
from time import sleep
import requests
import os

# Checking if the command is correct
if len(argv) > 1 and argv[1]:
    pass
else:
    print(
        "\nCommand usage:\npython3 convertsongs.py yourplaylist.csv\nMore info at https://github.com/therealmarius/Spotify-2-AppleMusic"
    )
    exit()


# Function to get contents of file if it exists
def get_connection_data(f, prompt):
    if os.path.exists(f):
        with open(f, "r") as file:
            return file.read().rstrip("\n")
    else:
        return input(prompt)


def create_apple_music_playlist(session, playlist_name):
    url = "https://amp-api.music.apple.com/v1/me/library/playlists"
    data = {
        "attributes": {
            "name": playlist_name,
            "description": "A new playlist created via API",
        }
    }
    # test if playlist exists and create it if not
    response = session.get(url)
    if response.status_code == 200:
        for playlist in response.json()["data"]:
            if playlist["attributes"]["name"] == playlist_name:
                print(f"Playlist {playlist_name} already exists!")
                return playlist["id"]
    response = session.post(url, json=data)
    if response.status_code == 201:
        sleep(1.5)
        return response.json()["data"][0]["id"]
    else:
        raise Exception(
            f"Error {response.status_code} while creating playlist {playlist_name}!"
        )
        return None


# Getting user's data for the connection
token = get_connection_data(
    "token.dat", "\nPlease enter your Apple Music Authorization (Bearer token):\n"
)
media_user_token = get_connection_data(
    "media_user_token.dat", "\nPlease enter your media user token:\n"
)
cookies = get_connection_data("cookies.dat", "\nPlease enter your cookies:\n")

# playlist_identifier = input("\nPlease enter the playlist identifier:\n")


# function to escape apostrophes
def escape_apostrophes(s):
    return s.replace("'", "\\'")


# Function to get the iTunes ID of a song
def get_itunes_id(title, artist, album):
    BASE_URL = "https://itunes.apple.com/search?country=FR&media=music&entity=song&limit=5&term="
    # Search the iTunes catalog for a song
    try:
        # Search for the title + artist + album
        url = BASE_URL + urllib.parse.quote(title + " " + artist + " " + album)
        request = urllib.request.Request(url)
        response = urllib.request.urlopen(request)
        data = json.loads(response.read().decode("utf-8"))
        # If no result, search for the title + artist
        if data["resultCount"] == 0:
            url = BASE_URL + urllib.parse.quote(title + " " + artist)
            request = urllib.request.Request(url)
            response = urllib.request.urlopen(request)
            data = json.loads(response.read().decode("utf-8"))
            # If no result, search for the title + album
            if data["resultCount"] == 0:
                url = BASE_URL + urllib.parse.quote(title + " " + album)
                request = urllib.request.Request(url)
                response = urllib.request.urlopen(request)
                data = json.loads(response.read().decode("utf-8"))
                # If no result, search for the title
                if data["resultCount"] == 0:
                    url = BASE_URL + urllib.parse.quote(title)
                    request = urllib.request.Request(url)
                    response = urllib.request.urlopen(request)
                    data = json.loads(response.read().decode("utf-8"))
    except:
        return print("An error occured with the request.")

    # Try to match the song with the results
    try:
        response = urllib.request.urlopen(request)
        data = json.loads(response.read().decode("utf-8"))

        for each in data["results"]:
            # Trying to match with the exact track name, the artist name and the album name
            if (
                each["trackName"].lower() == title.lower()
                and each["artistName"].lower() == artist.lower()
                and each["collectionName"].lower() == album.lower()
            ):
                return each["trackId"]
            # Trying to match with the exact track name and the artist name
            elif (
                each["trackName"].lower() == title.lower()
                and each["artistName"].lower() == artist.lower()
            ):
                return each["trackId"]
            # Trying to match with the exact track name and the album name
            elif (
                each["trackName"].lower() == title.lower()
                and each["collectionName"].lower() == album.lower()
            ):
                return each["trackId"]
            # Trying to match with the exact track name and the artist name, in the case artist name are different between Spotify and Apple Music
            elif each["trackName"].lower() == title.lower() and (
                each["artistName"].lower() in artist.lower()
                or artist.lower() in each["artistName"].lower()
            ):
                return each["trackId"]
            # Trying to match with the exact track name and the album name, in the case album name are different between Spotify and Apple Music
            elif each["trackName"].lower() == title.lower() and (
                each["collectionName"].lower() in album.lower()
                or album.lower() in each["collectionName"].lower()
            ):
                return each["trackId"]
            # Trying to match with the exact track name
            elif each["trackName"].lower() == title.lower():
                return each["trackId"]
            # Trying to match with the track name, in the case track name are different between Spotify and Apple Music
            elif (
                title.lower() in each["trackName"]
                or each["trackName"].lower() in title.lower()
            ):
                return each["trackId"]
        try:
            # If no result, return the first result
            return data["results"][0]["trackId"]
        except:
            # If no result, return None
            return None
    except:
        # The error is handled later in the code
        return None


# Function to add a song to a playlist
def add_song_to_playlist(session, song_id, playlist_id, playlist_name):
    try:
        request = session.post(
            f"https://amp-api.music.apple.com/v1/me/library/playlists/{playlist_id}/tracks",
            json={"data": [{"id": f"{song_id}", "type": "songs"}]},
        )
        # Checking if the request is successful
        if requests.codes.ok:
            print(f"Song {song_id} added to playlist {playlist_name}!")
            return True
        # If not, print the error code
        else:
            print(
                f"Error {request.status_code} while adding song {song_id} to playlist {playlist_name}!"
            )
            return False
    except:
        print(
            f"HOST ERROR: Apple Music might have blocked the connection during the add of {song_id} to playlist {playlist_name}!\nPlease wait a few minutes and try again.\nIf the problem persists, please contact the developer."
        )
        return False


def get_playlist_track_ids(session, playlist_id):
    # test if song is already in playlist
    try:
        response = session.get(
            f"https://amp-api.music.apple.com/v1/me/library/playlists/{playlist_id}/tracks"
        )
        if response.status_code == 200:
            # print(response.json()['data'])
            return [
                track["attributes"]["playParams"]["catalogId"]
                for track in response.json()["data"]
            ]
        elif response.status_code == 404:
            return []
        else:
            raise Exception(
                f"Error {response.status_code} while getting playlist {playlist_id}!"
            )
            return None
    except:
        raise Exception(f"Error while getting playlist {playlist_id}!")
        return None


from datetime import datetime


def parse_date(date_string):
    return datetime.strptime(date_string, "%Y-%m-%dT%H:%M:%SZ")


# Opening session
def create_playlist_and_add_song(file):
    with requests.Session() as s:
        s.headers.update(
            {
                "Authorization": f"{token}",
                "media-user-token": f"{media_user_token}",
                "Cookie": f"{cookies}".encode("utf-8"),
                "Host": "amp-api.music.apple.com",
                "Accept-Encoding": "gzip, deflate, br",
                "Referer": "https://music.apple.com/",
                "Origin": "https://music.apple.com",
                "Content-Length": "45",
                "Connection": "keep-alive",
                "Sec-Fetch-Dest": "empty",
                "Sec-Fetch-Mode": "cors",
                "Sec-Fetch-Site": "same-site",
                "TE": "trailers",
            }
        )

    # Getting the playlist name
    playlist_name = os.path.basename(file)
    playlist_name = playlist_name.split(".")
    playlist_name = playlist_name[0]
    playlist_name = playlist_name.replace("_", " ")

    playlist_identifier = create_apple_music_playlist(s, playlist_name)

    playlist_track_ids = get_playlist_track_ids(s, playlist_identifier)
    print(playlist_track_ids)
    # Opening the inputed CSV file
    with open(str(file), encoding="utf-8") as file:
        data = list(csv.DictReader(file))

    data.sort(key=lambda row: parse_date(row["Added At"]))
    converted = 0
    failed = 0
    n = 0
    for n, row in enumerate(data):
        # Trying to get the iTunes ID of the song
        title, artist, album = (
            escape_apostrophes(row["Track Name"]),
            escape_apostrophes(row["Artist Name(s)"]),
            escape_apostrophes(row["Album Name"]),
        )
        track_id = get_itunes_id(title, artist, album)
        # If the song is found, add it to the playlist
        if track_id:
            if str(track_id) in playlist_track_ids:
                print(f"\nN°{n} | {title} | {artist} | {album} => {track_id}")
                print(f"Song {track_id} already in playlist {playlist_name}!")
                failed += 1
                continue
            print(f"\nN°{n} | {title} | {artist} | {album} => {track_id}")
            sleep(0.5)
            if add_song_to_playlist(s, track_id, playlist_identifier, playlist_name):
                converted += 1
            else:
                failed += 1
        # If not, write it in a file
        else:
            print(f"N°{n} | {title} | {artist} | {album} => NOT FOUND")
            with open(f"{playlist_name}_noresult.txt", "a+", encoding="utf-8") as f:
                f.write(f"{title} | {artist} | {album} => NOT FOUND")
                f.write("\n")
            failed += 1
        sleep(1.5)

    # Printing the stats report
    converted_percentage = round(converted / n * 100) if n > 0 else 100
    print(
        f"\n - STAT REPORT -\nPlaylist Songs: {n}\nConverted Songs: {converted}\nFailed Songs: {failed}\nPlaylist converted at {converted_percentage}%"
    )


if __name__ == "__main__":
    if len(argv) > 1 and argv[1]:
        if ".csv" in argv[1]:
            create_playlist_and_add_song(argv[1])
        else:
            # get all csv files in the directory argv[1]
            files = [
                f
                for f in os.listdir(argv[1])
                if os.path.isfile(os.path.join(argv[1], f))
            ]
            # loop through all csv files
            for file in files:
                if ".csv" in file:
                    create_playlist_and_add_song(os.path.join(argv[1], file))

# Developped by @therealmarius on GitHub
# Based on the work of @simonschellaert on GitHub
# Github project page: https://github.com/therealmarius/Spotify-2-AppleMusic