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类Bot()的实例源码
def quote(ctx, message_id: str=None):
if message_id is None:
em = discord.Embed(
title="Error", description="The bot encountered an error; The error is no message id found.", colour=0xFF5959)
em.set_author(name=bot.user.display_name, icon_url=bot.user.avatar_url)
# em.set_footer(set the default datetime thing ive been using)
await bot.edit_message(ctx.message, embed=em)
else:
async for message in bot.logs_from(ctx.message.channel, limit=500):
if message.id == message_id:
if message is not None:
em = discord.Embed(title=message.content, colour=0x33CC66)
em.set_author(name=message.author.display_name,
icon_url=message.author.avatar_url)
em.set_footer(
text='Discordian Self-Bot at {}'.format(strftime("%Y-%m-%d %H:%M:%S", gmtime())))
await bot.edit_message(ctx.message, embed=em)
elif message is None:
print("Message with id of -{}- was not found".format(message_id))
# post the embed error but altered.`
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 on_command_error(ctx, error):
if isinstance(error, commands.CommandNotFound):
return
# if isinstance(error, commands.CommandInvokeError):
# return print(error)
errors = {
commands.DisabledCommand: 'Command has been disabled.',
commands.MissingPermissions: 'Invoker is missing permissions to run this command.',
commands.BotMissingPermissions: 'Bot is missing permissions to run this command.',
commands.CheckFailure: 'You are not allowed to run this command.'
}
for type, text in errors.items():
if isinstance(error, type):
return await ctx.send(errors[type])
# argument error
if isinstance(error, commands.UserInputError):
bot.formatter.context = ctx
bot.formatter.command = ctx.command
return await ctx.send(f'Invalid argument(s) provided.\n```{bot.formatter.get_command_signature()}```')
await ctx.send(f'An error occured in `{ctx.command.name}` invoked by {ctx.message.author}:\n```{error}```')
#traceback.print_exception(type(exception), exception, exception.__traceback__, file=sys.stderr)
# blacklist check
# check is user is blacklister, or if it's a bot
def __init__(self, bot):
self.bot = bot # type: commands.Bot
self.config = self.bot.cogs['Config'] # type: Config
# Uses the database proxy object for our db as we don't know which database provider to use until runtime.
self.db, self.models = models.get_models(self.config)
self.query = Query(self.models)
# Is there a nice way to scope all models within AryaORM? Still want them defined in a separate file.
# Have I got a surprise for you, my dear Tom
self.User = self.models.User
self.Message = self.models.Message
self.Channel = self.models.Channel
self.Server = self.models.Server
self.LoveTransaction = self.models.LoveTransaction
self.make_tables()
def __init__(self, bot):
self.bot: commands.Bot
self.bot = bot
# owner ids
self.owners = self.bot.owners
# praw magical thread magic
self.r = self.bot.r
# a class to hold all tg's precious global variables
self.v = self.bot.v
# aiohttp session for magix
self.session = self.bot.session
self.bot.loop.create_task(self.game_updater())
self.bot.loop.create_task(self.self_cleaner())
self.bot.loop.create_task(self.totmem())
self.bot.loop.create_task(self.new_part_checker())
# ===============
# bot commands
# ===============
def __init__(self, bot: Bot):
self.bot = bot
def setup(settings):
print("First time setup, prepare your anus for some questions.")
token = input("What's your Discord token? (To see how to get it go to https://github.com/PlanetTeamSpeakk/DiscordSelfBot#token)\n")
if token.startswith("\""):
token = token[1:]
if token.endswith("\""):
token = token[:(len(token) - 1)]
prefix = input("What should your prefix be?\n")
invite = input("What's the permanent invite link for you Discord server? Type None if you don't have one.\n")
settings['token'] = token
settings['prefix'] = prefix
settings['invite'] = invite
bot = commands.Bot(command_prefix=prefix, description=description, self_bot=True)
settings_file = None
with open("data/dsb/settings.json", "w") as settings_file:
json.dump(settings, settings_file, indent=4, sort_keys=True, separators=(',', ' : '))
print("You're all set! Bot is starting, don't mind the unclosed client session part, just wait a bit.")
def main():
# define bot
bot = Bot(
description=config.description,
verbose=config.verbose,
bleeding=config.bleeding,
reactive=config.reactive
)
bot.add_cog(cogs.Fun (bot))
bot.add_cog(cogs.Stats(bot))
bot.add_cog(cogs.Info (bot))
bot.add_cog(cogs.Admin(bot))
# launch bot
try:
bot.run(config.token)
except discord.errors.LoginFailure as e:
print(e, end='\n\n')
def on_ready():
print('Logged in as')
print(f'Bot-Name: {bot.user.name}')
print(f'Bot-ID: {bot.user.id}')
print(f'Dev Mode: {bot.dev}')
print('------')
for cog in loadconfig.__cogs__:
try:
bot.load_extension(cog)
except Exception:
print(f'Couldn\'t load cog {cog}')
bot.commands_used = Counter()
bot.startTime = time.time()
bot.botVersion = __version__
bot.userAgentHeaders = {'User-Agent': f'linux:shinobu_discordbot:v{__version__} (by Der-Eddy)'}
bot.gamesLoop = asyncio.ensure_future(_randomGame())
_setupDatabase('reaction.db')
def __init__(self, bot : commands.Bot):
print("initializing github tracker")
self.bot = bot
# Registered channel format:
# {channel_id : ("repo", "owner/repo")}
# {"167319706863140864" : ("repo", "professional-programmingers/GeBeO")}
self.registered_channels = {}
self.file_name = "cache/github.json"
self.api_base = "https://api.github.com/"
# Create a cache dir if it doesn't exists.
if not os.path.exists("cache"):
os.makedirs("cache")
# Check for cache file.
if os.path.isfile(self.file_name) and os.stat(self.file_name).st_size != 0:
f = open(self.file_name, "r")
self.registered_channels = json.loads(f.read())
else:
f = open(self.file_name, "w+")
f.close()
def add_sound(self, ctx, source, sourcetype):
""" Add a sound to a bot's queue. """
if ctx.message.author.voice == None:
await ctx.send("You're not in a voice channel!")
else:
# Check if any bots in guild. Warn user if not.
# TODO: Move to Bot Manager.
for helper in self.bot.helperList:
if helper.is_in_guild(ctx.guild):
break
else:
await ctx.send("No helper bots in this server. Do !invite to add them.")
return
vchan_id = ctx.message.author.voice.channel.id
sound = await self.parse_sound(source, sourcetype)
# Find a bot and add to its queue.
helper = self.choose_helper(vchan_id)
if helper != None:
await helper.queue_sound(vchan_id, sound)
await ctx.send("Queueing sound!")
else:
await ctx.send("Sorry, there are no available bots!")
def user(ctx, *, member: discord.Member = None):
if member is None:
member = ctx.message.author
embed = discord.Embed(color = embed_color)
embed.set_thumbnail(url = member.avatar_url)
embed.add_field(name="User ID:", value=member.id, inline=True)
embed.add_field(name="User Name:", value=member, inline=True)
embed.add_field(name="Is Bot?:", value=member.bot, inline=True)
embed.add_field(name="Join Date:", value=member.created_at, inline=True)
embed.add_field(name="Nickname:", value=member.display_name, inline=True)
await client.say(embed = embed)
else:
embed = discord.Embed(color = embed_color)
embed.set_thumbnail(url = member.avatar_url)
embed.add_field(name="User ID:", value=member.id, inline=True)
embed.add_field(name="User Name:", value=member, inline=True)
embed.add_field(name="Is Bot?:", value=member.bot, inline=True)
embed.add_field(name="Join Date:", value=member.created_at, inline=True)
embed.add_field(name="Nickname:", value=member.display_name, inline=True)
await client.say(embed = embed)
print(Fore.CYAN + "Command Successfully Executed |\n Command Ran In:[" + ctx.message.server.id + "]\n User:[" + ctx.message.author.id + "]\n Channel:[" + ctx.message.channel.id + "]")
def setup(bot: commands.Bot):
check_folders()
check_files()
if dateutil_available:
if pytz_available:
if tabulate_available:
bot.add_cog(Survey(bot))
else:
raise RuntimeError(
"You need to install `tabulate`: `pip install tabulate`.")
else:
raise RuntimeError(
"You need to install `pytz`: `pip install pytz`.")
else:
raise RuntimeError(
"You need to install `python-dateutil`:"
" `pip install python-dateutil`.")
def __init__(self, bot: commands.Bot):
self.bot = bot
self.settings_path = "data/rainbow6siege/settings.json"
self.settings = dataIO.load_json(self.settings_path)
self.platform_map = {
"uplay": r6sapi.Platforms.UPLAY,
"xbox": r6sapi.Platforms.XBOX,
"playstation": r6sapi.Platforms.PLAYSTATION
}
self.region_map = {
"na": r6sapi.RankedRegions.NA,
"eu": r6sapi.RankedRegions.EU,
"asia": r6sapi.RankedRegions.ASIA
}
self.operator_list = [
x.lower() for x in r6sapi.OperatorIcons if x != 'DEFAULT'
]
def main(loop=None):
if loop is None:
loop = asyncio.get_event_loop()
bot = Bot(loop=loop, command_prefix=core.get_prefixes(), description=__doc__, pm_help=core.is_help_private())
if not is_configured():
initial_config()
error = False
error_message = ""
try:
loop.run_until_complete(run(bot))
except discord.LoginFailure:
error = True
error_message = 'Invalid credentials'
choice = input(strings.invalid_credentials)
if choice.strip() == 'reset':
base.delete_token()
else:
base.disable_restarting()
except KeyboardInterrupt:
base.disable_restarting()
loop.run_until_complete(bot.logout())
except Exception as e:
error = True
print(e)
error_message = traceback.format_exc()
base.disable_restarting()
loop.run_until_complete(bot.logout())
finally:
if error:
print(error_message)
return bot
def __init__(self, sparcli:commands.Bot):
self.sparcli = sparcli
logChannel = getTokens()['BotLoggingChannel']
self.discordBotsToken = getTokens()['DiscordBotsPw']['Key']
self.logChannel = sparcli.get_channel(logChannel)
self.session = ClientSession(loop=sparcli.loop)
self.fserver = None
def setup(bot:commands.Bot):
x = BotLogger(bot)
bot.add_cog(x)
def __init__(self, sparcli:commands.Bot):
self.sparcli = sparcli
self.session = ClientSession(loop=sparcli.loop)
self.regexMatches = {
'OriginalRequest': r'(.+[a-zA-Z0-9]+\s[0-9]+:[0-9]+(-[0-9]+)?)',
'StripAuthor': r'(.+\b)([0-9]+:)',
'GetPassages': r'([0-9]+:[0-9]+)',
'GetMax': r'(-[0-9]+)',
'QuranMatch': r'([0-9]+:[0-9]+([-0-9]+)?)'
}
self.biblePicture = 'http://pacificbible.com/wp/wp-content/uploads/2015/03/holy-bible.png'
self.quranPicture = 'http://www.siotw.org/modules/burnaquran/images/quran.gif'
def on_ready():
print('Logged in as')
print(bot.user.name)
print(bot.user.id)
print('------')
logger.info('Bot started as ' + bot.user.name)
for server in bot.servers:
logger.info(' ' + server.name)
await bot.change_presence(game=discord.Game(name='eRepublik'))
def on_server_join(server):
logger.info('Bot joined: ' + server.name)
def on_server_remove(server):
logger.info('Bot left: ' + server.name)
def __init__(self, get, update_channel, **options):
"""Init AutomaBot.
:param get: The Queue reader side.
:param update_channel: The notification channel id
:param **options: Default commands.Bot parameters
"""
super().__init__(**options)
self.get = get
self.update_channel = update_channel
self.terminal_width = shutil.get_terminal_size((80, 20))[0]
def ping(ctx):
msg = discord.Embed(title='Pong!', colour=0x66CC99)
msg.set_author(name=bot.user.display_name, icon_url=bot.user.avatar_url)
msg.set_footer(
text='Discordian Self-Bot at {}'.format(strftime("%Y-%m-%d %H:%M:%S", gmtime())))
await bot.edit_message(ctx.message, embed=msg)
def setup(bot: commands.Bot) -> None:
bot.add_cog(AryasORM(bot))
def __init__(self, bot):
self.bot: commands.Bot = bot
self.orm: AryasORM = self.bot.cogs['AryasORM']
self.config: Config = self.bot.cogs['Config']
def setup(bot: commands.Bot):
bot.add_cog(Statistics(bot))
def __init__(self, bot):
self.bot: commands.Bot = bot
def setup(bot: commands.Bot) -> None:
bot.add_cog(Config())
def __init__(self, bot):
self.bot: commands.Bot = bot
self.orm: AryasORM = self.bot.cogs['AryasORM']
self.config: Config = self.bot.cogs['Config']
