Coverage for portality / bll / services / notifications.py: 68%

1# ~~Notifications:Service~~ 

2from portality import models 

3from portality.core import app 

4from portality import app_email 

5from portality.ui.messages import Messages 

6from portality.bll.exceptions import NoSuchObjectException, NoSuchPropertyException 

7from datetime import datetime 

8from portality.lib import dates 

9from portality.ui import templates 

10 

11class NotificationsService(object): 

12 def notify(self, notification: models.Notification): 

13 # first just save the notification 

14 notification.save() 

15 

16 # now send an email to the notification target 

17 acc = models.Account.pull(notification.who) 

18 if acc is None: 

19 raise NoSuchObjectException(Messages.EXCEPTION_NOTIFICATION_NO_ACCOUNT.format(x=notification.who)) 

20 

21 if acc.email is None: 

22 raise NoSuchPropertyException(Messages.EXCEPTION_NOTIFICATION_NO_EMAIL.format(x=notification.who)) 

23 

24 to = [acc.email] 

25 fro = app.config.get('SYSTEM_EMAIL_FROM', 'helpdesk@doaj.org') 

26 subject = app.config.get("SERVICE_NAME", "") + " - " + notification.short 

27 

28 action_intro = "" 

29 if notification.action is not None: 

30 action_intro = self.action_intro(notification.created_by) 

31 

32 # ~~-> Email:Library ~~ 

33 app_email.send_markdown_mail(to=to, 

34 fro=fro, 

35 subject=subject, 

36 template_name=templates.EMAIL_NOTIFICATION, 

37 markdown_template_name=templates.EMAIL_NOTIFICATION, # for the moment the markdown and plaintext templates are the same 

38 user=acc, 

39 message=notification.long, 

40 action=notification.action, 

41 action_intro=action_intro, 

42 url_root=app.config.get("BASE_URL")) 

43 

44 return notification 

45 

46 def long_notification(self, message_id): 

47 # ~~-> Notifications:Data ~~ 

48 return app.jinja_env.globals["data"]["notifications"].get(message_id, {}).get("long") 

49 

50 def short_notification(self, message_id): 

51 # ~~-> Notifications:Data ~~ 

52 return app.jinja_env.globals["data"]["notifications"].get(message_id, {}).get("short", Messages.NOTIFY__DEFAULT_SHORT_NOTIFICATION) 

53 

54 def action_intro(self, consumer_id): 

55 return app.jinja_env.globals["data"]["notifications"].get(consumer_id, {}).get("action_intro", 

56 Messages.NOTIFY__DEFAULT_ACTION_INTRO) 

57 

58 def top_notifications(self, account: models.Account, size: int = 10): 

59 # ~~-> TopNotifications:Query ~~ 

60 q = TopNotificationsQuery(account.id, size) 

61 top = models.Notification.object_query(q.query()) 

62 return top 

63 

64 def notification_seen(self, account: models.Account, notification_id: str): 

65 # ~~-> Notifications:Query ~~ 

66 note = models.Notification.pull(notification_id) 

67 if not note: 

68 raise NoSuchObjectException(Messages.EXCEPTION_NOTIFICATION_NO_NOTIFICATION.format(n=notification_id)) 

69 if account.id == note.who or account.is_super: 

70 if not note.is_seen(): 

71 note.set_seen() 

72 note.save() 

73 return True 

74 return False 

75 

76 def mark_all_as_seen(self, account: models.Account=None, until: datetime=None): 

77 account_id = None if account is None else account.id 

78 q = NotificationsQuery(account_id, until, False) 

79 for notification in models.Notification.scroll(q.query()): 

80 notification.set_seen() 

81 notification.save() 

82 

83 

84class TopNotificationsQuery(object): 

85 # ~~->$ TopNotifications:Query ~~ 

86 # ~~^-> Elasticsearch:Technology ~~ 

87 def __init__(self, account_id, size=10): 

88 self.account_id = account_id 

89 self.size = size 

90 

91 def query(self): 

92 return { 

93 "query": { 

94 "bool": { 

95 "must": [ 

96 {"term": {"who.exact": self.account_id}} 

97 ] 

98 } 

99 }, 

100 "sort": [{"created_date": {"order": "desc"}}], 

101 "size": self.size 

102 } 

103 

104 

105class NotificationsQuery(object): 

106 # ~~->$ Notifications:Query ~~ 

107 # ~~^-> Elasticsearch:Technology ~~ 

108 def __init__(self, account_id=None, until=None, seen=None): 

109 self._account_id = account_id 

110 self._until = until 

111 self._seen = seen 

112 

113 def query(self): 

114 musts = [] 

115 must_nots = [] 

116 if self._account_id: 

117 musts.append({"term": {"who.exact": self._account_id}}) 

118 

119 if self._until: 

120 musts.append({"range": {"created_date": {"lte": dates.format(self._until, format="%Y-%m-%dT%H:%M:%S")}}}) 

121 

122 if self._seen is not None: 

123 if self._seen is True: 

124 musts.append({"exists": {"field": "seen_date"}}) 

125 else: 

126 must_nots.append({"exists": {"field": "seen_date"}}) 

127 

128 if len(musts) > 0 or len(must_nots) > 0: 

129 q = {"query": {"bool": {}}} 

130 if len(musts) > 0: 

131 q["query"]["bool"]["must"] = musts 

132 if len(must_nots) > 0: 

133 q["query"]["bool"]["must_not"] = must_nots 

134 return q 

135 else: 

136 return {"query": {"match_all": {}}}