python类Message()的实例源码

def send_email(subject, recipients, template, **kwargs):
    if not isinstance(recipients, list):
        recipients = list(recipients)
    msg = Message(subject, reply_to=current_app.config['MAIL_DEFAULT_SENDER'],
                  recipients=recipients)
    msg.body = render_template(template + '.txt', **kwargs)
    msg.html = render_template(template + '.html', **kwargs)

    attachments = kwargs.get('attachments', [])
    mimes = MimeTypes()
    for file in attachments:
        path_ = os.path.join(current_app.config['APP_UPLOADS'], file)
        with current_app.open_resource(path_) as fp:
            mime = mimes.guess_type(fp.name)
            msg.attach(path_, mime[0], fp.read())

    app = current_app._get_current_object()
    t = Thread(target=send_async_email, args=[app, msg])
    t.start()
    return t

def send_email(to_email, subject, template_path, template_vars, attachment_path=None):

    if not ENV_DICT['SEND_EMAIL']:
        print '++ sending emails is currently disabled in this environment. Enable SEND_EMAIL to allow email sending'
        return

    # email address that emails will be sent from
    from_email = ENV_DICT['MAIL_DEFAULT_SENDER']

    # render HTML from template
    page_html = render_template(template_path, **template_vars)

    msg = Message(subject=subject,
                  sender=from_email,
                  recipients=[to_email],
                  html=page_html)

    if attachment_path:
        with open(attachment_path, 'r') as f:
            msg.attach("osf-results.json", "text/plain", f.read())

    mail.send(msg)

def send_error_mail(exception):
        """Sends an error mail to the admin containing the traceback and configuration.
        After that, a custom HTTP 500 page is shown.

        :param exception: the exception raised

        :type exception: ``Exception``

        :return:
          the HTML to send back in the response, and the HTTP code 500.

        :rtype: str, int
        """
        # Inspired from <https://github.com/jasonwyatt/Flask-ErrorMail>.
        message = Message("Join2 ORCID exception: %s" % exception, sender=CFG_SITE_ADMIN_EMAIL,
                          recipients=[CFG_SITE_ADMIN_EMAIL])
        message_contents = ["Traceback:", "=" * 80, traceback.format_exc(), "\n", "Request Information:", "=" * 80]
        environ = request.environ
        for key in sorted(environ.keys()):
            message_contents.append("%s: %s" % (key, environ.get(key)))
        message.body = "\n".join(message_contents) + "\n"
        mailer.send(message)
        return render_template("500.html"), 500

def send_email(device_data, enrolled=True):
    subject = 'Mobile Device Enrolled' if enrolled else 'Mobile Device Un-Enrolled'
    email_data = {
        'enrolled': enrolled,
        'name': device_data['deviceName'],
        'model': device_data['model'],
        'serial': device_data['serialNumber'],
        'url': os.path.join('{}/mobileDevices.html?id={}'.format(JSS_ADDRESS, device_data['jssID'])),
        'time': datetime.datetime.utcnow()
    }
    txt, html = build_email_body(email_data)
    msg = Message(
        subject,
        recipients=[DESTINATION_EMAIL]
    )
    msg.body = txt
    msg.html = html
    thr = Thread(target=send_async_email, args=[msg])
    thr.start()

def sendEmail(httpOrigin, email, activationId, mailer):
  '''
  Create a message and send it from our email to
  the passed in email. The message should contain
  a link built with the activationId
  '''
  if activationId is None:
    return
  #Change the URL to the appropriate environment
  host = getProperEnvironment(httpOrigin)
  url = host + '/activation/{0}'.format(activationId)
  message = render_template('registrationEmail.html', url=url)
  #Build Message
  msg = Message('Mentii: Thank You for Creating an Account!', recipients=[email],
      extra_headers={'Content-Transfer-Encoding': 'quoted-printable'}, html=message)
  #Send Email
  mailer.send(msg)

def sendClassRemovalEmail(dynamoDBInstance, mailer, jsonData):
  '''
  Create a message to send it from our email to
  the passed in email. The message should notify the user they were removed from a class
  '''
  email = jsonData.get('email')
  classCode = jsonData.get('classCode')
  cl = getClassByCode(classCode, dynamoDBInstance)
  classTitle = cl['title']
  message = render_template('removedEmail.html', classTitle=classTitle)

  #Build Message
  msg = Message('You have been removed from a class', recipients=[email],
        extra_headers={'Content-Transfer-Encoding': 'quoted-printable'}, html=message)

  mailer.send(msg)

def send_confirm_email(self):
        """
        Send an confirm email to user. contains a link to confirm the email
        confirm link is not provide by this app, a client must implement this endpoint to complete the confirmation.
        """
        from server import app, mail
        token = self.generate_confirm_email_token()
        confirm_url = '{0}://{1}/email-confirm?token={2}'.format(app.config['SITE_PROTOCOL'],
                                                                 app.config['SITE_HOST'],
                                                                 token)
        subject = '[{0}] Email Address Confirmation'.format(app.config['SITE_NAME'])
        email_content = render_template('email-confirm.html', info={
            'confirm_title': subject,
            'confirm_url': confirm_url,
            'site_name': app.config['SITE_NAME'],
            'user_name': self.name
        })
        msg = Message(subject, recipients=[self.email], html=email_content)
        try:
            mail.send(msg)
        except SMTPAuthenticationError:
            raise ServerError('SMTP authentication failed', 500)

def sendUpsellEmail(data):
        name = Utils.getUserName(data['user'])
        with webapp.app_context():
            consumer_mail = render_template('mailers/extend_order.html',
                            name = name,
                            book_name = data['book_name'],
                            order_id = data['order_id'],
                            items = data['items'], 
                            curated_items = data['curated_items'],
                            quote = data['quote'], 
                            quote_author = data['quote_author'])
            pre = Premailer(consumer_mail, remove_classes=False, strip_important=False)
            consumer_mail =  pre.transform()
            email = Message('Enjoying the book?',
                            recipients=[data['user'].email])
            email.html = consumer_mail
            mail.send(email)
        return True

def request_password_reset(username):
    """
    Emails a user a link to reset their password.

    Checks that a username was submitted to the function and grabs the relevant team info from the db.
    Generates a secure token and inserts it into the team's document as 'password_reset_token'.
    A link is emailed to the registered email address with the random token in the url.  The user can go to this
    link to submit a new password, if the token submitted with the new password matches the db token the password
    is hashed and updated in the db.

    Args:
        username: the username of the account
    """
    validate(password_reset_request_schema, {"username":username})
    user = safe_fail(api.user.get_user, name=username)
    if user is None:
        raise WebException("No registration found for '{}'.".format(username))

    token_value = api.token.set_token({"uid": user['uid']}, "password_reset")

    body = """We recently received a request to reset the password for the following {0} account:\n\n\t{2}\n\nOur records show that this is the email address used to register the above account.  If you did not request to reset the password for the above account then you need not take any further steps.  If you did request the password reset please follow the link below to set your new password. \n\n {1}/reset#{3} \n\n Best of luck! \n The {0} Team""".format(api.config.competition_name, api.config.competition_urls[0], username, token_value)

    subject = "{} Password Reset".format(api.config.competition_name)

    message = Message(body=body, recipients=[user['email']], subject=subject)
    mail.send(message)

def send_email_invite(gid, email, teacher=False):
    """
    Sends an email registration link that will automatically join into a group. This link will bypass the email filter.
    """

    group = api.group.get_group(gid=gid)

    token_value = api.token.set_token({"gid": group["gid"], "email": email, "teacher": teacher}, "registration_token")

    registration_link = "{}/#g={}&r={}".\
        format(api.config.competition_urls[0], group["gid"], token_value)

    body = """
You have been invited by the staff of the {1} organization to compete in {0}.
You will need to follow the registration link below to finish the account creation process.

If you believe this to be a mistake you can safely ignore this email.

Registration link: {2}

Good luck!
  The {0} Team.
    """.format(api.config.competition_name, group["name"], registration_link)

    subject = "{} Registration".format(api.config.competition_name)

    message = Message(body=body, recipients=[email], subject=subject)
    mail.send(message)

def send_email(organisation, recipients):
    sender = config.ADMINS[0]
    subject = 'Results have been generated for ' + organisation
    text_body = """ Greetings from Zense,

                    We have completed your task in regards to %s
                    Please visit us back

                    Regards,
                    Team Stalker, Zense """ % str(organisation)
    msg = Message(subject, sender = sender, recipients = recipients)
    msg.body = text_body
    with app.app_context():
        mail.send(msg)

def internal_server_error(e):
    message = repr(e)
    trace = traceback.format_exc()
    trace = string.split(trace, '\n')
    timestamp = (datetime.fromtimestamp(time.time())
                 .strftime('%Y-%m-%d %H:%M:%S'))
    if current_user.is_authenticated:
        user = current_user.username
    else:
        user = 'anonymous'
    gathered_data = ('message: {}\n\n\n'
                     'timestamp: {}\n'
                     'ip: {}\n'
                     'method: {}\n'
                     'request.scheme: {}\n'
                     'request.full_path: {}\n'
                     'user: {}\n\n\n'
                     'trace: {}'.format(message, timestamp,
                                        request.remote_addr, request.method,
                                        request.scheme, request.full_path,
                                        user, '\n'.join(trace)))
    # send email to admin
    if app.config['TESTING']:
        print(gathered_data)
    else:
        mail_message = gathered_data
        msg = Message('Error: ' + message[:40],
                      body=mail_message,
                      recipients=[app.config['ADMIN_MAIL']])
        mail.send(msg)
        flash(_('A message has been sent to the administrator'), 'info')
    bookcloud_before_request()
    return render_template('500.html', message=gathered_data), 500

def request_password_reset(username):
    """
    Emails a user a link to reset their password.

    Checks that a username was submitted to the function and grabs the relevant team info from the db.
    Generates a secure token and inserts it into the team's document as 'password_reset_token'.
    A link is emailed to the registered email address with the random token in the url.  The user can go to this
    link to submit a new password, if the token submitted with the new password matches the db token the password
    is hashed and updated in the db.

    Args:
        username: the username of the account
    """
    validate(password_reset_request_schema, {"username":username})
    user = safe_fail(api.user.get_user, name=username)
    if user is None:
        raise WebException("No registration found for '{}'.".format(username))

    token_value = api.token.set_token({"uid": user['uid']}, "password_reset")

    body = """We recently received a request to reset the password for the following {0} account:\n\n\t{2}\n\nOur records show that this is the email address used to register the above account.  If you did not request to reset the password for the above account then you need not take any further steps.  If you did request the password reset please follow the link below to set your new password. \n\n {1}/reset#{3} \n\n Best of luck! \n The {0} Team""".format(api.config.competition_name, api.config.competition_urls[0], username, token_value)

    subject = "{} Password Reset".format(api.config.competition_name)

    message = Message(body=body, recipients=[user['email']], subject=subject)
    mail.send(message)

def send_email_invite(gid, email, teacher=False):
    """
    Sends an email registration link that will automatically join into a group. This link will bypass the email filter.
    """

    group = api.group.get_group(gid=gid)

    token_value = api.token.set_token({"gid": group["gid"], "email": email, "teacher": teacher}, "registration_token")

    registration_link = "{}/#g={}&r={}".\
        format(api.config.competition_urls[0], group["gid"], token_value)

    body = """
You have been invited by the staff of the {1} organization to compete in {0}.
You will need to follow the registration link below to finish the account creation process.

If you believe this to be a mistake you can safely ignore this email.

Registration link: {2}

Good luck!
  The {0} Team.
    """.format(api.config.competition_name, group["name"], registration_link)

    subject = "{} Registration".format(api.config.competition_name)

    message = Message(body=body, recipients=[email], subject=subject)
    mail.send(message)

def send_email(to, subject, template, **kwargs):
    app = current_app._get_current_object()
    msg = Message(app.config['CIRCULATE_MAIL_SUBJECT_PREFIX'] + ' ' + subject,
                  sender=app.config['CIRCULATE_MAIL_SENDER'], recipients=[to])
    msg.body = render_template(template + '.txt', **kwargs)
    msg.html = render_template(template + '.html', **kwargs)
    thr = Thread(target=send_async_email, args=[app, msg])
    thr.start()
    return thr

def send_email(sender, recipients, subject, text_body, html_body=None,
               attach=None):
    msg = Message(subject, reply_to=sender, recipients=recipients)
    msg.body = text_body
    msg.html = html_body
    if attach:
        mimes = MimeTypes()
        for file in attach:
            path_ = os.path.join(current_app.config['APP_UPLOADS'], file)
            with current_app.open_resource(path_) as fp:
                mime = mimes.guess_type(fp.name)
                msg.attach(file, mime[0], fp.read())
    mail.send(msg)

def _do_on_login(sender, user, **extra):
    """Notify the user of a new login.

    This is only done if the ``notify_login`` attribute is set to ``True``.
    """
    if user.notify_login:
        notification = Message(
            'akamatsu - New session started',
            recipients=[user.email],
            body=render_template('mail/login.txt', user=user),
            html=render_template('mail/login.html', user=user)
        )

        mail.send(notification)

def send_email(to, subject, template, **kwargs):
    """Send email using multible threads.
    """
    app = current_app._get_current_object()
    msg = Message(app.config['MAIL_SUBJECT_PREFIX'] + ' ' + subject,
                  sender=app.config['MAIL_SENDER'], recipients=[to])
    msg.body = render_template(template + '.txt', **kwargs)
    msg.html = render_template(template + '.html', **kwargs)
    thread = Thread(target=send_async_email, args=[app, msg])
    thread.start()
    return thread

def send_email(subject, recipients, text_body, html_body, sender=None):
    msg = Message(subject, recipients=recipients, sender=sender)
    msg.body = text_body
    msg.html = html_body
    mail.send(msg)

def send_email(subject, recipients, text_body, html_body, sender=None):
    msg = Message(subject, recipients=recipients, sender=sender)
    msg.body = text_body
    msg.html = html_body
    mail.send(msg)

def send_email(subject, recipients, text_body, html_body, sender=None):
    msg = Message(subject, recipients=recipients, sender=sender)
    msg.body = text_body
    msg.html = html_body
    mail.send(msg)

def send_mails(recipients, cc, mail_title, mail_body):
    msg = Message(mail_title)
    msg.body = mail_body

    msg.sender = current_app._get_current_object().config['MAIL_USERNAME']
    msg.recipients = recipients
    msg.cc = cc

    mail.send(msg)

def send_email_thread(to, subject, template, **kwargs):
    """Send async email using threading.
    """
    app = current_app._get_current_object()
    msg = Message(app.config['APP_MAIL_SUBJECT_PREFIX'] + ' ' + subject,
                  sender=app.config['APP_MAIL_SENDER'], recipients=[to])
    msg.body = render_template(template + '.txt', **kwargs)
    msg.html = render_template(template + '.html', **kwargs)
    thr = Thread(target=send_async_email, args=[app, msg])
    thr.start()
    return thr

def send_email_celery(to, subject, template, countdown=None, **kwargs):
    """Send async email using Celery.
    """
    app = current_app._get_current_object()
    msg = Message(app.config['APP_MAIL_SUBJECT_PREFIX'] + ' ' + subject,
                  sender=app.config['APP_MAIL_SENDER'], recipients=[to])
    msg.body = render_template(template + '.txt', **kwargs)
    msg.html = render_template(template + '.html', **kwargs)
    send_celery_async_email.apply_async(args=[msg], countdown=countdown)

def send_reset_password_email(user: models.User) -> None:
    token = user.get_reset_token()
    html_body = current_app.config['EMAIL_TEMPLATE'].replace(
        '\n\n', '<br><br>'
    ).format(
        site_url=current_app.config["EXTERNAL_URL"],
        url=f'{psef.app.config["EXTERNAL_URL"]}/reset_'
        f'password/?user={user.id}&token={token}',
        user_id=user.id,
        token=token,
        user_name=html.escape(user.name),
        user_email=html.escape(user.email),
    )
    text_maker = html2text.HTML2Text(bodywidth=78)
    text_maker.inline_links = False
    text_maker.wrap_links = False

    message = Message(
        subject=f'Reset password on {psef.app.config["EXTERNAL_URL"]}',
        body=text_maker.handle(html_body),
        html=html_body,
        recipients=[user.email],
    )
    try:
        mail.send(message)
    except Exception:
        raise APIException(
            'Something went wrong sending the email, '
            'please contact your site admin',
            f'Sending email to {user.id} went wrong.',
            APICodes.UNKOWN_ERROR,
            500,
        )

def send_notif_message(self, notif):
        msg = Message('Update on ' + notif.source_type + ' ' + str(notif.source_id), recipients=[notif.user.email])
        msg.body = notif.to_text() + '\n\nsee it here: ' + notif.link
        options = {
            'notif': notif
        }
        msg.html = render_template('email/notif.html', **options)
        self.messages.append(msg)

def send_invite_message(self, email, sender):
        msg = Message('Invitation to Tingsmen', recipients=[email])
        msg.body = sender + ' invited you to join Tingsmen\n\nTingsmen is a social network where you can propose any ' \
                            'change to the platform and, supposing the community is ok with it, it will be developed' \
                            '\n\nWanna know more? visit: https://tingsmen.herokuapp.com'
        options = {
            'sender': sender
        }
        msg.html = render_template('email/invite.html', **options)
        self.messages.append(msg)

def send_email(to, subject, template, **kwargs):
    app = current_app._get_current_object()
    msg = Message(app.config['FLASKY_MAIL_SUBJECT_PREFIX'] + ' ' + subject,
                  sender=app.config['FLASKY_MAIL_SENDER'], recipients=[to])
    msg.body = render_template(template + '.txt', **kwargs)
    msg.html = render_template(template + '.html', **kwargs)
    thr = Thread(target=send_async_email, args=[app, msg])
    thr.start()
    return thr

def _send_email(app, email_addr):
    with app.app_context():
        msg = Message("[FreeTest] ??????",
                      recipients=[email_addr])
        msg.body = "This is body333."
        mail.send(msg)

def _broadcast_email(app, email_addrs):
    with app.app_context():
        for email_addr in email_addrs:
            msg = Message("[FreeTest] ??????",
                          recipients=[email_addr])
            msg.body = "This is body333."
            mail.send(msg)