-
Notifications
You must be signed in to change notification settings - Fork 0
/
utils.py
147 lines (114 loc) · 3.95 KB
/
utils.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
import json
import time
from io import BytesIO
from urllib.parse import urlencode
import discord
import requests
from custom_logger import Logger
from database import DatabaseReader
logger = Logger('gwaff.utils')
MAX_RETRIES = 5
def retry_request(request_func, url, **kwargs):
"""
Handles retry logic for making requests.
Args:
request_func (function): Function to call for the request (e.g., requests.get).
url (str): The URL to request.
kwargs (dict): Additional arguments for the request.
Returns:
requests.Response: The result of the request_func, or None on failure.
"""
count = 0
while count < MAX_RETRIES:
try:
response = request_func(url, **kwargs)
response.raise_for_status() # Raises an HTTPError for bad responses
return response
except requests.exceptions.RequestException as e:
logger.warning(f"Attempt {count + 1} failed: {str(e)}")
count += 1
if count < MAX_RETRIES:
time.sleep(1 << count)
else:
logger.error("Max retries reached, skipping.")
return None
def request_api(url: str, **kwargs) -> dict:
"""
Requests JSON data from the given API.
Args:
url (str): The URL to request.
kwargs (dict): Additional arguments for the request.
Returns:
dict: The requested data as a dictionary, or None on failure.
"""
if kwargs:
url = url_constructor(url, **kwargs)
# Timeout to avoid hanging
response = retry_request(requests.get, url, timeout=10)
if response:
try:
return response.json() # Parse JSON response
except json.JSONDecodeError as e:
logger.error(f"Failed to parse JSON: {str(e)}")
return None
return None
def request_img(url: str, **kwargs):
"""
Requests an image from the given URL.
Args:
url (str): The URL to request.
kwargs (dict): Additional arguments for the request.
Returns:
BytesIO: A file-like object of the image that matplotlib can read, or None on failure.
"""
if kwargs:
url = url_constructor(url, **kwargs)
headers = kwargs.get('headers', {"User-Agent": "Mozilla/5.0"})
response = retry_request(requests.get, url, headers=headers, timeout=10, stream=True)
if response:
return BytesIO(response.content) # Return image as a file-like object
return None
def url_constructor(base: str, **kwargs: dict) -> str:
"""
Constructs a URL from a base URL and several key-values.
Args:
base (str): The base URL.
kwargs (dict): Key-value pairs to append as query parameters.
Returns:
str: The constructed URL.
"""
query = urlencode(kwargs)
return f"{base}?{query}"
def resolve_member(interaction: discord.Interaction,
user: discord.User) -> discord.User:
"""
Chooses the member between a specified member and the triggering member.
Args:
interaction (discord.Interaction): The interaction that triggered the command.
user (discord.User): The specified user.
Returns:
discord.User: The resolved user, or None if no valid user is found.
"""
dbr = DatabaseReader()
if user is not None:
if dbr.get_profile_data(user.id):
return user
if interaction is not None:
if dbr.get_profile_data(interaction.user.id):
return interaction.user
return None
def ordinal(n: int) -> str:
"""
Converts an integer to its ordinal representation.
Args:
n (int): The integer to convert.
Returns:
str: The ordinal representation of the integer.
"""
suffixes = {1: "st", 2: "nd", 3: "rd"}
i = n if (n < 20) else (n % 10)
suffix = suffixes.get(i, 'th')
return str(n) + suffix
if __name__ == '__main__':
dbr = DatabaseReader()
print(dbr.get_profile_data(92029863090728960))