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
|
import asyncio
from typing import Any, List
import random
import config
from discord.ext import commands
async def await_n(it) -> List[Any]:
lst = [asyncio.create_task(task) for task in it]
if not lst:
return []
done, _ = await asyncio.wait(lst)
return [task.result() for task in done]
async def answer(ctx: commands.Context, msg: str):
await ctx.send(f'{ctx.author.mention} {msg}')
@commands.command(help='display this help message', aliases=('hepl', 'h', '?'))
async def help(ctx: commands.Context):
command_doc = '\n'.join(
f' * {config.COMMAND_PREFIX.strip()} {c.name:15} - {c.help}'
for c in ctx.bot.commands)
await answer(ctx, f'''```
{config.HELP_TEXT}\nThese are all available commands:\n{command_doc}```''')
@commands.command(help='create a new lobby', aliases=('create', 'inti', 'craete', 'cretae', 'c', 'i', '+'))
async def init(ctx: commands.Context):
ctx.bot.meta_channel = ctx.bot.get_meta_channel(ctx) or await ctx.bot.create_category(ctx)
ctx.bot.lobby_channel = ctx.bot.get_lobby_channel(ctx, ctx.bot.meta_channel) or await ctx.bot.create_lobby(ctx)
@commands.command(help=f'destruct all {config.NAME} channels', aliases=('kill', 'desctruction', 'genocide', '-'))
async def destroy(ctx: commands.Context):
await stop(ctx)
futures = []
meta_channel = ctx.bot.get_meta_channel(ctx)
for channel in (ctx.bot.get_lobby_channel(ctx, meta_channel), meta_channel):
if channel:
futures.append(channel.delete())
await await_n(futures)
ctx.bot.lobby_channel = None
ctx.bot.meta_channel = None
ctx.bot.pair_channels = []
@commands.command(help='start shuffling', aliases=('start', 'run', 'strat', 'rnu'))
async def shuffle(ctx: commands.Context, channel_size=2):
if channel_size < 1:
await answer(ctx, 'error: channel size must be at least 1 (you jerk)')
return False
channels = await ctx.bot.get_channels(ctx)
if not channels:
return False
meta_channel, lobby_channel = channels
members = await await_n(map(ctx.guild.fetch_member, lobby_channel.voice_states.keys()))
if not members:
await answer(ctx, 'error: nobody wants to get shuffeled :(')
return False
if len(members) < channel_size:
await answer(ctx, f'error: you are too few people ({len(members)}). Group size is {channel_size}')
return False
channel_count, excess_slots = divmod(len(members), channel_size)
normals = []
tutors = []
random.shuffle(members)
for member in members:
if member.id in ctx.bot.tutors:
tutors.append(member)
else:
normals.append(member)
futures = []
ctx.bot.pair_channels = await ctx.bot.create_pair_channels(ctx, meta_channel, channel_count)
random.shuffle(ctx.bot.pair_channels)
groups = [[] for _ in range(channel_count)]
for i, member in enumerate(tutors + normals):
n = i % channel_count
futures.append(member.move_to(ctx.bot.pair_channels[n]))
groups[n].append(member)
for group in groups:
for member in group:
futures.append(ctx.bot.send_panel(ctx, member, group))
await await_n(futures)
return True
@commands.command(help='move everyone back to lobby', aliases=('quit', 'exit', 'abort', 'back', 'return'))
async def stop(ctx: commands.Context, cancel_loop=True):
channels = await ctx.bot.get_channels(ctx)
if not channels:
return
meta_channel, lobby_channel = channels
pair_channels = ctx.bot.get_pair_channels(ctx, meta_channel)
if cancel_loop:
await ctx.bot.cancel_coroutine()
futures = []
for channel in pair_channels:
for member in channel.members:
futures.append(member.move_to(lobby_channel))
await await_n(futures)
await ctx.bot.destroy_pair_channels(ctx, meta_channel)
async def loop_cycle(ctx, t):
if not await shuffle(ctx):
return False
await asyncio.sleep(t)
await ctx.bot.update_scores(ctx.guild)
await stop(ctx, cancel_loop=False)
return True
@commands.command(
help=f'repeat "shuffle" and "stop" <n> (default: {config.DEFAULT_LOOP_COUNT}) times and <t>'
f' (default: {config.DEFAULT_LOOP_TIME}) seconds',
aliases=('repeat', 'shuffleloop'))
async def loop(ctx: commands.Context, n=config.DEFAULT_LOOP_COUNT, t=config.DEFAULT_LOOP_TIME):
try:
n, t = int(n), int(t)
except ValueError:
await answer(ctx, 'error: expecting positive integer arguments for <n> and <t> (e.g. "loop 5 60" to repeat 5 times)')
return
await answer(ctx, f'repeat shuffling {n} times and each {t} seconds')
for _ in range(n):
result = await ctx.bot.await_coroutine(loop_cycle(ctx, t))
message = {
'cancelled': 'cancelled loop command',
'already running': 'cannot start loop, another task is running...'
}.get(result, None)
if message is not None:
return await answer(ctx, f'error: {message}')
if not result:
break
@commands.command(
help='add a tutor',
aliases=('tutor', 'hello')
)
async def add(ctx: commands.Context, id=None):
if id is None:
return await answer(ctx, 'error: expecting user id')
try:
id = int(id)
except ValueError:
return await answer(ctx, 'error: invalid user id')
ctx.bot.add_tutor(id)
await answer(ctx, f'added user id {id} to the tutor list')
@commands.command(
help='remove a tutor',
aliases=('rm', 'del', 'delete', 'byebye', 'bb')
)
async def remove(ctx: commands.Context, id=None):
if id is None:
return await answer(ctx, 'error: expecting user id')
try:
id = int(id)
except ValueError:
return await answer(ctx, 'error: invalid user id')
ctx.bot.remove_tutor(id)
await answer(ctx, f'romved user id {id} from the tutor list')
@commands.command(
help='list all tutors',
aliases=('tutors',)
)
async def list(ctx: commands.Context):
if ctx.bot.tutors:
users = []
for tutor in ctx.bot.tutors:
try:
user = ctx.bot.get_user(tutor) or await ctx.bot.fetch_user(tutor)
users.append((str(user), tutor))
except Exception as e:
users.append(('*unknown*', tutor))
await answer(ctx, '```\n' + '\n'.join(f' • {name} ({id})' for name, id in users) + '\n```')
else:
await answer(ctx, 'there is no tutor :/')
bot_commands = [cmd for cmd in locals().values() if isinstance(cmd, commands.Command)]
|
