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
|
"""Various utilities for working with the Discord API."""
import asyncio
import typing
from urllib import parse
import httpx
from starlette.requests import Request
from backend import constants
from backend.models import Form, FormResponse
API_BASE_URL = "https://discord.com/api/v8/"
DISCORD_HEADERS = {
"Authorization": f"Bot {constants.DISCORD_BOT_TOKEN}"
}
async def make_request(
method: str,
url: str,
json: dict[str, any] = None,
headers: dict[str, str] = None
) -> httpx.Response:
"""Make a request to the discord API."""
url = parse.urljoin(API_BASE_URL, url)
_headers = DISCORD_HEADERS.copy()
_headers.update(headers if headers else {})
async with httpx.AsyncClient() as client:
if json is not None:
request = client.request(method, url, json=json, headers=_headers)
else:
request = client.request(method, url, headers=_headers)
resp = await request
# Handle Rate Limits
while resp.status_code == 429:
retry_after = float(resp.headers["X-Ratelimit-Reset-After"])
await asyncio.sleep(retry_after)
resp = await request
resp.raise_for_status()
return resp
async def fetch_bearer_token(code: str, redirect: str, *, refresh: bool) -> dict:
async with httpx.AsyncClient() as client:
data = {
"client_id": constants.OAUTH2_CLIENT_ID,
"client_secret": constants.OAUTH2_CLIENT_SECRET,
"redirect_uri": f"{redirect}/callback"
}
if refresh:
data["grant_type"] = "refresh_token"
data["refresh_token"] = code
else:
data["grant_type"] = "authorization_code"
data["code"] = code
r = await client.post(f"{API_BASE_URL}/oauth2/token", headers={
"Content-Type": "application/x-www-form-urlencoded"
}, data=data)
r.raise_for_status()
return r.json()
async def fetch_user_details(bearer_token: str) -> dict:
r = await make_request("GET", "users/@me", headers={"Authorization": f"Bearer {bearer_token}"})
r.raise_for_status()
return r.json()
def build_ctx_variables(
form: Form,
response: FormResponse,
user: Request.user
) -> dict[str, str]:
"""Parses contextual information from a request. See SCHEMA.md for all variables."""
try:
mention = user.discord_mention
except AttributeError:
mention = "User"
return {
"user": mention,
"response_id": response.id,
"form": form.name,
"form_id": form.id,
"time": response.timestamp,
}
def format_message(ctx: dict[str, str], message: str) -> str:
"""Formats a message using contextual information."""
for key, val in ctx.items():
message = message.replace(f"{{{key}}}", str(val))
return message
async def send_submission_webhook(
form: Form,
response: FormResponse,
request_user: Request.user
) -> None:
"""Helper to send a submission message to a discord webhook."""
# Stop if webhook is not available
if form.webhook is None:
raise ValueError("Got empty webhook.")
ctx = build_ctx_variables(form, response, request_user)
user = response.user
mention = ctx.get("mention")
# Build Embed
embed = {
"title": "New Form Response",
"description": f"{mention} submitted a response to `{form.name}`.",
"url": f"{constants.FRONTEND_URL}/path_to_view_form/{response.id}", # noqa: E501 # TODO: Enter Form View URL
"timestamp": response.timestamp,
"color": 7506394,
}
# Add author to embed
if request_user.is_authenticated:
embed["author"] = {"name": request_user.display_name}
if user and user.avatar:
url = f"https://cdn.discordapp.com/avatars/{user.id}/{user.avatar}.png"
embed["author"]["icon_url"] = url
# Build Hook
hook = {
"embeds": [embed],
"allowed_mentions": {"parse": ["users", "roles"]},
"username": form.name or "Python Discord Forms"
}
# Set hook message
message = form.webhook.message
if message:
hook["content"] = format_message(ctx, message)
# Post hook
await make_request("POST", form.webhook.url, hook)
async def assign_role(form: Form, request_user: Request.user) -> None:
"""Assigns Discord role to user when user submitted response."""
if not form.discord_role:
raise ValueError("Got empty Discord role ID.")
url = (
f"guilds/{constants.DISCORD_GUILD}"
f"/members/{request_user.payload['id']}/roles/{form.discord_role}"
)
await make_request("PUT", url)
async def get_direct_message_channel(user_id: int) -> typing.Optional[int]:
"""Get the ID of a direct message channel."""
try:
response = await make_request("POST", "users/@me/channels", {"recipient_id": user_id})
return response.json().get("id")
except httpx.HTTPStatusError as e:
# This is most likely caused by an incorrect ID
if e.response.status_code == 400:
return
raise e
async def send_direct_message(form: Form, response: FormResponse, user: Request.user) -> None:
"""A helper method to assign a discord role to a user."""
channel = await get_direct_message_channel(user.payload['id'])
if not channel:
return
message = format_message(build_ctx_variables(form, response, user), form.dm_message)
try:
await make_request("POST", f"channels/{channel}/messages", {"content": message})
except httpx.HTTPStatusError as e:
# This is most likely caused by closed DMs
if e.response.status_code == 403:
return
raise e
|
