Coverage for otaku_info/external/anilist.py: 24%

1"""LICENSE 

2Copyright 2020 Hermann Krumrey <hermann@krumreyh.com> 

3 

4This file is part of otaku-info. 

5 

6otaku-info is free software: you can redistribute it and/or modify 

7it under the terms of the GNU General Public License as published by 

8the Free Software Foundation, either version 3 of the License, or 

9(at your option) any later version. 

10 

11otaku-info is distributed in the hope that it will be useful, 

12but WITHOUT ANY WARRANTY; without even the implied warranty of 

13MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

14GNU General Public License for more details. 

15 

16You should have received a copy of the GNU General Public License 

17along with otaku-info. If not, see <http://www.gnu.org/licenses/>. 

18LICENSE""" 

19 

20import time 

21from requests import ConnectionError 

22from requests.exceptions import ChunkedEncodingError 

23from typing import Optional, List 

24from puffotter.graphql import GraphQlClient 

25from otaku_info.enums import MediaType, ListService 

26from otaku_info.external.entities.AnilistItem import AnilistItem 

27from otaku_info.external.entities.AnilistUserItem import AnilistUserItem 

28 

29 

30MEDIA_QUERY = """ 

31 id 

32 idMal 

33 chapters 

34 volumes 

35 episodes 

36 status 

37 format 

38 title { 

39 english 

40 romaji 

41 } 

42 coverImage { 

43 large 

44 } 

45 nextAiringEpisode { 

46 episode 

47 airingAt 

48 } 

49 relations { 

50 edges { 

51 node { 

52 id 

53 type 

54 } 

55 relationType 

56 } 

57 } 

58""" 

59 

60 

61def guess_latest_manga_chapter(anilist_id: int) -> Optional[int]: 

62 """ 

63 Guesses the latest chapter number based on anilist user activity 

64 :param anilist_id: The anilist ID to check 

65 :return: The latest chapter number 

66 """ 

67 graphql = GraphQlClient("https://graphql.anilist.co") 

68 query = """ 

69 query ($id: Int) { 

70 Page(page: 1) { 

71 activities(mediaId: $id, sort: ID_DESC) { 

72 ... on ListActivity { 

73 progress 

74 userId 

75 status 

76 media { 

77 chapters 

78 } 

79 } 

80 } 

81 } 

82 } 

83 """ 

84 try: 

85 resp = graphql.query(query, {"id": anilist_id}) 

86 except (ChunkedEncodingError, ConnectionError): 

87 return None 

88 

89 if resp is None: 

90 return None 

91 

92 data = resp["data"]["Page"]["activities"] 

93 

94 progresses = [] 

95 for entry in data: 

96 progress = entry["progress"] 

97 status = entry["status"] 

98 chapters = entry["media"]["chapters"] 

99 

100 if status == "completed": 

101 progresses.append(chapters) 

102 elif progress is not None: 

103 progress = entry["progress"].split(" - ")[-1] 

104 progresses.append(int(progress)) 

105 

106 progresses = progresses[0:20] 

107 progresses.sort(key=lambda x: progresses.count(x), reverse=True) 

108 progresses = sorted(progresses, key=progresses.count, reverse=True) 

109 time.sleep(0.5) 

110 

111 try: 

112 return progresses[0] 

113 except IndexError: 

114 return None 

115 

116 

117def load_anilist( 

118 username: str, 

119 media_type: MediaType 

120) -> List[AnilistUserItem]: 

121 """ 

122 Loads the anilist for a user 

123 :param username: The username 

124 :param media_type: The media type, either MANGA or ANIME 

125 :return: The anilist list items for the user and media type 

126 """ 

127 graphql = GraphQlClient("https://graphql.anilist.co") 

128 query = """ 

129 query ($username: String, $media_type: MediaType) { 

130 MediaListCollection(userName: $username, type: $media_type) { 

131 lists { 

132 name 

133 entries { 

134 progress 

135 progressVolumes 

136 score 

137 status 

138 media {@{MEDIA_QUERY}} 

139 } 

140 } 

141 } 

142 } 

143 """.replace("@{MEDIA_QUERY}", MEDIA_QUERY) 

144 

145 try: 

146 resp = graphql.query(query, { 

147 "username": username, 

148 "media_type": media_type.value.upper() 

149 }) 

150 except (ChunkedEncodingError, ConnectionError): 

151 return [] 

152 if resp is None: 

153 return [] 

154 user_lists = resp["data"]["MediaListCollection"]["lists"] 

155 

156 anilist_items: List[AnilistUserItem] = [] 

157 for entries, list_name in [ 

158 (y["entries"], y["name"]) for y in user_lists 

159 ]: 

160 for entry in entries: 

161 entry["list_name"] = list_name 

162 anilist_items.append(AnilistUserItem.from_query(media_type, entry)) 

163 

164 return anilist_items 

165 

166 

167def load_anilist_info( 

168 service_id: int, 

169 media_type: MediaType, 

170 service: ListService = ListService.ANILIST 

171) -> Optional[AnilistItem]: 

172 """ 

173 Loads information for a single anilist media item 

174 :param service_id: The anilist or myanimelist media ID 

175 :param media_type: The media type 

176 :param service: The service the ID belongs to 

177 (either anilist or myanimelist) 

178 :return: The fetched AnilistItem 

179 """ 

180 graphql = GraphQlClient("https://graphql.anilist.co") 

181 query = """ 

182 query ($id: Int, $media_type: MediaType) { 

183 Media(@{ID}: $id, type: $media_type) { 

184 @{MEDIA_QUERY} 

185 } 

186 } 

187 """.replace("@{MEDIA_QUERY}", MEDIA_QUERY) 

188 if service == ListService.ANILIST: 188 ↛ 190line 188 didn't jump to line 190, because the condition on line 188 was never false

189 query = query.replace("@{ID}", "id") 

190 elif service == ListService.MYANIMELIST: 

191 query = query.replace("@{ID}", "idMal") 

192 else: 

193 return None 

194 

195 try: 

196 resp = graphql.query( 

197 query, 

198 {"id": service_id, "media_type": media_type.value.upper()} 

199 ) 

200 except (ChunkedEncodingError, ConnectionError): 

201 return None 

202 

203 if resp is None: 203 ↛ 204line 203 didn't jump to line 204, because the condition on line 203 was never true

204 return None 

205 else: 

206 return AnilistItem.from_query(media_type, resp["data"]["Media"])