import log import json from typing import Optional, Any from base import Search, Module, ParsedJson from post import Url, Post, PostType, Media, Video from query_parser import QueryParser from yt_dlp import YoutubeDL from yt_dlp.cookies import extract_cookies_from_browser class YDLLogger(): def debug(self, message: str) -> None: if message.startswith('[debug] '): log.debug(message) else: log.info(message) def info(self, message: str) -> None: log.info(message) def warning(self, message: str) -> None: log.warn(message) def error(self, message: str) -> None: log.error(message) ydl_opts = { 'quiet': False, 'logger': YDLLogger(), 'cachedir': False, 'socket_timeout': 10, 'extractor_args': {'youtube': {'skip': ['hls'], 'player_client': ['android_vr']}} # 'extractor_args': {'youtube': {'skip': ['hls'], 'player_client': ['tv', 'ios']}} # 'cookiefile': '' } #print(extract_cookies_from_browser('firefox')) # https://github.com/yt-dlp/yt-dlp/issues/4103 ydl = YoutubeDL(ydl_opts) def get_playback_url(data: ParsedJson, video: bool = True) -> Optional[str]: data['formats'] = list(filter(lambda f: not f['protocol'].startswith('m3u8'), data['formats'])) if len(data['formats']) == 0: return None result = data['formats'][-1]['url'] has_audio = list(filter(lambda f: (f['acodec'] and f['acodec'] != 'none') or (f['audio_ext'] and f['audio_ext'] != 'none') or (f['abr']), data['formats'])) if video: data['formats'] = list(filter(lambda f: (f['vcodec'] and f['vcodec'] != 'none') or (f['video_ext'] and f['video_ext'] != 'none') or (f['vbr']), has_audio)) else: data['formats'] = list(filter(lambda f: (f['vcodec'] and f['vcodec'] == 'none') and (f['video_ext'] and f['video_ext'] == 'none') and (not f['vbr']), has_audio)) if len(data['formats']) > 0: selection = max(data['formats'], key=lambda f: 0 if 'quality' not in f else f['quality']) log.info(json.dumps(selection, indent=4)) result = selection['url'] else: log.warn('Defaulting in Youtube get_playback_url().') return result class YoutubeBase(Search): def __init__(self, userdata: Any) -> None: super().__init__() self.module: YoutubeModule = userdata self.guessed_index: int = 0 def get_info(self) -> Optional[ParsedJson]: return None def request_page(self, num: int) -> bool: info = self.get_info() if not info: return False if 'entries' in info: if len(info['entries']) == 0: return False if len(info['entries']) <= self.guessed_index: self.guessed_index = 0 info = info['entries'][self.guessed_index] if 'formats' not in info and 'url' in info: self.link = info['url'] info = self.get_info() if not info: return False if info.get('direct'): url = info['url'] else: url = get_playback_url(info, video=False) if not url: return False unique_id = f'youtube:v:{info['id']}' media: dict[str, Media] = { '0': Video(url=Url(url)) } self.module.add_to_map(unique_id, Post(type=PostType.POST, unique_id=unique_id, url='', title=info['title'], text='', media=media)) self.pages[num] = [unique_id] return True class YoutubeSearch(YoutubeBase): def __init__(self, userdata: Any, arg: str) -> None: super().__init__(userdata) self.query: str = arg def get_info(self) -> Optional[ParsedJson]: ydl.params['extract_flat'] = False try: info = ydl.extract_info(f'ytsearch1:{self.query}', download=False) except Exception as e: log.error(repr(e)) return None return info class YoutubeLink(YoutubeBase): def __init__(self, userdata: Any, arg: str) -> None: super().__init__(userdata) self.link: str = arg def get_info(self) -> Optional[ParsedJson]: index = self.link.find('&index=') if index >= 0: sub = self.link[index + 7:] end = sub.find('&') if end >= 0: sub = sub[:end] try: self.guessed_index = int(sub, 10) - 1 except ValueError: pass log.info(f'Guessed index: {self.guessed_index}') ydl.params['extract_flat'] = 'in_playlist' try: info = ydl.extract_info(self.link, download=False) except Exception as e: log.error(repr(e)) return None return info class YoutubeModule(Module): def __init__(self) -> None: super().__init__() self.parser: QueryParser = QueryParser(self, 'search') self.parser.add_command('search', YoutubeSearch) self.parser.add_command('link', YoutubeLink) def search(self, query: str, *extra_args: Any) -> Optional[Search]: return self.parser.parse_query(query) def get_download(self, unique_id: str, key: str) -> Optional[dict[str, Any]]: if unique_id not in self.unique_id_map: return None return None