-
Notifications
You must be signed in to change notification settings - Fork 2
/
main.py
226 lines (175 loc) · 6.79 KB
/
main.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
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
"""
Main Script, run from here
Parses Some Args from command line run,
--test=BOOL : Sets config.ini path to use config_test.ini
--loglevel=LEVEL [-l] : Sets loglevel, DEBUG, WARN, INFO etc.
"""
# external imports
import discord
from discord.ext import commands
import logging
import logging.handlers
import sys
import traceback
import asyncio
import argparse
import pathlib
import os
# internal imports
import modules.config as cfg
import modules.accounts_handler
import modules.discord_obj as d_obj
import modules.database
import modules.loader as loader
import modules.signal
import modules.elo_ranks_handler as elo_ranks
import classes
import display
import modules.spam_detector as spam
# parse commandline args
ap = argparse.ArgumentParser()
ap.add_argument('--test', default=False, type=bool)
ap.add_argument('-l', '--loglevel', default='INFO', type=str)
c_args = vars(ap.parse_args())
print(c_args.get('loglevel'))
numeric_level = getattr(logging, c_args.get('loglevel'), None)
if not isinstance(numeric_level, int):
raise ValueError('Invalid log level: %s' % c_args.get('loglevel'))
# Logs Setup
log = logging.getLogger('fs_bot')
log.setLevel(numeric_level)
log_formatter = logging.Formatter('%(asctime)s:%(levelname)s:%(name)s: %(message)s')
# Log to file
log_path = f'{pathlib.Path(__file__).parent.absolute()}/../FSBotData/Logs/'
if not os.path.exists(log_path):
os.makedirs(log_path)
# Log to console
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setFormatter(log_formatter)
log.addHandler(console_handler)
# discord logs
discord_logger = logging.getLogger('discord')
discord_logger.setLevel(logging.INFO)
discord_logger.addHandler(console_handler)
# Auraxium Logging
sh = logging.StreamHandler()
sh.setFormatter(log_formatter)
sh.setLevel(logging.DEBUG if cfg.TEST else logging.WARNING)
auraxium_logger = logging.getLogger('auraxium')
auraxium_logger.setLevel(logging.DEBUG)
auraxium_logger.addHandler(sh)
# Log to file only if not testing
if not c_args.get('test'):
# single_log_handler = logging.FileHandler(filename=log_path, encoding='utf-8', mode='w') # single log
log_handler = logging.handlers.TimedRotatingFileHandler(f'{log_path}fs_bot.log', when='D',
interval=1) # rotating log files, every day
log_handler.setFormatter(log_formatter)
log.addHandler(log_handler)
discord_logger.addHandler(log_handler)
else:
# Auraxium Logging
fh = logging.FileHandler(filename=f"{log_path}fs_auraxium.log", encoding='utf-8', mode='w+')
fh.setFormatter(log_formatter)
auraxium_logger.addHandler(fh)
class StreamToLogger(object):
"""
Fake file-like stream object that redirects writes to a logger instance.
"""
def __init__(self, logger, log_level=logging.INFO):
self.logger = logger
self.log_level = log_level
self.linebuf = ''
def write(self, buf):
for line in buf.rstrip().splitlines():
self.logger.log(self.log_level, line.rstrip())
def flush(self):
pass
# Redirect stdout and stderr to log:
sys.stdout = StreamToLogger(log, logging.INFO)
sys.stderr = StreamToLogger(log, logging.ERROR)
if c_args.get('test'):
cfg.get_config('config_test.ini', test=True)
else:
cfg.get_config('config.ini')
intents = discord.Intents.all()
# TODO move to subclassed bot. Will allow passing more variables through to cogs if needed
bot = commands.Bot(intents=intents)
bot.activity = discord.Game(name="Hello Pilots!")
@bot.event
async def on_ready():
if loader.is_all_loaded():
return
log.info(f"Logged in as {bot.user} (ID: {bot.user.id})")
modules.signal.init(bot)
d_obj.init(bot)
bot.loop.create_task(modules.accounts_handler.init(cfg.GAPI_SERVICE, cfg.TEST), name="Accounts Handler Init")
# loader.load_secondary(bot)
await loader.load_all(bot)
bot.loop.create_task(elo_ranks.init_elo_ranks(), name="Elo Ranks Init")
loader.unlock_all()
loader.set_all_loaded()
# Global Bot Interaction Check
@bot.check
async def global_interaction_check(ctx):
if loader.is_all_locked():
memb = d_obj.guild.get_member(ctx.user.id)
if d_obj.is_admin(memb):
return True
else:
raise AllLocked
if await spam.is_spam(ctx):
return False
# Allow timed out users to use only /freeme command
if not ctx.command.full_parent_name == "freeme" and await d_obj.is_timeout_check(ctx):
return False
return True
# unlock user from spam filter
@bot.after_invoke
async def global_after_invoke(interaction):
spam.unlock(interaction.user.id)
class AllLocked(discord.CheckFailure):
pass
class UserDisabled(discord.CheckFailure):
pass
@bot.event
async def on_application_command_error(context, exception):
command = context.command
if command and command.has_error_handler():
return
cog = context.cog
if cog and cog.has_error_handler():
return
if isinstance(exception, AllLocked):
await display.AllStrings.ALL_LOCKED.send_priv(context)
elif isinstance(exception, UserDisabled):
await display.AllStrings.DISABLED_PLAYER.send_priv(context)
elif isinstance(exception, discord.ext.commands.PrivateMessageOnly):
await display.AllStrings.DM_ONLY.send(context)
elif isinstance(exception, discord.CheckFailure) and not d_obj.is_timeout(context.user):
await display.AllStrings.CHECK_FAILURE.send_priv(context)
else:
try:
await display.AllStrings.LOG_GENERAL_ERROR.send_priv(context, exception)
except (discord.errors.InteractionResponded, discord.errors.NotFound):
pass
finally:
await d_obj.d_log(source=context.user.name, message=f"Ignoring exception in command {context.command}",
error=exception)
# traceback.print_exception(type(exception), exception, exception.__traceback__, file=sys.stderr)
@bot.event
async def on_member_join(member):
"""Ensure proper roles are applied to players on server join and send Join message if Member already verified"""
await d_obj.role_update(member)
if member.pending is False:
await display.AllStrings.SERVER_JOIN.send(d_obj.guild.system_channel, member.mention, mention=member.mention)
@bot.event
async def on_member_update(before: discord.Member, after: discord.Member):
"""Send Join message after member is verified"""
if before.pending is True and after.pending is False:
await display.AllStrings.SERVER_JOIN.send(d_obj.guild.system_channel, after.mention, mention=after.mention)
# database init
modules.database.init(cfg.database)
modules.database.get_all_elements(classes.Player.new_from_data, 'users')
log.info("Loaded Players from Database: %s", len(classes.Player.get_all_players()))
loader.init(bot)
bot.run(cfg.general['token'])