apple_to_google_music.py

from argparse import ArgumentParser
from gmusicapi import Mobileclient
from multiprocessing import Pool, Manager, cpu_count
from time import sleep
from re import sub as re_sub


class bcolors:
    OKBLUE = "\033[94m"
    OKGREEN = "\033[92m"
    FAIL = "\033[91m"
    ENDC = "\033[0m"


REDUNDANT_PARTS = "\(Original Motion Picture Soundtrack\)|-? Single|-? EP"
AUTOGENERATED_AM_TXT_LINE = ["music", "", "", ""]

def _get_args():
    parser = ArgumentParser()
    required = parser.add_argument_group("required arguments")
    required.add_argument("-e", "--email", metavar="\b", required=True,
                          help="Google Play Music email.")
    required.add_argument("-p", "--password", metavar="\b", required=True,
                          help="Generated app password. See: https://support.google.com/mail/answer/185833?hl=en")
    required.add_argument("-f", "--file", metavar="\b", required=True,
                          help="Path to Apple Music playlist file in txt format.")
    return parser.parse_args()


def _get_songs_count_from_file(path):
    try:
        with open(path) as file:
            return sum(1 for line in file) - 1
    except Exception as error:
        print(error)


def _search_tracks(line, with_artist=True):
    columns = line.split("\t")
    artist = columns[1] if with_artist else ""
    album = re_sub(REDUNDANT_PARTS, "", columns[3])
    query = "%s %s %s" % (columns[0], artist, album)

    try:
        track_id = api.search(query)["song_hits"][0]["track"]["storeId"]
        print(bcolors.OKGREEN + query + bcolors.ENDC)
        return track_id
    except Exception:
        if with_artist:
            sleep(3)
            # Sometimes songs can be found only without artist name
            _search_tracks(line, False)
        else:
            print(bcolors.FAIL + query + bcolors.ENDC)
            ignored_tracks.append("%s %s %s" %
                                  (columns[0], columns[1], columns[3]))


def _add_track_to_gm(track_id, tried=False):
    try:
        api.add_store_tracks([track_id])
    except Exception as error:
        if not tried:
            _add_track_to_gm(track_id, True)
        else:
            print(error)


def _print_message():
    print("\n--------------------------------------------------------\n\n")
    print(bcolors.OKGREEN + "Successfully connected to your account." +
          bcolors.ENDC)
    print("\n\nBelow you will see track full names in different colors. " +
          "Here's explanations:")
    print(bcolors.OKGREEN + "Track was found on Google Music " +
          "and will be added to your library." + bcolors.ENDC)
    print(bcolors.FAIL + "Track was not found on Google Music " +
          "and will be added to 'ignored_tracks.txt' file" + bcolors.ENDC)
    print("\n\n--------------------------------------------------------\n")
    print("\nStarting process...\n\n\n")


def _get_actual_songs_lines_from(lines):
    if lines[-1].split("\t")[0:4] == AUTOGENERATED_AM_TXT_LINE:
        return lines[1:-1]
    else:
        return lines[1:]


def _init_global():
    global api, ignored_tracks, lock
    api = Mobileclient()
    ignored_tracks = Manager().list([])


def main():
    args = _get_args()
    am_songs_count = _get_songs_count_from_file(args.file)
    if am_songs_count:
        print("\n\n--------------------------------------------------------\n")
        print(bcolors.OKBLUE +
              "Found %d songs in passed file" % am_songs_count + bcolors.ENDC)
        print("\n--------------------------------------------------------\n\n")
        _init_global()
        if api.login(args.email, args.password, Mobileclient.FROM_MAC_ADDRESS):
            _print_message()
            gm_track_ids = []
            with open(args.file) as file:
                pool = Pool(processes=cpu_count() + 1)
                lines = _get_actual_songs_lines_from(file.readlines())
                gm_track_ids = pool.map(_search_tracks, lines)
                pool.close()
                pool.join()

                gm_track_ids = [i for i in gm_track_ids if i is not None]

                if ignored_tracks:
                    with open("ignored_tracks.txt", "w") as ignored_list:
                        ignored_list.write("\n".join(ignored_tracks))

                if gm_track_ids:
                    print(bcolors.OKBLUE + "Found %d tracks from %d in Google Music" %
                          (len(gm_track_ids), am_songs_count) + bcolors.ENDC)
                    print("Adding them...")
                    pool = Pool(processes=cpu_count() + 1)
                    pool.map(_add_track_to_gm, gm_track_ids)
                    pool.close()
                    pool.join()
                    print(bcolors.OKGREEN + "Done!" + bcolors.ENDC)
        else:
            print(bcolors.FAIL + "Could not connect to your account. " +
                  "Please check credentials." + bcolors.ENDC)

if __name__ == "__main__":
    main()