feat: add test email and improve on email notifications (#3018)
This commit is contained in:
@@ -24,7 +24,7 @@ from lnbits.core.models.notifications import (
|
||||
)
|
||||
from lnbits.core.services.nostr import fetch_nip5_details, send_nostr_dm
|
||||
from lnbits.core.services.websockets import websocket_manager
|
||||
from lnbits.helpers import check_callback_url
|
||||
from lnbits.helpers import check_callback_url, is_valid_email_address
|
||||
from lnbits.settings import settings
|
||||
from lnbits.utils.nostr import normalize_private_key
|
||||
|
||||
@@ -111,41 +111,56 @@ async def send_telegram_message(token: str, chat_id: str, message: str) -> dict:
|
||||
return response.json()
|
||||
|
||||
|
||||
async def send_email_notification(message: str) -> dict:
|
||||
await send_email(
|
||||
settings.lnbits_email_notifications_server,
|
||||
settings.lnbits_email_notifications_port,
|
||||
settings.lnbits_email_notifications_password,
|
||||
settings.lnbits_email_notifications_email,
|
||||
settings.lnbits_email_notifications_to_emails,
|
||||
"LNbits Notification",
|
||||
message,
|
||||
)
|
||||
return {"status": "ok"}
|
||||
async def send_email_notification(
|
||||
message: str, subject: str = "LNbits Notification"
|
||||
) -> dict:
|
||||
if not settings.lnbits_email_notifications_enabled:
|
||||
return {"status": "error", "message": "Email notifications are disabled"}
|
||||
try:
|
||||
await send_email(
|
||||
settings.lnbits_email_notifications_server,
|
||||
settings.lnbits_email_notifications_port,
|
||||
settings.lnbits_email_notifications_username,
|
||||
settings.lnbits_email_notifications_password,
|
||||
settings.lnbits_email_notifications_email,
|
||||
settings.lnbits_email_notifications_to_emails,
|
||||
subject,
|
||||
message,
|
||||
)
|
||||
return {"status": "ok"}
|
||||
except Exception as e:
|
||||
logger.error(f"Error sending email notification: {e}")
|
||||
return {"status": "error", "message": str(e)}
|
||||
|
||||
|
||||
async def send_email(
|
||||
server: str,
|
||||
port: int,
|
||||
username: str,
|
||||
password: str,
|
||||
from_email: str,
|
||||
to_emails: list,
|
||||
to_emails: list[str],
|
||||
subject: str,
|
||||
message: str,
|
||||
):
|
||||
) -> bool:
|
||||
if not is_valid_email_address(from_email):
|
||||
raise ValueError(f"Invalid from email address: {from_email}")
|
||||
if len(to_emails) == 0:
|
||||
raise ValueError("No email addresses provided")
|
||||
for email in to_emails:
|
||||
if not is_valid_email_address(email):
|
||||
raise ValueError(f"Invalid email address: {email}")
|
||||
msg = MIMEMultipart()
|
||||
msg["From"] = from_email
|
||||
msg["To"] = ", ".join(to_emails)
|
||||
msg["Subject"] = subject
|
||||
msg.attach(MIMEText(message, "plain"))
|
||||
try:
|
||||
with smtplib.SMTP(server, port) as smtp_server:
|
||||
smtp_server.starttls()
|
||||
smtp_server.login(from_email, password)
|
||||
smtp_server.sendmail(from_email, to_emails, msg.as_string())
|
||||
logger.debug(f"Emails sent successfully to: {', '.join(to_emails)}")
|
||||
except Exception as e:
|
||||
logger.debug(f"Failed to send email: {e}")
|
||||
username = username if len(username) > 0 else from_email
|
||||
with smtplib.SMTP(server, port) as smtp_server:
|
||||
smtp_server.starttls()
|
||||
smtp_server.login(username, password)
|
||||
smtp_server.sendmail(from_email, to_emails, msg.as_string())
|
||||
return True
|
||||
|
||||
|
||||
def is_message_type_enabled(message_type: NotificationType) -> bool:
|
||||
|
||||
Reference in New Issue
Block a user