File size: 1,668 Bytes
21bc372 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 |
import random
import string
from config import *
from akn.utils.database import *
from datetime import datetime, timedelta
TEST_VIA_EMAIL_TEXT = """
Your login code: <spoiler>{otp}</spoiler> Never give this code to anyone,
even if they say it is from **AkenoAi**
"""
def generate_otp(length=6):
return ''.join(random.choices(string.digits, k=length))
async def store_otp(email: str, otp: str):
expiration_time = datetime.utcnow() + timedelta(minutes=5)
otp_data = {
"email": email,
"otp": otp,
"expires_at": expiration_time
}
await otp_collection.update_one(
{"email": email},
{"$set": otp_data},
upsert=True
)
async def verify_otp_in_db(email: str, otp: str):
otp_record = await otp_collection.find_one({"email": email})
if otp_record and otp_record["otp"] == otp:
if otp_record["expires_at"] > datetime.utcnow():
return True
return False
async def email_send_otp(email: str):
otp = generate_otp()
if email == str(ME_GMAIL):
return {"message": f"can't be blacklisted"}
await store_otp(email, otp)
try:
await send_check_otp_email(
receiver_email=email,
text=TEST_VIA_EMAIL_TEXT.format(
otp=otp
)
)
return {"message": f"OTP sent to check your email"}
except Exception as e:
return {"message": f"Error: {e}"}
async def email_verify_otp(email: str, otp: str):
is_valid = await verify_otp_in_db(email, otp)
if is_valid:
return {"message": "OTP verified successfully!"}
else:
return {"message": "Invalid or expired OTP"}
|