Coverage for otaku_info/background/notifications.py: 16%

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, List, Tuple 

22from jerrycan.base import db, app 

23from otaku_info.db.MediaUserState import MediaUserState 

24from otaku_info.db.MediaNotification import MediaNotification 

25from otaku_info.db.NotificationSetting import NotificationSetting 

26from otaku_info.wrappers.UpdateWrapper import UpdateWrapper 

27from otaku_info.enums import MediaType, MediaSubType, NotificationType, \ 

28 ConsumingState 

29 

30 

31def send_new_update_notifications(): 

32 """ 

33 Sends out telegram notifications for media updates 

34 :return: None 

35 """ 

36 start = time.time() 

37 app.logger.info("Starting check for notifications") 

38 

39 user_states: List[MediaUserState] = MediaUserState.query.filter_by( 

40 consuming_state=ConsumingState.CURRENT 

41 ).options( 

42 db.joinedload(MediaUserState.media_notification) 

43 ).all() 

44 

45 notification_settings: Dict[ 

46 Tuple[int, NotificationType], 

47 NotificationSetting 

48 ] = { 

49 (x.user_id, x.notification_type): x 

50 for x in NotificationSetting.query.all() 

51 } 

52 

53 for user_state in user_states: 

54 

55 target_type = { 

56 MediaType.ANIME: NotificationType.NEW_ANIME_EPISODES, 

57 MediaType.MANGA: NotificationType.NEW_MANGA_CHAPTERS 

58 }.get(user_state.media_type) 

59 

60 settings = notification_settings.get(( 

61 user_state.user_id, target_type 

62 )) 

63 if settings is None or not settings.value: 

64 continue 

65 else: 

66 handle_notification(user_state, settings) 

67 

68 db.session.commit() 

69 app.logger.info(f"Completed check for notifications in " 

70 f"{time.time() - start}s.") 

71 

72 

73def handle_notification( 

74 media_user_state: MediaUserState, 

75 settings: NotificationSetting 

76): 

77 """ 

78 Handles a single notification 

79 :param media_user_state: The user state for which to notify 

80 :param settings: The notification settings 

81 :return: None 

82 """ 

83 chat = media_user_state.user.telegram_chat_id 

84 if chat is None: 

85 return 

86 

87 update = UpdateWrapper(media_user_state) 

88 notification = media_user_state.media_notification 

89 if notification is None: 

90 notification = MediaNotification( 

91 service=media_user_state.service, 

92 service_id=media_user_state.service_id, 

93 media_type=media_user_state.media_type, 

94 user_id=media_user_state.user_id, 

95 last_update=update.latest 

96 ) 

97 db.session.add(notification) 

98 

99 if update.diff <= 0: 

100 return 

101 

102 if notification.last_update < update.latest: 

103 notification.last_update = update.latest 

104 

105 show_update = \ 

106 (update.score is not None 

107 and update.score >= settings.minimum_score) \ 

108 or settings.minimum_score == 0 

109 

110 if show_update: 

111 media_item = media_user_state.media_item 

112 if media_item.media_type == MediaType.ANIME: 

113 keyword = "Episode" 

114 elif media_item.media_subtype == MediaSubType.NOVEL: 

115 keyword = "Volume" 

116 else: 

117 keyword = "Chapter" 

118 

119 chat.send_message( 

120 f"New {keyword} for {update.title}\n\n" 

121 f"{keyword} {update.progress}/{update.latest} " 

122 f"(+{update.diff})\n\n" 

123 f"{update.url}" 

124 )