__init__.py

import smtplib
import threading
import logging
from typing import Sequence, Union
from sqlalchemy import select
from flask import current_app
from compendium_v2.db import db
from compendium_v2.db.auth_model import User, ROLES

logger = logging.getLogger(__name__)


USER_NOTIFICATION_TEMPLATE = """Hello{name},

Thank you for registering for the new GÉANT Compendium Survey site.

The Compendium team have been notified and will assign you to your NREN shortly. If you are a long-time contributor this should be fairly quick (same working day). For new users, we will have to check with your colleagues first to make sure you are authorised to answer on behalf of your NREN.

Best regards,

Daniel and Jennifer (the Compendium admins)
"""  # noqa: E501


def _send_mail(smtp_server, port, sender_email, recipients, message):

    try:
        with smtplib.SMTP(smtp_server, port) as server:
            server.sendmail(from_addr=sender_email, to_addrs=recipients, msg=message)
        logger.debug('Successfully sent email')
    except Exception:
        logger.exception('Unable to send email:')


def send_mail(
    contents: str,
    subject: str,
    recipients: Union[str, Sequence[str]] = ''
):
    if not current_app.config['MAIL_ENABLE']:
        logger.warning('No mail configuration, cannot send email.')
        return

    if not contents or not isinstance(contents, str):
        raise ValueError('Contents must be a non-empty string.')

    excluded_admins = set(email.lower() for email in current_app.config['MAIL_EXCLUDED_ADMINS'])

    admins = db.session.scalars(select(User).where(User.roles == ROLES.admin)).unique().all()

    admin_emails = [admin.email for admin in admins if admin.email.lower() not in excluded_admins]

    if not recipients:
        recipients = admin_emails

    subject = subject.replace('\n', ' ')
    message = f"""Subject: {subject}\n\n{contents}"""

    smtp_server = current_app.config['MAIL_SERVER']
    port = current_app.config['MAIL_PORT']
    sender_email = current_app.config['MAIL_SENDER_EMAIL']

    # spin off a thread since this can take some time..
    logger.debug('Sending email')
    thread = threading.Thread(target=_send_mail, args=(smtp_server, port, sender_email, recipients, message))
    thread.start()


def send_admin_signup_notification(user: User):
    fullname = user.fullname
    email = user.email
    oidc_sub = user.oidc_sub

    contents = f"""{fullname} has just signed up with the email {email} and provider ID {oidc_sub}"""
    send_mail(contents=contents, subject='New user signed up for Compendium')


def send_user_signup_notification(user: User):
    fullname = user.fullname
    email = user.email

    first_name = ' ' + fullname.split()[0] if fullname else ''
    contents = USER_NOTIFICATION_TEMPLATE.format(name=first_name)
    send_mail(contents=contents, subject='You have signed up for the Compendium Survey', recipients=email)