import smtplib from email.message import EmailMessage from sqlalchemy.orm import Session from app.core.config import get_settings from app.core.security import decrypt_secret, encrypt_secret from app.models.entities import MailSetting from app.schemas.common import MailSettingsIn, MailSettingsOut def get_mail_settings(db: Session) -> MailSetting: settings = db.get(MailSetting, 1) if not settings: settings = MailSetting(id=1) db.add(settings) db.flush() return settings def serialize_mail_settings(settings: MailSetting) -> MailSettingsOut: return MailSettingsOut( smtp_host=settings.smtp_host, smtp_port=settings.smtp_port, smtp_user=settings.smtp_user, has_password=bool(settings.smtp_password_encrypted), use_tls=settings.use_tls, use_starttls=settings.use_starttls, sender_address=settings.sender_address, sender_name=settings.sender_name, ) def update_mail_settings(db: Session, payload: MailSettingsIn) -> MailSetting: settings = get_mail_settings(db) settings.smtp_host = payload.smtp_host settings.smtp_port = payload.smtp_port settings.smtp_user = payload.smtp_user if payload.smtp_password is not None: settings.smtp_password_encrypted = encrypt_secret(payload.smtp_password) settings.use_tls = payload.use_tls settings.use_starttls = payload.use_starttls settings.sender_address = str(payload.sender_address) if payload.sender_address else None settings.sender_name = payload.sender_name return settings def send_mail(db: Session, to: str, subject: str, body: str) -> None: settings = get_mail_settings(db) if not settings.smtp_host or not settings.sender_address: raise RuntimeError("SMTP is not configured") message = EmailMessage() message["From"] = f"{settings.sender_name} <{settings.sender_address}>" message["To"] = to message["Subject"] = subject message.set_content(body) password = decrypt_secret(settings.smtp_password_encrypted) client_cls = smtplib.SMTP_SSL if settings.use_tls and not settings.use_starttls else smtplib.SMTP with client_cls(settings.smtp_host, settings.smtp_port, timeout=20) as smtp: if settings.use_starttls: smtp.starttls() if settings.smtp_user and password: smtp.login(settings.smtp_user, password) smtp.send_message(message) def invite_body(token: str) -> str: link = f"{get_settings().instance_url.rstrip('/')}/accept-invite?token={token}" return f"Welcome to NexaPantry.\n\nOpen this invitation link to set your password:\n{link}\n\nThe link expires automatically."