"""Email settings service helpers.""" from __future__ import annotations from typing import Any, Dict, Tuple from ..database import get_app_settings, update_app_setting from ..settings import ( SMTP_SETTINGS, EMAIL_NOTIFY_CONTACT_FORM, EMAIL_NOTIFY_NEWSLETTER_SIGNUPS, ) from ..utils import normalize_recipients, is_valid_email EMAIL_SETTING_PREFIX = "email_" # Definition metadata describing each configurable field. EMAIL_SETTINGS_DEFINITIONS: Dict[str, Dict[str, Any]] = { "smtp_host": { "type": "string", "required": True, "default": SMTP_SETTINGS.get("host") or "", "description": "Hostname or IP address of the SMTP server.", }, "smtp_port": { "type": "int", "required": True, "default": int(SMTP_SETTINGS.get("port") or 587), "description": "Port number used to connect to the SMTP server.", }, "smtp_username": { "type": "string", "required": False, "default": SMTP_SETTINGS.get("username") or "", "description": "Username for authenticating with the SMTP server (optional).", }, "smtp_password": { "type": "string", "required": False, "default": SMTP_SETTINGS.get("password") or "", "description": "Password for authenticating with the SMTP server (optional).", }, "smtp_sender": { "type": "string", "required": True, "default": SMTP_SETTINGS.get("sender") or SMTP_SETTINGS.get("username") or "", "description": "Default email address used in the From header when sending mail.", }, "smtp_use_tls": { "type": "bool", "required": True, "default": bool(SMTP_SETTINGS.get("use_tls", True)), "description": "Enable TLS when communicating with the SMTP server.", }, "smtp_recipients": { "type": "list", "required": False, "default": SMTP_SETTINGS.get("recipients") or [], "description": "Comma separated list of notification recipient emails.", }, "notify_contact_form": { "type": "bool", "required": True, "default": EMAIL_NOTIFY_CONTACT_FORM, "description": "Send notification emails for new contact form submissions.", }, "notify_newsletter_signups": { "type": "bool", "required": True, "default": EMAIL_NOTIFY_NEWSLETTER_SIGNUPS, "description": "Send notification emails for new newsletter subscriber signups.", }, } def _storage_key(field: str) -> str: """Return the persistent storage key for a given field.""" return f"{EMAIL_SETTING_PREFIX}{field}" def _ensure_bool(value: Any) -> bool: """Coerce user-supplied values into boolean form.""" if isinstance(value, bool): return value if value is None: return False return str(value).strip().lower() in {"1", "true", "yes", "on"} def _ensure_int(value: Any) -> Tuple[bool, int | None]: """Try to coerce a value into an integer.""" try: return True, int(value) except (TypeError, ValueError): return False, None def _serialize(field: str, value: Any) -> str: """Serialize a typed value to its string representation for storage.""" definition = EMAIL_SETTINGS_DEFINITIONS[field] if definition["type"] == "bool": return "true" if bool(value) else "false" if definition["type"] == "int": return str(int(value)) if definition["type"] == "list": if isinstance(value, str): value = normalize_recipients(value) return ", ".join(value) return str(value or "") def _deserialize(field: str, raw_value: Any) -> Any: """Deserialize the stored string representation for a field.""" definition = EMAIL_SETTINGS_DEFINITIONS[field] if raw_value is None or raw_value == "": return definition["default"] if definition["type"] == "bool": return _ensure_bool(raw_value) if definition["type"] == "int": ok, coerced = _ensure_int(raw_value) return coerced if ok and coerced is not None else definition["default"] if definition["type"] == "list": if isinstance(raw_value, (list, tuple)): return [item for item in raw_value if item] return normalize_recipients(str(raw_value)) return str(raw_value) def load_email_settings() -> Dict[str, Any]: """Load email settings, combining stored overrides with defaults.""" stored = get_app_settings() results: Dict[str, Any] = {} for field in EMAIL_SETTINGS_DEFINITIONS: raw_value = stored.get(_storage_key(field)) results[field] = _deserialize(field, raw_value) return results def load_effective_smtp_settings() -> Dict[str, Any]: """Return SMTP configuration merging persisted settings with environment defaults.""" effective = load_email_settings() recipients_value = effective.get("smtp_recipients") if isinstance(recipients_value, list): effective_recipients = [item for item in recipients_value if item] else: effective_recipients = normalize_recipients(str(recipients_value or "")) return { "host": effective.get("smtp_host") or SMTP_SETTINGS.get("host"), "port": effective.get("smtp_port") or SMTP_SETTINGS.get("port"), "username": effective.get("smtp_username") or SMTP_SETTINGS.get("username"), "password": effective.get("smtp_password") or SMTP_SETTINGS.get("password"), "sender": effective.get("smtp_sender") or SMTP_SETTINGS.get("sender") or SMTP_SETTINGS.get("username"), "use_tls": bool(effective.get("smtp_use_tls")) if effective.get("smtp_use_tls") is not None else SMTP_SETTINGS.get("use_tls", True), "recipients": effective_recipients, "notify_contact_form": bool(effective.get("notify_contact_form")), "notify_newsletter_signups": bool(effective.get("notify_newsletter_signups")), } def validate_email_settings(payload: Dict[str, Any]) -> Dict[str, str]: """Validate user supplied email settings. Returns mapping of field -> error.""" errors: Dict[str, str] = {} smtp_host = str(payload.get("smtp_host", "")).strip() if not smtp_host: errors["smtp_host"] = "SMTP host is required." ok, port_value = _ensure_int(payload.get("smtp_port")) if not ok or port_value is None or not (1 <= port_value <= 65535): errors["smtp_port"] = "SMTP port must be a valid integer between 1 and 65535." sender = str(payload.get("smtp_sender", "")).strip() if not sender: errors["smtp_sender"] = "Sender email is required." elif not is_valid_email(sender): errors["smtp_sender"] = "Sender must be a valid email address." recipients_raw = payload.get("smtp_recipients") if isinstance(recipients_raw, list): recipient_list = [item.strip() for item in recipients_raw if item] else: recipient_list = normalize_recipients(str(recipients_raw or "")) if recipient_list: invalid = [addr for addr in recipient_list if not is_valid_email(addr)] if invalid: errors["smtp_recipients"] = "Invalid recipient email(s): " + ", ".join( invalid) for flag_field in ("smtp_use_tls", "notify_contact_form", "notify_newsletter_signups"): value = payload.get(flag_field) if value is None: continue # Ensure value is interpretable as boolean. if not isinstance(value, bool) and str(value).strip().lower() not in {"1", "0", "true", "false", "yes", "no", "on", "off"}: errors[flag_field] = "Provide a boolean value." return errors def persist_email_settings(payload: Dict[str, Any]) -> Dict[str, Any]: """Persist validated email settings and return the normalized payload.""" normalized: Dict[str, Any] = {} for field, meta in EMAIL_SETTINGS_DEFINITIONS.items(): value = payload.get(field, meta["default"]) if meta["type"] == "bool": value = _ensure_bool(value) elif meta["type"] == "int": ok, coerced = _ensure_int(value) value = coerced if ok and coerced is not None else meta["default"] elif meta["type"] == "list": if isinstance(value, list): value = [item.strip() for item in value if item] else: value = normalize_recipients(str(value or "")) else: value = str(value or "").strip() # Persist using string representation update_app_setting(_storage_key(field), _serialize(field, value)) normalized[field] = value return normalized