snegov/robocyp
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Linter, 80chars lines

Browse Source
This commit is contained in:
Maks Snegov 2024-03-29 18:19:19 -07:00
parent 51c7b11809
commit 5590a794b3
1 changed files with 60 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

84
robocyp.py
View File

@ -80,7 +80,9 @@ def exit_on_exceeded_quota(e: HttpError):
def read_playlists_file(): def read_playlists_file():
""" Read playlists.csv and return a dictionary of playlist names to playlist IDs """ """
Read playlists.csv and return a dictionary of playlist names to playlist IDs
"""
global _playlists global _playlists
playlists_file = os.path.join(program_dir, 'playlists.csv') playlists_file = os.path.join(program_dir, 'playlists.csv')
if not os.path.exists(playlists_file): if not os.path.exists(playlists_file):
@ -141,7 +143,8 @@ def add_video_to_playlist(yt_api, video_id: str, playlist_id: str,
video_title = _truncate_title(video_info['snippet']['title']) video_title = _truncate_title(video_info['snippet']['title'])
try: try:
if dry_run: if dry_run:
print(f"Would add video '{video_title}' [{video_id}] to playlist {playlist_name}") print(f"Would add video '{video_title}' [{video_id}]"
f" to playlist {playlist_name}")
return True return True
yt_api.playlistItems().insert( yt_api.playlistItems().insert(
part='snippet', part='snippet',
@ -155,11 +158,13 @@ def add_video_to_playlist(yt_api, video_id: str, playlist_id: str,
} }
} }
).execute() ).execute()
print(f"Added video '{video_title}' [{video_id}] to playlist {playlist_name}") print(f"Added video '{video_title}' [{video_id}]"
f" to playlist {playlist_name}")
return True return True
except HttpError as e: except HttpError as e:
exit_on_exceeded_quota(e) exit_on_exceeded_quota(e)
print(f"Error adding video '{video_title}' [{video_id}] to playlist {playlist_name}: {e}") print(f"Error adding video '{video_title}' [{video_id}]"
f" to playlist {playlist_name}: {e}")
return False return False
@ -209,7 +214,8 @@ def copy_playlist_items(yt_api,
add_video_to_playlist(yt_api, video_id, dst_playlist_id, dry_run) add_video_to_playlist(yt_api, video_id, dst_playlist_id, dry_run)
was_processed = True was_processed = True
if delete_from_src: if delete_from_src:
remove_video_from_playlist(yt_api, src_pl_item["id"], src_playlist_id, dry_run) remove_video_from_playlist(yt_api, src_pl_item["id"],
src_playlist_id, dry_run)
was_processed = True was_processed = True
if was_processed: if was_processed:
processed_amt += 1 processed_amt += 1
@ -219,7 +225,8 @@ def get_video_info(youtube, video_id: str):
try: try:
# TODO maybe remove 'status' # TODO maybe remove 'status'
response = youtube.videos().list( response = youtube.videos().list(
part="localizations,snippet,contentDetails,statistics,status,topicDetails", part="localizations,snippet,contentDetails,"
"statistics,status,topicDetails",
id=video_id id=video_id
).execute() ).execute()
except HttpError as e: except HttpError as e:
@ -258,36 +265,51 @@ def main():
parser_add.add_argument('playlist', help='Playlist name/ID') parser_add.add_argument('playlist', help='Playlist name/ID')
parser_add.add_argument('video_ids', nargs='*', help='Video IDs to add') parser_add.add_argument('video_ids', nargs='*', help='Video IDs to add')
parser_add_csv = subparsers.add_parser('add-csv', help='Add videos to a playlist from a CSV file') parser_add_csv = subparsers.add_parser(
'add-csv', help='Add videos to a playlist from a CSV file'
)
parser_add_csv.add_argument('playlist', help='Playlist name/ID') parser_add_csv.add_argument('playlist', help='Playlist name/ID')
parser_add_csv.add_argument('csv', help='CSV file with video IDs') parser_add_csv.add_argument('csv', help='CSV file with video IDs')
parser_add_csv.add_argument('-l', '--limit', type=int, default=-1, parser_add_csv.add_argument('-l', '--limit', type=int, default=-1,
help='Limit number of videos to process') help='Limit number of videos to process')
parser_copy = subparsers.add_parser('copy', help='Copy videos from one playlist to another') parser_copy = subparsers.add_parser(
'copy', help='Copy videos from one playlist to another'
)
parser_copy.add_argument('src_playlist', help='Source playlist name/ID') parser_copy.add_argument('src_playlist', help='Source playlist name/ID')
parser_copy.add_argument('dst_playlist', help='Destination playlist name/ID') parser_copy.add_argument('dst_playlist',
help='Destination playlist name/ID')
parser_copy.add_argument('-l', '--limit', type=int, default=-1, parser_copy.add_argument('-l', '--limit', type=int, default=-1,
help='Limit number of videos to process') help='Limit number of videos to process')
parser_move = subparsers.add_parser('move', help='Move videos from one playlist to another') parser_move = subparsers.add_parser(
'move',
help='Move videos from one playlist to another'
)
parser_move.add_argument('src_playlist', help='Source playlist name/ID') parser_move.add_argument('src_playlist', help='Source playlist name/ID')
parser_move.add_argument('dst_playlist', help='Destination playlist name/ID') parser_move.add_argument('dst_playlist',
help='Destination playlist name/ID')
parser_move.add_argument('-l', '--limit', type=int, default=-1, parser_move.add_argument('-l', '--limit', type=int, default=-1,
help='Limit number of videos to process') help='Limit number of videos to process')
parser_dups = subparsers.add_parser('dups', help='Remove duplicate videos in a playlist') parser_dups = subparsers.add_parser(
'dups', help='Remove duplicate videos in a playlist'
)
parser_dups.add_argument('playlist', help='Playlist name/ID') parser_dups.add_argument('playlist', help='Playlist name/ID')
parser_dups.add_argument('-l', '--limit', type=int, default=-1, parser_dups.add_argument('-l', '--limit', type=int, default=-1,
help='Limit number of videos to process') help='Limit number of videos to process')
parser_download = subparsers.add_parser('download', help='Download videos from a playlist') parser_download = subparsers.add_parser(
'download', help='Download videos from a playlist'
)
parser_download.add_argument('playlist', help='Playlist name/ID') parser_download.add_argument('playlist', help='Playlist name/ID')
parser_download.add_argument('dst_folder', help='Destination folder') parser_download.add_argument('dst_folder', help='Destination folder')
parser_download.add_argument('-l', '--limit', type=int, default=-1, parser_download.add_argument('-l', '--limit', type=int, default=-1,
help='Limit number of videos to process') help='Limit number of videos to process')
parser_download.add_argument('-r', '--remove-from-playlist', action='store_true', parser_download.add_argument(
help='Remove downloaded videos from the playlist') '-r', '--remove-from-playlist',
action='store_true', help='Remove downloaded videos from the playlist'
)
args = parser.parse_args() args = parser.parse_args()
@ -317,8 +339,11 @@ def main():
elif args.command == 'add': elif args.command == 'add':
playlist_id = get_playlist_id(args.playlist) playlist_id = get_playlist_id(args.playlist)
# {video_id: video_title} for videos already in the playlist # {video_id: video_title} for videos already in the playlist
pl_videos = {pl_item['snippet']['resourceId']['videoId']: pl_item['snippet']['title'] pl_videos = {
for pl_item in list_playlist(youtube, playlist_id)} pl_item['snippet']['resourceId']['videoId']:
pl_item['snippet']['title']
for pl_item in list_playlist(youtube, playlist_id)
}
for video_id in args.video_ids: for video_id in args.video_ids:
if video_id in pl_videos: if video_id in pl_videos:
@ -336,15 +361,20 @@ def main():
video_ids.extend(row[0] for row in reader if row) video_ids.extend(row[0] for row in reader if row)
playlist_id = get_playlist_id(args.playlist) playlist_id = get_playlist_id(args.playlist)
# {video_id: video_title} for videos already in the playlist # {video_id: video_title} for videos already in the playlist
pl_videos = {pl_item['snippet']['resourceId']['videoId']: pl_item['snippet']['title'] pl_videos = {
for pl_item in list_playlist(youtube, playlist_id)} pl_item['snippet']['resourceId']['videoId']:
pl_item['snippet']['title']
for pl_item in list_playlist(youtube, playlist_id)
}
processed = 0 processed = 0
for video_id in video_ids: for video_id in video_ids:
if 0 <= args.limit <= processed: if 0 <= args.limit <= processed:
break break
if video_id in pl_videos: if video_id in pl_videos:
continue continue
processed += int(add_video_to_playlist(youtube, video_id, playlist_id, args.dry_run)) processed += int(add_video_to_playlist(
youtube, video_id, playlist_id, args.dry_run
))
elif args.command == "dups": elif args.command == "dups":
processed = 0 processed = 0
@ -356,7 +386,8 @@ def main():
break break
video_id = plitem["snippet"]["resourceId"]["videoId"] video_id = plitem["snippet"]["resourceId"]["videoId"]
if video_id in plitems_processed: if video_id in plitems_processed:
remove_video_from_playlist(youtube, plitem["id"], playlist_id, args.dry_run) remove_video_from_playlist(youtube, plitem["id"], playlist_id,
args.dry_run)
processed += 1 processed += 1
else: else:
plitems_processed.add(video_id) plitems_processed.add(video_id)
@ -391,17 +422,22 @@ def main():
video_title = _truncate_title(video_info['snippet']['title']) video_title = _truncate_title(video_info['snippet']['title'])
if args.dry_run: if args.dry_run:
print(f"Would download video '{video_title}' [{video_id}]" print(f"Would download video '{video_title}' [{video_id}]"
f" from playlist {args.playlist} to folder {args.dst_folder}") f" from playlist {args.playlist}"
f" to folder {args.dst_folder}")
else: else:
# download video # download video
with YoutubeDL(ydl_opts) as ydl: with YoutubeDL(ydl_opts) as ydl:
ydl.download(['https://www.youtube.com/watch?v=' + video_id]) ydl.download(
['https://www.youtube.com/watch?v=' + video_id]
)
db.insert(video_info) db.insert(video_info)
# remove video from playlist # remove video from playlist
if args.remove_from_playlist: if args.remove_from_playlist:
remove_video_from_playlist(youtube, plitem["id"], playlist_id, args.dry_run) remove_video_from_playlist(
youtube, plitem["id"], playlist_id, args.dry_run
)
return 0 return 0