import subprocess from pathlib import Path from typing import Union import yt_dlp from mergedeep import merge class YDL: 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') 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) def playlist_contents(self, url: str) -> Union[dict, bool]: ydl_opts = { 'extract_flat': True, 'skip_download': True, 'ignoreerrors': True, 'logger': self.ydl_opts['logger'], } with yt_dlp.YoutubeDL(ydl_opts) as ydl: info = self.get_info(url) if not info: return False entries = [] if info['_type'] == 'playlist': if 'entries' in info.keys(): # When downloading a channel youtube-dl returns a playlist for videos and another for shorts. # We need to combine all the videos into one list. for item in info['entries']: if item['_type'] in ('video', 'url'): entries.append(item) elif item['_type'] == 'playlist': for video in self.get_info(item['webpage_url'])['entries']: entries.append(video) else: raise ValueError(f"Unknown sub-media type: {item['_type']}") 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. entries.append(info) entries[0]['url'] = f"https://www.youtube.com/watch?v={info['id']}" 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' 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) 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)) def __call__(self, *args, **kwargs): return self.yt_dlp.download(*args, **kwargs) def update_ytdlp(): package_name = 'yt-dlp' 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) return False 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 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