#!/usr/bin/env python
import argparse
from pytube import Playlist, YouTube
from prefect import task, flow
from pydantic import HttpUrl
from prefect_dask import DaskTaskRunner

@task(retries=3, retry_delay_seconds=4)
def download_video(video: YouTube):
    video.streams.get_highest_resolution().download()
    return

@flow(task_runner=DaskTaskRunner)
def playlist_downloader(playlist_url: HttpUrl):
    p = Playlist(playlist_url)
    for video in p.videos:
        download_video.submit(video)
    return


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description='Downloads all youtubes videos in a playlist in parallel using the highest available resolution')
    parser.add_argument('playlist', help='a playlist url to download videos')
    
    args = parser.parse_args()
    playlist_downloader(args.playlist)