"""Email notification service for domain and price alerts."""
import logging
import asyncio
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from typing import Optional
from dataclasses import dataclass
import aiosmtplib
from app.config import get_settings
logger = logging.getLogger(__name__)
settings = get_settings()
@dataclass
class EmailConfig:
"""Email configuration."""
smtp_host: str = ""
smtp_port: int = 587
smtp_user: str = ""
smtp_password: str = ""
from_email: str = "noreply@pounce.dev"
from_name: str = "Pounce Alerts"
class EmailService:
"""
Async email service for sending notifications.
Supports:
- Domain availability alerts
- Price change notifications
- Weekly digest emails
"""
def __init__(self, config: EmailConfig = None):
"""Initialize email service."""
self.config = config or EmailConfig(
smtp_host=getattr(settings, 'smtp_host', ''),
smtp_port=getattr(settings, 'smtp_port', 587),
smtp_user=getattr(settings, 'smtp_user', ''),
smtp_password=getattr(settings, 'smtp_password', ''),
)
self._enabled = bool(self.config.smtp_host and self.config.smtp_user)
@property
def is_enabled(self) -> bool:
"""Check if email service is configured."""
return self._enabled
async def send_email(
self,
to_email: str,
subject: str,
html_body: str,
text_body: str = None,
) -> bool:
"""
Send an email.
Args:
to_email: Recipient email address
subject: Email subject
html_body: HTML content
text_body: Plain text content (optional)
Returns:
True if sent successfully, False otherwise
"""
if not self.is_enabled:
logger.warning("Email service not configured, skipping send")
return False
try:
message = MIMEMultipart("alternative")
message["From"] = f"{self.config.from_name} <{self.config.from_email}>"
message["To"] = to_email
message["Subject"] = subject
# Add text and HTML parts
if text_body:
message.attach(MIMEText(text_body, "plain"))
message.attach(MIMEText(html_body, "html"))
# Send via SMTP
await aiosmtplib.send(
message,
hostname=self.config.smtp_host,
port=self.config.smtp_port,
username=self.config.smtp_user,
password=self.config.smtp_password,
use_tls=True,
)
logger.info(f"Email sent to {to_email}: {subject}")
return True
except Exception as e:
logger.error(f"Failed to send email to {to_email}: {e}")
return False
async def send_domain_available_alert(
self,
to_email: str,
domain: str,
user_name: str = None,
) -> bool:
"""Send alert when a watched domain becomes available."""
subject = f"🎉 Domain Available: {domain}"
html_body = f"""
• pounce
Great news{f', {user_name}' if user_name else ''}!
A domain you're watching just became available:
{domain}
This is your chance to register it before someone else does!
Register Now →
"""
text_body = f"""
Great news{f', {user_name}' if user_name else ''}!
A domain you're watching just became available: {domain}
This is your chance to register it before someone else does!
Register now: https://porkbun.com/checkout/search?q={domain}
— The Pounce Team
"""
return await self.send_email(to_email, subject, html_body, text_body)
async def send_price_change_alert(
self,
to_email: str,
tld: str,
old_price: float,
new_price: float,
change_percent: float,
registrar: str = "average",
) -> bool:
"""Send alert when TLD price changes significantly."""
direction = "📈 increased" if new_price > old_price else "📉 decreased"
color = "#f97316" if new_price > old_price else "#00d4aa"
subject = f"TLD Price Alert: .{tld} {direction} by {abs(change_percent):.1f}%"
html_body = f"""
• pounce
TLD Price Change Alert
.{tld}
${old_price:.2f} →
${new_price:.2f}
{change_percent:+.1f}% ({registrar})
{"Now might be a good time to register!" if new_price < old_price else "Prices have gone up - act fast if you need this TLD."}
"""
text_body = f"""
TLD Price Change Alert
.{tld} has {direction} by {abs(change_percent):.1f}%
Old price: ${old_price:.2f}
New price: ${new_price:.2f}
Source: {registrar}
{"Now might be a good time to register!" if new_price < old_price else "Prices have gone up."}
— The Pounce Team
"""
return await self.send_email(to_email, subject, html_body, text_body)
async def send_weekly_digest(
self,
to_email: str,
user_name: str,
watched_domains: list[dict],
price_changes: list[dict],
) -> bool:
"""Send weekly summary email."""
subject = "📊 Your Weekly Pounce Digest"
# Build domain status HTML
domains_html = ""
for d in watched_domains[:10]:
status_color = "#00d4aa" if d.get("is_available") else "#888"
status_text = "Available!" if d.get("is_available") else "Taken"
domains_html += f'| {d["domain"]} | {status_text} |
'
# Build price changes HTML
prices_html = ""
for p in price_changes[:5]:
change = p.get("change_percent", 0)
color = "#00d4aa" if change < 0 else "#f97316"
prices_html += f'| .{p["tld"]} | {change:+.1f}% | ${p.get("new_price", 0):.2f} |
'
html_body = f"""
• pounce
Weekly Digest
Hi {user_name}, here's your weekly summary:
Your Watched Domains
| Domain | Status |
|---|
{domains_html if domains_html else '| No domains being watched |
'}
Notable Price Changes
| TLD | Change | Price |
|---|
{prices_html if prices_html else '| No significant changes this week |
'}
"""
return await self.send_email(to_email, subject, html_body)
# Singleton instance
email_service = EmailService()