107 lines
3.1 KiB
Python
107 lines
3.1 KiB
Python
"""Email notification functionality."""
|
|
|
|
import logging
|
|
import smtplib
|
|
|
|
from . import datetime_str
|
|
from email.mime.text import MIMEText
|
|
|
|
|
|
class EmailService:
|
|
"""Email service for sending notifications."""
|
|
|
|
def __init__(self, config):
|
|
"""
|
|
Initialize email service.
|
|
|
|
Args:
|
|
config: Application configuration dictionary.
|
|
"""
|
|
self.config = config.get("email", {})
|
|
self.enabled = self.config.get("enabled", False)
|
|
|
|
def send(self, to, subject, body):
|
|
"""
|
|
Send email using configured SMTP server.
|
|
|
|
Args:
|
|
to: Recipient email address.
|
|
subject: Email subject.
|
|
body: Email body text.
|
|
|
|
Returns:
|
|
True if sent successfully, False otherwise.
|
|
"""
|
|
if not self.enabled:
|
|
logging.debug("Email disabled, skipping")
|
|
return False
|
|
|
|
try:
|
|
msg = MIMEText(body)
|
|
msg["Subject"] = subject
|
|
msg["From"] = self.config["from_address"]
|
|
msg["To"] = to
|
|
|
|
smtp_host = self.config["smtp_host"]
|
|
smtp_port = self.config["smtp_port"]
|
|
|
|
server = smtplib.SMTP(smtp_host, smtp_port)
|
|
if self.config.get("smtp_starttls", False):
|
|
server.starttls()
|
|
|
|
try:
|
|
if self.config.get("smtp_user"):
|
|
server.login(
|
|
self.config["smtp_user"],
|
|
self.config["smtp_password"]
|
|
)
|
|
server.sendmail(msg["From"], [to], msg.as_string())
|
|
logging.info(f"Email sent: to={to} subject={subject}")
|
|
return True
|
|
finally:
|
|
server.quit()
|
|
|
|
except Exception as e:
|
|
logging.error(f"Email send failed: to={to} error={e}")
|
|
return False
|
|
|
|
def send_expiry_notification(
|
|
self,
|
|
email,
|
|
hostname,
|
|
last_ipv4,
|
|
last_ipv6,
|
|
expiry_ttl
|
|
):
|
|
"""
|
|
Send hostname expiry notification email.
|
|
|
|
Args:
|
|
email: Recipient email.
|
|
hostname: Expired hostname.
|
|
last_ipv4: Tuple containing last IPv4 address and last update timestamp.
|
|
last_ipv6: Tuple containing last IPv6 address and last update timestamp.
|
|
expiry_ttl: Expiry TTL in seconds.
|
|
|
|
Returns:
|
|
True if sent successfully.
|
|
"""
|
|
subject = f"DDNS hostname expired: {hostname}"
|
|
body = f"""Your dynamic DNS entry has expired due to inactivity.
|
|
|
|
Hostname: {hostname}
|
|
"""
|
|
if last_ipv4:
|
|
ip = last_ipv4[0]
|
|
last_update = datetime_str(last_ipv4[1])
|
|
body += f"IPv4 address: {ip} (last update: {last_update})\n"
|
|
if last_ipv6:
|
|
ip = last_ipv6[0]
|
|
last_update = datetime_str(last_ipv6[1])
|
|
body += f"IPv6 address: {ip} (last update: {last_update})\n"
|
|
body += f"""Expiry TTL: {expiry_ttl} seconds
|
|
|
|
The DNS records have been removed. Update your client to restore them.
|
|
"""
|
|
return self.send(email, subject, body)
|