-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathutil.py
199 lines (156 loc) · 6.13 KB
/
util.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
"""Utility functions for use throughout the code."""
import aiohttp
import nextcord
from steam import steamid
from steam.steamid import SteamID
from nextcord.ext import application_checks
from database import get_server, is_server_setup, BotCollection
config_db = BotCollection("guilds", "config")
class ServerNotSetupError(Exception):
"""Exception raised when the server setup process has not been completed.
Attributes:
message -- explanation of the error
"""
def __init__(
self, message="Server is not setup. Please run /setup."
): # pylint: disable=super-init-not-called
self.message = message
class NoServemeKey(Exception):
"""Exception raised when the server does not have a serveme API key setup.
Attributes:
message -- explanation of the error
"""
def __init__( # pylint: disable=super-init-not-called
self, message="No serveme key setup for the server. Please run /serveme."
):
self.message = message
def get_steam64(arg: str) -> str:
"""Converts a steam id, steam link, or RGL link to a steam64 id.
Args:
arg (str): The steam id, steam link, or RGL link to convert.
Returns:
str: The steam64 id.
"""
if arg.startswith("https://steamcommunity.com/id/"):
steam64 = steamid.steam64_from_url(arg)
elif arg.startswith("[U:1:"):
obj = SteamID(arg)
steam64 = obj.as_64
elif arg.startswith("STEAM_"):
obj = SteamID(arg)
steam64 = obj.as_64
elif arg.startswith("7656119"):
steam64 = arg
elif arg.startswith("https://rgl.gg/Public/PlayerProfile.aspx?"):
args = arg.split("=")
steam64 = args[1].replace("&r", "")
else:
steam64 = steamid.steam64_from_url(f"https://steamcommunity.com/id/{arg}")
if steam64 is None:
raise ValueError(f"Invalid steam id: {arg}")
return steam64
def is_runner():
"""A decorator to check if the user has the runner role for the guild."""
async def predicate(interaction: nextcord.Interaction):
if not is_server_setup(interaction.guild.id):
raise ServerNotSetupError(
"Guild id: " + str(interaction.guild.id) + " is not setup."
)
try:
required_role = get_server(interaction.guild.id)["role"]
except KeyError:
await interaction.send(
"A runner role could not be found for this server. Please run /setup."
)
return False
if required_role not in [role.id for role in interaction.user.roles]:
raise application_checks.ApplicationMissingRole(required_role)
return True
return nextcord.ext.application_checks.check(predicate)
def is_setup():
"""A decorator to check if the server has gone through the setup process."""
def predicate(interaction: nextcord.Interaction):
if not is_server_setup(interaction.guild.id):
raise ServerNotSetupError(
"Guild id: " + str(interaction.guild.id) + " is not setup."
)
if get_server(interaction.guild.id).get("serveme") is None:
raise NoServemeKey(
f"Guild id: {str(interaction.guild.id)} does not have a serveme key setup."
)
return True
return nextcord.ext.application_checks.check(predicate)
def guild_config_check():
"""A decorator to check if the server has guild configs setup."""
async def predicate(interaction: nextcord.Interaction):
try:
await config_db.find_item({"guild": interaction.guild.id})
except LookupError:
await interaction.send(
"This server has not been setup yet. Please run /setup first."
)
return False
return True
return nextcord.ext.application_checks.check(predicate)
async def get_exec_command(reservation: dict, tf_map: str) -> str:
"""Creates the correct exec command depending on the desired map and reservation.
Args:
reservation (dict): The reservation data retrieved from the serveme API.
map (str): The desired map to switch to.
Raises:
Exception: If the whitelist ID in the reservation doesn't match a RGL whitelist.
Returns:
str: The exec command to run.
"""
whitelist_id: int = reservation["whitelist_id"]
custom_whitelist: str = reservation["custom_whitelist_id"]
new_config: str
if whitelist_id == 20: # 6s whitelist ID
if tf_map.startswith("cp_"):
new_config = "rgl_6s_5cp_scrim"
elif tf_map.startswith("koth_"):
new_config = "rgl_6s_koth_bo5"
else:
new_config = "rgl_off"
elif whitelist_id == 22: # HL whitelist ID
if tf_map.startswith("pl_"):
new_config = "rgl_hl_stopwatch"
elif tf_map.startswith("koth_"):
new_config = "rgl_hl_koth_bo5"
else:
new_config = "rgl_off"
elif whitelist_id == 26: # PT whitelist ID
new_config = "rgl_pt_push"
elif custom_whitelist in ("13798", "13797"):
if tf_map.startswith("cp_"):
return f"changelevel {tf_map}"
if tf_map.startswith("koth_"):
new_config = "rgl_6s_koth_bo5"
else:
new_config = "rgl_off"
else:
return "changelevel " + tf_map
command: str = "exec " + new_config + "; changelevel " + tf_map
return command
async def get_log(log_id: int):
"""Returns log data from the logs.tf API.
Returns:
log: dict of log
"""
async with aiohttp.ClientSession() as session:
async with session.get("https://logs.tf/api/v1/log/" + str(log_id)) as resp:
log = await resp.json()
return log
async def get_total_logs(steam_id: str):
"""Returns the total number of logs for a player
Args:
steam_id (int): The steam ID to look up
Returns:
int: Total number of logs
"""
async with aiohttp.ClientSession() as session:
async with session.get(
"https://logs.tf/api/v1/log?limit=10000&player=" + str(steam_id)
) as resp:
logs = await resp.json()
return logs["results"]