2023-01-20 21:42:36 -07:00
import subprocess
from pathlib import Path
from typing import Union
import yt_dlp
from mergedeep import merge
class YDL :
2023-02-06 23:01:47 -07:00
def __init__ ( self , ydl_opts : dict = None , extra_ydlp_opts : dict = None ) :
self . ydl_opts = ydl_opts if ydl_opts else { }
extra_ydlp_opts = extra_ydlp_opts if extra_ydlp_opts else { }
self . ydl_opts = merge ( ydl_opts , extra_ydlp_opts )
self . ydl_opts [ ' logger ' ] = self . ydl_opts . get ( ' logger ' )
2023-01-20 21:42:36 -07:00
self . yt_dlp = yt_dlp . YoutubeDL ( ydl_opts )
def get_formats ( self , url : Union [ str , Path ] ) - > tuple :
"""
Not used since we ' re letting youtube-dl manage filesize filters for us.
"""
sizes = [ ]
with self . yt_dlp as ydl :
for video in ydl . extract_info ( url , download = False ) [ ' formats ' ] :
d = {
' format_id ' : video [ ' format_id ' ] ,
' format_note ' : video [ ' format_note ' ] ,
}
if video . get ( ' filesize ' ) :
d [ ' filesize ' ] = round ( video [ ' filesize ' ] / 1e+6 , 1 ) # MB
else :
d [ ' filesize ' ] = - 1
sizes . append ( d )
return tuple ( sizes )
2023-02-02 20:40:19 -07:00
def playlist_contents ( self , url : str ) - > Union [ dict , bool ] :
2023-02-06 23:01:47 -07:00
ydl_opts = {
2023-01-20 21:42:36 -07:00
' extract_flat ' : True ,
2023-02-06 23:01:47 -07:00
' skip_download ' : True ,
' ignoreerrors ' : True ,
' logger ' : self . ydl_opts [ ' logger ' ] ,
}
2023-01-20 21:42:36 -07:00
with yt_dlp . YoutubeDL ( ydl_opts ) as ydl :
2023-02-07 00:13:44 -07:00
info = self . get_info ( url )
2023-02-02 20:35:37 -07:00
if not info :
return False
2023-01-20 21:42:36 -07:00
entries = [ ]
if info [ ' _type ' ] == ' playlist ' :
if ' entries ' in info . keys ( ) :
2023-02-06 23:01:47 -07:00
# When downloading a channel youtube-dl returns a playlist for videos and another for shorts.
2023-02-07 00:13:44 -07:00
# We need to combine all the videos into one list.
2023-02-06 23:01:47 -07:00
for item in info [ ' entries ' ] :
2023-02-06 23:20:28 -07:00
if item [ ' _type ' ] in ( ' video ' , ' url ' ) :
2023-02-06 23:01:47 -07:00
entries . append ( item )
elif item [ ' _type ' ] == ' playlist ' :
2023-02-07 00:13:44 -07:00
for video in self . get_info ( item [ ' webpage_url ' ] ) [ ' entries ' ] :
2023-02-06 23:01:47 -07:00
entries . append ( video )
else :
raise ValueError ( f " Unknown sub-media type: { item [ ' _type ' ] } " )
2023-01-20 21:42:36 -07:00
elif info [ ' _type ' ] == ' video ' :
# `info` doesn't seem to contain the `url` key so we'll add it manually.
# If any issues arise in the future make sure to double check there isn't any weirdness going on here.
2023-01-21 18:19:03 -07:00
entries . append ( info )
entries [ 0 ] [ ' url ' ] = f " https://www.youtube.com/watch?v= { info [ ' id ' ] } "
2023-01-20 21:42:36 -07:00
else :
raise ValueError ( f " Unknown media type: { info [ ' _type ' ] } " )
return {
' title ' : info [ ' title ' ] ,
' id ' : info [ ' id ' ] ,
' entries ' : entries ,
}
# def filter_filesize(self, info, *, incomplete):
# duration = info.get('duration')
# if duration and duration < 60:
# return 'The video is too short'
2023-02-02 20:35:37 -07:00
def extract_info ( self , * args , * * kwargs ) :
return self . yt_dlp . extract_info ( * args , * * kwargs )
def prepare_filename ( self , * args , * * kwargs ) :
return self . yt_dlp . prepare_filename ( * args , * * kwargs )
def process_info ( self , * args , * * kwargs ) :
return self . yt_dlp . process_info ( * args , * * kwargs )
2023-02-07 00:13:44 -07:00
def get_info ( self , url ) :
ydl_opts = {
' extract_flat ' : True ,
' skip_download ' : True ,
' ignoreerrors ' : True ,
' logger ' : self . ydl_opts [ ' logger ' ] ,
}
ydl = yt_dlp . YoutubeDL ( ydl_opts )
return ydl . sanitize_info ( ydl . extract_info ( url , download = False ) )
2023-02-06 23:01:47 -07:00
2023-02-02 20:35:37 -07:00
def __call__ ( self , * args , * * kwargs ) :
return self . yt_dlp . download ( * args , * * kwargs )
2023-01-20 21:42:36 -07:00
def update_ytdlp ( ) :
2023-05-06 15:30:34 -06:00
package_name = ' yt-dlp '
2023-05-06 18:24:47 -06:00
try :
result = subprocess . run (
[ " pip " , " install " , " --disable-pip-version-check " , " --upgrade " , package_name ] ,
capture_output = True ,
text = True ,
check = True
)
if f " Successfully installed { package_name } " in result . stdout :
# print(f"{package_name} was updated.")
return True
else :
# print(f"{package_name} was not updated.")
return False
except subprocess . CalledProcessError as e :
print ( f " An error occurred while updating { package_name } : " )
print ( e . output )
2023-05-06 15:30:34 -06:00
return False
2023-01-21 18:19:03 -07:00
class ytdl_no_logger ( object ) :
def debug ( self , msg ) :
return
def info ( self , msg ) :
return
def warning ( self , msg ) :
return
def error ( self , msg ) :
return
2023-02-07 19:45:11 -07:00
2023-02-07 20:52:13 -07:00
def get_output_templ ( video_id : str = None , title : str = None , uploader : str = None , uploader_id : str = None , include_ext : bool = True ) :
return f ' [ { video_id if video_id else " %(id)s " } ] [ { title if title else " %(title)s " } ] [ { uploader if uploader else " %(uploader)s " } ] [ { uploader_id if uploader_id else " %(uploader_id)s " } ] { " . %(ext)s " if include_ext else " " } ' # leading dash can cause issues due to bash args so we surround the variables in brackets