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
|
import itertools
import logging
from collections import namedtuple
from typing import Dict, Set, ValuesView
from discord import Guild
from discord.ext.commands import Bot
log = logging.getLogger(__name__)
# These objects are declared as namedtuples because tuples are hashable,
# something that we make use of when diffing site roles against guild roles.
Role = namedtuple('Role', ('id', 'name', 'colour', 'permissions'))
User = namedtuple('User', ('id', 'name', 'discriminator', 'avatar_hash', 'roles', 'in_guild'))
def get_roles_for_update(guild_roles: Set[Role], api_roles: Set[Role]) -> Set[Role]:
"""
Determine which roles should be updated on the site.
Arguments:
guild_roles (Set[Role]):
Roles that were found on the guild at startup.
api_roles (Set[Role]):
Roles that were retrieved from the API at startup.
Returns:
Set[Role]:
Roles to be sent to the site for an update or insert.
"""
return guild_roles - api_roles
async def sync_roles(bot: Bot, guild: Guild):
"""
Synchronize roles found on the given `guild` with the ones on the API.
"""
roles = await bot.api_client.get('bot/roles')
api_roles = {Role(**role_dict) for role_dict in roles}
guild_roles = {
Role(
id=role.id, name=role.name,
colour=role.colour.value, permissions=role.permissions.value
)
for role in guild.roles
}
roles_to_update = get_roles_for_update(guild_roles, api_roles)
for role in roles_to_update:
log.info(f"Updating role `{role.name}` on the site.")
await bot.api_client.put(
'bot/roles',
json={
'id': role.id,
'name': role.name,
'colour': role.colour,
'permissions': role.permissions
}
)
def get_users_for_sync(
guild_users: Dict[int, User], api_users: Dict[int, User]
) -> ValuesView[User]:
"""
Obtain a set of users to update on the website.
"""
users_to_create = set()
users_to_update = set()
for api_user in api_users.values():
guild_user = guild_users.get(api_user.id)
if guild_user is not None:
if api_user != guild_user:
users_to_update.add(guild_user)
else:
# User left
api_user._replace(in_guild=False)
users_to_update.add(guild_user)
new_user_ids = set(guild_users.keys()) - set(api_users.keys())
for user_id in new_user_ids:
# User joined
new_user = guild_users[user_id]
users_to_create.add(new_user)
return users_to_create, users_to_update
async def sync_users(bot: Bot, guild: Guild):
"""
Synchronize users found on the given
`guild` with the ones on the API.
"""
current_users = await bot.api_client.get('bot/users')
api_users = {
user_dict['id']: User(
roles=tuple(sorted(user_dict.pop('roles'))),
**user_dict
)
for user_dict in current_users
}
guild_users = {
member.id: User(
id=member.id, name=member.name,
discriminator=int(member.discriminator), avatar_hash=member.avatar,
roles=tuple(sorted(role.id for role in member.roles)), in_guild=True
)
for member in guild.members
}
users_to_create, users_to_update = get_users_for_sync(guild_users, api_users)
log.info("Creating a total of `%d` users on the site.", len(users_to_create))
for user in users_to_create:
await bot.api_client.post(
'bot/users',
json={
'avatar_hash': user.avatar_hash,
'discriminator': user.discriminator,
'id': user.id,
'in_guild': user.in_guild,
'name': user.name,
'roles': list(user.roles)
}
)
log.info("User creation complete.")
log.info("Updating a total of `%d` users on the site.", len(users_to_update))
for user in users_to_update:
if user is None: # ??
continue
await bot.api_client.put(
'bot/users/' + str(user.id),
json={
'avatar_hash': user.avatar_hash,
'discriminator': user.discriminator,
'id': user.id,
'in_guild': user.in_guild,
'name': user.name,
'roles': list(user.roles)
}
)
log.info("User update complete.")
|
