Coverage for otaku_info/background/ln_releases.py: 13%

Shortcuts on this page

r m x p   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

55 statements  

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 Dict 

22from jerrycan.base import app, db 

23from otaku_info.db import MediaIdMapping 

24from otaku_info.db.MediaItem import MediaItem 

25from otaku_info.enums import ListService, MediaType 

26from otaku_info.external.reddit import load_ln_releases 

27from otaku_info.utils.object_conversion import anime_list_item_to_media_item, \ 

28 reddit_ln_release_to_ln_release 

29from otaku_info.external.myanimelist import load_myanimelist_item 

30from otaku_info.external.anilist import load_anilist_info 

31 

32 

33def update_ln_releases(): 

34 """ 

35 Updates the light novel releases 

36 :return: None 

37 """ 

38 start = time.time() 

39 app.logger.info("Starting Reddit LN Update") 

40 

41 existing_myanimelist_items: Dict[int, MediaItem] = { 

42 int(x.service_id): x 

43 for x in MediaItem.query.filter_by( 

44 service=ListService.MYANIMELIST 

45 ).options(db.joinedload(MediaItem.id_mappings)).all() 

46 } 

47 existing_anilist_items: Dict[int, MediaItem] = { 

48 int(x.service_id): x 

49 for x in MediaItem.query.filter_by( 

50 service=ListService.ANILIST 

51 ).options(db.joinedload(MediaItem.id_mappings)).all() 

52 } 

53 myanimelist_anilist_items = {} 

54 for anilist_id, anilist_item in existing_anilist_items.items(): 

55 mal_mapping = anilist_item.ids.get(ListService.MYANIMELIST) 

56 if mal_mapping is not None: 

57 mal_id = int(mal_mapping.service_id) 

58 myanimelist_anilist_items[mal_id] = anilist_item 

59 

60 ln_releases = load_ln_releases() 

61 for ln_release in ln_releases: 

62 

63 items = [] 

64 if ln_release.myanimelist_id is not None: 

65 mal_id = ln_release.myanimelist_id 

66 mal_item = existing_myanimelist_items.get(mal_id) 

67 anilist_item = myanimelist_anilist_items.get(mal_id) 

68 

69 if mal_item is None: 

70 mal_info = load_myanimelist_item(mal_id, MediaType.MANGA) 

71 if mal_info is not None: 

72 mal_item = anime_list_item_to_media_item(mal_info) 

73 app.logger.debug( 

74 f"Upserting myanimelist: {mal_item.english_title}" 

75 ) 

76 mal_item = db.session.merge(mal_item) 

77 existing_myanimelist_items[mal_id] = mal_item 

78 if anilist_item is None: 

79 anilist_info = load_anilist_info( 

80 int(mal_id), MediaType.MANGA, ListService.MYANIMELIST 

81 ) 

82 if anilist_info is not None: 

83 anilist_item = anime_list_item_to_media_item(anilist_info) 

84 app.logger.debug( 

85 f"Upserting anilist: {anilist_item.english_title}" 

86 ) 

87 anilist_item = db.session.merge(anilist_item) 

88 myanimelist_anilist_items[mal_id] = anilist_item 

89 

90 if anilist_item is not None and mal_item is not None: 

91 for one, two in [ 

92 (anilist_item, mal_item), 

93 (mal_item, anilist_item) 

94 ]: 

95 app.logger.debug( 

96 f"Upserting mapping: " 

97 f"{one.service.value}:{one.service_id}" 

98 f"->{two.service.value}:{two.service_id}" 

99 ) 

100 db.session.merge(MediaIdMapping( 

101 service=one.service, 

102 service_id=one.service_id, 

103 media_type=one.media_type, 

104 parent_service=two.service, 

105 parent_service_id=two.service_id 

106 )) 

107 items += [x for x in [anilist_item, mal_item] if x is not None] 

108 

109 if len(items) == 0: 

110 items = [None] 

111 

112 app.logger.debug( 

113 f"Upserting ln releases: {ln_release.series_name} " 

114 f"volume {ln_release.volume}" 

115 ) 

116 for item in items: 

117 release = reddit_ln_release_to_ln_release(ln_release, item) 

118 

119 db.session.merge(release) 

120 

121 db.session.commit() 

122 

123 app.logger.info(f"Finished Reddit LN Update in {time.time() - start}s.")