refactor(maintenance): one SMTP settings reader and sender for the nightly digest and radio alerts

send_nightly_email, send_zombie_alert_email and send_radio_offline_alert_email each read the same notif.* SMTP settings, parsed the port the same way, and repeated the same SSL/STARTTLS/plain transport. maintenance.read_smtp_settings returns an SmtpSettings record and send_smtp_message delivers a message with it. Each sender keeps its own enable checks, recipients, validation messages and body.

Full suite: all pass except tests/unit/test_wx_location_prefix_budget.py::test_a_prefix_is_actually_in_play, which makes a live Nominatim call and fails the same way on dev at the moment.

modules/maintenance.py

@@ -8,6 +8,7 @@ import json
 import os
 import sqlite3
 import time
+from dataclasses import dataclass
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Callable
 
@@ -109,6 +110,67 @@ def _count_log_errors_last_24h(log_path: Path) -> tuple[int | str, int | str]:
         return 'n/a', 'n/a'
 
 
+
+@dataclass(frozen=True)
+class SmtpSettings:
+    """The [notif.*] SMTP settings shared by the nightly digest and the radio alerts."""
+
+    host: str
+    security: str
+    user: str
+    password: str
+    from_name: str
+    from_email: str
+    port: int
+    allow_local: bool
+
+    @property
+    def from_header(self) -> str:
+        return f'{self.from_name} <{self.from_email}>'
+
+
+def read_smtp_settings(get_notif: Callable[[str], str]) -> SmtpSettings:
+    """Read the SMTP settings; port defaults to 465 for ssl, else 587 (587 when unparseable)."""
+    security = get_notif('smtp_security') or 'starttls'
+    try:
+        port = int(get_notif('smtp_port') or (465 if security == 'ssl' else 587))
+    except ValueError:
+        port = 587
+    return SmtpSettings(
+        host=get_notif('smtp_host'),
+        security=security,
+        user=get_notif('smtp_user'),
+        password=get_notif('smtp_password'),
+        from_name=get_notif('from_name') or 'MeshCore Bot',
+        from_email=get_notif('from_email'),
+        port=port,
+        allow_local=get_notif('allow_local_smtp').lower() == 'true',
+    )
+
+
+def send_smtp_message(settings: SmtpSettings, msg: Any) -> None:
+    """Deliver *msg* over SSL, STARTTLS or plain SMTP per the settings; raises on failure."""
+    import smtplib
+    import ssl as _ssl
+
+    context = _ssl.create_default_context()
+    smtp_timeout = 30
+    if settings.security == 'ssl':
+        with smtplib.SMTP_SSL(settings.host, settings.port, context=context, timeout=smtp_timeout) as s:
+            if settings.user and settings.password:
+                s.login(settings.user, settings.password)
+            s.send_message(msg)
+    else:
+        with smtplib.SMTP(settings.host, settings.port, timeout=smtp_timeout) as s:
+            if settings.security == 'starttls':
+                s.ehlo()
+                s.starttls(context=context)
+                s.ehlo()
+            if settings.user and settings.password:
+                s.login(settings.user, settings.password)
+            s.send_message(msg)
+
+
 class MaintenanceRunner:
     """Runs data retention, nightly email, log rotation hot-apply, and DB backups."""
 
@@ -415,41 +477,28 @@ class MaintenanceRunner:
 
     def send_nightly_email(self) -> None:
         """Build and dispatch the nightly maintenance digest if enabled."""
-        import smtplib
-        import ssl as _ssl
         from email.message import EmailMessage
 
         if self.get_notif('nightly_enabled') != 'true':
             return
 
-        smtp_host = self.get_notif('smtp_host')
-        smtp_security = self.get_notif('smtp_security') or 'starttls'
-        smtp_user = self.get_notif('smtp_user')
-        smtp_password = self.get_notif('smtp_password')
-        from_name = self.get_notif('from_name') or 'MeshCore Bot'
-        from_email = self.get_notif('from_email')
+        smtp = read_smtp_settings(self.get_notif)
         recipients = [r.strip() for r in self.get_notif('recipients').split(',') if r.strip()]
 
-        if not smtp_host or not from_email or not recipients:
+        if not smtp.host or not smtp.from_email or not recipients:
             self.logger.warning(
                 "Nightly email enabled but SMTP settings incomplete "
-                f"(host={smtp_host!r}, from={from_email!r}, recipients={recipients})"
+                f"(host={smtp.host!r}, from={smtp.from_email!r}, recipients={recipients})"
             )
             return
 
-        allow_local = self.get_notif('allow_local_smtp').lower() == 'true'
-        if not validate_external_url(f'http://{smtp_host}', allow_private=allow_local):
+        if not validate_external_url(f'http://{smtp.host}', allow_private=smtp.allow_local):
             self.logger.error(
                 "Nightly email aborted: SMTP host %r resolves to a private or reserved address",
-                smtp_host,
+                smtp.host,
             )
             return
 
-        try:
-            smtp_port = int(self.get_notif('smtp_port') or (465 if smtp_security == 'ssl' else 587))
-        except ValueError:
-            smtp_port = 587
-
         now_utc = _utc_now()
         yesterday = now_utc - datetime.timedelta(days=1)
         period_start = yesterday.strftime('%Y-%m-%d %H:%M UTC')
@@ -461,7 +510,7 @@ class MaintenanceRunner:
 
             msg = EmailMessage()
             msg['Subject'] = f'MeshCore Bot — Nightly Report {now_utc.strftime("%Y-%m-%d")}'
-            msg['From'] = f'{from_name} <{from_email}>'
+            msg['From'] = smtp.from_header
             msg['To'] = ', '.join(recipients)
             msg.set_content(body)
 
@@ -478,22 +527,7 @@ class MaintenanceRunner:
                         except Exception as attach_err:
                             self.logger.warning(f"Could not attach log file to nightly email: {attach_err}")
 
-            context = _ssl.create_default_context()
-            _smtp_timeout = 30
-            if smtp_security == 'ssl':
-                with smtplib.SMTP_SSL(smtp_host, smtp_port, context=context, timeout=_smtp_timeout) as s:
-                    if smtp_user and smtp_password:
-                        s.login(smtp_user, smtp_password)
-                    s.send_message(msg)
-            else:
-                with smtplib.SMTP(smtp_host, smtp_port, timeout=_smtp_timeout) as s:
-                    if smtp_security == 'starttls':
-                        s.ehlo()
-                        s.starttls(context=context)
-                        s.ehlo()
-                    if smtp_user and smtp_password:
-                        s.login(smtp_user, smtp_password)
-                    s.send_message(msg)
+            send_smtp_message(smtp, msg)
 
             self.logger.info(
                 f"Nightly maintenance email sent to {recipients} "

modules/scheduler.py

@@ -23,7 +23,7 @@ from apscheduler.triggers.date import DateTrigger
 from meshcore.events import EventType
 
 from .flood_scope import scope_key_hex
-from .maintenance import MaintenanceRunner
+from .maintenance import MaintenanceRunner, read_smtp_settings, send_smtp_message
 from .models import CHANNEL_REGIONAL_FLOOD_SCOPE_BODY_OVERHEAD, channel_body_limit, self_info_name
 from .scheduled_message_cron import (
     is_valid_legacy_hhmm,
@@ -1744,8 +1744,6 @@ class MessageScheduler:
         This method is intentionally synchronous so it can be run in a thread
         executor from the async event loop without blocking it.
         """
-        import smtplib
-        import ssl as _ssl
         from email.message import EmailMessage
 
         zombie_alert_enabled = self.bot.config.getboolean(
@@ -1770,12 +1768,7 @@ class MessageScheduler:
         if not zombie_alert_enabled:
             return
 
-        smtp_host     = self._get_notif('smtp_host')
-        smtp_security = self._get_notif('smtp_security') or 'starttls'
-        smtp_user     = self._get_notif('smtp_user')
-        smtp_password = self._get_notif('smtp_password')
-        from_name     = self._get_notif('from_name') or 'MeshCore Bot'
-        from_email    = self._get_notif('from_email')
+        smtp = read_smtp_settings(self._get_notif)
 
         # Alert recipients: dedicated config key, falls back to nightly recipients
         alert_email_cfg = zombie_alert_email_cfg or self.bot.config.get(
@@ -1788,27 +1781,21 @@ class MessageScheduler:
         else:
             recipients = [r.strip() for r in self._get_notif('recipients').split(',') if r.strip()]
 
-        if not smtp_host or not from_email or not recipients:
+        if not smtp.host or not smtp.from_email or not recipients:
             self.bot.logger.warning(
                 "Zombie alert email enabled but SMTP settings incomplete "
-                f"(host={smtp_host!r}, from={from_email!r}, recipients={recipients}) "
+                f"(host={smtp.host!r}, from={smtp.from_email!r}, recipients={recipients}) "
                 "— alert email not sent"
             )
             return
 
-        allow_local = self._get_notif('allow_local_smtp').lower() == 'true'
-        if not validate_external_url(f'http://{smtp_host}', allow_private=allow_local):
+        if not validate_external_url(f'http://{smtp.host}', allow_private=smtp.allow_local):
             self.bot.logger.error(
                 "Zombie alert email aborted: SMTP host %r resolves to a private or reserved address",
-                smtp_host,
+                smtp.host,
             )
             return
 
-        try:
-            smtp_port = int(self._get_notif('smtp_port') or (465 if smtp_security == 'ssl' else 587))
-        except ValueError:
-            smtp_port = 587
-
         now_utc         = datetime.datetime.now(datetime.timezone.utc)
         connection_type = self.bot.config.get('Connection', 'connection_type', fallback='unknown')
         serial_port     = self.bot.config.get('Connection', 'serial_port', fallback='n/a')
@@ -1850,27 +1837,11 @@ class MessageScheduler:
         try:
             msg = EmailMessage()
             msg['Subject'] = subject
-            msg['From']    = f'{from_name} <{from_email}>'
+            msg['From']    = smtp.from_header
             msg['To']      = ', '.join(recipients)
             msg.set_content(body)
 
-            context = _ssl.create_default_context()
-            _smtp_timeout = 30
-
-            if smtp_security == 'ssl':
-                with smtplib.SMTP_SSL(smtp_host, smtp_port, context=context, timeout=_smtp_timeout) as s:
-                    if smtp_user and smtp_password:
-                        s.login(smtp_user, smtp_password)
-                    s.send_message(msg)
-            else:
-                with smtplib.SMTP(smtp_host, smtp_port, timeout=_smtp_timeout) as s:
-                    if smtp_security == 'starttls':
-                        s.ehlo()
-                        s.starttls(context=context)
-                        s.ehlo()
-                    if smtp_user and smtp_password:
-                        s.login(smtp_user, smtp_password)
-                    s.send_message(msg)
+            send_smtp_message(smtp, msg)
 
             self.bot.logger.info(
                 f"Zombie radio alert email sent to {recipients}"
@@ -1903,8 +1874,6 @@ class MessageScheduler:
 
         Intentionally synchronous — intended to be run in a daemon thread.
         """
-        import smtplib
-        import ssl as _ssl
         from email.message import EmailMessage
 
         alert_enabled = self.bot.config.getboolean(
@@ -1915,12 +1884,7 @@ class MessageScheduler:
         if not alert_enabled:
             return
 
-        smtp_host     = self._get_notif('smtp_host')
-        smtp_security = self._get_notif('smtp_security') or 'starttls'
-        smtp_user     = self._get_notif('smtp_user')
-        smtp_password = self._get_notif('smtp_password')
-        from_name     = self._get_notif('from_name') or 'MeshCore Bot'
-        from_email    = self._get_notif('from_email')
+        smtp = read_smtp_settings(self._get_notif)
 
         alert_email_cfg = self.bot.config.get(
             'Connection',
@@ -1932,27 +1896,21 @@ class MessageScheduler:
         else:
             recipients = [r.strip() for r in self._get_notif('recipients').split(',') if r.strip()]
 
-        if not smtp_host or not from_email or not recipients:
+        if not smtp.host or not smtp.from_email or not recipients:
             self.bot.logger.warning(
                 "Radio-offline alert email enabled but SMTP settings incomplete "
-                f"(host={smtp_host!r}, from={from_email!r}, recipients={recipients}) "
+                f"(host={smtp.host!r}, from={smtp.from_email!r}, recipients={recipients}) "
                 "— alert email not sent"
             )
             return
 
-        allow_local = self._get_notif('allow_local_smtp').lower() == 'true'
-        if not validate_external_url(f'http://{smtp_host}', allow_private=allow_local):
+        if not validate_external_url(f'http://{smtp.host}', allow_private=smtp.allow_local):
             self.bot.logger.error(
                 "Radio-offline alert email aborted: SMTP host %r resolves to a private or reserved address",
-                smtp_host,
+                smtp.host,
             )
             return
 
-        try:
-            smtp_port = int(self._get_notif('smtp_port') or (465 if smtp_security == 'ssl' else 587))
-        except ValueError:
-            smtp_port = 587
-
         now_utc         = datetime.datetime.now(datetime.timezone.utc)
         connection_type = self.bot.config.get('Connection', 'connection_type', fallback='unknown')
         serial_port     = self.bot.config.get('Connection', 'serial_port', fallback='n/a')
@@ -1992,27 +1950,11 @@ class MessageScheduler:
         try:
             msg = EmailMessage()
             msg['Subject'] = subject
-            msg['From']    = f'{from_name} <{from_email}>'
+            msg['From']    = smtp.from_header
             msg['To']      = ', '.join(recipients)
             msg.set_content(body)
 
-            context = _ssl.create_default_context()
-            _smtp_timeout = 30
-
-            if smtp_security == 'ssl':
-                with smtplib.SMTP_SSL(smtp_host, smtp_port, context=context, timeout=_smtp_timeout) as s:
-                    if smtp_user and smtp_password:
-                        s.login(smtp_user, smtp_password)
-                    s.send_message(msg)
-            else:
-                with smtplib.SMTP(smtp_host, smtp_port, timeout=_smtp_timeout) as s:
-                    if smtp_security == 'starttls':
-                        s.ehlo()
-                        s.starttls(context=context)
-                        s.ehlo()
-                    if smtp_user and smtp_password:
-                        s.login(smtp_user, smtp_password)
-                    s.send_message(msg)
+            send_smtp_message(smtp, msg)
 
             self.bot.logger.info(f"Radio-offline alert email sent to {recipients}")
         except Exception as e: