Coverage for otaku_info/background/anilist.py: 15%

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 typing import List, Optional, Dict 

22from jerrycan.base import app, db 

23 

24from otaku_info.db import MediaList, MediaListItem, MediaIdMapping 

25from otaku_info.enums import ListService, MediaType 

26from otaku_info.utils.object_conversion import anime_list_item_to_media_item, \ 

27 anilist_user_item_to_media_user_state 

28from otaku_info.db.ServiceUsername import ServiceUsername 

29from otaku_info.external.anilist import load_anilist 

30from otaku_info.external.entities.AnilistUserItem import AnilistUserItem 

31 

32 

33def update_anilist_data(usernames: Optional[List[ServiceUsername]] = None): 

34 """ 

35 Retrieves all entries on the anilists of all users that provided 

36 an anilist username 

37 :param usernames: Can be used to override the usernames to use 

38 :return: None 

39 """ 

40 start = time.time() 

41 app.logger.info("Starting Anilist Update") 

42 

43 if usernames is None: 

44 usernames = ServiceUsername.query\ 

45 .filter_by(service=ListService.ANILIST).all() 

46 

47 anilist_data: Dict[ 

48 ServiceUsername, 

49 Dict[MediaType, List[AnilistUserItem]] 

50 ] = { 

51 username: { 

52 media_type: load_anilist(username.username, media_type) 

53 for media_type in MediaType 

54 } 

55 for username in usernames 

56 } 

57 __update_data(anilist_data) 

58 app.logger.info(f"Finished Anilist Update in {time.time() - start}s.") 

59 

60 

61def __update_data( 

62 anilist_data: Dict[ 

63 ServiceUsername, 

64 Dict[MediaType, List[AnilistUserItem]] 

65 ] 

66): 

67 """ 

68 Updates the anilist data in the database 

69 :param anilist_data: The anilist data to enter 

70 :return: None 

71 """ 

72 media_items = {} 

73 user_states = [] 

74 user_lists = {} 

75 user_list_items = [] 

76 mal_mappings = [] 

77 

78 for username, anilist_info in anilist_data.items(): 

79 for media_type, anilist_items in anilist_info.items(): 

80 for anilist_item in anilist_items: 

81 media_item = anime_list_item_to_media_item(anilist_item) 

82 user_state = anilist_user_item_to_media_user_state( 

83 anilist_item, username.user_id 

84 ) 

85 media_list = MediaList( 

86 service=ListService.ANILIST, 

87 media_type=anilist_item.media_type, 

88 user_id=username.user_id, 

89 name=anilist_item.list_name 

90 ) 

91 media_list_item = MediaListItem( 

92 media_list_service=media_list.service, 

93 media_list_media_type=media_list.media_type, 

94 media_list_user_id=media_list.user_id, 

95 media_list_name=media_list.name, 

96 user_state_service=user_state.service, 

97 user_state_media_type=user_state.media_type, 

98 user_state_user_id=user_state.user_id, 

99 user_state_service_id=user_state.service_id 

100 ) 

101 media_item_tuple = ( 

102 media_item.service, 

103 media_item.service_id, 

104 media_item.media_type 

105 ) 

106 media_list_tuple = ( 

107 media_list.service, 

108 media_list.media_type, 

109 media_list.user_id, 

110 media_list.name 

111 ) 

112 media_items[media_item_tuple] = media_item 

113 user_states.append(user_state) 

114 user_lists[media_list_tuple] = media_list 

115 user_list_items.append(media_list_item) 

116 if anilist_item.myanimelist_id is not None: 

117 mal_mapping = MediaIdMapping( 

118 service=ListService.MYANIMELIST, 

119 service_id=str(anilist_item.myanimelist_id), 

120 parent_service=ListService.ANILIST, 

121 parent_service_id=media_item.service_id, 

122 media_type=media_item.media_type 

123 ) 

124 mal_mappings.append(mal_mapping) 

125 

126 for media_item in media_items.values(): 

127 app.logger.debug(f"Upserting anilist item {media_item.title}") 

128 db.session.merge(media_item) 

129 for user_state in user_states: 

130 db.session.merge(user_state) 

131 for media_list in user_lists.values(): 

132 db.session.merge(media_list) 

133 for media_list_item in user_list_items: 

134 db.session.merge(media_list_item) 

135 for mal_mapping in mal_mappings: 

136 db.session.merge(mal_mapping) 

137 app.logger.debug(f"Upserting id mapping: " 

138 f"anilist:{mal_mapping.parent_service_id} " 

139 f"-> myanimelist:{mal_mapping.service_id}") 

140 db.session.commit()