"""Email service for sending notifications via SMTP."""
import logging
from email.message import EmailMessage
import aiosmtplib
from sqlalchemy import select
from scribe.config import Config
from scribe.models import async_session
from scribe.models.setting import Setting
from scribe.models.user import User
logger = logging.getLogger(__name__)
# Inline SVG logo for email (white palette, renders on indigo header)
_EMAIL_LOGO_SVG = (
''
)
def _email_html(title: str, body: str) -> str:
"""Wrap email body content in the standard Fabled Scribe template."""
return f"""
{_EMAIL_LOGO_SVG}Fabled Scribe
{title}
{body}
Sent by your Fabled Scribe instance.
"""
SMTP_SETTING_KEYS = [
"smtp_host",
"smtp_port",
"smtp_username",
"smtp_password",
"smtp_from_address",
"smtp_from_name",
"smtp_use_tls",
]
async def get_smtp_config() -> dict[str, str]:
"""Get SMTP config from admin's settings, falling back to env vars."""
config: dict[str, str] = {
"smtp_host": Config.SMTP_HOST,
"smtp_port": str(Config.SMTP_PORT),
"smtp_username": Config.SMTP_USERNAME,
"smtp_password": Config.SMTP_PASSWORD,
"smtp_from_address": Config.SMTP_FROM_ADDRESS,
"smtp_from_name": Config.SMTP_FROM_NAME,
"smtp_use_tls": "true" if Config.SMTP_USE_TLS else "false",
}
# Override with DB settings from admin user
async with async_session() as session:
result = await session.execute(
select(Setting)
.join(User, Setting.user_id == User.id)
.where(User.role == "admin", Setting.key.in_(SMTP_SETTING_KEYS))
)
for setting in result.scalars().all():
config[setting.key] = setting.value
return config
async def is_smtp_configured() -> bool:
"""Check if SMTP is configured (has a host set)."""
config = await get_smtp_config()
return bool(config.get("smtp_host"))
async def get_base_url() -> str:
"""Get the application base URL from admin settings, falling back to Config.BASE_URL."""
async with async_session() as session:
result = await session.execute(
select(Setting)
.join(User, Setting.user_id == User.id)
.where(User.role == "admin", Setting.key == "base_url")
)
setting = result.scalars().first()
if setting and setting.value:
return setting.value.rstrip("/")
return Config.BASE_URL.rstrip("/")
async def send_email(to: str, subject: str, html_body: str) -> None:
"""Send an email via SMTP."""
config = await get_smtp_config()
host = config.get("smtp_host")
if not host:
logger.debug("SMTP not configured, skipping email to %s", to)
return
port = int(config.get("smtp_port", "587"))
username = config.get("smtp_username", "")
password = config.get("smtp_password", "")
from_address = config.get("smtp_from_address", "")
from_name = config.get("smtp_from_name", "Fabled Scribe")
use_tls = config.get("smtp_use_tls", "true") == "true"
msg = EmailMessage()
msg["Subject"] = subject
msg["From"] = f"{from_name} <{from_address}>" if from_name else from_address
msg["To"] = to
msg.set_content(subject) # plain text fallback
msg.add_alternative(html_body, subtype="html")
try:
await aiosmtplib.send(
msg,
hostname=host,
port=port,
username=username or None,
password=password or None,
start_tls=use_tls and port != 465,
use_tls=port == 465,
)
logger.info("Email sent to %s: %s", to, subject)
except Exception:
logger.exception("Failed to send email to %s: %s", to, subject)
raise
async def send_test_email(to: str) -> None:
"""Send a branded test email."""
body = """
SMTP is configured correctly
Your Fabled Scribe instance can send email notifications.