def send_notification_message(build_id, user_id):
reset_database_connection()
build = Build.objects.get(id=build_id)
user = User.objects.get(id=user_id)
try:
log_lines = build.flows.order_by('-date_end')[0].log.split('\n')
except:
log_lines = build.log.split('\n')
log_lines = '\n'.join(log_lines[-25:])
template_txt = get_template('build/email.txt')
template_html = get_template('build/email.html')
context = {
'build': build,
'log_lines': log_lines,
}
subject = '[{}] Build #{} of {} {} - {}'.format(build.repo.name, build.id, build.branch.name, build.plan.name, build.get_status().upper())
message = template_txt.render(Context(context))
html_message = template_html.render(Context(context))
return send_mail(subject, message, settings.FROM_EMAIL, [user.email], html_message=html_message)
python类send_mail()的实例源码
def emailVerify(email, send_type='registe'):
email_record = EmailVerifyRecord()
code = random_str(8)
email_record.code = code
email_record.email = email
email_record.send_type = send_type
email_record.save()
email_title = ''
email_body = ''
if send_type == 'registe':
email_title = '??????'
email_body = '???????????????http://127.0.0.1:8000/active/{0}'.format(code)
send_status = send_mail(email_title, email_body, EMAIL_FROM, [email])
if send_status:
pass
elif send_type == 'forget':
email_title = '??????'
email_body = '???????????????http://127.0.0.1:8000/reset/{0}'.format(code)
send_status = send_mail(email_title, email_body, EMAIL_FROM, [email])
if send_status:
pass
def send_register_email(email, send_type = 'register'):
email_record = EmailVerifyRecord()
code = random_str(16)
email_record.code = code
email_record.email = email
email_record.send_type = send_type
email_record.save()
email_title = ''
email_body = ''
if send_type == 'register':
email_title = '???????????'
email_body = '??????????????: http://127.0.0.1:8000/active/{0}'.format(code)
send_status = send_mail(email_title, email_body, EMAIL_FROM, [email])
if send_status:
pass
if send_type == 'forget':
email_title = ' ???????????'
email_body = '??????????????: http://127.0.0.1:8000/reset/{0}'.format(code)
send_status = send_mail(email_title,email_body,EMAIL_FROM,[email])
if send_status:
pass
def send_activation_email(self):
if not self.activated:
self.activation_key = code_generator()# 'somekey' #gen key
self.save()
#path_ = reverse()
path_ = reverse('activate', kwargs={"code": self.activation_key})
full_path = "https://muypicky.com" + path_
subject = 'Activate Account'
from_email = settings.DEFAULT_FROM_EMAIL
message = f'Activate your account here: {full_path}'
recipient_list = [self.user.email]
html_message = f'<p>Activate your account here: {full_path}</p>'
print(html_message)
sent_mail = send_mail(
subject,
message,
from_email,
recipient_list,
fail_silently=False,
html_message=html_message)
sent_mail = False
return sent_mail
def advert_add(request):
"""
Renders a form, with which visitor can report their adverts.
Form data is send by email to the Admin, for further verification.
"""
form = ReportAdvertForm()
if request.method == 'POST':
form = ReportAdvertForm(request.POST)
if form.is_valid():
# Sending email with advert data
# TODO: Send mail as Celery task
# test_receiver = None
# msg_html = render_to_string('adverts/mail/advert_report_message.html', form.cleaned_data)
# msg_text = render_to_string('adverts/mail/advert_report_message.txt', form.cleaned_data)
# send_mail('Zg?oszono now? ofert?', msg_text, settings.EMAIL_HOST_USER, [test_receiver],
# html_message=msg_html, fail_silently=False)
return redirect(to=advert_add_success)
fields = list(form)
personal_form, estate_form = fields[:4], fields[4:]
return render(request, template_name='adverts/report_advert.html', context={'personal_form': personal_form,
'estate_form': estate_form})
def get_message_dict(self):
"""
Generate the various parts of the message and return them in a
dictionary, suitable for passing directly as keyword arguments
to ``django.core.mail.send_mail()``.
By default, the following values are returned:
* ``from_email``
* ``message``
* ``recipient_list``
* ``subject``
"""
if not self.is_valid():
raise ValueError("Message cannot be sent from invalid contact form")
message_dict = {}
for message_part in ('from_email', 'message', 'recipient_list', 'subject'):
attr = getattr(self, message_part)
message_dict[message_part] = callable(attr) and attr() or attr
return message_dict
def send_register_email(email, send_type='register'):
email_record = EmailVerifyRecord()
code = random_str(16)
email_record.code = code
email_record.email = email
email_record.send_type = send_type
email_record.save()
if send_type == 'register':
email_title = '???????????'
email_body = '???????????????http://127.0.0.1:8000/active/{0}'.format(code)
send_status = send_mail(email_title, email_body, EMAIL_FROM, [email])
if send_status:
# TODO
print('????')
elif send_type == 'forget':
email_title = '???????????'
email_body = '???????????????http://127.0.0.1:8000/reset/{0}'.format(code)
send_status = send_mail(email_title, email_body, EMAIL_FROM, [email])
if send_status:
# TODO ??????
pass
def send_email(self, request):
'''
Sends an email after being successfully added to the database (for user
submitted exercises only)
'''
try:
user = User.objects.get(username=self.license_author)
except User.DoesNotExist:
return
if self.license_author and user.email:
translation.activate(user.userprofile.notification_language.short_name)
url = request.build_absolute_uri(self.get_absolute_url())
subject = _('Exercise was successfully added to the general database')
context = {
'exercise': self.name,
'url': url,
'site': Site.objects.get_current().domain
}
message = render_to_string('exercise/email_new.tpl', context)
mail.send_mail(subject,
message,
settings.WGER_SETTINGS['EMAIL_FROM'],
[user.email],
fail_silently=True)
def send_email(self, request):
'''
Sends an email after being successfully added to the database (for user
submitted ingredients only)
'''
if self.user and self.user.email:
translation.activate(self.user.userprofile.notification_language.short_name)
url = request.build_absolute_uri(self.get_absolute_url())
subject = _('Ingredient was successfully added to the general database')
context = {
'ingredient': self.name,
'url': url,
'site': Site.objects.get_current().domain
}
message = render_to_string('ingredient/email_new.tpl', context)
mail.send_mail(subject,
message,
settings.WGER_SETTINGS['EMAIL_FROM'],
[self.user.email],
fail_silently=True)
def send_email(user, last_entry, datediff):
'''
Notify a user to input the weight entry
:type user User
:type last_entry Date
'''
# Compose and send the email
translation.activate(user.userprofile.notification_language.short_name)
context = {'site': Site.objects.get_current(),
'date': last_entry,
'days': datediff,
'user': user}
subject = _('You have to enter your weight')
message = loader.render_to_string('workout/email_weight_reminder.tpl', context)
mail.send_mail(subject,
message,
settings.WGER_SETTINGS['EMAIL_FROM'],
[user.email],
fail_silently=True)
def post_share(request, post_id):
post = get_object_or_404(Post, id=post_id, status='published')
sent = False
if request.method == 'POST':
form = EmailPostForm(request.POST)
if form.is_valid():
cd = form.cleaned_data
post_url = request.build_absolute_uri(post.get_absolute_url())
subject = '{} ({}) recommends you reading "{}"'.format(cd['name'], cd['email'], post.title)
message = 'Read "{}" at {}\n\n{}\'s comments: {}'.format(post.title, post_url, cd['name'], cd['comments'])
send_mail(subject, message, config('DEFAULT_FROM_EMAIL'), [cd['to']])
sent = True
else:
form = EmailPostForm()
return render(request, 'blog/post/share.html', {'post': post, 'form': form, 'sent': sent })
def process(instance, parent, created):
for info in Event.get(instance, parent, created):
for recipient in info.recipients:
if recipient == info.actor:
continue
print("Notify {0} about {1} {2}".format(recipient, instance, info.verb))
notify.send(info.actor, verb=info.verb, action_object=instance, target=parent, recipient=recipient, template=info.template_name, url=info.url)
if info.is_user_subscribed(recipient):
subject = info.email_subject
template = loader.get_template(join('assets', 'notification', info.email_body_template))
context = info.email_context
context['user'] = recipient
body = template.render(context)
if subject and body:
print("Send mail to {0}: {1}".format(recipient.email, subject).encode('utf-8'))
send_mail(subject, body, settings.DEFAULT_FROM_EMAIL, [recipient.email], fail_silently = False)
### Specific events
def send_email_confirmation(self, custom_url=None):
if not self.email_confirmed:
self.email_activation_key = hashlib.sha1(str(timezone.now())).hexdigest()
self.email_activation_key_created_at = timezone.now()
self.save()
try:
response = send_mail(
grauth_settings.GRAUTH_EMAIL_CONFIRMATION_SUBJECT,
'Clique no link para ativar sua conta. %s' % self.create_email_validation_url(custom_url),
grauth_settings.GRAUTH_EMAIL_SENDER,
[self.email],
fail_silently=False,
)
return response
except:
return None
def send_password_email_reset(self, custom_url=None):
salt = uuid.uuid4().hex
self.password_reset_key = hashlib.sha1(salt.encode() + str(timezone.now())).hexdigest()
self.password_reset_user_reference = hashlib.sha1(salt.encode() + str(self.email)).hexdigest()
self.password_reset_key_created_at = timezone.now()
self.save()
try:
response = send_mail(
grauth_settings.GRAUTH_PASSWORD_RESET_SUBJECT,
'Clique no link para redefinir a senha. %s' % self.create_reset_url_password(custom_url),
grauth_settings.GRAUTH_EMAIL_SENDER,
[self.email],
fail_silently=False,
)
return response
except:
return None
def send_register_email(email, send_type='register'):
email_record = EmailVerifyRecord()
code = random_str(16)
email_record.code = code
email_record.email = email
email_record.send_type = send_type
email_record.save()
if send_type == 'register':
email_title = '???????????'
# http://118.89.105.65 ???????? IP ????????????????? IP ? ??
email_body = '???????????????http://127.0.0.1:8000/active/{0}??????????? http://127.0.0.1:8000 ??? http://118.89.105.65 ?'.format(code)
send_status = send_mail(email_title, email_body, EMAIL_FROM, [email])
if send_status:
# TODO
print('????')
elif send_type == 'forget':
email_title = '???????????'
email_body = '???????????????http://127.0.0.1:8000/reset/{0}??????????? http://127.0.0.1:8000 ??? http://118.89.105.65 ?'.format(code)
send_status = send_mail(email_title, email_body, EMAIL_FROM, [email])
if send_status:
# TODO ??????
pass
def send_register_email(email, send_type="register"):
email_record = EmailVerifyRecord()
code = random_str(16)
email_record.code = code
email_record.email = email
email_record.send_type = send_type
email_record.save()
email_title = ""
email_body = ""
if send_type == "register":
email_title = "???????????"
email_body = "???????????????http://127.0.0.1:8000/active/{0}".format(code)
send_status = send_mail(email_title, email_body, EMAIL_FROM, [email])
if send_status:
pass
elif send_type == "forget":
email_title = "???????????"
email_body = "???????????????http://127.0.0.1:8000/reset/{0}".format(code)
send_status = send_mail(email_title, email_body, EMAIL_FROM, [email])
if send_status:
pass
def send_activation_email(self):
if not self.activated:
self.activation_key = code_generator()
self.save()
#path_ = reverse()
path_ = reverse('activate', kwargs={"code": self.activation_key})
full_path = "http://00f3nd.pythonanywhere.com" + path_
subject = '????????? ????????'
from_email = settings.DEFAULT_FROM_EMAIL
message = f'??????????? ???? ??????? ?????: {full_path}'
recipient_list = [self.user.email]
html_message = f'<p>??????????? ???? ??????? ?????: {full_path}</p>'
sent_mail = send_mail(
subject,
message,
from_email,
recipient_list,
fail_silently=False,
html_message=html_message
)
sent_mail = False
print(html_message)
return sent_mail
def test_missing_api_key(settings):
settings.POSTMARK = {}
with pytest.raises(ImproperlyConfigured) as exc:
send_mail(**SEND_KWARGS)
assert str(exc.value) == 'You should specify TOKEN to use Postmark email backend'
def test_on_exception(self, catch_signal):
with patch('requests.Session.request', side_effect=ConnectTimeout):
with catch_signal(on_exception) as handler:
send_mail(fail_silently=True, **SEND_KWARGS)
assert handler.called
kwargs = handler.call_args[1]
assert kwargs['sender'] == EmailBackend
assert kwargs['signal'] == on_exception
assert isinstance(kwargs['exception'], ConnectTimeout)
assert len(kwargs['raw_messages']) == 1
def send_ask_confirmation_mail(self, confirmation_token, email, username):
confirmation_url = "{}?token={}".format(self.request.build_absolute_uri(reverse('email-confirmation')),
confirmation_token)
context_params = {'username': username, 'confirmation_url': confirmation_url}
plain_text_message = get_template('ask_confirmation_email.txt').render(context_params)
html_message = get_template('ask_confirmation_email.html').render(context_params)
subject, origin_email, target_email = 'Abidria account confirmation', settings.EMAIL_HOST_USER, email
mail.send_mail(subject,
plain_text_message,
origin_email, [target_email, ],
html_message=html_message,
fail_silently=False)
def send_mail(self):
subject = self.email_subject
from_ = self.email_from
to = self.get_email_to()
template_name = self.get_email_template_name()
context = self.get_email_context_data()
body = render_to_string(template_name, context)
return mail.send_mail(subject, body, from_, to)
def form_valid(self, form):
response = super().form_valid(form)
self.send_mail()
return response
def _send_mail(subject, sender, to, template_name, context):
body = render_to_string(template_name, context)
mail.send_mail(subject, body, sender, to)
def send_email(self):
template = self.get_email_template_name()
subject = self.email_subject
from_ = self.email_from
to = self.get_email_to()
context = self.get_email_context_data()
body = render_to_string(template, context)
mail.send_mail(subject, body, from_, [from_, to])
def send_mail(self):
# Send subscription email
subject = self.email_subject
from_ = self.email_from
to = self.get_email_to()
template_name = self.get_email_template_name()
context = self.get_email_context_data()
body = render_to_string(template_name, context)
mail.send_mail(subject, body, from_, [from_, to])
def form_valid(self, form):
response = super().form_valid(form)
self.send_mail()
return response
def send(self, to=None, from_email=None, **context_args):
content = self.render(**context_args)
from_email = from_email or self.default_from_email
# does not need to check for remote connection permission because
# emails are diverted by default in a test environment.
return mail.send_mail(
subject=content.subject,
message=content.body,
from_email=from_email,
recipient_list=to
)
def password_reset_email(self, request):
htmlmsg = render_to_string('closedverse_main/help/email.html', {
'menulogo': request.build_absolute_uri(settings.STATIC_URL + 'img/menu-logo.png'),
'contact': request.build_absolute_uri(reverse('main:help-contact')),
'link': request.build_absolute_uri(reverse('main:forgot-passwd')) + "?token=" + base64.urlsafe_b64encode(bytes(self.password, 'utf-8')).decode(),
})
subj = 'Closedverse password reset for "{0}"'.format(self.username)
return send_mail(subject=subj, message="Bro, do you even HTML E-Mail?", html_message=htmlmsg, from_email="Closedverse not Openverse <{0}>".format(settings.DEFAULT_FROM_EMAIL), recipient_list=[self.email], fail_silently=False)
return EmailMessage(subj, htmlmsg, to=(self.email)).send()
