Import local M3U file list as YouTube playlist

yt_playlist_import.py

"""
Import local M3U playlists into YouTube as playlist.

The script accepts one argument, path to an M3U playlist, such as one produces with:

    find mp3/genre/directory -type f  >  My_genre_name_playlist_name.m3u

It then searches for the best video match to each item in the playlist
and adds it to a newly created PRIVATE YouTube playlist bearing the
M3U playlist file name.

You need to provide additional client_id.json from Google API console. See below.
The script asks for Oath2 authorization every time before execution.
Improvements welcome!

The script is NOT fault tolerant.
"""
import os
import argparse
import re

from googleapiclient.discovery import build
from google_auth_oauthlib.flow import InstalledAppFlow

# Contains OAuth2 info for this app
# https://developers.google.com/youtube/v3/quickstart/python#step_1_turn_on_the_api_name
# https://console.developers.google.com/apis/credentials
CLIENT_SECRETS_FILE = "client_id.json"


def main():
    args = parse_args()
    print(args)

    yt = YouTube()

    for playlist_name, queries in parse_playlists(args.m3u_file):
        print(playlist_name)
        playlist_id = yt.playlists_insert(
            dict(snippet=dict(title=playlist_name,
                              description='Autogenerated by yt_playlist_import.py\n\nhttps://gist.github.com/kernc/ba931731add1f1485e211c3d1c67c1c2'),
                 status=dict(privacyStatus='private'))
        )['id']
        print(playlist_id)
        print()
        for query in queries:
            print(query, end='\t')
            try:
                item = yt.search_list_by_keyword(query)['items'][0]
            except:
                print('%%%%')
                continue

            video_id = item['id']['videoId']
            print(video_id)

            item = yt.playlist_items_insert(
                dict(snippet=dict(playlistId=playlist_id,
                                  resourceId=dict(kind='youtube#video',
                                                  videoId=video_id)))
            )
            print(item['id'])


def parse_args():
    p = argparse.ArgumentParser()
    p.add_argument('m3u_file', nargs='+')
    args = p.parse_args()
    return args


def parse_playlists(pls_files):
    for filename in pls_files:
        with open(filename) as f:
            items = f.read().split('\n')
        yield playlist_name(filename), list(queries_from_items(items))


def playlist_name(filename):
    name, _ext = os.path.splitext(os.path.basename(filename.replace('_', ' ')))
    return name


def queries_from_items(items):
    for item in items:
        item = item.strip()
        if item.startswith('#'):
            # m3u Comment
            continue

        name, ext = os.path.splitext(item)
        name = os.path.basename(name)

        query = name.replace(' - from YouTube by Offliberty', '')
        query = query.replace('-', ' ')
        query = query.replace('.', ' ')
        query = query + get_ytid(name, ext)

        query = re.sub('\s+', ' ', query)
        query = query.strip()
        if query:
            yield query


YT_EXTENSIONS = ('.opus', '.m4a')


def get_ytid(name, ext, _ytid=re.compile('\S-([\w]{11})$').search):
    if ext in YT_EXTENSIONS:
        try:
            return ' id:' + _ytid(name).group(1)
        except AttributeError:
            pass
    return ''


class YouTube:
    SCOPES = ['https://www.googleapis.com/auth/youtube.force-ssl']
    API_SERVICE_NAME = 'youtube'
    API_VERSION = 'v3'

    def __init__(self):
        self.client = self.get_authenticated_service()

    def get_authenticated_service(self):
        flow = InstalledAppFlow.from_client_secrets_file(CLIENT_SECRETS_FILE, self.SCOPES)
        credentials = flow.run_console()
        return build(self.API_SERVICE_NAME, self.API_VERSION, credentials=credentials)

    def playlists_insert(self, properties):
        # https://developers.google.com/youtube/v3/docs/playlists/insert
        response = self.client.playlists().insert(
            part=','.join(properties.keys()),
            body=properties,
        ).execute()
        return response

    def search_list_by_keyword(self, query):
        # https://developers.google.com/youtube/v3/docs/search/list
        response = self.client.search().list(
            part='snippet',
            type='video',
            maxResults=1,
            q=query,
        ).execute()
        return response

    def playlist_items_insert(self, properties):
        # https://developers.google.com/youtube/v3/docs/playlistItems/insert
        response = self.client.playlistItems().insert(
            part=','.join(properties.keys()),
            body=properties,
        ).execute()
        return response


if __name__ == '__main__':
    main()