Imports YouTube Takeout playlists CSVs in Python

import_youtube_takeout.py

#!/usr/bin/env python
#
# Copyright (c) 2023 Adam Strzelecki
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

"""Imports playlists from YouTube Takeout's CSV files"""

# pip install google-api-python-client google_auth_oauthlib
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError

# https://developers.google.com/youtube/v3/getting-started
# https://googleapis.github.io/google-api-python-client/docs/epy/googleapiclient-module.html
SERVICE = 'youtube'
VERSION = 'v3'
REDIRECT_URI = 'http://localhost:8080/' # probably not needed
# obtain Desktop App OAuth from https://console.cloud.google.com/apis/dashboard
SECRET = 'client_secret.apps.googleusercontent.com.json'
SCOPES = [
    'https://www.googleapis.com/auth/youtube',
    'https://www.googleapis.com/auth/youtube.force-ssl',
    'https://www.googleapis.com/auth/youtubepartner'
]

import os
os.environ['OAUTHLIB_INSECURE_TRANSPORT'] = '1' # workaround for force-ssl


def credentials(secret):
    import os.path
    import glob
    cred = None
    token_file = 'token_' + secret
    if os.path.exists(token_file):
        cred = Credentials.from_authorized_user_file(token_file, SCOPES)
    if not cred or not cred.valid:
        if cred and cred.expired and cred.refresh_token:
            cred.refresh(Request())
        else:
            flow = InstalledAppFlow.from_client_secrets_file(secret, SCOPES, redirect_uri=REDIRECT_URI)
            cred = flow.run_local_server(port=0)
        with open(token_file, 'w') as token:
            token.write(cred.to_json())
    return cred

def playlists(service):
    pls = {}
    for item in service.playlists().list(part='snippet,status', mine=True).execute()['items']:
        pls[item['snippet']['title']] = item['id']
    return pls

def create_playlist(service, title):
    return service.playlists().insert(part='snippet,status', body={
        'snippet': {
            'title': title,
        },
        'status': {
            'privacyStatus': 'private',
        },
    }).execute()

def add_to_playlist(service, playlist_id, video_id, published_at):
    try:
        service.playlistItems().insert(part='snippet', body={
            'snippet': {
                'playlistId': playlist_id,
                'resourceId': {
                    'kind': 'youtube#video',
                    'videoId': video_id,
                    'publishedAt': published_at,
                    # 'position': 0
                },
            },
        }).execute()
    except HttpError as err:
        if err.resp.status == 404:
            print(err)
        else:
            raise

def list_playlist(service, playlist_id):
    res = service.playlistItems().list(part='snippet',
                                       maxResults=50,
                                       playlistId=playlist_id).execute()
    ids = []
    for item in res['items']:
        ids.append(item['snippet']['resourceId']['videoId'])
    while 'nextPageToken' in res:
        res = service.playlistItems().list(part='snippet',
                                           maxResults=50,
                                           pageToken=res['nextPageToken'],
                                           playlistId=playlist_id).execute()
        for item in res['items']:
            ids.append(item['snippet']['resourceId']['videoId'])
    return ids

def main(skip_nonexisting=True, skip_existing=True):
    import glob
    import os.path
    import csv

    cred = credentials(SECRET)
    service = build(SERVICE, VERSION, credentials=cred)
    pls = playlists(service)
    print('Playlists', pls)
    incomplete_pls = ['DIY']
    for fn in glob.glob(os.path.join('playlists', '*.csv')):
        print('---')
        n = os.path.splitext(os.path.basename(fn))[0]
        if n.startswith('Uploads from') or n.startswith('Watch later'):
            continue
        video_ids = []
        if n in pls:
            if skip_existing:
                print('Skipping', n, 'that already exists')
                continue
            print('Loading', n, 'that already exists')
            pid = pls[n]
            video_ids = list_playlist(service, pid)
        else:
            if skip_nonexisting:
                print('Skipping', n, 'that does not exist yet')
                continue
            print('Creating', n, 'that does not exist yet')
            pid = create_playlist(service, title=n)['id']
        print('Playlist', n, 'file', fn)
        with open(fn, newline='') as f:
            c = csv.reader(f)
            line = 0
            for row in c:
                if line < 4:
                    line += 1
                    continue
                if len(row) == 0:
                    continue
                video_id, published_at = row[0], row[1]
                if video_id in video_ids:
                    print('- video', video_id, 'from', published_at)
                else:
                    print('+ video', video_id, 'from', published_at)
                    add_to_playlist(service, pid, video_id, published_at)


if __name__ == '__main__':
    main(skip_nonexisting=False, skip_existing=False)