Python discord 模块，FFmpegPCMAudio() 实例源码

def advance(self, error=None):
        # uh oh
        if error is not None:
            log.error('State.advance() threw an exception. %s', error)

        # if we are looping and have a song to loop, don't pop the queue. instead, make a ffmpeg source
        # and play that.
        if self.looping and self.to_loop:
            log.debug('Bypassing State.advance() logic, looping %s.', self.to_loop['url'])
            self.play(
                VolumeTransformer(FFmpegPCMAudio(self.to_loop['url']))
            )
            return

        # out of sources in the queue -- oh well.
        if not self.queue:
            log.debug('Out of queue items.')
            return

        # get the latest source in the queue, then play it.
        next_up = self.queue.pop(0)
        self.play(next_up)

def play_file(self, ctx, sound, volume=1):
        """Plays a file from the local filesystem"""

        if ctx.voice_client is None:
            if ctx.author.voice.channel:
                await ctx.author.voice.channel.connect()
            else:
                return await ctx.send("Not connected to a voice channel.")

        if ctx.voice_client.is_playing():
            ctx.voice_client.stop()

        def done(e):
            if e:
                print("Player error: %s" % e)
            if not ctx.voice_client.is_playing():
                ctx.bot.loop.create_task(ctx.voice_client.disconnect())

        source = discord.PCMVolumeTransformer(
            discord.FFmpegPCMAudio(join(folder, sound)))
        source.volume = volume
        ctx.voice_client.play(source, after=done)

        # await ctx.send('Now playing: {}'.format(query))

def create_player(self, voice_client):
        await self.download()
        if self.location:
            audio_source = discord.FFmpegPCMAudio(self.location)
            if not voice_client.is_playing():
                voice_client.play(audio_source)

def create_source(entry):
        source = PCMVolumeTransformer(FFmpegPCMAudio(entry["download_url"], **entry["kwargs"]))
        source.download_url = entry["download_url"]
        source.url = entry["url"]
        source.yt = entry["yt"]
        source.is_live = entry["is_live"]
        source.title = entry["title"]
        source.duration = entry["duration"]
        source.description = entry["description"]
        source.upload_date = entry["upload_date"]
        source.kwargs = entry["kwargs"]
        return source

def make_player(self, voice, item):
        location = item['url']
        if item['type'] == 0:
            file_location = self.download_yt_data(location)
        elif item['type'] == 1:
            file_location = await self.download_sc_data(location)
        elif item['type'] == 2:
            file_location = await self.download_bc_data(item['sound'])
        else:
            file_location = location
        source = discord.FFmpegPCMAudio(file_location, executable='ffmpeg')
        voice.play(source)

def play(self, ctx):
        """Play a youtube video"""
        voice_channel = None
        if not ctx.author.voice is None:
            voice_channel = ctx.author.voice.channel
        if not voice_channel is None:
            vc = await voice_channel.connect()
            vc.play(discord.FFmpegPCMAudio('data/music/tmp/CSvFpBOe8eY.m4a'))

def on_member_ban(self, guild, member):
        if member.voice.channel is None:
            return
        guild_config = await self.bot.get_guild_config(guild.id)
        if guild_config["teamspeakBanSound"]:
            vc = await member.voice.channel.connect()
            src = discord.FFmpegPCMAudio("bot/resources/user_banned.mp3")
            vc.play(src)
            await asyncio.sleep(5)
            await vc.disconnect()

def from_url(cls, url, *, loop=None):
        loop = loop or asyncio.get_event_loop()
        data = await loop.run_in_executor(None, ytdl.extract_info, url)

        if 'entries' in data:
            # take first item from a playlist
            data = data['entries'][0]

        filename = ytdl.prepare_filename(data)
        return cls(discord.FFmpegPCMAudio(filename, **ffmpeg_options),
                   data=data)

def play_file(self, ctx, *, query):
        """Plays a file from the local filesystem"""

        if ctx.voice_client is None:
            if ctx.author.voice.channel:
                await ctx.author.voice.channel.connect()
            else:
                return await ctx.send("Not connected to a voice channel.")

        if ctx.voice_client.is_playing():
            ctx.voice_client.stop()

        source = discord.PCMVolumeTransformer(discord.FFmpegPCMAudio(query))
        ctx.voice_client.play(source, after=lambda e: print(
            'Player error: %s' % e) if e else None)

        await ctx.send('Now playing: {}'.format(query))

def play_next_clip(self):
        clip = self.next_clip()

        self.voice.play(discord.FFmpegPCMAudio(clip.audiopath), after=self.done_talking)
        self.voice.source = discord.PCMVolumeTransformer(self.voice.source)
        self.voice.source.volume = clip.volume
        print("playing: " + clip.audiopath)
        if self.last_clip != None and clip.audiopath != self.last_clip.audiopath:
            remove_if_temp(self.last_clip.audiopath)
        self.last_clip = clip

    # try queueing an mp3 to play

def play_next_sound(self):
        """ Play the next sound in queue. If none, disconnect """
        if len(self.soundQueue) != 0:
            sound = self.soundQueue[0]
            del self.soundQueue[0]
            self.currentVoiceClient.play(discord.FFmpegPCMAudio(sound.location), after=self.after_sound_clip)
        else:
            await self.disconnect()

def process_song_data(self, ydl, url, entry):
        """
        Just packing some code into a function to make fetch_song_data more readable
        :param ydl: youtube_dl object
        :param url: original url of the request
        :param entry: request data
        :return: dictionary of parameters
        """
        t_dict = DotDict({})
        self.logger.info(f'processing URL {entry["title"]}')
        if self.plugin_config["download_songs"]:
            filename = ydl.prepare_filename(entry)
            self.storage["stored_songs"][filename] = time.time()
        else:
            filename = entry['url']
        # t_source = PCMVolumeTransformer(FFmpegPCMAudio(filename, **kwargs))  # out, out, damn spot!
        t_dict['download_url'] = filename
        t_dict['url'] = entry.get('webpage_url')
        t_dict['yt'] = ydl
        t_dict['is_live'] = bool(entry.get('is_live'))
        t_dict['duration'] = ceil(entry.get('duration', 0))

        is_twitch = 'twitch' in url

        if is_twitch:
            # twitch has 'title' and 'description' sort of mixed up.
            t_dict['title'] = entry.get('description')
            t_dict['description'] = None
        else:
            t_dict['title'] = entry.get('title')
            t_dict['description'] = entry.get('description')

        # upload date handling
        date = entry.get('upload_date')
        if date:
            try:
                date = datetime.datetime.strptime(date, '%Y%M%d').date()
            except ValueError:
                date = None

        t_dict['upload_date'] = date
        return t_dict