BastiTee · April 5, 2021 12:48
diff --git a/mp3_to_spotify.py b/mp3_to_spotify.py
 # -*- coding: utf-8 -*-
 """MP3 folder to Spotify.

 Creates a Spotify playlist with the first song of each album found in the
 provided directory or its children by the containing MP3s.
 That's the only way in Spotify to have some way of record crates by genre or
 theme.
 Identification happens by the ID3 tags. If multiple tracks of the same
 album are found amongst the MP3s then the subsequent files are ignored.

 Requires:
 - Python 3
 - https://pypi.org/project/click/
 - https://pypi.org/project/eyed3/
 - https://pypi.org/project/spotipy/

 """

 import asyncio
 from glob import glob
 from os import environ, path
 from re import sub
 from typing import Any, List, Optional, Tuple

 import click
 import eyed3
 import spotipy


 @click.command(name='mp3-to-spotify', help=__doc__)
 @click.option('--input-dir', '-i', metavar='DIR',
              help='MP3 source folder', required=True)
 @click.option('--playlist-title', '-p', metavar='NAME',
              help='Title of playlist', required=True)
 @click.option('--simulate-only', '-s',
              help='Do not create final playlist', is_flag=True)
 def main(
    input_dir: str,
    playlist_title: str,
    simulate_only: bool
 ) -> None:
    print('Setting up spotify link...')
    sp = __authenticate_spotify()

    print(f'Reading albums from directory {input_dir} ...')
    album_infos = asyncio.run(__read_albums(input_dir))

    print('Looking up albums on Spotify...')
    track_metas = asyncio.run(__search_albums(album_infos, sp))

    print(f' - Found {len(track_metas)}/{len(album_infos)} records at Spotify')

    print(f'Creating target playlist "{playlist_title}"...')
    if not simulate_only:
        playlist = sp.user_playlist_create(
            user=sp.me()['id'],
            name=playlist_title,
            public=False, collaborative=False,
            description='Auto-generated Playlist based on MP3 folder.'
        )
        target_pid = playlist['id']

    print('Adding tracks to playlist...')
    for track_meta in [
        # We need to batch as spotify allows 100 additions per call
        track_metas[i:i + 100] for i in range(0, len(track_metas), 100)
    ]:
        for track in track_meta:
            print(f' - {track[1]} "{track[2]}" ({track[3]})')
        if not simulate_only:
            sp.playlist_add_items(
                target_pid,
                [track[0] for track in track_meta]
            )


 async def __search_albums(
    album_infos: List[Tuple[str, str, str]],
    sp: spotipy.Spotify
 ) -> List[List[str]]:
    tasks = []
    for album_info in album_infos:
        tasks.append(__resolve_album(album_info, sp))
    results = list(filter(
        lambda x: x is not None,
        await asyncio.gather(*tasks)
    ))
    return results


 async def __resolve_album(
    album_info: Tuple[str, str, str],
    sp: spotipy.Spotify
 ) -> Optional[List[str]]:
    # Strict search
    query_strict = album_info[0].lower() + ' ' + album_info[1].lower()
    album = __search_album(query_strict, sp)
    if not album:
        # Loose search
        # - Remove '& the ...' parts of artists
        art = sub(r'&.*', '', album_info[0].lower()).strip()
        art = sub(r' and .*', '', art).strip()
        # - Remove '(ep)' or other suffixes
        alb = sub(r'\([^\)]+\)', '', album_info[1].lower()).strip()
        album = __search_album(f'{art} {alb}', sp)
    if not album:
        __print(f' - !!! Could not resolve album: {album_info}', 'yellow')
        return None
    album_track = sp.album_tracks(album['id'], limit=1)['items'][0]
    return [
        album_track['id'],
        album_info[0],
        album_track['name'],
        album_info[1]
    ]


 def __search_album(query: str, sp: spotipy.Spotify) -> Optional[Any]:
    result = sp.search(q=query, limit=1, type='album')
    if len(result['albums']['items']) == 0:
        print(f' - No search results for query: "{query}"')
        return None
    album = result['albums']['items'][0]
    return album


 async def __read_albums(input_dir: str) -> List[Tuple[str, str, str]]:
    tasks = []
    for mp3_path in glob(path.join(input_dir, '**', '*.mp3'), recursive=True):
        tasks.append(__extract_id3tags(mp3_path))
    results = sorted(
        list(set(await asyncio.gather(*tasks))), key=lambda x: x[0]
    )
    print(f' - Read {len(results)} different albums from disk')
    return results


 async def __extract_id3tags(mp3_path: str) -> Tuple[str, str, str]:
    mp3: eyed3.core.AudioFile = eyed3.load(mp3_path)
    return str(mp3.tag.artist), str(mp3.tag.album), str(mp3.tag.genre)


 def __authenticate_spotify() -> spotipy.Spotify:
    try:
        client = spotipy.Spotify(auth_manager=spotipy.SpotifyOAuth(
            client_id=environ.get('SPOTIPY_CLIENT_ID'),
            client_secret=environ.get('SPOTIPY_CLIENT_SECRET'),
            redirect_uri='http://localhost:8000',
            scope='playlist-modify-private'
        ))
        # Make sure searching is possible
        client.search(q='search', limit=1)
        # Check user authentication
        user_name = client.me()['display_name']
        user_id = client.me()['id']
        print(f' - Logged in as {user_name} ({user_id})')
        return client
    except spotipy.SpotifyOauthError as ex:
        __print('Could not login to Spotify. Bad credentials?', 'red')
        print(ex)
        exit(1)


 def __print(msg: str, color: str) -> None:
    click.echo(click.style(msg, fg=color, bold=True))
	# -- coding: utf-8 --
	"""MP3 folder to Spotify.

	Creates a Spotify playlist with the first song of each album found in the
	provided directory or its children by the containing MP3s.
	That's the only way in Spotify to have some way of record crates by genre or
	theme.
	Identification happens by the ID3 tags. If multiple tracks of the same
	album are found amongst the MP3s then the subsequent files are ignored.

	Requires:
	- Python 3
	- https://pypi.org/project/click/
	- https://pypi.org/project/eyed3/
	- https://pypi.org/project/spotipy/

	"""

	import asyncio
	from glob import glob
	from os import environ, path
	from re import sub
	from typing import Any, List, Optional, Tuple

	import click
	import eyed3
	import spotipy


	@click.command(name='mp3-to-spotify', help=__doc__)
	@click.option('--input-dir', '-i', metavar='DIR',
	help='MP3 source folder', required=True)
	@click.option('--playlist-title', '-p', metavar='NAME',
	help='Title of playlist', required=True)
	@click.option('--simulate-only', '-s',
	help='Do not create final playlist', is_flag=True)
	def main(
	input_dir: str,
	playlist_title: str,
	simulate_only: bool
	) -> None:
	print('Setting up spotify link...')
	sp = __authenticate_spotify()

	print(f'Reading albums from directory {input_dir} ...')
	album_infos = asyncio.run(__read_albums(input_dir))

	print('Looking up albums on Spotify...')
	track_metas = asyncio.run(__search_albums(album_infos, sp))

	print(f' - Found {len(track_metas)}/{len(album_infos)} records at Spotify')

	print(f'Creating target playlist "{playlist_title}"...')
	if not simulate_only:
	playlist = sp.user_playlist_create(
	user=sp.me()['id'],
	name=playlist_title,
	public=False, collaborative=False,
	description='Auto-generated Playlist based on MP3 folder.'
	)
	target_pid = playlist['id']

	print('Adding tracks to playlist...')
	for track_meta in [
	# We need to batch as spotify allows 100 additions per call
	track_metas[i:i + 100] for i in range(0, len(track_metas), 100)
	]:
	for track in track_meta:
	print(f' - {track[1]} "{track[2]}" ({track[3]})')
	if not simulate_only:
	sp.playlist_add_items(
	target_pid,
	[track[0] for track in track_meta]
	)


	async def __search_albums(
	album_infos: List[Tuple[str, str, str]],
	sp: spotipy.Spotify
	) -> List[List[str]]:
	tasks = []
	for album_info in album_infos:
	tasks.append(__resolve_album(album_info, sp))
	results = list(filter(
	lambda x: x is not None,
	await asyncio.gather(*tasks)
	))
	return results


	async def __resolve_album(
	album_info: Tuple[str, str, str],
	sp: spotipy.Spotify
	) -> Optional[List[str]]:
	# Strict search
	query_strict = album_info[0].lower() + ' ' + album_info[1].lower()
	album = __search_album(query_strict, sp)
	if not album:
	# Loose search
	# - Remove '& the ...' parts of artists
	art = sub(r'&.*', '', album_info[0].lower()).strip()
	art = sub(r' and .*', '', art).strip()
	# - Remove '(ep)' or other suffixes
	alb = sub(r'\([^\)]+\)', '', album_info[1].lower()).strip()
	album = __search_album(f'{art} {alb}', sp)
	if not album:
	__print(f' - !!! Could not resolve album: {album_info}', 'yellow')
	return None
	album_track = sp.album_tracks(album['id'], limit=1)['items'][0]
	return [
	album_track['id'],
	album_info[0],
	album_track['name'],
	album_info[1]
	]


	def __search_album(query: str, sp: spotipy.Spotify) -> Optional[Any]:
	result = sp.search(q=query, limit=1, type='album')
	if len(result['albums']['items']) == 0:
	print(f' - No search results for query: "{query}"')
	return None
	album = result['albums']['items'][0]
	return album


	async def __read_albums(input_dir: str) -> List[Tuple[str, str, str]]:
	tasks = []
	for mp3_path in glob(path.join(input_dir, '*', '.mp3'), recursive=True):
	tasks.append(__extract_id3tags(mp3_path))
	results = sorted(
	list(set(await asyncio.gather(*tasks))), key=lambda x: x[0]
	)
	print(f' - Read {len(results)} different albums from disk')
	return results


	async def __extract_id3tags(mp3_path: str) -> Tuple[str, str, str]:
	mp3: eyed3.core.AudioFile = eyed3.load(mp3_path)
	return str(mp3.tag.artist), str(mp3.tag.album), str(mp3.tag.genre)


	def __authenticate_spotify() -> spotipy.Spotify:
	try:
	client = spotipy.Spotify(auth_manager=spotipy.SpotifyOAuth(
	client_id=environ.get('SPOTIPY_CLIENT_ID'),
	client_secret=environ.get('SPOTIPY_CLIENT_SECRET'),
	redirect_uri='http://localhost:8000',
	scope='playlist-modify-private'
	))
	# Make sure searching is possible
	client.search(q='search', limit=1)
	# Check user authentication
	user_name = client.me()['display_name']
	user_id = client.me()['id']
	print(f' - Logged in as {user_name} ({user_id})')
	return client
	except spotipy.SpotifyOauthError as ex:
	__print('Could not login to Spotify. Bad credentials?', 'red')
	print(ex)
	exit(1)


	def __print(msg: str, color: str) -> None:
	click.echo(click.style(msg, fg=color, bold=True))