Coverage for an_website / discord / discord.py: 72.727%

1# This program is free software: you can redistribute it and/or modify 

2# it under the terms of the GNU Affero General Public License as 

3# published by the Free Software Foundation, either version 3 of the 

4# License, or (at your option) any later version. 

5# 

6# This program is distributed in the hope that it will be useful, 

7# but WITHOUT ANY WARRANTY; without even the implied warranty of 

8# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

9# GNU Affero General Public License for more details. 

10# 

11# You should have received a copy of the GNU Affero General Public License 

12# along with this program. If not, see <https://www.gnu.org/licenses/>. 

13 

14"""A permanent redirect to an invite of the Discord guild.""" 

15 

16 

17import time 

18from typing import Final 

19 

20import orjson as json 

21from tornado.httpclient import AsyncHTTPClient 

22from tornado.web import HTTPError 

23 

24from .. import CA_BUNDLE_PATH 

25from ..utils.request_handler import APIRequestHandler, HTMLRequestHandler 

26from ..utils.utils import ModuleInfo 

27 

28GUILD_ID: Final[str] = "367648314184826880" 

29 

30INVITE_CACHE: dict[ 

31 str, 

32 tuple[float, str, str] | tuple[float, HTTPError], 

33] = {} 

34 

35 

36def get_module_info() -> ModuleInfo: 

37 """Create and return the ModuleInfo for this module.""" 

38 return ModuleInfo( 

39 handlers=( 

40 (r"/discord", ANDiscord), 

41 (r"/api/discord", ANDiscordAPI), 

42 (f"/api/discord/({GUILD_ID})", ANDiscordAPI), 

43 (r"/api/discord/(\d+)", DiscordAPI), 

44 ), 

45 name="Discord-Einladung", 

46 short_name="Discord", 

47 description="Eine permanente Einladung zu unserer Discord-Gilde", 

48 path="/discord", 

49 keywords=( 

50 "Discord", 

51 "Server", 

52 "Guild", 

53 "Gilde", 

54 "Invite", 

55 "Einladung", 

56 ), 

57 ) 

58 

59 

60async def url_returns_200(url: str) -> bool: 

61 """Check whether a URL returns a status code of 200.""" 

62 response = await AsyncHTTPClient().fetch( 

63 url, 

64 method="HEAD", 

65 raise_error=False, 

66 ca_certs=CA_BUNDLE_PATH, 

67 ) 

68 return response.code == 200 

69 

70 

71async def get_invite(guild_id: str = GUILD_ID) -> tuple[str, str]: 

72 """ 

73 Get the invite to a Discord guild and return it with the source. 

74 

75 How to get the invite: 

76 - from the widget (has to be enabled in guild settings) 

77 - from DISBOARD (the guild needs to set it up first) 

78 

79 If the invite couldn't be fetched an HTTPError is raised. 

80 """ 

81 reason = "Invite not found." 

82 

83 # try getting the invite from the widget 

84 url = f"https://discord.com/api/guilds/{guild_id}/widget.json" 

85 response = await AsyncHTTPClient().fetch( 

86 url, raise_error=False, ca_certs=CA_BUNDLE_PATH 

87 ) 

88 if response.code == 200: 

89 response_json = json.loads(response.body) 

90 if invite := response_json["instant_invite"]: 

91 return invite, url 

92 reason = f"No instant invite in widget ({url}) found." 

93 

94 # try getting the invite from DISBOARD 

95 url = f"https://disboard.org/site/get-invite/{guild_id}" 

96 response = await AsyncHTTPClient().fetch( 

97 url, raise_error=False, ca_certs=CA_BUNDLE_PATH 

98 ) 

99 if response.code == 200: 

100 return ( 

101 json.loads(response.body), 

102 f"https://disboard.org/server/{guild_id}", 

103 ) 

104 

105 # check if Top.gg lists the guild 

106 url = f"https://top.gg/servers/{guild_id}/join" 

107 if await url_returns_200(url): 

108 return url, f"https://top.gg/servers/{guild_id}/" 

109 

110 # check if Discords.com lists the guild 

111 if await url_returns_200( 

112 # API endpoint that only returns 200 if the guild exists 

113 f"https://discords.com/api-v2/server/{guild_id}/relevant" 

114 ): 

115 return ( 

116 f"https://discords.com/servers/{guild_id}/join", 

117 f"https://discords.com/servers/{guild_id}/", 

118 ) 

119 

120 raise HTTPError(404, reason=reason) 

121 

122 

123async def get_invite_with_cache( 

124 guild_id: str = GUILD_ID, 

125) -> tuple[str, str]: 

126 """Get an invite from cache or from get_invite().""" 

127 if guild_id in INVITE_CACHE: 

128 cache_entry = INVITE_CACHE[guild_id] 

129 if cache_entry[0] > time.monotonic() - 300: 

130 if isinstance(cache_entry[1], HTTPError): 

131 raise cache_entry[1] 

132 return cache_entry[1], cache_entry[2] 

133 

134 try: 

135 invite, source = await get_invite(guild_id) 

136 except HTTPError as exc: 

137 INVITE_CACHE[guild_id] = (time.monotonic(), exc) 

138 raise exc 

139 

140 INVITE_CACHE[guild_id] = (time.monotonic(), invite, source) 

141 

142 return invite, source 

143 

144 

145class ANDiscord(HTMLRequestHandler): 

146 """The request handler that gets the Discord invite and redirects to it.""" 

147 

148 RATELIMIT_GET_LIMIT = 10 

149 

150 async def get(self, *, head: bool = False) -> None: 

151 """Get the Discord invite.""" 

152 invite = (await get_invite_with_cache(GUILD_ID))[0] 

153 if not self.user_settings.ask_before_leaving: 

154 return self.redirect(invite) 

155 if head: 

156 return 

157 return await self.render( 

158 "pages/redirect.html", 

159 send_referrer=True, 

160 redirect_url=invite, 

161 from_url=None, 

162 discord=True, 

163 ) 

164 

165 

166class DiscordAPI(APIRequestHandler): 

167 """The API request handler that gets the Discord invite and returns it.""" 

168 

169 RATELIMIT_GET_LIMIT = 5 

170 RATELIMIT_GET_COUNT_PER_PERIOD = 10 

171 

172 async def get( 

173 self, guild_id: str = GUILD_ID, *, head: bool = False 

174 ) -> None: 

175 """Get the Discord invite and render it as JSON.""" 

176 # pylint: disable=unused-argument 

177 invite, source_url = await get_invite_with_cache(guild_id) 

178 return await self.finish_dict(invite=invite, source=source_url) 

179 

180 

181class ANDiscordAPI(DiscordAPI): 

182 """The API request handler only for the AN Discord guild.""" 

183 

184 RATELIMIT_GET_LIMIT = 10 

185 RATELIMIT_GET_COUNT_PER_PERIOD = 30