"""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)', '\n', html) text = re.sub(r'(?i)', '\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'

{_escape(intro)}

' if intro else '' html = ( '
' f'

{_escape(title)}

' f'{intro_html}{body_html}' '
' '

Sent by ShopDB.

' '
' ) 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'{_escape(label)}' for _key, label in normalized) body_rows = [] for row in rows or []: cells = ''.join( f'' f'{_escape(row.get(key) if isinstance(row, dict) else row)}' for key, _label in normalized) body_rows.append(f'{cells}') table = ( '' f'{header_cells}' f'{"".join(body_rows) or ""}' '
No data
' ) count_line = f'

{len(rows or [])} row(s).

' return render_email(title, table + count_line, intro=intro)