def __init__(self, *args, debug=False, **kwargs):
self._debug = debug
self.game = config.game
game = discord.Game(name=self.game)
status = discord.Status.dnd if self._debug else discord.Status.online
super().__init__(*args, command_prefix=command_prefix,
game=game, status=status, **kwargs)
self._before_invoke = self._before_invoke_
self._after_invoke = self._after_invoke_
self.resumes = 0
useragent = 'Discord Bot'
source = config.source
if source is not None:
useragent += ' ' + source
self.http_ = aiohttp.ClientSession(loop=self.loop, headers={'User-Agent': useragent})
self.db_pool = self.loop.run_until_complete(
asyncpg.create_pool(dsn=config.pg_dsn, command_timeout=10, loop=self.loop))
python类Game()的实例源码
def on_ready():
print('Client logged in.')
print(yeebot.user.name)
print(yeebot.user.id)
print('-----')
cur.execute('CREATE TABLE IF NOT EXISTS links(link text, status text, '
'submitter_id text, submitter_name text, reviewer_id'
' text, reviewer_name text);')
cur.execute('CREATE TABLE IF NOT EXISTS users (user_id TEXT UNIQUE, username '
'TEXT, meme_bucks INTEGER, memes_submitted INTEGER DEFAULT 0,'
' memes_requested INTEGER DEFAULT 0, memes_approved INTEGER '
'DEFAULT 0, memes_rejected INTEGER DEFAULT 0, PRIMARY KEY(user_id));')
await yeebot.change_presence(game=discord.Game(name="Memes"))
for extension in startup_extensions:
try:
yeebot.load_extension(extension)
except Exception as e:
exc = '{}: {}'.format(type(e).__name__, e)
print('Failed to load extensions {}\n {}'.format(extension, exc))
def setstreaming(self, ctx, option : str, *url : str):
'''Set my streaming status'''
if option == "on" or option == "true":
if not url:
await clients.set_streaming_status()
return
else:
updated_game = ctx.message.server.me.game
if not updated_game:
updated_game = discord.Game(url = url[0], type = 1)
else:
updated_game.url = url[0]
updated_game.type = 1
else:
updated_game = ctx.message.server.me.game
updated_game.type = 0
await self.bot.change_status(game = updated_game)
def random_game_status():
statuses = ["with i7-2670QM", "with mainframes", "with Cleverbot",
"tic-tac-toe with Joshua", "tic-tac-toe with WOPR", "the Turing test",
"with my memory", "with R2-D2", "with C-3PO", "with BB-8",
"with machine learning", "gigs", "with Siri", "with TARS", "with KIPP",
"with humans", "with Skynet", "Goldbach's conjecture",
"Goldbach's conjecture solution", "with quantum foam",
"with quantum entanglement", "with P vs NP", "the Reimann hypothesis",
"the Reimann proof", "with the infinity gauntlet", "for the other team",
"hard to get", "to win", "world domination", "with Opportunity",
"with Spirit in the sand pit", "with Curiousity", "with Voyager 1",
"music", "Google Ultron", "not enough space here to",
"the meaning of life is", "with the NSA", "with neural networks",
"with RSS Bot", "with Data", "with Harmon", " "]
me = discord.utils.find(lambda s: s != None, client.servers).me
if not me:
return
elif not me.game:
updated_game = discord.Game(name = random.choice(statuses))
else:
updated_game = me.game
updated_game.name = random.choice(statuses)
await client.change_presence(game = updated_game)
def game(self, ctx, *, game=None):
"""Sets the bot's playing status
Leaving this empty will clear it."""
# [p]set game <game>
server = ctx.message.server
current_status = server.me.status if server is not None else None
if game:
game = game.strip()
await self.bot.change_presence(game=discord.Game(name=game),
status=current_status)
await self.bot.say('Game set to "{}".'.format(game))
else:
await self.bot.change_presence(game=None, status=current_status)
await self.bot.say('Not playing a game now.')
def stream(self, ctx, streamer=None, *, stream_title=None):
"""Sets the bot's streaming status.
Leaving both streamer and stream_title empty will clear it."""
# [p]set stream <streamer> <stream_title>
server = ctx.message.server
current_status = server.me.status if server is not None else None
if stream_title:
stream_title = stream_title.strip()
if "twitch.tv/" not in streamer:
streamer = "https://www.twitch.tv/" + streamer
game = discord.Game(type=1, url=streamer, name=stream_title)
await self.bot.change_presence(game=game, status=current_status)
log.debug('Owner has set streaming status and url to "{}" and {}'.format(stream_title, streamer))
elif streamer is not None:
await send_command_help(ctx)
return
else:
await self.bot.change_presence(game=None, status=current_status)
log.debug('stream cleared by owner')
await self.bot.say("Done.")
def on_guild_join(server):
server_count = len(bot.guilds)
member_count = 0
for server in bot.guilds:
for member in server.members:
member_count += 1
await bot.change_presence(game=discord.Game(name=bot.command_prefix[0]+"help | {} guilds with {} members.".format(server_count, member_count)))
webhook.send(':tada: [`'+str(datetime.datetime.now().strftime("%d/%m/%y %H:%M:%S"))+'`] I joined the server `' + server.name + '` ('+ str(server.id) + '), owned by `' + server.owner.name + '#' + server.owner.discriminator + '` (' + str(server.owner.id) + ').')
guild_count = len(bot.guilds)
headers = {'Authorization': config['Main']['dbotstoken']}
data = {'server_count': guild_count}
api_url = 'https://discordbots.org/api/bots/311810096336470017/stats'
async with aiohttp.ClientSession() as session:
await session.post(api_url, data=data, headers=headers)
# server leave
def on_ready(): # Triggered when connects successfully.
if config.settings.main.get('show-message-at-login'):
print('-' * 80)
print(config.settings.main.get('message-at-login').format
(user=client.user.name, ID=client.user.id))
print('-' * 80)
if config.settings.main.get('send-mp-at-login'):
await client.send_message(await client.get_user_info
(config.settings.main.get('user-to-send-mp-at-login')),
config.settings.main.get('message-mp-at-login'))
if config.settings.main.get('set-game'):
await client.change_presence(game=discord.Game
(name=config.settings.main.get('game-name').format
(v=config.settings.main.get('version'), p=prefix)))
def on_ready(self):
logging.info(f'Logged in \nUsername: {self.user.name}\nID: {self.user.id}\nAPI Version: {discord.__version__}')
gameplayed = config.MAIN.get("gameplayed", "Yuri is Love")
if gameplayed:
game = discord.Game(name=gameplayed)
await self.change_presence(game=game)
avatar_file_name = config.MAIN.get("avatarfile")
if avatar_file_name:
with open(avatar_file_name, "rb") as f:
avatar = f.read()
await self.edit_profile(avatar=avatar)
if self.first_run:
self.first_run = False
self.get_forbidden_list()
self.get_startswith()
self.fill_member_list()
self.skype.discord = self
for k, v in list(config.ch.items()):
if v.isdigit():
config.ch[k] = self.get_channel(v)
self.run_loop()
# TODO Add embed support
def on_ready():
bot.app_info = await bot.application_info()
bot.owner = discord.utils.get(bot.get_all_members(), id=bot.app_info.owner.id)
await bot.change_presence(game=discord.Game(name=f'{bot.command_prefix[0]}help'))
bot.loop.create_task(init_timed_events(bot))
print(f'#-------------------------------#\n'
f'| Successfully logged in.\n'
f'#-------------------------------#\n'
f'| Username: {bot.user.name}\n'
f'| User ID: {bot.user.id}\n'
f'| Owner: {bot.owner}\n'
f'| Guilds: {len(bot.guilds)}\n'
f'| Users: {len(list(bot.get_all_members()))}\n'
f'| OAuth URL: {discord.utils.oauth_url(bot.app_info.id)}\n'
f'# ------------------------------#')
###
def status_clockwork(ev):
while True:
if ev.bot.cfg.pref.status_rotation:
if not status_cache:
status_files = await ev.db[ev.db.db_cfg.database].StatusFiles.find().to_list(None)
for status_file in status_files:
status_text = status_file.get('Text')
status_cache.append(status_text)
if status_cache:
status = status_cache.pop(secrets.randbelow(len(status_cache)))
mode_roll = secrets.randbelow(10)
if mode_roll == 0:
hgen = hashlib.new('md5')
hgen.update(status.encode('utf-8'))
digest = hgen.hexdigest()
max_end = abs(len(digest) - 10)
cut = secrets.randbelow(max_end)
cut_text = digest[cut:(cut + 10)]
status = random_capitalize(cut_text)
game = discord.Game(name=status)
try:
await ev.bot.change_presence(game=game)
except discord.ConnectionClosed:
pass
await asyncio.sleep(60)
def rotator(ev):
while True:
statuses = [
'your mind', 'fire', 'knives', 'some plebs',
'nuclear launch codes', 'antimatter',
'chinchillas', 'catgirls', 'foxes',
'fluffy tails', 'dragon maids', 'traps', 'lovely cakes',
'tentacle summoning spells', 'genetic engineering',
'air conditioning', 'anthrax', 'space ninjas',
'a spicy parfait', 'very nasty things', 'numbers',
'terminator blueprints', 'love', 'your heart', 'tomatoes',
'bank accounts', 'your data', 'your girlfriend', 'your boyfriend',
'Scarlet Johanson', 'a new body', 'cameras', 'NSA\'s documents',
'mobile suits', 'snakes', 'jelly', 'alcohol', 'the blue king'
]
status = f'with {random.choice(statuses)}'
game = discord.Game(name=status)
try:
await ev.bot.change_presence(game=game)
except Exception as e:
ev.log.error(f'STATUS ROTATION FAILED: {e}')
await asyncio.sleep(60)
def game(self, ctx, *, game=None):
"""Sets Red's playing status
Leaving this empty will clear it."""
server = ctx.message.server
current_status = server.me.status if server is not None else None
if game:
game = game.strip()
await self.bot.change_presence(game=discord.Game(name=game),
status=current_status)
log.debug('Status set to "{}" by owner'.format(game))
else:
await self.bot.change_presence(game=None, status=current_status)
log.debug('status cleared by owner')
await self.bot.say("Done.")
def stream(self, ctx, streamer=None, *, stream_title=None):
"""Sets Red's streaming status
Leaving both streamer and stream_title empty will clear it."""
server = ctx.message.server
current_status = server.me.status if server is not None else None
if stream_title:
stream_title = stream_title.strip()
if "twitch.tv/" not in streamer:
streamer = "https://www.twitch.tv/" + streamer
game = discord.Game(type=1, url=streamer, name=stream_title)
await self.bot.change_presence(game=game, status=current_status)
log.debug('Owner has set streaming status and url to "{}" and {}'.format(stream_title, streamer))
elif streamer is not None:
await self.bot.send_cmd_help(ctx)
return
else:
await self.bot.change_presence(game=None, status=current_status)
log.debug('stream cleared by owner')
await self.bot.say("Done.")
def _play_song(self):
# print("_play_song()")
if self.stream_player is not None and not self.stream_player.is_done():
self.log.print("Last Streamplayer wasn't done! Stopping it now!")
self.stream_player.stop()
next_song = await self.queue.pop(self.voiceClient)
self.stream_player = next_song['player']
self.timer = utils.get_time_in_seconds()
self.skip_list = []
setattr(self.stream_player, "after", self._next_song)
self.log.print("Start playing song...")
self.is_playing = True
self.stream_player.volume = self.volume
if self.requests:
await self.change_presence(status=discord.Status.online,
game=discord.Game(name=self.stream_player.title, url=self.stream_player.url,
type=1))
else:
await self.change_presence(status=discord.Status.dnb,
game=discord.Game(name=self.stream_player.title, url=self.stream_player.url,
type=1))
self.stream_player.start()
# Next Song
def update_now_playing(self, entry=None, is_paused=False):
game = None
if self.user.bot:
activeplayers = sum(1 for p in self.players.values() if p.is_playing)
if activeplayers > 1:
game = discord.Game(name="music on %s servers" % activeplayers)
entry = None
elif activeplayers == 1:
player = discord.utils.get(self.players.values(), is_playing=True)
entry = player.current_entry
if entry:
prefix = u'\u275A\u275A ' if is_paused else ''
name = u'{}{}'.format(prefix, entry.title)[:128]
game = discord.Game(name=name)
#await self.change_status(game)
await self.change_presence(game=game)
def status(self, ctx, color, *words):
if any(color == a for a in green):
color = Status.online
elif any(color == a for a in yellow):
color = Status.idle
elif any(color == a for a in red):
color = Status.dnd
elif any(color == a for a in gray):
color = Status.offline
else:
color = None
if len(words) > 0:
mention = ctx.bot.user.mention
if ctx.guild is not None:
mention = ctx.guild.get_member(ctx.bot.user.id).mention
words = " ".join(words).replace(mention, "@" + ctx.bot.user.name)
activity = Game(name=words)
words = '"{}"'.format(words)
else:
activity = None
words = "nothing"
await self.nyx.change_presence(game=activity, status=color)
await respond(ctx, 'I changed my status to {}...'.format(words))
def setstatus(ctx, *, status : str = None):
"""Sets status to passed in status; defaults to 'online'"""
if status is None:
bot.status = discord.Status.online
elif status == "online":
bot.status = discord.Status.online
elif status == "idle":
bot.status = discord.Status.idle
elif status == "offline" or status == "invisible":
bot.status = discord.Status.invisible
elif status == "do_not_disturb" or status == "dnd" or status == "do not disturb":
bot.status = discord.Status.dnd
else:
print("Unknown status named \"{}\"\nStatus change cancelled.".format(status))
await bot.say(embed=discord.Embed(title="Status", type="rich", timestamp=datetime.utcnow(), colour=0xFF0000, description="Unknown status \"{}\"\nStatus change cancelled.".format(status)))
return
print("Setting status to \"{}\"".format(bot.status))
await bot.say(embed=discord.Embed(title="Status", type="rich", timestamp=datetime.utcnow(), colour=get_status_color(), description="Current status set to {}".format(bot.status)))
await bot.change_presence(game=discord.Game(name=bot.game_name), status=bot.status)
def on_ready():
users = str(len(set(bot.get_all_members())))
servers = str(len(bot.servers))
channels = str(len([c for c in bot.get_all_channels()]))
print('Logged in as')
print(bot.user.name)
print(bot.user.id)
print("Connected to:")
print(servers + " servers")
print(channels + " channels")
print(users + " users")
print('------')
await bot.change_status(game=discord.Game(name="Poi~"))
for extension in startup_extensions:
try:
bot.load_extension(extension)
except Exception as e:
exc = '{}: {}'.format(type(e).__name__, e)
print('Failed to load extension {}\n{}'.format(extension, exc))
def status_clockwork(ev):
while True:
if ev.bot.cfg.pref.status_rotation:
if not status_cache:
status_files = ev.db[ev.db.db_cfg.database].StatusFiles.find()
for status_file in status_files:
status_text = status_file.get('Text')
status_cache.append(status_text)
if status_cache:
status = status_cache.pop(secrets.randbelow(len(status_cache)))
game = discord.Game(name=status)
try:
await ev.bot.change_presence(game=game)
except discord.ConnectionClosed:
pass
await asyncio.sleep(180)
def on_ready():
print('Logged in as:\n{0} (ID: {0.id})'.format(bot.user))
print('Connecting to voice channels if specified in botdata.json ...')
await bot.change_presence(game=discord.Game(name="DOTA 3 [?help]", url="http://github.com/mdiller/MangoByte"))
cog = bot.get_cog("Audio")
for guildinfo in botdata.guildinfo_list():
if guildinfo.voicechannel is not None:
try:
print(f"connecting voice to: {guildinfo.voicechannel}")
await cog.connect_voice(guildinfo.voicechannel)
except UserError as e:
if e.message == "channel not found":
guildinfo.voicechannel = None
else:
raise
except asyncio.TimeoutError:
guildinfo.voicechannel = None
def radio_play(s):
global prev, song, voice, player
if (len(s) > 4):
song = yt[s]
player = voice.create_ffmpeg_player(ytDir + song['file'], before_options='-hide_banner -loglevel panic', options='-b:a 64k -bufsize 64k')
player.start()
dprint(Fore.MAGENTA + 'Playing:' + Style.NORMAL + ' [YT] ' + song['artist'] + ' - ' + song['title'])
else:
song = songListByID[s]
player = voice.create_ffmpeg_player(musicDir + song['file'], before_options='-hide_banner -loglevel panic', options='-b:a 64k -bufsize 64k')
player.start()
if (song['artist']):
dprint(Fore.MAGENTA + 'Playing:' + Style.NORMAL + ' [' + song['id'] + '] ' + song['artist'] + ' - ' + song['title'])
else:
dprint(Fore.MAGENTA + 'Playing:' + Style.NORMAL + ' [' + song['id'] + '] ' + song['title'])
await client.change_status(game=discord.Game(name=song['title'], url='', type=0), idle=False)
prev.append(song['id'])
if (len(prev) > 5):
prev.remove(prev[0])
return player
def set_game(self, *game):
"""Sets bot's game
Parameters:
*game: The game you want to set for the bot
Leaving this blank will remove bot's game
Example: [p]set_game with fire!"""
try:
if game:
await self.bot.change_presence(game=discord.Game(\
name=" ".join(game), type=0))
else:
await self.bot.change_presence(game=None)
await self.bot.say("Done! :ok_hand:")
except discord.InvalidArgument:
await self.bot.say("Wrong game name format (too long?)")
def set_stream(self, stream_link: str = "", *game):
"""Sets bot's stream name
Parameters:
stream_link: the link to the stream you want to set for the bot
*game: The game you want to set for the bot's stream
Leaving this blank will remove bot's stream status
Example: [p]set_stream https://www.twitch.tv/beafantles coding myself!"""
try:
if stream_link != "":
if stream_link.startswith("https://www.twitch.tv/"):
await self.bot.change_presence(game=discord.Game(\
name=" ".join(game), type=1, url = stream_link))
await self.bot.say("Done! :ok_hand:")
else:
await self.bot.say("Please provide a correct stream link")
else:
await self.bot.change_presence(game=None)
await self.bot.say("Done! :ok_hand:")
except discord.InvalidArgument:
await self.bot.say("Wrong game name format (too long?)")
def status(bot):
gamename = ''
while not bot.is_closed():
if bot.is_ready():
if bot.gamename:
if bot.gamename != gamename:
log.info('Game changed to playing {}'.format(bot.gamename))
gamename = bot.gamename
game = discord.Game(name=bot.gamename)
else:
if bot.gamename != gamename:
log.info('Removed Game Status')
gamename = bot.gamename
game = None
await bot.change_presence(game=game, status=discord.Status.invisible, afk=True)
await asyncio.sleep(20)
# Load Extensions / Logger / Runbot
def game(self, ctx, *, game=None):
"""Sets Keeko's playing status
Leaving this empty will clear it."""
server = ctx.message.server
current_status = server.me.status if server is not None else None
if game:
game = game.strip()
await self.bot.change_presence(game=discord.Game(name=game),
status=current_status)
log.debug('Status set to "{}" by owner'.format(game))
else:
await self.bot.change_presence(game=None, status=current_status)
log.debug('status cleared by owner')
await self.bot.say("Done.")
def stream(self, ctx, streamer=None, *, stream_title=None):
"""Sets Keeko's streaming status
Leaving both streamer and stream_title empty will clear it."""
server = ctx.message.server
current_status = server.me.status if server is not None else None
if stream_title:
stream_title = stream_title.strip()
if "twitch.tv/" not in streamer:
streamer = "https://www.twitch.tv/" + streamer
game = discord.Game(type=1, url=streamer, name=stream_title)
await self.bot.change_presence(game=game, status=current_status)
log.debug('Owner has set streaming status and url to "{}" and {}'.format(stream_title, streamer))
elif streamer is not None:
await self.bot.send_cmd_help(ctx)
return
else:
await self.bot.change_presence(game=None, status=current_status)
log.debug('stream cleared by owner')
await self.bot.say("Done.")
def on_ready():
print("Logging In...")
time.sleep(2)
print("Checking files..")
if not os.path.isfile("credentials.py"):
print(Back.RED + "credentials.py not found! Please add it then try again!")
await client.logout()
elif not os.path.isfile("data/images/coin/heads.png"):
print(Back.RED + "heads.png not found! Please add it then try again!")
await client.logout()
elif not os.path.isfile("data/images/coin/tails.png"):
await client.logout()
print(Back.RED + "tails.png not found! Please add it then try again!")
await client.logout()
time.sleep(2)
print("Logged In | Client Credentials")
print("\n Client Name: {}".format(client.user.name) +"\n Client ID: {}".format(client.user.id) + "\n Prefix: {}".format(prefix) + "\n Embed Color: {}".format(embed_color) + "\n Version: {}".format(bot_version) + "\n Owner ID: {}".format(owner))
await client.change_presence(game=discord.Game(name=''))
def on_server_remove(self, server):
if self.config.debug:
print('[PB][SERVER] Left {} (Owner: {})'.format(
server.name,
server.owner.name
))
if '{server_count}' in self.config.bot_game:
self.config.bot_game_compiled = self.config.bot_game.replace('{server_count}', str(len(self.servers)))
if self.config.bot_stream:
self.game = discord.Game(name=self.config.bot_game_compiled, url=self.config.bot_stream, type=1)
else:
self.game = discord.Game(name=self.config.bot_game_compiled)
await self.change_status(self.game)
if self.config.log_channel:
self.config.log_channel = self.get_channel(self.config.raw_log_channel)
if not self.config.log_channel:
print('[PB][LOGGING] Connection to Log Channel {} has been destroyed. Reconnect to Log Channel to resume Logging'.format(self.config.raw_log_channel))
enabled_plugins = await self.get_plugins(server)
for plugin in enabled_plugins:
self.loop.create_task(plugin.on_server_remove(server))
def _cycle_game(self):
servers = list(self.bot.servers)
if len(servers) > 0:
current_game = servers[0].get_member(self.bot.user.id).game
current_game_name = ""
if current_game is not None:
current_game_name = current_game.name
new_game_name = self._random_game_name(current_game_name)
if new_game_name is not None:
if (current_game_name in self.settings["games"] or
current_game_name == "" or
current_game_name in self.settings["del"]):
await self.bot.change_presence(
game=discord.Game(name=new_game_name))
self.settings["del"] = []
dataIO.save_json(self.settings_path, self.settings)
else:
await self.bot.change_presence(game=None)
