Python youtube_dl 模块，DownloadError() 实例源码

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

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)