Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/quant_platform_kit/notifications/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
SmtpEmailChannel,
TelegramChatChannel,
)
from .email import parse_email_recipients, send_smtp_email
from ._email import parse_email_recipients, send_smtp_email
from .events import NotificationPublisher, RenderedNotification, publish_rendered_notification
from .push import parse_push_recipients, send_ntfy_push, send_pushover_push, send_strategy_plugin_push
from .sms import normalize_sms_recipient, parse_sms_recipients, send_twilio_sms
Expand Down
67 changes: 67 additions & 0 deletions src/quant_platform_kit/notifications/_email.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,67 @@
"""SMTP email notification helpers."""

from __future__ import annotations

import smtplib
from collections.abc import Sequence
from email.message import EmailMessage


def parse_email_recipients(raw_value: str | Sequence[str] | None) -> tuple[str, ...]:
if raw_value is None:
return ()
if isinstance(raw_value, str):
values = raw_value.replace(";", ",").replace("\n", ",").split(",")
else:
values = raw_value
recipients = []
seen = set()
for value in values:
recipient = str(value or "").strip()
if not recipient or recipient in seen:
continue
recipients.append(recipient)
seen.add(recipient)
return tuple(recipients)


def send_smtp_email(
*,
subject: str,
body: str,
smtp_host: str | None,
smtp_port: int,
sender: str | None,
recipients: Sequence[str],
username: str | None = None,
password: str | None = None,
use_starttls: bool = True,
use_ssl: bool = False,
timeout: float = 10.0,
smtp_module=smtplib,
printer=print,
) -> bool:
resolved_recipients = parse_email_recipients(recipients)
host = str(smtp_host or "").strip()
from_addr = str(sender or "").strip()
if not host or not from_addr or not resolved_recipients:
return False

message = EmailMessage()
message["From"] = from_addr
message["To"] = ", ".join(resolved_recipients)
message["Subject"] = str(subject or "").strip() or "strategy alert"
message.set_content(str(body or "").strip())

try:
smtp_cls = smtp_module.SMTP_SSL if use_ssl else smtp_module.SMTP
with smtp_cls(host, int(smtp_port), timeout=timeout) as smtp:
if use_starttls and not use_ssl:
smtp.starttls()
if username:
smtp.login(str(username), str(password or ""))
smtp.send_message(message)
return True
except Exception as exc:
printer(f"Email send failed: {exc}", flush=True)
return False
71 changes: 9 additions & 62 deletions src/quant_platform_kit/notifications/email.py
Original file line number Diff line number Diff line change
@@ -1,67 +1,14 @@
"""SMTP email notification helpers."""
"""Backward-compatibility shim — delegates to _email to avoid stdlib collision.

from __future__ import annotations
The SMTP helpers live in ``_email.py`` because ``email.py`` shadows
Python's stdlib ``email`` package, breaking ``smtplib`` imports in some
environments (e.g. when pytest's plugin loader runs alongside this package).

import smtplib
from collections.abc import Sequence
from email.message import EmailMessage
Prefer importing from the package root::

from quant_platform_kit.notifications import parse_email_recipients, send_smtp_email
"""

def parse_email_recipients(raw_value: str | Sequence[str] | None) -> tuple[str, ...]:
if raw_value is None:
return ()
if isinstance(raw_value, str):
values = raw_value.replace(";", ",").replace("\n", ",").split(",")
else:
values = raw_value
recipients = []
seen = set()
for value in values:
recipient = str(value or "").strip()
if not recipient or recipient in seen:
continue
recipients.append(recipient)
seen.add(recipient)
return tuple(recipients)
from ._email import parse_email_recipients, send_smtp_email


def send_smtp_email(
*,
subject: str,
body: str,
smtp_host: str | None,
smtp_port: int,
sender: str | None,
recipients: Sequence[str],
username: str | None = None,
password: str | None = None,
use_starttls: bool = True,
use_ssl: bool = False,
timeout: float = 10.0,
smtp_module=smtplib,
printer=print,
) -> bool:
resolved_recipients = parse_email_recipients(recipients)
host = str(smtp_host or "").strip()
from_addr = str(sender or "").strip()
if not host or not from_addr or not resolved_recipients:
return False

message = EmailMessage()
message["From"] = from_addr
message["To"] = ", ".join(resolved_recipients)
message["Subject"] = str(subject or "").strip() or "strategy alert"
message.set_content(str(body or "").strip())

try:
smtp_cls = smtp_module.SMTP_SSL if use_ssl else smtp_module.SMTP
with smtp_cls(host, int(smtp_port), timeout=timeout) as smtp:
if use_starttls and not use_ssl:
smtp.starttls()
if username:
smtp.login(str(username), str(password or ""))
smtp.send_message(message)
return True
except Exception as exc:
printer(f"Email send failed: {exc}", flush=True)
return False
__all__ = ["parse_email_recipients", "send_smtp_email"]
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@
from pathlib import Path
from typing import Any

from .email import send_smtp_email
from ._email import send_smtp_email
from .push import send_strategy_plugin_push
from .sms import send_twilio_sms
from .telegram import send_strategy_plugin_telegram
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@
)

from .alert_marker import CloudAlertMarkerStore, _clean_relative_key
from .email import parse_email_recipients, send_smtp_email
from ._email import parse_email_recipients, send_smtp_email


_DEFAULT_EMAIL_SMTP_HOST = "smtp.gmail.com"
Expand Down
Loading