shopdb-flask/shopdb/utils/mailer.py

"""Email sending service (stdlib smtplib/ssl/email only).

Reads SMTP configuration settings-first (via the cached settings map) with an
environment-variable fallback when any SMTP_* env var is present. When email is
disabled or the host is unset the sender is a graceful no-op that logs a warning
and returns False, so an unconfigured site never crashes on a send attempt.

Public helpers:
    send_email(to, subject, html, text=None) -> bool
    try_send(to, subject, html, text=None) -> (bool, error_or_None)
    send_alert(subject, html, text=None) -> bool
    render_email(title, body_html, intro=None) -> (html, text)
    render_table_email(title, columns, rows, intro=None) -> (html, text)

The SMTP password is never logged.
"""

import os
import re
import smtplib
import ssl
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.utils import formataddr

from flask import current_app, has_app_context

# Env var names that back each SMTP setting when settings are blank.
_ENV_MAP = {
    'smtp_host': 'SMTP_HOST',
    'smtp_port': 'SMTP_PORT',
    'smtp_username': 'SMTP_USERNAME',
    'smtp_password': 'SMTP_PASSWORD',
    'smtp_from_address': 'SMTP_FROM_ADDRESS',
    'smtp_from_name': 'SMTP_FROM_NAME',
    'alert_recipients': 'SMTP_ALERT_RECIPIENTS',
}

# Connect/send timeout in seconds. Keeps a wedged relay from hanging a request.
_SMTP_TIMEOUT = 10


def _log():
    """App logger when in an app context, else a module logger."""
    if has_app_context():
        return current_app.logger
    import logging
    return logging.getLogger('shopdb.mailer')


def _env_active():
    """True when the deployment supplies SMTP_* env overrides."""
    return any(k.startswith('SMTP_') for k in os.environ)


def get_smtp_config():
    """Resolve the SMTP config settings-first with env fallback.

    Returns a dict with typed fields. `enabled` is False when the site has not
    turned email on; callers should treat that as a no-op signal.
    """
    settings = {}
    if has_app_context():
        # Local import avoids a circular import at module load.
        from shopdb.core.api.settings import get_cached_settings
        try:
            settings = get_cached_settings() or {}
        except Exception:
            settings = {}

    env_active = _env_active()

    def pick(key, default=''):
        val = settings.get(key)
        if (val is None or val == '') and env_active:
            val = os.environ.get(_ENV_MAP.get(key, ''), default)
        return default if val is None else val

    enabled = bool(settings.get('smtp_enabled'))
    if not enabled and env_active:
        enabled = os.environ.get('SMTP_ENABLED', '').lower() in ('true', '1', 'yes')

    use_tls = settings.get('smtp_use_tls')
    if use_tls is None:
        if env_active:
            use_tls = os.environ.get('SMTP_USE_TLS', 'true').lower() in ('true', '1', 'yes')
        else:
            use_tls = True

    try:
        port = int(pick('smtp_port', 587) or 587)
    except (ValueError, TypeError):
        port = 587

    return {
        'enabled': enabled,
        'host': pick('smtp_host'),
        'port': port,
        'username': pick('smtp_username'),
        'password': pick('smtp_password'),
        'use_tls': bool(use_tls),
        'from_address': pick('smtp_from_address'),
        'from_name': pick('smtp_from_name') or 'ShopDB',
        'alert_recipients': pick('alert_recipients'),
    }


def _normalize_recipients(to):
    """Coerce a recipient spec (string, comma/semicolon list, or iterable) to a
    clean list of addresses."""
    if not to:
        return []
    if isinstance(to, str):
        parts = re.split(r'[,;]', to)
    else:
        parts = list(to)
    return [p.strip() for p in parts if p and p.strip()]


def try_send(to, subject, html, text=None):
    """Send an email. Returns (ok, error).

    ok is False with error=None when email is not configured (a graceful
    no-op). ok is False with an error string when a real send failed. The SMTP
    password is never included in the error.
    """
    config = get_smtp_config()
    recipients = _normalize_recipients(to)

    if not config['enabled'] or not config['host']:
        _log().warning('Email not sent: SMTP is disabled or host is unset.')
        return False, None
    if not recipients:
        _log().warning('Email not sent: no recipients.')
        return False, 'No recipients specified'
    if not config['from_address']:
        _log().warning('Email not sent: from address is unset.')
        return False, 'From address is not configured'

    message = MIMEMultipart('alternative')
    message['Subject'] = subject
    message['From'] = formataddr((config['from_name'], config['from_address']))
    message['To'] = ', '.join(recipients)
    # Plaintext first so alternative-aware clients prefer the HTML part.
    message.attach(MIMEText(text or _html_to_text(html), 'plain', 'utf-8'))
    message.attach(MIMEText(html, 'html', 'utf-8'))

    try:
        context = ssl.create_default_context()
        if config['port'] == 465:
            server = smtplib.SMTP_SSL(
                config['host'], config['port'],
                timeout=_SMTP_TIMEOUT, context=context)
        else:
            server = smtplib.SMTP(
                config['host'], config['port'], timeout=_SMTP_TIMEOUT)
        with server:
            if config['port'] != 465 and config['use_tls']:
                server.starttls(context=context)
            if config['username']:
                server.login(config['username'], config['password'])
            server.sendmail(config['from_address'], recipients, message.as_string())
        _log().info('Email sent to %d recipient(s): %s', len(recipients), subject)
        return True, None
    except Exception as exception:
        # Never let the password reach the log or the caller.
        error = _scrub(str(exception), config['password'])
        _log().error('Email send failed: %s', error)
        return False, error


def send_email(to, subject, html, text=None):
    """Send an email. Returns True on success, False otherwise (no-op safe)."""
    ok, _error = try_send(to, subject, html, text=text)
    return ok


def send_alert(subject, html, text=None):
    """Send an alert to the site's configured alert_recipients. Returns False
    when email is off or no alert recipients are configured."""
    config = get_smtp_config()
    recipients = _normalize_recipients(config['alert_recipients'])
    if not recipients:
        _log().warning('Alert not sent: no alert_recipients configured.')
        return False
    return send_email(recipients, subject, html, text=text)


def _scrub(value, secret):
    """Remove a secret substring from a string (defensive log hygiene)."""
    if secret and secret in value:
        return value.replace(secret, '***')
    return value


def _html_to_text(html):
    """Very small HTML-to-text fallback for the plaintext alternative."""
    text = re.sub(r'(?i)<br\s*/?>', '\n', html)
    text = re.sub(r'(?i)</(p|tr|div|h[1-6]|li)>', '\n', text)
    text = re.sub(r'<[^>]+>', '', text)
    text = re.sub(r'\n{3,}', '\n\n', text)
    return text.strip()


def _escape(value):
    """HTML-escape a cell value."""
    return (str('' if value is None else value)
            .replace('&', '&amp;').replace('<', '&lt;').replace('>', '&gt;'))


def render_email(title, body_html, intro=None):
    """Wrap body HTML in a simple branded shell. Returns (html, text)."""
    intro_html = f'<p style="margin:0 0 16px;color:#444;">{_escape(intro)}</p>' if intro else ''
    html = (
        '<div style="font-family:Arial,Helvetica,sans-serif;max-width:640px;'
        'margin:0 auto;color:#222;">'
        f'<h2 style="color:#1a1a1a;margin:0 0 12px;">{_escape(title)}</h2>'
        f'{intro_html}{body_html}'
        '<hr style="border:none;border-top:1px solid #ddd;margin:24px 0 12px;">'
        '<p style="font-size:12px;color:#888;margin:0;">Sent by ShopDB.</p>'
        '</div>'
    )
    return html, _html_to_text(html)


def render_table_email(title, columns, rows, intro=None):
    """Render tabular report data as an HTML table email. Returns (html, text).

    columns: list of {'key','label'} dicts or of plain strings.
    rows: list of dicts keyed by the column keys.
    """
    normalized = []
    for column in columns or []:
        if isinstance(column, dict):
            normalized.append((column.get('key'), column.get('label', column.get('key'))))
        else:
            normalized.append((column, column))

    header_cells = ''.join(
        f'<th style="text-align:left;padding:8px 10px;border-bottom:2px solid #ccc;'
        f'background:#f4f4f4;">{_escape(label)}</th>'
        for _key, label in normalized)

    body_rows = []
    for row in rows or []:
        cells = ''.join(
            f'<td style="padding:8px 10px;border-bottom:1px solid #eee;">'
            f'{_escape(row.get(key) if isinstance(row, dict) else row)}</td>'
            for key, _label in normalized)
        body_rows.append(f'<tr>{cells}</tr>')

    table = (
        '<table style="border-collapse:collapse;width:100%;font-size:14px;">'
        f'<thead><tr>{header_cells}</tr></thead>'
        f'<tbody>{"".join(body_rows) or "<tr><td>No data</td></tr>"}</tbody>'
        '</table>'
    )
    count_line = f'<p style="color:#666;font-size:13px;">{len(rows or [])} row(s).</p>'
    return render_email(title, table + count_line, intro=intro)