def send_share_notification(share_id):
"""Super simple you're content has been shared notification to a user."""
if settings.DEBUG:
return
try:
content = Content.objects.get(id=share_id, content_type=ContentType.SHARE, share_of__local=True)
except Content.DoesNotExist:
logger.warning("No share content found with id %s", share_id)
return
content_url = "%s%s" % (settings.SOCIALHOME_URL, content.share_of.get_absolute_url())
subject = _("New share of: %s" % content.share_of.short_text_inline)
context = get_common_context()
context.update({
"subject": subject, "actor_name": content.author.name_or_handle,
"actor_url": "%s%s" % (settings.SOCIALHOME_URL, content.author.get_absolute_url()),
"content_url": content_url, "name": content.share_of.author.name_or_handle,
})
send_mail(
"%s%s" % (settings.EMAIL_SUBJECT_PREFIX, subject),
render_to_string("notifications/share.txt", context=context),
settings.DEFAULT_FROM_EMAIL,
[content.share_of.author.user.email],
fail_silently=False,
html_message=render_to_string("notifications/share.html", context=context),
)
python类EMAIL_SUBJECT_PREFIX的实例源码
def _send_invitation_mail(request, invitation, subject, template_name):
if not invitation.invitee.email:
return
old_lang = translation.get_language()
translation.activate(invitation.invitee.language)
template = loader.get_template('groups/mail_{0}.txt'.format(template_name))
message = template.render({
'invitation': invitation,
'site': get_current_site(request)
})
translation.activate(old_lang)
send_mail(settings.EMAIL_SUBJECT_PREFIX + subject,
message,
settings.DEFAULT_FROM_EMAIL,
[invitation.invitee.email],
fail_silently=True)
def test_reviews_create(self):
url = reverse('pages:review_create')
response = self.client.post(url, {
'text': 'test_text',
'contacts': 'client contacts'
})
self.assertContains(
response, 'message'
)
review = Review.objects.filter(
text__startswith='test_text',
text__contains='client contacts',
is_enabled=False
)
self.assertGreater(review.count(), 0)
self.assertEqual(mail.outbox[0].subject, settings.EMAIL_SUBJECT_PREFIX + 'New client review.')
def test_send_expiration_mails(mailoutbox, mocker, now, cluster_factory):
cluster = cluster_factory(
expires_at=now + timedelta(minutes=59), # 1 hours is the cut-off
most_recent_status=models.Cluster.STATUS_WAITING,
)
assert len(mailoutbox) == 0
tasks.send_expiration_mails()
assert len(mailoutbox) == 1
message = mailoutbox[0]
assert message.subject == (
'%sCluster %s is expiring soon!' %
(settings.EMAIL_SUBJECT_PREFIX, cluster.identifier)
)
assert message.from_email == settings.DEFAULT_FROM_EMAIL
assert list(message.to) == [cluster.created_by.email]
cluster.refresh_from_db()
assert cluster.expiration_mail_sent
def send_email(self, rendered_notification, notification):
from django.utils.safestring import SafeText
assert (isinstance(rendered_notification, SafeText))
try:
# TODO: send_mass_mail to send to multiple recipients
# mail_managers(u'{}'.format(notification.title),
# u'{}'.format(''),
# fail_silently=False, html_message=rendered_notification)
send_mail('%s%s' % (settings.EMAIL_SUBJECT_PREFIX, notification.title),
u'{}'.format(''),
from_email=settings.SERVER_EMAIL,
recipient_list=[self.email_to, ],
fail_silently=False,
html_message=rendered_notification)
except Exception as e:
raise e
def mail_admins(subject, message, fail_silently=False, priority="medium"):
from django.utils.encoding import force_unicode
from django.conf import settings
from mailer.models import Message
priority = PRIORITY_MAPPING[priority]
subject = settings.EMAIL_SUBJECT_PREFIX + force_unicode(subject)
message = force_unicode(message)
if len(subject) > 100:
subject = u"%s..." % subject[:97]
for name, to_address in settings.ADMINS:
Message(to_address=to_address,
from_address=settings.SERVER_EMAIL,
subject=subject,
message_body=message,
priority=priority).save()
def mail_managers(subject, message, fail_silently=False, priority="medium"):
from django.utils.encoding import force_unicode
from django.conf import settings
from mailer.models import Message
priority = PRIORITY_MAPPING[priority]
subject = settings.EMAIL_SUBJECT_PREFIX + force_unicode(subject)
message = force_unicode(message)
if len(subject) > 100:
subject = u"%s..." % subject[:97]
for name, to_address in settings.MANAGERS:
Message(to_address=to_address,
from_address=settings.SERVER_EMAIL,
subject=subject,
message_body=message,
priority=priority).save()
def send_recover_mail(self):
from sentry.http import get_server_hostname
from sentry.utils.email import MessageBuilder
context = {
'user': self.user,
'domain': get_server_hostname(),
'url': absolute_uri(reverse(
'sentry-account-recover-confirm',
args=[self.user.id, self.hash]
)),
}
msg = MessageBuilder(
subject='%sPassword Recovery' % (settings.EMAIL_SUBJECT_PREFIX,),
template='sentry/emails/recover_account.txt',
context=context,
)
msg.send_async([self.user.email])
def status_mail(request):
form = TestEmailForm(request.POST or None)
if form.is_valid():
body = """This email was sent as a request to test the Sentry outbound email configuration."""
try:
send_mail(
'%s Test Email' % (settings.EMAIL_SUBJECT_PREFIX,),
body, settings.SERVER_EMAIL, [request.user.email],
fail_silently=False
)
except Exception as e:
form.errors['__all__'] = [six.text_type(e)]
return render_to_response('sentry/admin/status/mail.html', {
'form': form,
'EMAIL_HOST': settings.EMAIL_HOST,
'EMAIL_HOST_PASSWORD': bool(settings.EMAIL_HOST_PASSWORD),
'EMAIL_HOST_USER': settings.EMAIL_HOST_USER,
'EMAIL_PORT': settings.EMAIL_PORT,
'EMAIL_USE_TLS': settings.EMAIL_USE_TLS,
'SERVER_EMAIL': settings.SERVER_EMAIL,
}, request)
def send_follow_notification(follower_id, followed_id):
"""Super simple you've been followed notification to a user."""
if settings.DEBUG:
return
try:
user = User.objects.get(profile__id=followed_id, is_active=True)
except User.DoesNotExist:
logger.warning("No active user with profile %s found for follow notification", followed_id)
return
try:
follower = Profile.objects.get(id=follower_id)
except Profile.DoesNotExist:
logger.warning("No follower profile %s found for follow notifications", follower_id)
return
logger.info("send_follow_notification - Sending mail to %s", user.email)
subject = _("New follower: %s" % follower.handle)
context = get_common_context()
context.update({
"subject": subject, "actor_name": follower.name_or_handle,
"actor_url": "%s%s" % (settings.SOCIALHOME_URL, follower.get_absolute_url()),
"name": user.profile.name_or_handle,
})
send_mail(
"%s%s" % (settings.EMAIL_SUBJECT_PREFIX, subject),
render_to_string("notifications/follow.txt", context=context),
settings.DEFAULT_FROM_EMAIL,
[user.email],
fail_silently=False,
html_message=render_to_string("notifications/follow.html", context=context),
)
def send_reply_notifications(content_id):
"""Super simple reply notification to content local participants.
Until proper notifications is supported, just pop out an email.
"""
if settings.DEBUG:
return
try:
content = Content.objects.get(id=content_id, content_type=ContentType.REPLY)
except Content.DoesNotExist:
logger.warning("No reply content found with id %s", content_id)
return
root_content = content.root
exclude_user = content.author.user if content.local else None
participants = get_root_content_participants(root_content, exclude_user=exclude_user)
if not participants:
return
subject = _("New reply to: %s" % root_content.short_text_inline)
# TODO use fragment url to reply directly when available
content_url = "%s%s" % (settings.SOCIALHOME_URL, root_content.get_absolute_url())
context = get_common_context()
context.update({
"subject": subject, "actor_name": content.author.name_or_handle,
"actor_url": "%s%s" % (settings.SOCIALHOME_URL, content.author.get_absolute_url()),
"reply_text": content.text, "reply_rendered": content.rendered, "reply_url": content_url,
})
for participant in participants:
context["name"] = participant.profile.name_or_handle
logger.info("send_reply_notifications - Sending mail to %s", participant.email)
send_mail(
"%s%s" % (settings.EMAIL_SUBJECT_PREFIX, subject),
render_to_string("notifications/reply.txt", context=context),
settings.DEFAULT_FROM_EMAIL,
[participant.email],
fail_silently=False,
html_message=render_to_string("notifications/reply.html", context=context),
)
def run(self):
if not settings.ADMINS:
return
mail = EmailMultiAlternatives('%s%s' % (settings.EMAIL_SUBJECT_PREFIX, self.subject),
self.message, settings.SERVER_EMAIL, [a[1] for a in settings.ADMINS],
connection=self.connection)
if self.html_message:
mail.attach_alternative(self.html_message, 'text/html')
try:
mail.send(fail_silently=self.fail_silently)
except Exception as e:
logger.exception(e)
def mail_admins(subject, message, fail_silently=False, connection=None,
html_message=None):
"""Sends a message to the admins, as defined by the ADMINS setting."""
if not settings.ADMINS:
return
mail = EmailMultiAlternatives('%s%s' % (settings.EMAIL_SUBJECT_PREFIX, subject),
message, settings.SERVER_EMAIL, [a[1] for a in settings.ADMINS],
connection=connection)
if html_message:
mail.attach_alternative(html_message, 'text/html')
mail.send(fail_silently=fail_silently)
def mail_managers(subject, message, fail_silently=False, connection=None,
html_message=None):
"""Sends a message to the managers, as defined by the MANAGERS setting."""
if not settings.MANAGERS:
return
mail = EmailMultiAlternatives('%s%s' % (settings.EMAIL_SUBJECT_PREFIX, subject),
message, settings.SERVER_EMAIL, [a[1] for a in settings.MANAGERS],
connection=connection)
if html_message:
mail.attach_alternative(html_message, 'text/html')
mail.send(fail_silently=fail_silently)
def mail_admins(subject, message, fail_silently=False, connection=None,
html_message=None):
"""Sends a message to the admins, as defined by the ADMINS setting."""
if not settings.ADMINS:
return
mail = EmailMultiAlternatives(
'%s%s' % (settings.EMAIL_SUBJECT_PREFIX, subject), message,
settings.SERVER_EMAIL, [a[1] for a in settings.ADMINS],
connection=connection,
)
if html_message:
mail.attach_alternative(html_message, 'text/html')
mail.send(fail_silently=fail_silently)
def mail_managers(subject, message, fail_silently=False, connection=None,
html_message=None):
"""Sends a message to the managers, as defined by the MANAGERS setting."""
if not settings.MANAGERS:
return
mail = EmailMultiAlternatives(
'%s%s' % (settings.EMAIL_SUBJECT_PREFIX, subject), message,
settings.SERVER_EMAIL, [a[1] for a in settings.MANAGERS],
connection=connection,
)
if html_message:
mail.attach_alternative(html_message, 'text/html')
mail.send(fail_silently=fail_silently)
def mail_admins(subject, message, fail_silently=False, connection=None,
html_message=None):
"""Sends a message to the admins, as defined by the ADMINS setting."""
if not settings.ADMINS:
return
mail = EmailMultiAlternatives(
'%s%s' % (settings.EMAIL_SUBJECT_PREFIX, subject), message,
settings.SERVER_EMAIL, [a[1] for a in settings.ADMINS],
connection=connection,
)
if html_message:
mail.attach_alternative(html_message, 'text/html')
mail.send(fail_silently=fail_silently)
def mail_managers(subject, message, fail_silently=False, connection=None,
html_message=None):
"""Sends a message to the managers, as defined by the MANAGERS setting."""
if not settings.MANAGERS:
return
mail = EmailMultiAlternatives(
'%s%s' % (settings.EMAIL_SUBJECT_PREFIX, subject), message,
settings.SERVER_EMAIL, [a[1] for a in settings.MANAGERS],
connection=connection,
)
if html_message:
mail.attach_alternative(html_message, 'text/html')
mail.send(fail_silently=fail_silently)
def mail_client(subject, template, data, email=None, client=None):
send_mail(
recipient_list=[email] if email else [client.email],
from_email=settings.DEFAULT_FROM_EMAIL,
subject='{prefix}{text}'.format(
prefix=settings.EMAIL_SUBJECT_PREFIX, text=subject),
message='',
html_message=render_to_string(template, data))
logging.getLogger('profit').info(
'Send mail to client. Subject: {}; client: {}; email: {};'.format(
subject, client, email))
def mail_admins(subject, message, fail_silently=False, connection=None,
html_message=None):
"""Sends a message to the admins, as defined by the ADMINS setting."""
if not settings.ADMINS:
return
mail = EmailMultiAlternatives(
'%s%s' % (settings.EMAIL_SUBJECT_PREFIX, subject), message,
settings.SERVER_EMAIL, [a[1] for a in settings.ADMINS],
connection=connection,
)
if html_message:
mail.attach_alternative(html_message, 'text/html')
mail.send(fail_silently=fail_silently)
def mail_managers(subject, message, fail_silently=False, connection=None,
html_message=None):
"""Sends a message to the managers, as defined by the MANAGERS setting."""
if not settings.MANAGERS:
return
mail = EmailMultiAlternatives(
'%s%s' % (settings.EMAIL_SUBJECT_PREFIX, subject), message,
settings.SERVER_EMAIL, [a[1] for a in settings.MANAGERS],
connection=connection,
)
if html_message:
mail.attach_alternative(html_message, 'text/html')
mail.send(fail_silently=fail_silently)
def send_validation(strategy, backend, code, partial_token):
url = '{0}?verification_code={1}&partial_token={2}'.format(
reverse('social:complete', args=(backend.name,)),
code.code,
partial_token
)
url = strategy.request.build_absolute_uri(url)
msg = EmailMultiAlternatives(
subject='{0} Validate your account'.format(settings.EMAIL_SUBJECT_PREFIX),
body='Validate your account {0}'.format(url),
from_email=settings.VALIDATION_EMAIL_SENDER,
to=[code.email],
)
msg.esp_extra = {"sender_domain": settings.EMAIL_SENDER_DOMAIN}
msg.send()
def mail_admins(subject, message, fail_silently=False, connection=None,
html_message=None):
"""Sends a message to the admins, as defined by the ADMINS setting."""
if not settings.ADMINS:
return
mail = EmailMultiAlternatives(
'%s%s' % (settings.EMAIL_SUBJECT_PREFIX, subject), message,
settings.SERVER_EMAIL, [a[1] for a in settings.ADMINS],
connection=connection,
)
if html_message:
mail.attach_alternative(html_message, 'text/html')
mail.send(fail_silently=fail_silently)
def mail_managers(subject, message, fail_silently=False, connection=None,
html_message=None):
"""Sends a message to the managers, as defined by the MANAGERS setting."""
if not settings.MANAGERS:
return
mail = EmailMultiAlternatives(
'%s%s' % (settings.EMAIL_SUBJECT_PREFIX, subject), message,
settings.SERVER_EMAIL, [a[1] for a in settings.MANAGERS],
connection=connection,
)
if html_message:
mail.attach_alternative(html_message, 'text/html')
mail.send(fail_silently=fail_silently)
def mail_admins(subject, message, fail_silently=False, connection=None,
html_message=None):
"""Sends a message to the admins, as defined by the ADMINS setting."""
if not settings.ADMINS:
return
mail = EmailMultiAlternatives('%s%s' % (settings.EMAIL_SUBJECT_PREFIX, subject),
message, settings.SERVER_EMAIL, [a[1] for a in settings.ADMINS],
connection=connection)
if html_message:
mail.attach_alternative(html_message, 'text/html')
mail.send(fail_silently=fail_silently)
def mail_managers(subject, message, fail_silently=False, connection=None,
html_message=None):
"""Sends a message to the managers, as defined by the MANAGERS setting."""
if not settings.MANAGERS:
return
mail = EmailMultiAlternatives('%s%s' % (settings.EMAIL_SUBJECT_PREFIX, subject),
message, settings.SERVER_EMAIL, [a[1] for a in settings.MANAGERS],
connection=connection)
if html_message:
mail.attach_alternative(html_message, 'text/html')
mail.send(fail_silently=fail_silently)
def mail_user(subject, template, data, email=None, user=None):
send_mail(
recipient_list=[email] if email else user.email(),
from_email=settings.DEFAULT_FROM_EMAIL,
subject='{prefix}{text}'.format(prefix=settings.EMAIL_SUBJECT_PREFIX, text=subject),
message='',
html_message=render_to_string(template, data)
)
def test_mail_user(self):
subject = 'Test user message'
user = User.objects.first()
Mailer.mail_user(subject, 'emails/base.html',
{'content': 'test content'}, user)
self.assertEqual(len(mail.outbox), 1)
self.assertIn(user, mail.outbox[0].recipients())
self.assertEqual(mail.outbox[0].subject,
settings.EMAIL_SUBJECT_PREFIX + subject)
def test_mail_managers_task(self):
"""
Test mail_managers_task
"""
mail_managers_task.delay(
subject='New message via contact form',
template='emails/contact_form.html',
data={'name': 'test name', 'contacts': 'test contacts', 'message': 'test message'}
)
self.assertEqual(len(mail.outbox), 1)
self.assertEqual(mail.outbox[0].subject, settings.EMAIL_SUBJECT_PREFIX + 'New message via contact form')
def test_new_sms_notifications_task(self):
"""
Test test_sms_notifications
"""
new_sms_notifications_task.delay()
self.assertEqual(len(mail.outbox), 1)
self.assertEqual(mail.outbox[0].subject,
settings.EMAIL_SUBJECT_PREFIX + 'New sms waiting')