def _get_stream_info(self):
func = functools.partial(self._ytdl.extract_info, self._stream_url, download=False)
try:
info = await self._bot.loop.run_in_executor(None, func)
except youtube_dl.DownloadError as e:
await self._bot.message('Failed to obtain stream information: {}'.format(str(e)))
return False
if not self._stream_title:
if 'twitch' in self._stream_url: # TODO: regex should be much better
self._stream_title = info.get('description')
else:
self._stream_title = info.get('title')
if not self._stream_title:
self._stream_title = '<untitled stream>'
if 'url' not in info:
await self._bot.message('Failed to extract stream URL, is the link valid?')
return False
self._stream_url = info['url']
return True
python类DownloadError()的实例源码
def add(self, song_url: str, voice_client: discord.VoiceClient, user=None):
try:
song_player = await voice_client.create_ytdl_player(song_url, ytdl_options=ytdl_format_options)
except youtube_dl.DownloadError:
return youtube_dl.DownloadError
except youtube_dl.SameFileError:
return youtube_dl.SameFileError
except youtube_dl.utils.ExtractorError:
return youtube_dl.utils.ExtractorError
except youtube_dl.utils.UnavailableVideoError:
return youtube_dl.utils.UnavailableVideoError
user_name = "********"
if user is not None:
user_name = user.display_name
song = {"player": None, "url": song_player.url, "title": song_player.title,
"uploader": song_player.uploader, "user": user_name}
await self.__queue.put(song)
return song
# Get next song and delete it
def run(self):
song_urls = iter(self.song_urls)
errors = []
with youtube_dl.YoutubeDL(self.opts) as yt:
while not self.is_aborted():
try:
song_url = next(song_urls)
logger.info("Downloading audio/video from {url}".format(url=song_url))
try:
yt.download([song_url])
except youtube_dl.DownloadError:
errors.append(song_url)
self.downloaded += 100
wx.CallAfter(self.parent.download_update, message=self.downloaded)
except StopIteration:
wx.CallAfter(self.parent.download_complete, errors=errors)
break
def yt_extract(links):
"""Extract individual URLs from a playlist.
Args:
links (list[str]): A list of the URLs.
Returns:
list[str]: The list of extracted URLs.
"""
songs = []
with youtube_dl.YoutubeDL() as yt_dl:
for url in links:
try:
result = yt_dl.extract_info(url, download=False, process=False)
except youtube_dl.DownloadError:
return songs
if 'entries' in result:
songs.extend(vid['url'] for vid in list(result['entries']))
else:
songs.append(url)
return songs
def episode(request):
""" Get all the information and the video links from a given episode.
How: use youtube-dl to get the information
"""
episode_url = request.matchdict['episode_url']
url = EITB_VIDEO_BASE_URL + episode_url
try:
playlist_title, playlist_id, video_title, video_id = episode_url.split('/')
except ValueError:
return {}
result = {
'@context': 'http://www.w3.org/ns/hydra/context.jsonld',
'@id': request.route_url('episode', episode_url=episode_url),
'@type': 'Episode',
'parent': request.route_url('playlist', playlist_id=playlist_id),
}
try:
ydl = youtube_dl.YoutubeDL({'outtmpl': '%(id)s%(ext)s'})
video_data = ydl.extract_info(url, download=False)
except youtube_dl.DownloadError:
return result
result.update(video_data)
return result
def _play(self, ctx: DogbotContext, url, *, search=False):
msg = await ctx.send(f'\N{INBOX TRAY} {random.choice(SEARCHING_TEXT)}')
# grab the source
url = 'ytsearch:' + url if search else url
try:
source = await YouTubeDLSource.create(url, ctx.bot)
except youtube_dl.DownloadError:
return await msg.edit(content='\U0001f4ed YouTube gave me nothing.')
except YouTubeError as yterr:
return await msg.edit(content='\N{CROSS MARK} ' + str(yterr))
disp = '**{}**'.format(source.title)
state = self.state_for(ctx.guild)
if state.is_playing():
# add it to the queue
log.debug('Adding to queue.')
state.queue.append(source)
await msg.edit(content=f'\N{LINKED PAPERCLIPS} Added {disp} to queue.')
else:
# play immediately since we're not playing anything
log.debug('Playing immediately.')
state.play(source)
await msg.edit(content=f'\N{MULTIPLE MUSICAL NOTES} Playing {disp}.')
def get_video_info(url):
"""
Retrieve the YouTube videos' information without downloading it.
Source: http://stackoverflow.com/questions/18054500/how-to-use-youtube-dl-\
from-a-python-programm/18947879#18947879
"""
ydl = youtube_dl.YoutubeDL()
ydl.add_default_info_extractors()
try:
return ydl.extract_info(url, download=False)
except youtube_dl.DownloadError:
return None
def start_download(self, dic):
writeln('[' + color('DOWN', 'cyan') + '] Starting download of %s from %s, saving as ID %d'
% (dic['name'], dic['url'], dic['id']))
# cur_option = self.options
# cur_option['progress_hooks'] = [partial(self.download_progress, dic['id'])]
# cur_option['outtmpl'] = 'video/' + str(dic['id']) + '/' + str(dic['id']) + r'.%(title)s-%(id)s.%(ext)s'
# downloader = youtube_dl.YoutubeDL(cur_option)
# try:
# downloader.download([dic['url']])
# self.download_progress(dic['id'], {'status': 'complete'})
# except youtube_dl.DownloadError as e:
# writeln('[' + color('ERROR', 'red') + '] youtube_dl error for %s: ' % dic['name'] + e.message)
# self.download_progress(dic['id'], {'status': 'error'})
self.download_progress(dic['id'], {'status': 'downloading'})
outpath = 'video/' + str(dic['id']) + '/'
try:
os.makedirs(outpath)
except:
pass
log = open(outpath + 'log.txt', 'w')
process = subprocess.Popen(["you-get", dic['url']], stdout=log, stderr=subprocess.STDOUT, cwd=outpath)
retcode = process.wait()
log.close()
log = open(outpath + 'log.txt', 'r')
if retcode != 0 or ' '.join(log.readlines()).find('error') != -1:
self.download_progress(dic['id'], {'status': 'error'})
else:
self.download_progress(dic['id'], {'status': 'complete'})
def play(message, url):
"""
Play a URL.
Example::
/play https://www.youtube.com/watch?v=U9DZkj8Rq6g
"""
m = LINK_PATTERN.search(url)
if not m:
raise CommandError("You need to provide a URL to play.")
url = m.group(1)
# check to see if we can play this URL
opts = {
'format': 'webm[abr>0]/bestaudio/best',
'prefer_ffmpeg': True,
'source_address': source_address(),
'noplaylist': True,
}
ydl = youtube_dl.YoutubeDL(opts)
func = functools.partial(ydl.extract_info, url, download=False)
try:
info = await asyncio.get_event_loop().run_in_executor(None, func)
# get metadata
is_twitch = 'twitch' in url
if is_twitch:
# twitch has 'title' and 'description' sort of mixed up
title = info.get('description')
description = None
else:
title = info.get('title')
description = info.get('description')
except DownloadError as e:
raise CommandError("Can't play <{}>. It might not be a supported site.".format(url))
# get the voice channel
voice_client = await get_voice_client(message.author)
queue = queue_map.get(voice_client.channel)
# queue that stuff up
async def factory(entry: QueueEntry):
return await voice_client.create_ytdl_player(url, ytdl_options=opts, after=entry.on_end,
options=['-af', 'loudnorm=I=-16:TP=-1.5:LRA=11'])
meta = EntryMeta(title=title, description=description, url=url)
entry = await queue.add(factory, channel=voice_client.channel, meta=meta)
# tell the user
entries = queue.entries()
index = entries.index(entry)
if index == 0:
return "Now playing **{}**...".format(meta)
else:
return "Queued **{}** at position #{} for play.".format(meta, index)
