~launchpad-pqm/launchpad/devel

# Copyright 2009 Canonical Ltd.  This software is licensed under the

# GNU Affero General Public License version 3 (see the file LICENSE).

"""Notifications for the Answers system."""

__metaclass__ = type

__all__ = [

    'QuestionNotification',

    ]

import os

from zope.component import getUtility

from lp.services.config import config

from lp.services.webapp.publisher import canonical_url

from lp.answers.enums import (

    QuestionAction,

    QuestionRecipientSet,

    )

from lp.answers.interfaces.questionjob import IQuestionEmailJobSource

from lp.registry.interfaces.person import IPerson

from lp.services.mail.mailwrapper import MailWrapper

from lp.services.propertycache import cachedproperty

def get_email_template(filename):

    """Returns the email template with the given file name.

    The templates are located in 'emailtemplates'.

    """

    base = os.path.dirname(__file__)

    fullpath = os.path.join(base, 'emailtemplates', filename)

    return open(fullpath).read()

class QuestionNotification:

    """Base class for a notification related to a question.

    Creating an instance of that class will build the notification and

    send it to the appropriate recipients. That way, subclasses of

    QuestionNotification can be registered as event subscribers.

    """

    recipient_set = QuestionRecipientSet.ASKER_SUBSCRIBER

    def __init__(self, question, event):

        """Base constructor.

        It saves the question and event in attributes and then call

        the initialize() and send() method.

        """

        self.question = question

        self.event = event

        self._user = IPerson(self.event.user)

        self.initialize()

        self.job = None

        if self.shouldNotify():

            self.job = self.enqueue()

    @property

    def user(self):

        """Return the user from the event. """

        return self._user

    def getSubject(self):

        """Return the subject of the notification.

        Default to [Question #dd]: Title

        """

        return '[Question #%s]: %s' % (self.question.id, self.question.title)

    def getBody(self):

        """Return the content of the notification message.

        This method must be implemented by a subclass.

        """

        raise NotImplementedError

    def getHeaders(self):

        """Return additional headers to add to the email.

        Default implementation adds a X-Launchpad-Question header.

        """

        question = self.question

        headers = dict()

        if self.question.distribution:

            if question.sourcepackagename:

                sourcepackage = question.sourcepackagename.name

            else:

                sourcepackage = 'None'

            target = 'distribution=%s; sourcepackage=%s;' % (

                question.distribution.name, sourcepackage)

        else:

            target = 'product=%s;' % question.product.name

        if question.assignee:

            assignee = question.assignee.name

        else:

            assignee = 'None'

        headers['X-Launchpad-Question'] = (

            '%s status=%s; assignee=%s; priority=%s; language=%s' % (

                target, question.status.title, assignee,

                question.priority.title, question.language.code))

        headers['Reply-To'] = 'question%s@%s' % (

            self.question.id, config.answertracker.email_domain)

        return headers

    def initialize(self):

        """Initialization hook for subclasses.

        This method is called before send() and can be use for any

        setup purpose.

        Default does nothing.

        """

        pass

    def shouldNotify(self):

        """Return if there is something to notify about.

        When this method returns False, no notification will be sent.

        By default, all event trigger a notification.

        """

        return True

    def enqueue(self):

        """Create a job to send email about the event."""

        subject = self.getSubject()

        body = self.getBody()

        headers = self.getHeaders()

        job_source = getUtility(IQuestionEmailJobSource)

        job = job_source.create(

            self.question, self.user, self.recipient_set,

            subject, body, headers)

        return job

    @property

    def unsupported_language(self):

        """Whether the question language is unsupported or not."""

        supported_languages = self.question.target.getSupportedLanguages()

        return self.question.language not in supported_languages

    @property

    def unsupported_language_warning(self):

        """Warning about the fact that the question is written in an

        unsupported language."""

        return get_email_template(

                'question-unsupported-language-warning.txt') % {

                'question_language': self.question.language.englishname,

                'target_name': self.question.target.displayname}

class QuestionAddedNotification(QuestionNotification):

    """Notification sent when a question is added."""

    @property

    def user(self):

        """Return the question owner.

        Questions can be created by other users for the owner; the

        question is from the owner.

        """

        return self.question.owner

    def getBody(self):

        """See QuestionNotification."""

        question = self.question

        body = get_email_template('question-added-notification.txt') % {

            'target_name': question.target.displayname,

            'question_id': question.id,

            'question_url': canonical_url(question),

            'comment': question.description}

        if self.unsupported_language:

            body += self.unsupported_language_warning

        return body

class QuestionModifiedDefaultNotification(QuestionNotification):

    """Base implementation of a notification when a question is modified."""

    recipient_set = QuestionRecipientSet.SUBSCRIBER

    # Email template used to render the body.

    body_template = "question-modified-notification.txt"

    def initialize(self):

        """Save the old question for comparison. It also set the new_message

        attribute if a new message was added.

        """

        self.old_question = self.event.object_before_modification

        new_messages = set(

            self.question.messages).difference(self.old_question.messages)

        assert len(new_messages) <= 1, (

                "There shouldn't be more than one message for a "

                "notification.")

        if new_messages:

            self.new_message = new_messages.pop()

        else:

            self.new_message = None

        self.wrapper = MailWrapper()

    @cachedproperty

    def metadata_changes_text(self):

        """Textual representation of the changes to the question metadata."""

        question = self.question

        old_question = self.old_question

        indent = 4 * ' '

        info_fields = []

        if question.status != old_question.status:

            info_fields.append(indent + 'Status: %s => %s' % (

                old_question.status.title, question.status.title))

        if question.target != old_question.target:

            info_fields.append(

                indent + 'Project: %s => %s' % (

                old_question.target.displayname, question.target.displayname))

        if question.assignee != old_question.assignee:

            if old_question.assignee is None:

                old_assignee = None

            else:

                old_assignee = old_question.assignee.displayname

            if question.assignee is None:

                assignee = None

            else:

                assignee = question.assignee.displayname

            info_fields.append(indent + 'Assignee: %s => %s' % (

               old_assignee, assignee))

        old_bugs = set(old_question.bugs)

        bugs = set(question.bugs)

        for linked_bug in bugs.difference(old_bugs):

            info_fields.append(

                indent + 'Linked to bug: #%s\n' % linked_bug.id +

                indent + '%s\n' % canonical_url(linked_bug) +

                indent + '"%s"' % linked_bug.title)

        for unlinked_bug in old_bugs.difference(bugs):

            info_fields.append(

                indent + 'Removed link to bug: #%s\n' % unlinked_bug.id +

                indent + '%s\n' % canonical_url(unlinked_bug) +

                indent + '"%s"' % unlinked_bug.title)

        if question.faq != old_question.faq:

            if question.faq is None:

                info_fields.append(

                    indent + 'Related FAQ was removed:\n' +

                    indent + old_question.faq.title + '\n' +

                    indent + canonical_url(old_question.faq))

            else:

                info_fields.append(

                    indent + 'Related FAQ set to:\n' +

                    indent + question.faq.title + '\n' +

                    indent + canonical_url(question.faq))

        if question.title != old_question.title:

            info_fields.append('Summary changed to:\n%s' % question.title)

        if question.description != old_question.description:

            info_fields.append(

                'Description changed to:\n%s' % (

                    self.wrapper.format(question.description)))

        question_changes = '\n\n'.join(info_fields)

        return question_changes

    def getSubject(self):

        """The reply subject line."""

        line = super(QuestionModifiedDefaultNotification, self).getSubject()

        return 'Re: %s' % line

    def getHeaders(self):

        """Add a References header."""

        headers = QuestionNotification.getHeaders(self)

        if self.new_message:

            # XXX flacoste 2007-02-02 bug=83846:

            # The first message cannot contain a References

            # because we don't create a Message instance for the

            # question description, so we don't have a Message-ID.

            messages = list(self.question.messages)

            assert self.new_message in messages, (

                "Question %s: message id %s not in %s." % (

                    self.question.id, self.new_message.id,

                    [m.id for m in messages]))

            index = messages.index(self.new_message)

            if index > 0:

                headers['References'] = (

                    self.question.messages[index - 1].rfc822msgid)

        return headers

    def shouldNotify(self):

        """Only send a notification when a message was added or some

        metadata was changed.

        """

        return self.new_message or self.metadata_changes_text

    def getBody(self):

        """See QuestionNotification."""

        body = self.metadata_changes_text

        replacements = dict(

            question_id=self.question.id,

            target_name=self.question.target.displayname,

            question_url=canonical_url(self.question))

        if self.new_message:

            if body:

                body += '\n\n'

            body += self.getNewMessageText()

            replacements['new_message_id'] = list(

                self.question.messages).index(self.new_message)

        replacements['body'] = body

        return get_email_template(self.body_template) % replacements

    # Header template used when a new message is added to the question.

    action_header_template = {

        QuestionAction.REQUESTINFO:

            '%(person)s requested more information:',

        QuestionAction.CONFIRM:

            '%(person)s confirmed that the question is solved:',

        QuestionAction.COMMENT:

            '%(person)s posted a new comment:',

        QuestionAction.GIVEINFO:

            '%(person)s gave more information on the question:',

        QuestionAction.REOPEN:

            '%(person)s is still having a problem:',

        QuestionAction.ANSWER:

            '%(person)s proposed the following answer:',

        QuestionAction.EXPIRE:

            '%(person)s expired the question:',

        QuestionAction.REJECT:

            '%(person)s rejected the question:',

        QuestionAction.SETSTATUS:

            '%(person)s changed the question status:',

    }

    def getNewMessageText(self):

        """Should return the notification text related to a new message."""

        if not self.new_message:

            return ''

        header = self.action_header_template.get(

            self.new_message.action, '%(person)s posted a new message:') % {

            'person': self.new_message.owner.displayname}

        return '\n'.join([

            header, self.wrapper.format(self.new_message.text_contents)])

class QuestionModifiedOwnerNotification(QuestionModifiedDefaultNotification):

    """Notification sent to the owner when his question is modified."""

    recipient_set = QuestionRecipientSet.ASKER

    # These actions will be done by the owner, so use the second person.

    action_header_template = dict(

        QuestionModifiedDefaultNotification.action_header_template)

    action_header_template.update({

        QuestionAction.CONFIRM:

            'You confirmed that the question is solved:',

        QuestionAction.GIVEINFO:

            'You gave more information on the question:',

        QuestionAction.REOPEN:

            'You are still having a problem:',

        })

    body_template = 'question-modified-owner-notification.txt'

    body_template_by_action = {

        QuestionAction.ANSWER: "question-answered-owner-notification.txt",

        QuestionAction.EXPIRE: "question-expired-owner-notification.txt",

        QuestionAction.REJECT: "question-rejected-owner-notification.txt",

        QuestionAction.REQUESTINFO: (

            "question-info-requested-owner-notification.txt"),

    }

    def initialize(self):

        """Set the template based on the new comment action."""

        QuestionModifiedDefaultNotification.initialize(self)

        if self.new_message:

            self.body_template = self.body_template_by_action.get(

                self.new_message.action, self.body_template)

    def getBody(self):

        """See QuestionNotification."""

        body = QuestionModifiedDefaultNotification.getBody(self)

        if self.unsupported_language:

            body += self.unsupported_language_warning

        return body

class QuestionUnsupportedLanguageNotification(QuestionNotification):

    """Notification sent to answer contacts for unsupported languages."""

    recipient_set = QuestionRecipientSet.CONTACT

    def getSubject(self):

        """See QuestionNotification."""

        return '[Question #%s]: (%s) %s' % (

            self.question.id, self.question.language.englishname,

            self.question.title)

    def shouldNotify(self):

        """Return True when the question is in an unsupported language."""

        return self.unsupported_language

    def getBody(self):

        """See QuestionNotification."""

        question = self.question

        return get_email_template(

                'question-unsupported-languages-added.txt') % {

            'target_name': question.target.displayname,

            'question_id': question.id,

            'question_url': canonical_url(question),

            'question_language': question.language.englishname,

            'comment': question.description}