fb188fd302
Support teams gain a webhookurl (migration 7d29 + API + settings-page field), so a team is a notification target. send_webhook(url=) lets a caller override the site default with a team's webhook. Printedparts gains a 'alert support team' setting (printedparts_alert_supportteamid) + selector on its settings page; low-stock alerts post to that team's webhook, falling back to the site alert_webhook_url. Email leg unchanged. Same pattern extends to other alerting plugins (printers low-toner next).
340 lines
12 KiB
Python
340 lines
12 KiB
Python
"""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
|
|
|
|
import requests
|
|
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 get_webhook_config():
|
|
"""Alert-webhook config from settings: the URL and the payload format."""
|
|
settings = {}
|
|
if has_app_context():
|
|
from shopdb.core.api.settings import get_cached_settings
|
|
try:
|
|
settings = get_cached_settings() or {}
|
|
except Exception:
|
|
settings = {}
|
|
return {
|
|
'url': (settings.get('alert_webhook_url') or '').strip(),
|
|
# 'teams' = classic Incoming Webhook (MessageCard); 'adaptivecard' =
|
|
# Teams Workflow (Power Automate); 'json' = generic {title,text}.
|
|
'format': (settings.get('alert_webhook_format') or 'teams').strip().lower(),
|
|
}
|
|
|
|
|
|
def _webhook_payload(fmt, title, text):
|
|
if fmt == 'adaptivecard':
|
|
return {
|
|
'type': 'message',
|
|
'attachments': [{
|
|
'contentType': 'application/vnd.microsoft.card.adaptive',
|
|
'content': {
|
|
'type': 'AdaptiveCard',
|
|
'$schema': 'http://adaptivecards.io/schemas/adaptive-card.json',
|
|
'version': '1.4',
|
|
'body': [
|
|
{'type': 'TextBlock', 'weight': 'Bolder',
|
|
'size': 'Medium', 'text': title, 'wrap': True},
|
|
{'type': 'TextBlock', 'text': text, 'wrap': True},
|
|
],
|
|
},
|
|
}],
|
|
}
|
|
if fmt == 'json':
|
|
return {'title': title, 'text': text}
|
|
# default 'teams' = classic Incoming Webhook connector MessageCard
|
|
return {
|
|
'@type': 'MessageCard',
|
|
'@context': 'https://schema.org/extensions',
|
|
'summary': title,
|
|
'themeColor': 'D93F3F',
|
|
'title': title,
|
|
'text': text,
|
|
}
|
|
|
|
|
|
def send_webhook(title, text, url=None):
|
|
"""POST an alert to a webhook (Teams, etc.). `url` overrides the site
|
|
alert_webhook_url (e.g. a support team's own webhook); the payload format
|
|
still follows alert_webhook_format. Best-effort: returns (ok, error), a
|
|
no-op ((False, None)) when no URL resolves, and never raises."""
|
|
config = get_webhook_config()
|
|
target = (url or '').strip() or config['url']
|
|
if not target:
|
|
return False, None
|
|
try:
|
|
response = requests.post(
|
|
target, json=_webhook_payload(config['format'], title, text),
|
|
timeout=10)
|
|
if response.status_code >= 400:
|
|
return False, f'HTTP {response.status_code}'
|
|
_log().info('Alert webhook posted: %s', title)
|
|
return True, None
|
|
except Exception as exception:
|
|
error = str(exception)
|
|
_log().warning('Alert webhook failed: %s', error)
|
|
return False, error
|
|
|
|
|
|
def send_alert(subject, html, text=None):
|
|
"""Fan an alert out to the configured channels: the site's alert_recipients
|
|
(email) and the alert webhook (Teams, etc.). Each channel is independent and
|
|
best-effort; returns True if the EMAIL leg sent."""
|
|
body = text or _html_to_text(html)
|
|
# Webhook fans out alongside email, independent of SMTP being configured.
|
|
try:
|
|
send_webhook(subject, body)
|
|
except Exception:
|
|
pass
|
|
config = get_smtp_config()
|
|
recipients = _normalize_recipients(config['alert_recipients'])
|
|
if not recipients:
|
|
_log().warning('Alert email 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('&', '&').replace('<', '<').replace('>', '>'))
|
|
|
|
|
|
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)
|