-
Notifications
You must be signed in to change notification settings - Fork 2k
/
email_notifications.py
145 lines (107 loc) · 5.01 KB
/
email_notifications.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
'''
Code for generating email notifications for users (e.g. email notifications for
new activities in your dashboard activity stream) and emailing them to the
users.
'''
import datetime
import ckan.model as model
import ckan.logic as logic
def _notifications_for_activities(activities):
'''Return one or more email notifications covering the given activities.
This function handles grouping multiple activities into a single digest
email.
:param activities: the activities to consider
:type activities: list of activity dicts like those returned by
ckan.logic.action.get.dashboard_activity_list()
:returns: a list of email notifications
:rtype: list of dicts each with keys 'subject' and 'body'
'''
if not activities:
return []
# We just group all activities into a single "new activity" email that
# doesn't say anything about _what_ new activities they are.
# TODO: Here we could generate some smarter content for the emails e.g.
# say something about the contents of the activities, or single out
# certain types of activity to be sent in their own individual emails,
# etc.
notifications = [{
'subject': "You have new activity",
'body': "You have new activity"
}]
return notifications
def _notifications_from_dashboard_activity_list(user_id, since):
'''Return any email notifications from user_id's dashboard activity list
since `since`.
'''
# Get the user's dashboard activity stream.
context = {'model': model, 'session': model.Session, 'user': user_id}
activity_list = logic.get_action('dashboard_activity_list')(context, {})
# Filter out the user's own activities., so they don't get an email every
# time they themselves do something (we are not Trac).
activity_list = [activity for activity in activity_list
if activity['user_id'] != user_id]
# Filter out the old activities.
strptime = datetime.datetime.strptime
fmt = '%Y-%m-%dT%H:%M:%S.%f'
activity_list = [activity for activity in activity_list
if strptime(activity['timestamp'], fmt) > since]
return _notifications_for_activities(activity_list)
# A list of functions that provide email notifications for users from different
# sources. Add to this list if you want to implement a new source of email
# notifications.
_notifications_functions = [
_notifications_from_dashboard_activity_list,
]
def get_notifications(user_id, since):
'''Return any email notifications for `user_id` since `since`.
For example email notifications about activity streams will be returned for
any activities the occurred since `since`.
:param user_id: id of the user to return notifications for
:type user_id: string
:param since: datetime after which to return notifications from
:rtype since: datetime.datetime
:returns: a list of email notifications
:rtype: list of dicts with keys 'subject' and 'body'
'''
notifications = []
for function in _notifications_functions:
notifications.extend(function(user_id, since))
return notifications
def send_notification(user, email_dict):
'''Email `email_dict` to `user`.'''
import ckan.lib.mailer
if not user.get('email'):
# FIXME: Raise an exception.
return
try:
ckan.lib.mailer.mail_recipient(user['display_name'], user['email'],
email_dict['subject'], email_dict['body'])
except ckan.lib.mailer.MailerException:
raise
def get_and_send_notifications_for_user(user):
if user['email_notifications']:
# FIXME: We are accessing model from lib here but I'm not sure what
# else to do unless we add a get_email_last_sent() logic function which
# would only be needed by this lib.
email_last_sent = model.Dashboard.get(user['id']).email_last_sent
activity_stream_last_viewed = (
model.Dashboard.get(user['id']).activity_stream_last_viewed)
since = max(email_last_sent, activity_stream_last_viewed)
notifications = get_notifications(user['id'], since)
# TODO: Handle failures from send_email_notification.
for notification in notifications:
send_notification(user, notification)
# Whether the user had har 'email_notifications' preference turned on or
# not, we still update her email_last_sent time. This prevents users from
# getting emails about old activities when they turn on email
# notifications.
# FIXME: We are accessing model from lib here but I'm not sure what
# else to do unless we add a update_email_last_sent()
# logic function which would only be needed by this lib.
model.Dashboard.get(user['id']).email_last_sent = datetime.datetime.now()
def get_and_send_notifications_for_all_users():
context = {'model': model, 'session': model.Session, 'ignore_auth': True,
'keep_sensitive_data': True}
users = logic.get_action('user_list')(context, {})
for user in users:
get_and_send_notifications_for_user(user)