import logging from dataclasses import dataclass from datetime import datetime, timedelta from pathlib import Path from typing import Any import emails from jinja2 import Template from jose import JWTError, jwt from app.core.config import settings @dataclass class EmailData: html_content: str subject: str def render_email_template(*, template_name: str, context: dict[str, Any]): template_str = (Path(settings.EMAIL_TEMPLATES_DIR) / template_name).read_text() html_content = Template(template_str).render(context) return html_content def send_email( *, email_to: str, subject: str = "", html_content: str = "", ) -> None: assert settings.EMAILS_ENABLED, "no provided configuration for email variables" message = emails.Message( subject=subject, html=html_content, mail_from=(settings.EMAILS_FROM_NAME, settings.EMAILS_FROM_EMAIL), ) smtp_options = {"host": settings.SMTP_HOST, "port": settings.SMTP_PORT} if settings.SMTP_TLS: smtp_options["tls"] = True if settings.SMTP_USER: smtp_options["user"] = settings.SMTP_USER if settings.SMTP_PASSWORD: smtp_options["password"] = settings.SMTP_PASSWORD response = message.send(to=email_to, smtp=smtp_options) logging.info(f"send email result: {response}") def generate_test_email(email_to: str) -> EmailData: project_name = settings.PROJECT_NAME subject = f"{project_name} - Test email" html_content = render_email_template( template_name="test_email.html", context={"project_name": settings.PROJECT_NAME, "email": email_to}, ) return EmailData(html_content=html_content, subject=subject) def generate_reset_password_email(email_to: str, email: str, token: str) -> EmailData: project_name = settings.PROJECT_NAME subject = f"{project_name} - Password recovery for user {email}" server_host = settings.SERVER_HOST link = f"{server_host}/reset-password?token={token}" html_content = render_email_template( template_name="reset_password.html", context={ "project_name": settings.PROJECT_NAME, "username": email, "email": email_to, "valid_hours": settings.EMAIL_RESET_TOKEN_EXPIRE_HOURS, "link": link, }, ) return EmailData(html_content=html_content, subject=subject) def generate_new_account_email( email_to: str, username: str, password: str ) -> EmailData: project_name = settings.PROJECT_NAME subject = f"{project_name} - New account for user {username}" link = settings.SERVER_HOST html_content = render_email_template( template_name="new_account.html", context={ "project_name": settings.PROJECT_NAME, "username": username, "password": password, "email": email_to, "link": link, }, ) return EmailData(html_content=html_content, subject=subject) def generate_password_reset_token(email: str) -> str: delta = timedelta(hours=settings.EMAIL_RESET_TOKEN_EXPIRE_HOURS) now = datetime.utcnow() expires = now + delta exp = expires.timestamp() encoded_jwt = jwt.encode( {"exp": exp, "nbf": now, "sub": email}, settings.SECRET_KEY, algorithm="HS256", ) return encoded_jwt def verify_password_reset_token(token: str) -> str | None: try: decoded_token = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"]) return decoded_token["sub"] except JWTError: return None