74 lines
2.2 KiB
Python
74 lines
2.2 KiB
Python
import smtplib
|
|
import random
|
|
import string
|
|
|
|
from pydantic import EmailStr
|
|
|
|
|
|
from app.config import settings
|
|
from app.tasks.celery import celery
|
|
from app.tasks.email_templates import (
|
|
create_registration_confirmation_template,
|
|
create_password_change_confirmation_template,
|
|
create_password_recover_template,
|
|
)
|
|
from app.users.auth import encode_invitation_token
|
|
from app.users.schemas import SInvitationData
|
|
|
|
|
|
def generate_confirmation_code(length=6) -> str:
|
|
characters = string.ascii_letters + string.digits
|
|
confirmation_code = "".join(random.choice(characters) for _ in range(length))
|
|
return confirmation_code
|
|
|
|
|
|
@celery.task
|
|
def send_registration_confirmation_email(user_data: SInvitationData):
|
|
invitation_token = encode_invitation_token(user_data)
|
|
|
|
confirmation_link = settings.INVITATION_LINK_HOST + "/api/users/email_verification/" + invitation_token
|
|
|
|
msg_content = create_registration_confirmation_template(
|
|
username=user_data.username, email_to=user_data.email_to, confirmation_link=confirmation_link
|
|
)
|
|
|
|
with smtplib.SMTP_SSL(settings.SMTP_HOST, settings.SMTP_PORT) as server:
|
|
server.login(settings.SMTP_USER, settings.SMTP_PASS)
|
|
server.send_message(msg_content)
|
|
|
|
|
|
@celery.task
|
|
def send_password_change_email(username: str, email_to: EmailStr, MODE: str):
|
|
if MODE == "TEST":
|
|
return True
|
|
msg_content = create_password_change_confirmation_template(username=username, email_to=email_to)
|
|
|
|
with smtplib.SMTP_SSL(settings.SMTP_HOST, settings.SMTP_PORT) as server:
|
|
server.login(settings.SMTP_USER, settings.SMTP_PASS)
|
|
server.send_message(msg_content)
|
|
|
|
return True
|
|
|
|
|
|
@celery.task
|
|
def send_password_recover_email(username: str, email_to: EmailStr, MODE: str):
|
|
confirmation_code = generate_confirmation_code()
|
|
if MODE == "TEST":
|
|
return confirmation_code
|
|
msg_content = create_password_recover_template(username=username, email_to=email_to, confirmation_code=confirmation_code)
|
|
|
|
with smtplib.SMTP_SSL(settings.SMTP_HOST, settings.SMTP_PORT) as server:
|
|
server.login(settings.SMTP_USER, settings.SMTP_PASS)
|
|
server.send_message(msg_content)
|
|
|
|
return confirmation_code
|
|
|
|
|
|
@celery.task
|
|
def send_data_change_confirmation_email(username: str, email_to: EmailStr, MODE: str):
|
|
pass
|
|
|
|
|
|
@celery.task
|
|
def send_data_change_email(username: str, email_to: EmailStr, MODE: str):
|
|
pass
|