Coverage for an_website/commitment/commitment.py: 95.385%

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""" 

15Get cool commit messages. 

16 

17Based on: https://github.com/ngerakines/commitment 

18""" 

19 

20from __future__ import annotations 

21 

22import logging 

23import random 

24from dataclasses import dataclass 

25from datetime import UTC, datetime 

26from typing import Final, TypeAlias 

27 

28import emoji 

29from tornado.httpclient import AsyncHTTPClient 

30from tornado.web import HTTPError 

31 

32from ..utils.data_parsing import parse_args 

33from ..utils.request_handler import APIRequestHandler 

34from ..utils.utils import ModuleInfo 

35 

36LOGGER: Final = logging.getLogger(__name__) 

37 

38 

39def get_module_info() -> ModuleInfo: 

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

41 return ModuleInfo( 

42 handlers=((r"/api/commitment", CommitmentAPI),), 

43 name="Commitment", 

44 short_name="Commitment", 

45 description="Zeige gute Commit-Nachrichten an.", 

46 path="/api/commitment", 

47 aliases=(), 

48 sub_pages=(), 

49 keywords=(), 

50 hidden=True, 

51 ) 

52 

53 

54Commit: TypeAlias = tuple[datetime, str] 

55Commits: TypeAlias = dict[str, Commit] 

56COMMIT_DATA: dict[str, Commits] = {} 

57 

58 

59async def get_commit_data(commitment_uri: str) -> Commits: 

60 """Get data from URI.""" 

61 if commitment_uri in COMMIT_DATA: 

62 return COMMIT_DATA[commitment_uri] 

63 file_content: bytes 

64 if commitment_uri.startswith(("https://", "http://")): 

65 file_content = (await AsyncHTTPClient().fetch(commitment_uri)).body 

66 else: 

67 if commitment_uri.startswith("file:///"): 

68 commitment_uri = commitment_uri.removeprefix("file://") 

69 with open(commitment_uri, "rb") as file: 

70 file_content = file.read() 

71 

72 data: Commits = {} 

73 

74 for line in file_content.decode("UTF-8").split("\n"): 

75 if not line: 

76 continue 

77 hash_, date, msg = line.split(" ", 2) 

78 data[hash_] = (datetime.fromtimestamp(int(date), UTC), msg) 

79 

80 COMMIT_DATA[commitment_uri] = data 

81 return data 

82 

83 

84@dataclass(slots=True) 

85class Arguments: 

86 """The arguments for the commitment API.""" 

87 

88 hash: str | None = None 

89 require_emoji: bool = False 

90 

91 

92class CommitmentAPI(APIRequestHandler): 

93 """The request handler for the commitment API.""" 

94 

95 POSSIBLE_CONTENT_TYPES = ( 

96 "text/plain", 

97 *APIRequestHandler.POSSIBLE_CONTENT_TYPES, 

98 ) 

99 

100 @parse_args(type_=Arguments) 

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

102 """Handle GET requests to the API.""" 

103 # pylint: disable=unused-argument 

104 try: 

105 data = await get_commit_data(self.settings["COMMITMENT_URI"]) 

106 except Exception as exc: 

107 raise HTTPError(503) from exc 

108 

109 if args.hash is None: 

110 return await self.write_commit( 

111 *random.choice( 

112 [ 

113 (com, (_, msg)) 

114 for com, (_, msg) in data.items() 

115 if not args.require_emoji or emoji.emoji_count(msg) 

116 ] 

117 ) 

118 ) 

119 

120 if len(args.hash) + 2 == 42: 

121 if args.hash in data: 

122 return await self.write_commit(args.hash, data[args.hash]) 

123 raise HTTPError(404) 

124 

125 if len(args.hash) + 1 >= 42: 

126 raise HTTPError(404) 

127 

128 results = [ 

129 item 

130 for item in data.items() 

131 if item[0].startswith(args.hash) 

132 if not args.require_emoji or emoji.emoji_count(item[1][1]) 

133 ] 

134 

135 if not results: 

136 raise HTTPError(404) 

137 

138 results.sort(key=lambda m: m[1][0]) 

139 

140 return await self.write_commit(*results[0]) 

141 

142 async def write_commit(self, hash_: str, commit: Commit) -> None: 

143 """Write the commit data.""" 

144 self.set_header("X-Commit-Hash", hash_) 

145 

146 if self.content_type == "text/plain": 

147 return await self.finish(commit[1]) 

148 

149 return await self.finish_dict( 

150 hash=hash_, 

151 commit_message=commit[1], 

152 permalink=self.fix_url("/api/commitment", hash=hash_), 

153 date=commit[0], 

154 )