143 lines
4.7 KiB
Python
Executable File
143 lines
4.7 KiB
Python
Executable File
#!/usr/bin/env python
|
|
import argparse
|
|
import os
|
|
import sys
|
|
|
|
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
|
|
|
|
|
|
def get_yt_creds():
|
|
""" Get YouTube API credentials """
|
|
creds = None
|
|
client_secrets_file = "client_secrets_file.json"
|
|
scopes = ["https://www.googleapis.com/auth/youtube.force-ssl"]
|
|
|
|
# The file token.json stores the user's access and refresh tokens, and is
|
|
# created automatically when the authorization flow completes for the first
|
|
# time.
|
|
if os.path.exists("token.json"):
|
|
creds = Credentials.from_authorized_user_file("token.json", scopes)
|
|
|
|
# If there are no (valid) credentials available, let the user log in.
|
|
if not creds or not creds.valid:
|
|
if creds and creds.expired and creds.refresh_token:
|
|
creds.refresh(Request())
|
|
else:
|
|
flow = InstalledAppFlow.from_client_secrets_file(
|
|
client_secrets_file, scopes
|
|
)
|
|
creds = flow.run_local_server(port=0)
|
|
# Save the credentials for the next run
|
|
with open("token.json", "w") as token:
|
|
token.write(creds.to_json())
|
|
|
|
return creds
|
|
|
|
|
|
def get_videos(yt_api, playlist_id):
|
|
videos = []
|
|
fetched = 0
|
|
overall = -1
|
|
page_token = ""
|
|
try:
|
|
while fetched < overall or overall == -1:
|
|
response = yt_api.playlistItems().list(
|
|
part="snippet,contentDetails",
|
|
maxResults=50,
|
|
playlistId=playlist_id,
|
|
pageToken=page_token
|
|
).execute()
|
|
if overall == -1:
|
|
overall = response['pageInfo']['totalResults']
|
|
fetched += len(response['items'])
|
|
page_token = response.get('nextPageToken', "")
|
|
for item in response['items']:
|
|
videos.append(item)
|
|
except HttpError as e:
|
|
print(f'Error getting video IDs: {e}')
|
|
|
|
print(f'Fetched {fetched} videos from playlist {playlist_id}')
|
|
return videos
|
|
|
|
|
|
def add_video_to_playlist(yt_api, video, playlist_id) -> bool:
|
|
video_id = video['snippet']['resourceId']['videoId']
|
|
try:
|
|
yt_api.playlistItems().insert(
|
|
part='snippet',
|
|
body={
|
|
'snippet': {
|
|
'playlistId': playlist_id,
|
|
'resourceId': {
|
|
'kind': 'youtube#video',
|
|
'videoId': video_id
|
|
}
|
|
}
|
|
}
|
|
).execute()
|
|
print(f'Added video {video_id} to playlist {playlist_id}')
|
|
return True
|
|
except HttpError as e:
|
|
print(f'Error adding video {video_id} to playlist {playlist_id}: {e}')
|
|
return False
|
|
|
|
|
|
def remove_video_from_playlist(yt_api, video, playlist_id) -> bool:
|
|
video_id = video['snippet']['resourceId']['videoId']
|
|
try:
|
|
yt_api.playlistItems().delete(
|
|
id=video['id']
|
|
).execute()
|
|
print(f'Removed video {video_id} from playlist {playlist_id}')
|
|
return True
|
|
except HttpError as e:
|
|
print(f'Error removing video {video_id} from playlist {playlist_id}: {e}')
|
|
return False
|
|
|
|
|
|
def move_all_videos(yt_api, src_playlist: str, dst_playlist: str, limit: int = -1):
|
|
if limit < 0:
|
|
limit = len(src_playlist)
|
|
|
|
src_videos = get_videos(yt_api, src_playlist)
|
|
dst_videos = get_videos(yt_api, dst_playlist)
|
|
|
|
dst_video_map = {video['snippet']['resourceId']['videoId']: video['id']
|
|
for video in dst_videos}
|
|
|
|
for src_video in src_videos[:limit]:
|
|
video_id = src_video['snippet']['resourceId']['videoId']
|
|
if video_id not in dst_video_map:
|
|
add_video_to_playlist(yt_api, src_video, dst_playlist)
|
|
remove_video_from_playlist(yt_api, src_video, src_playlist)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('src_playlist', help='Source playlist ID')
|
|
parser.add_argument('dst_playlist', help='Destination playlist ID')
|
|
parser.add_argument(
|
|
'-l', '--limit', help='Limit number of videos to move', type=int, default=-1
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
# Disable OAuthlib's HTTPS verification when running locally.
|
|
# *DO NOT* leave this option enabled in production.
|
|
os.environ["OAUTHLIB_INSECURE_TRANSPORT"] = "1"
|
|
|
|
api_service_name = "youtube"
|
|
api_version = "v3"
|
|
|
|
creds = get_yt_creds()
|
|
youtube = build(api_service_name, api_version, credentials=creds)
|
|
|
|
move_all_videos(youtube, args.src_playlist, args.dst_playlist, limit=args.limit)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|