OSIT-AE-API-FastAPI/app/lib_email.py

import html2text
import smtplib, ssl
import logging
from email.message import EmailMessage
from email.headerregistry import Address
from typing import Optional

from app.log import logger_reset
from app.config import settings

log = logging.getLogger(__name__)

# ### BEGIN ### API Lib Email ### send_email() ###
# Moved from lib_general.py 2026-01-07
@logger_reset
def send_email(
        from_email: str,
        to_email: str,
        subject: str,
        body_html: str,

        from_name: str = '',
        reply_to_email: str = '',
        reply_to_name: str = '',
        to_name: str = '',
        cc_email: str = '',
        cc_name: str = '',
        bcc_email: str = '',
        bcc_name: str = '',
        body_text: str = '',

        test: bool = False,
        log_lvl: int = logging.WARNING, # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
        ):
    log.setLevel(log_lvl)
    log.debug(locals())

    if test:
        log.setLevel(logging.DEBUG)
        log.debug('[TESTING] Running with send_email() in TEST mode')

    message = EmailMessage()
    if subject:
        message['Subject'] = subject
    else:
        return False

    if from_email and from_name:
        try:
            message['From'] = Address(display_name=from_name, addr_spec=from_email)
        except Exception as e:
            log.exception('**** *** ** * ### BEGIN ### Exception Happened: Returning False * ** *** ****')
            return False
    elif from_email:
        try:
            message['From'] = Address(display_name=from_email, addr_spec=from_email)
        except Exception as e:
            log.exception('**** *** ** * ### BEGIN ### Exception Happened: Returning False * ** *** ****')
            return False
    else:
        return False

    if reply_to_email and reply_to_name:
        try:
            message['Reply-To'] = Address(display_name=reply_to_name, addr_spec=reply_to_email)
        except Exception as e:
            log.exception('**** *** ** * ### BEGIN ### Exception Happened: Returning False * ** *** ****')
            return False
    elif reply_to_email:
        try:
            message['Reply-To'] = Address(display_name=reply_to_email, addr_spec=reply_to_email)
        except Exception as e:
            log.exception('**** *** ** * ### BEGIN ### Exception Happened: Returning False * ** *** ****')
            return False

    if to_email and to_name:
        try:
            message['To'] = Address(display_name=to_name, addr_spec=to_email)
        except Exception as e:
            log.exception('**** *** ** * ### BEGIN ### Exception Happened: Returning False * ** *** ****')
            return False
    elif to_email:
        try:
            message['To'] = Address(display_name=to_email, addr_spec=to_email)
        except Exception as e:
            log.exception('**** *** ** * ### BEGIN ### Exception Happened: Returning False * ** *** ****')
            return False
    else:
        return False

    if cc_email and cc_name:
        try:
            message['Cc'] = Address(display_name=cc_name, addr_spec=cc_email)
        except Exception as e:
            log.exception('**** *** ** * ### BEGIN ### Exception Happened: Returning False * ** *** ****')
            return False
    elif cc_email:
        try:
            message['Cc'] = Address(display_name=cc_email, addr_spec=cc_email)
        except Exception as e:
            log.exception('**** *** ** * ### BEGIN ### Exception Happened: Returning False * ** *** ****')
            return False

    if bcc_email and bcc_name:
        try:
            message['Bcc'] = Address(display_name=bcc_name, addr_spec=bcc_email)
        except Exception as e:
            log.exception('**** *** ** * ### BEGIN ### Exception Happened: Returning False * ** *** ****')
            return False
    elif bcc_email:
        try:
            message['Bcc'] = Address(display_name=bcc_email, addr_spec=bcc_email)
        except Exception as e:
            log.exception('**** *** ** * ### BEGIN ### Exception Happened: Returning False * ** *** ****')
            return False

    html_version = """
<html>
<body>

"""+body_html+"""

</body>
</html>
    """

    if body_text:
        text_version = body_text
    else:
        text_version = html2text.html2text(html_version)

    message.set_content(text_version)
    message.add_alternative(html_version, subtype='html')

    log.info('Sending email...')
    log.debug(settings.SMTP)

    log.info(f'Subject: {subject}')
    log.info(f'From: {from_email} Reply To: {reply_to_email} To: {to_email} CC: {cc_email} BCC: {bcc_email}')

    log.debug('Message:')
    log.debug(message.as_string())

    log.info('Creating SMTP SSL connection...')
    context = ssl.create_default_context()

    log.info('SMTP configuration, connect, and send')
    log.info(f'Server: {settings.SMTP["server"]} Port: {settings.SMTP["port"]} Username: {settings.SMTP["username"]}')

    log.info('Trying smtplib.SMTP_SSL in send_email()...')
    if test:
        log.info('[TESTING] Email will NOT actually be sent! [TEST MODE]')
    try:
        with smtplib.SMTP_SSL(settings.SMTP['server'], settings.SMTP['port'], context=context) as server:
            log.info('SMTP log in...')
            log.debug(f'Server: {settings.SMTP["server"]} Port: {settings.SMTP["port"]} Username: {settings.SMTP["username"]} Password: {settings.SMTP["password"]}')
            server.login(settings.SMTP['username'], settings.SMTP['password'])
            log.info('SMTP send message...')
            if not test:
                log.info('Email sent! Returning True')
                server.send_message(message)
            else:
                log.info('[TESTING] Email (NOT) sent! Returning True [TEST MODE]')
        return True
    except:
        log.error('Error: Unable to send email. Returning False')
        return False
# ### END ### API Lib Email ### send_email() ###