Python django.core.mail 模块，send_mail() 实例源码

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)

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()