Accept playlist names as CLI arguments

This commit is contained in:
Maks Snegov 2024-02-02 10:27:22 -08:00
parent 056a40684d
commit 6f4cdc66c7
2 changed files with 40 additions and 7 deletions

1
.gitignore vendored
View File

@ -2,3 +2,4 @@
.python-version .python-version
client_secrets_file.json client_secrets_file.json
token.json token.json
playlists.csv

46
main.py
View File

@ -1,5 +1,6 @@
#!/usr/bin/env python #!/usr/bin/env python
import argparse import argparse
import csv
import os import os
import sys import sys
@ -9,6 +10,8 @@ from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build from googleapiclient.discovery import build
from googleapiclient.errors import HttpError from googleapiclient.errors import HttpError
_playlists = {}
def get_yt_creds(): def get_yt_creds():
""" Get YouTube API credentials """ """ Get YouTube API credentials """
@ -38,7 +41,31 @@ def get_yt_creds():
return creds return creds
def read_playlists_file():
""" Read playlists.csv and return a dictionary of playlist names to playlist IDs """
global _playlists
if not os.path.exists('playlists.csv'):
print('playlists.csv not found')
return {}
with open('playlists.csv', newline='') as csvfile:
reader = csv.DictReader(csvfile)
_playlists = {row['name']: row['playlist_id'] for row in reader}
def get_playlist_id(playlist_name: str) -> str:
if not _playlists:
read_playlists_file()
return _playlists.get(playlist_name, playlist_name)
def get_playlist_name(playlist_id: str) -> str:
if not _playlists:
read_playlists_file()
return next((name for name, id in _playlists.items() if id == playlist_id), playlist_id)
def get_videos(yt_api, playlist_id): def get_videos(yt_api, playlist_id):
playlist_name = get_playlist_name(playlist_id)
videos = [] videos = []
fetched = 0 fetched = 0
overall = -1 overall = -1
@ -58,13 +85,14 @@ def get_videos(yt_api, playlist_id):
for item in response['items']: for item in response['items']:
videos.append(item) videos.append(item)
except HttpError as e: except HttpError as e:
print(f'Error getting video IDs: {e}') print(f'Error getting video IDs from playlist {playlist_name}: {e}')
print(f'Fetched {fetched} videos from playlist {playlist_id}') print(f'Fetched {fetched} videos from playlist {playlist_name}')
return videos return videos
def add_video_to_playlist(yt_api, video, playlist_id) -> bool: def add_video_to_playlist(yt_api, video, playlist_id) -> bool:
playlist_name = get_playlist_name(playlist_id)
video_id = video['snippet']['resourceId']['videoId'] video_id = video['snippet']['resourceId']['videoId']
try: try:
yt_api.playlistItems().insert( yt_api.playlistItems().insert(
@ -79,23 +107,24 @@ def add_video_to_playlist(yt_api, video, playlist_id) -> bool:
} }
} }
).execute() ).execute()
print(f'Added video {video_id} to playlist {playlist_id}') print(f'Added video {video_id} to playlist {playlist_name}')
return True return True
except HttpError as e: except HttpError as e:
print(f'Error adding video {video_id} to playlist {playlist_id}: {e}') print(f'Error adding video {video_id} to playlist {playlist_name}: {e}')
return False return False
def remove_video_from_playlist(yt_api, video, playlist_id) -> bool: def remove_video_from_playlist(yt_api, video, playlist_id) -> bool:
playlist_name = get_playlist_name(playlist_id)
video_id = video['snippet']['resourceId']['videoId'] video_id = video['snippet']['resourceId']['videoId']
try: try:
yt_api.playlistItems().delete( yt_api.playlistItems().delete(
id=video['id'] id=video['id']
).execute() ).execute()
print(f'Removed video {video_id} from playlist {playlist_id}') print(f'Removed video {video_id} from playlist {playlist_name}')
return True return True
except HttpError as e: except HttpError as e:
print(f'Error removing video {video_id} from playlist {playlist_id}: {e}') print(f'Error removing video {video_id} from playlist {playlist_name}: {e}')
return False return False
@ -129,13 +158,16 @@ def main():
# *DO NOT* leave this option enabled in production. # *DO NOT* leave this option enabled in production.
os.environ["OAUTHLIB_INSECURE_TRANSPORT"] = "1" os.environ["OAUTHLIB_INSECURE_TRANSPORT"] = "1"
src_playlist = get_playlist_id(args.src_playlist)
dst_playlist = get_playlist_id(args.dst_playlist)
api_service_name = "youtube" api_service_name = "youtube"
api_version = "v3" api_version = "v3"
creds = get_yt_creds() creds = get_yt_creds()
youtube = build(api_service_name, api_version, credentials=creds) youtube = build(api_service_name, api_version, credentials=creds)
move_all_videos(youtube, args.src_playlist, args.dst_playlist, limit=args.limit) move_all_videos(youtube, src_playlist, dst_playlist, limit=args.limit)
if __name__ == '__main__': if __name__ == '__main__':