-
-
Notifications
You must be signed in to change notification settings - Fork 9
/
bot.py
165 lines (149 loc) · 6.33 KB
/
bot.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
import discord
import os
import asyncpg
from datetime import datetime
from discord.ext import commands
class Bot(commands.Bot):
CC_LOGO_URL = 'https://avatars.githubusercontent.com/u/63065397?v=4'
INTRO_IMG_URL = 'https://user-images.githubusercontent.com/63065397/156466208-ffb6db84-f0c0-4860-ab6d-48ad0f2cd5f7.png'
EXHAUSTED_FACE = 'https://user-images.githubusercontent.com/63065397/156922064-95c73c2a-b6cb-402e-b24b-d79fe7bf520a.png'
DEX_YELLOW = 0x8e38ce
REPOSITORY_URL = 'https://github.com/code-chaser/dex/'
def __init__(self, *args, **kwargs):
super().__init__(
command_prefix=self.get_pref,
intents=discord.Intents.all(),
activity=discord.Activity(
type=discord.ActivityType.listening,
name="$dex help",
large_image_url=self.EXHAUSTED_FACE,
small_image_url=self.EXHAUSTED_FACE,
start=datetime(2022, 2, 24),
),
)
self.DB_CONNECTION = None
self.DATABASE = {}
self.loop.create_task(self.startup())
for file in os.listdir('./src/cogs'):
if file.endswith('.py'):
self.load_extension(f'src.cogs.{file[:-3]}')
async def connect_to_db(self) -> None:
self.DB_CONNECTION = await asyncpg.connect(
host=os.getenv('DEX_DB_HOST'),
database=os.getenv('DEX_DB_NAME'),
user=os.getenv('DEX_DB_USER'),
port=os.getenv('DEX_DB_PORT'),
password=os.getenv('DEX_DB_PASSWORD')
)
print("\nDATABASE CONNECTED\n")
async def clone_database(self):
await self.DB_CONNECTION.execute('CREATE TABLE IF NOT EXISTS guilds (guild_id VARCHAR(27) NOT NULL, prefix VARCHAR(108) NOT NULL, tag_messages SWITCH NOT NULL, PRIMARY KEY (guild_id));')
self.DATABASE['guilds'] = {}
self.DATABASE['guilds'] = {result['guild_id']: {k: v for k, v in result.items() if k != 'guild_id'} for result in await self.DB_CONNECTION.fetch("SELECT * FROM guilds")}
print("\nDATABASE CLONED\n")
print("\n\n****DATABASE DICTIONARY****\n\n")
print(self.DATABASE)
print("\n\n")
return
async def startup(self):
print("\nINSIDE Bot.startup()\n")
await self.connect_to_db()
await self.clone_database()
def get_pref(self, _, message):
return self.DATABASE['guilds'][str(message.guild.id)]['prefix']
def run(self) -> None:
super().run(os.getenv('DEX_BOT_TOKEN'))
async def on_ready(self):
print('Logged in as {0.user}'.format(self))
async def on_guild_join(self, guild):
if str(guild.id) in self.DATABASE['guilds'].keys():
return
await self.DB_CONNECTION.execute('INSERT INTO guilds (guild_id,prefix,tag_messages) VALUES (\'' +
str(guild.id)+'\', \'$dex \', \'on\');')
self.DATABASE['guilds'][str(guild.id)] = {}
self.DATABASE['guilds'][str(guild.id)] = {
'prefix': '$dex ', 'tag_messages': 'on'}
for channel in guild.text_channels:
if channel.permissions_for(guild.me).send_messages:
general = channel
if general is not None:
await general.send(embed=self.intro_msg_embed(guild))
return
async def on_guild_remove(self, guild):
await self.DB_CONNECTION.execute('DELETE FROM guilds WHERE guild_id = \'' +
str(guild.id) + '\';')
if str(guild.id) in self.DATABASE['guilds'].keys():
self.DATABASE['guilds'].pop(str(guild.id))
return
async def on_message(self, message):
await self.process_commands(message)
tag_switch = self.DATABASE['guilds'][str(
message.guild.id)]['tag_messages']
target = message.author
if target == self.user:
return
print("\n\n-----------------------\n-----------------------\n\n" +
str(message.content) + "\n-----------------------\n")
if tag_switch == 'off':
return
embed = discord.Embed(
title='Message Tagged',
colour=target.colour,
timestamp=datetime.utcnow(),
)
embed.set_footer(
text=''.join('`<prefix> tags off` -to turn this off'),
)
embed.add_field(name='Message', value=message.content, inline=False)
embed.add_field(name='Author', value=target.mention, inline=True)
embed.set_thumbnail(url=target.avatar_url)
await message.channel.send(embed=embed)
async def on_command_error(self, ctx, error) -> None:
embed = discord.Embed(
title='Status',
colour=0xff0000,
timestamp=datetime.utcnow(),
)
if isinstance(error, commands.MissingPermissions):
n = 'Error'
v = 'Missing Permissions'
elif isinstance(error, commands.MissingRequiredArgument):
n = 'Error'
v = 'Missing required arguements'
elif isinstance(error, commands.MemberNotFound):
n = 'Error'
v = "Requested member not found or Dex doesn't have access to them"
elif isinstance(error, commands.BotMissingPermissions):
n = 'Error'
v = 'Missing Permissions'
elif isinstance(error, commands.CommandNotFound):
n = 'Error'
v = 'Invalid Command'
else:
raise error
embed.add_field(name=n, value=v, inline=False)
await ctx.send(embed=embed)
def intro_msg_embed(self, guild):
description = ('\nThanks for adding me to ' + guild.name + '!')
description += ('\nUse `$dex help` to get started!')
description += ('\nSource Code: [Link](')
description += (self.REPOSITORY_URL)
description += (')')
embed = discord.Embed(
title='**GREETINGS!**',
description=description,
color=self.DEX_YELLOW,
timestamp=datetime.utcnow(),
)
embed.set_image(url=self.INTRO_IMG_URL)
embed.set_author(
name='dex',
url=self.REPOSITORY_URL,
icon_url=self.user.avatar_url,
)
embed.set_footer(
text='made by codechaser',
icon_url=self.CC_LOGO_URL,
)
embed.set_thumbnail(url=guild.icon_url)
return embed