'''Manage notifications and subscriptions
When an artifact is modified:
- Notification generated by tool app
- Search is made for subscriptions matching the notification
- Notification is added to each matching subscriptions' queue
Periodically:
- For each subscriptions with notifications and direct delivery:
- For each notification, enqueue as a separate email message
- Clear subscription's notification list
- For each subscription with notifications and delivery due:
- Enqueue one email message with all notifications
- Clear subscription's notification list
Notifications are also available for use in feeds
'''
import logging
from bson import ObjectId
from datetime import datetime, timedelta
from collections import defaultdict
from webhelpers import feedgenerator as FG
from pylons import c, g
from tg import config
import pymongo
import jinja2
from ming import schema as S
from ming.orm import FieldProperty, ForeignIdProperty, RelationProperty, session
from ming.orm.declarative import MappedClass
from allura.lib import helpers as h
from allura.lib import security
import allura.tasks.mail_tasks
from .session import main_orm_session, project_orm_session
from .auth import User
log = logging.getLogger(__name__)
MAILBOX_QUIESCENT=None # Re-enable with [#1384]: timedelta(minutes=10)
class Notification(MappedClass):
class __mongometa__:
session = main_orm_session
name = 'notification'
indexes = [ ('neighborhood_id', 'tool_name', 'pubdate') ]
_id = FieldProperty(str, if_missing=h.gen_message_id)
# Classify notifications
neighborhood_id = ForeignIdProperty('Neighborhood', if_missing=lambda:c.project.neighborhood._id)
project_id = ForeignIdProperty('Project', if_missing=lambda:c.project._id)
app_config_id = ForeignIdProperty('AppConfig', if_missing=lambda:c.app.config._id)
tool_name = FieldProperty(str, if_missing=lambda:c.app.config.tool_name)
ref_id = ForeignIdProperty('ArtifactReference')
topic = FieldProperty(str)
unique_id = FieldProperty(str, if_missing=lambda:h.nonce(40))
# Notification Content
in_reply_to=FieldProperty(str)
from_address=FieldProperty(str)
reply_to_address=FieldProperty(str)
subject=FieldProperty(str)
text=FieldProperty(str)
link=FieldProperty(str)
author_id=ForeignIdProperty('User')
feed_meta=FieldProperty(S.Deprecated)
artifact_reference = FieldProperty(S.Deprecated)
pubdate = FieldProperty(datetime, if_missing=datetime.utcnow)
ref = RelationProperty('ArtifactReference')
view = jinja2.Environment(
loader=jinja2.PackageLoader('allura', 'templates'))
def author(self):
return User.query.get(_id=self.author_id) or User.anonymous()
@classmethod
def post(cls, artifact, topic, **kw):
'''Create a notification and send the notify message'''
import allura.tasks.notification_tasks
n = cls._make_notification(artifact, topic, **kw)
if n:
allura.tasks.notification_tasks.notify.post(
n._id, artifact.index_id(), topic)
return n
@classmethod
def post_user(cls, user, artifact, topic, **kw):
'''Create a notification and deliver directly to a user's flash
mailbox'''
try:
mbox = Mailbox(user_id=user._id, is_flash=True,
project_id=None,
app_config_id=None)
session(mbox).flush(mbox)
except pymongo.errors.DuplicateKeyError:
session(mbox).expunge(mbox)
mbox = Mailbox.query.get(user_id=user._id, is_flash=True)
n = cls._make_notification(artifact, topic, **kw)
if n:
mbox.queue.append(n._id)
return n
@classmethod
def _make_notification(cls, artifact, topic, **kwargs):
from allura.model import Project
idx = artifact.index()
subject_prefix = '[%s:%s] ' % (
c.project.shortname, c.app.config.options.mount_point)
if topic == 'message':
post = kwargs.pop('post')
text = post.text
file_info = kwargs.pop('file_info', None)
if file_info is not None:
file_info.file.seek(0, 2)
bytecount = file_info.file.tell()
file_info.file.seek(0)
text = "%s\n\n\nAttachment: %s (%s; %s)" % (text, file_info.filename, h.do_filesizeformat(bytecount), file_info.type)
subject = post.subject or ''
if post.parent_id and not subject.lower().startswith('re:'):
subject = 'Re: ' + subject
author = post.author()
d = dict(
_id=artifact.url()+post._id,
from_address=str(author._id) if author != User.anonymous() else None,
reply_to_address='"%s" <%s>' % (
subject_prefix, getattr(artifact, 'email_address', 'noreply@in.sf.net')),
subject=subject_prefix + subject,
text=text,
in_reply_to=post.parent_id,
author_id=author._id,
pubdate=datetime.utcnow())
else:
subject = kwargs.pop('subject', '%s modified by %s' % (
idx['title_s'],c.user.get_pref('display_name')))
reply_to = '"%s" <%s>' % (
idx['title_s'],
getattr(artifact, 'email_address', 'noreply@in.sf.net'))
d = dict(
from_address=reply_to,
reply_to_address=reply_to,
subject=subject_prefix + subject,
text=kwargs.pop('text', subject),
author_id=c.user._id,
pubdate=datetime.utcnow())
if c.user.get_pref('email_address'):
d['from_address'] = '"%s" <%s>' % (
c.user.get_pref('display_name'),
c.user.get_pref('email_address'))
elif c.user.email_addresses:
d['from_address'] = '"%s" <%s>' % (
c.user.get_pref('display_name'),
c.user.email_addresses[0])
if not d.get('text'):
d['text'] = ''
try:
''' Add addional text to the notification e-mail based on the artifact type '''
template = cls.view.get_template('mail/' + artifact.type_s + '.txt')
d['text'] += template.render(dict(c=c, g=g, config=config, data=artifact))
except jinja2.TemplateNotFound:
pass
except:
''' Catch any errors loading or rendering the template,
but the notification still gets sent if there is an error
'''
log.warn('Could not render notification template %s' % artifact.type_s, exc_info=True)
assert d['reply_to_address'] is not None
project = Project.query.get(_id=d.get('project_id', c.project._id))
if project.notifications_disabled:
log.info('Notifications disabled for project %s, not sending %s(%r)',
project.shortname, topic, artifact)
return None
n = cls(ref_id=artifact.index_id(),
topic=topic,
link=kwargs.pop('link', artifact.url()),
**d)
return n
@classmethod
def feed(cls, q, feed_type, title, link, description,
since=None, until=None, offset=None, limit=None):
"""Produces webhelper.feedgenerator Feed"""
d = dict(title=title, link=h.absurl(link), description=description, language=u'en')
if feed_type == 'atom':
feed = FG.Atom1Feed(**d)
elif feed_type == 'rss':
feed = FG.Rss201rev2Feed(**d)
query = defaultdict(dict)
query.update(q)
if since is not None:
query['pubdate']['$gte'] = since
if until is not None:
query['pubdate']['$lte'] = until
cur = cls.query.find(query)
cur = cur.sort('pubdate', pymongo.DESCENDING)
if limit is None: limit = 10
query = cur.limit(limit)
if offset is not None: query = cur.offset(offset)
for r in cur:
feed.add_item(title=r.subject,
link=h.absurl(r.link.encode('utf-8')),
pubdate=r.pubdate,
description=r.text,
unique_id=r.unique_id,
author_name=r.author().display_name,
author_link=h.absurl(r.author().url()))
return feed
def footer(self):
template = self.view.get_template('mail/footer.txt')
return template.render(dict(
notification=self,
prefix=config.get('forgemail.url', 'https://sourceforge.net')))
def send_simple(self, toaddr):
allura.tasks.mail_tasks.sendsimplemail.post(
toaddr=toaddr,
fromaddr=self.from_address,
reply_to=self.reply_to_address,
subject=self.subject,
message_id=self._id,
in_reply_to=self.in_reply_to,
text=(self.text or '') + self.footer())
def send_direct(self, user_id):
user = User.query.get(_id=ObjectId(user_id))
artifact = self.ref.artifact
# Don't send if user doesn't have read perms to the artifact
if user and artifact and \
not security.has_access(artifact, 'read', user)():
log.debug("Skipping notification - User %s doesn't have read "
"access to artifact %s" % (user_id, str(self.ref_id)))
return
allura.tasks.mail_tasks.sendmail.post(
destinations=[str(user_id)],
fromaddr=self.from_address,
reply_to=self.reply_to_address,
subject=self.subject,
message_id=self._id,
in_reply_to=self.in_reply_to,
text=(self.text or '') + self.footer())
@classmethod
def send_digest(self, user_id, from_address, subject, notifications,
reply_to_address=None):
if not notifications: return
# Filter out notifications for which the user doesn't have read
# permissions to the artifact.
user = User.query.get(_id=ObjectId(user_id))
artifact = self.ref.artifact
def perm_check(notification):
return not (user and artifact) or \
security.has_access(artifact, 'read', user)()
notifications = filter(perm_check, notifications)
if reply_to_address is None:
reply_to_address = from_address
text = [ 'Digest of %s' % subject ]
for n in notifications:
text.append('From: %s' % n.from_address)
text.append('Subject: %s' % (n.subject or '(no subject)'))
text.append('Message-ID: %s' % n._id)
text.append('')
text.append(n.text or '-no text-')
text.append(n.footer())
text = '\n'.join(text)
allura.tasks.mail_tasks.sendmail.post(
destinations=[str(user_id)],
fromaddr=from_address,
reply_to=reply_to_address,
subject=subject,
message_id=h.gen_message_id(),
text=text)
@classmethod
def send_summary(self, user_id, from_address, subject, notifications):
if not notifications: return
text = [ 'Digest of %s' % subject ]
for n in notifications:
text.append('From: %s' % n.from_address)
text.append('Subject: %s' % (n.subject or '(no subject)'))
text.append('Message-ID: %s' % n._id)
text.append('')
text.append(h.text.truncate(n.text or '-no text-', 128))
text.append(n.footer())
text = '\n'.join(text)
allura.tasks.mail_tasks.sendmail.post(
destinations=[str(user_id)],
fromaddr=from_address,
reply_to=from_address,
subject=subject,
message_id=h.gen_message_id(),
text=text)
class Mailbox(MappedClass):
class __mongometa__:
session = main_orm_session
name = 'mailbox'
unique_indexes = [
('user_id', 'project_id', 'app_config_id',
'artifact_index_id', 'topic', 'is_flash'),
]
indexes = [
('project_id', 'artifact_index_id'),
('is_flash', 'user_id'),
('type', 'next_scheduled')]
_id = FieldProperty(S.ObjectId)
user_id = ForeignIdProperty('User', if_missing=lambda:c.user._id)
project_id = ForeignIdProperty('Project', if_missing=lambda:c.project._id)
app_config_id = ForeignIdProperty('AppConfig', if_missing=lambda:c.app.config._id)
# Subscription filters
artifact_title = FieldProperty(str)
artifact_url = FieldProperty(str)
artifact_index_id = FieldProperty(str)
topic = FieldProperty(str)
# Subscription type
is_flash = FieldProperty(bool, if_missing=False)
type = FieldProperty(S.OneOf('direct', 'digest', 'summary', 'flash'))
frequency = FieldProperty(dict(
n=int,unit=S.OneOf('day', 'week', 'month')))
next_scheduled = FieldProperty(datetime, if_missing=datetime.utcnow)
# Actual notification IDs
last_modified = FieldProperty(datetime, if_missing=datetime(2000,1,1))
queue = FieldProperty([str])
project = RelationProperty('Project')
app_config = RelationProperty('AppConfig')
@classmethod
def subscribe(
cls,
user_id=None, project_id=None, app_config_id=None,
artifact=None, topic=None,
type='direct', n=1, unit='day'):
if user_id is None: user_id = c.user._id
if project_id is None: project_id = c.project._id
if app_config_id is None: app_config_id = c.app.config._id
tool_already_subscribed = cls.query.get(user_id=user_id,
project_id=project_id,
app_config_id=app_config_id,
artifact_index_id=None)
if tool_already_subscribed:
log.debug('Tried to subscribe to artifact %s, while there is a tool subscription', artifact)
return
if artifact is None:
artifact_title = 'All artifacts'
artifact_url = None
artifact_index_id = None
else:
i = artifact.index()
artifact_title = i['title_s']
artifact_url = artifact.url()
artifact_index_id = i['id']
artifact_already_subscribed = cls.query.get(user_id=user_id,
project_id=project_id,
app_config_id=app_config_id,
artifact_index_id=artifact_index_id)
if artifact_already_subscribed:
return
d = dict(user_id=user_id, project_id=project_id, app_config_id=app_config_id,
artifact_index_id=artifact_index_id, topic=topic)
sess = session(cls)
try:
mbox = cls(
type=type, frequency=dict(n=n, unit=unit),
artifact_title=artifact_title,
artifact_url=artifact_url,
**d)
sess.flush(mbox)
except pymongo.errors.DuplicateKeyError:
sess.expunge(mbox)
mbox = cls.query.get(**d)
mbox.artifact_title = artifact_title
mbox.artifact_url = artifact_url
mbox.type = type
mbox.frequency.n = n
mbox.frequency.unit = unit
sess.flush(mbox)
if not artifact_index_id:
# Unsubscribe from individual artifacts when subscribing to the tool
for other_mbox in cls.query.find(dict(
user_id=user_id, project_id=project_id, app_config_id=app_config_id)):
if other_mbox is not mbox:
other_mbox.delete()
@classmethod
def unsubscribe(
cls,
user_id=None, project_id=None, app_config_id=None,
artifact_index_id=None, topic=None):
if user_id is None: user_id = c.user._id
if project_id is None: project_id = c.project._id
if app_config_id is None: app_config_id = c.app.config._id
cls.query.remove(dict(
user_id=user_id,
project_id=project_id,
app_config_id=app_config_id,
artifact_index_id=artifact_index_id,
topic=topic))
@classmethod
def subscribed(
cls, user_id=None, project_id=None, app_config_id=None,
artifact=None, topic=None):
if user_id is None: user_id = c.user._id
if project_id is None: project_id = c.project._id
if app_config_id is None: app_config_id = c.app.config._id
if artifact is None:
artifact_index_id = None
else:
i = artifact.index()
artifact_index_id = i['id']
return cls.query.find(dict(
user_id=user_id,
project_id=project_id,
app_config_id=app_config_id,
artifact_index_id=artifact_index_id)).count() != 0
@classmethod
def deliver(cls, nid, artifact_index_id, topic):
'''Called in the notification message handler to deliver notification IDs
to the appropriate mailboxes. Atomically appends the nids
to the appropriate mailboxes.
'''
d = {
'project_id':c.project._id,
'app_config_id':c.app.config._id,
'artifact_index_id':{'$in':[None, artifact_index_id]},
'topic':{'$in':[None, topic]}
}
for mbox in cls.query.find(d):
mbox.query.update(
{'$push':dict(queue=nid),
'$set':dict(last_modified=datetime.utcnow())})
# Make sure the mbox doesn't stick around to be flush()ed
session(mbox).expunge(mbox)
@classmethod
def fire_ready(cls):
'''Fires all direct subscriptions with notifications as well as
all summary & digest subscriptions with notifications that are ready
'''
now = datetime.utcnow()
# Queries to find all matching subscription objects
q_direct = dict(
type='direct',
queue={'$ne':[]})
if MAILBOX_QUIESCENT:
q_direct['last_modified']={'$lt':now - MAILBOX_QUIESCENT}
q_digest = dict(
type={'$in': ['digest', 'summary']},
next_scheduled={'$lt':now})
for mbox in cls.query.find(q_direct):
mbox = cls.query.find_and_modify(
query=dict(_id=mbox._id),
update={'$set': dict(
queue=[])},
new=False)
mbox.fire(now)
for mbox in cls.query.find(q_digest):
next_scheduled = now
if mbox.frequency.unit == 'day':
next_scheduled += timedelta(days=mbox.frequency.n)
elif mbox.frequency.unit == 'week':
next_scheduled += timedelta(days=7 * mbox.frequency.n)
elif mbox.frequency.unit == 'month':
next_scheduled += timedelta(days=30 * mbox.frequency.n)
mbox = cls.query.find_and_modify(
query=dict(_id=mbox._id),
update={'$set': dict(
next_scheduled=next_scheduled,
queue=[])},
new=False)
mbox.fire(now)
def fire(self, now):
notifications = Notification.query.find(dict(_id={'$in':self.queue}))
notifications = notifications.all()
if self.type == 'direct':
ngroups = defaultdict(list)
for n in notifications:
if n.topic == 'message':
n.send_direct(self.user_id)
# Messages must be sent individually so they can be replied
# to individually
else:
key = (n.subject, n.from_address, n.reply_to_address, n.author_id)
ngroups[key].append(n)
# Accumulate messages from same address with same subject
for (subject, from_address, reply_to_address, author_id), ns in ngroups.iteritems():
if len(ns) == 1:
n.send_direct(self.user_id)
else:
Notification.send_digest(
self.user_id, from_address, subject, ns, reply_to_address)
elif self.type == 'digest':
Notification.send_digest(
self.user_id, 'noreply@in.sf.net', 'Digest Email',
notifications)
elif self.type == 'summary':
Notification.send_summary(
self.user_id, 'noreply@in.sf.net', 'Digest Email',
notifications)