284 lines
8.4 KiB
Python
284 lines
8.4 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
认证状态
|
||
"""
|
||
from asyncio import sleep
|
||
from email.header import Header
|
||
from email.mime.text import MIMEText
|
||
import re
|
||
from typing import AsyncGenerator
|
||
|
||
from aiosmtplib import SMTP, SMTPException
|
||
import reflex as rx
|
||
from reflex.event import EventCallback
|
||
|
||
from application.states.conversation import ConversationState
|
||
from application.states.database import DatabaseState
|
||
|
||
# 验证码HTML模板
|
||
CAPTCHA_HTML_TEMPLATE = """
|
||
<p>验证码：<strong>{captcha}</strong></p>
|
||
"""
|
||
|
||
|
||
class AuthState(rx.State):
|
||
"""
|
||
认证状态
|
||
"""
|
||
|
||
# 用户唯一标识（通过本地存储同步）
|
||
user_id: str = rx.LocalStorage("user_id", sync=True)
|
||
|
||
# 邮箱
|
||
email: str = ""
|
||
# 验证码
|
||
captcha: str = ""
|
||
# 协议同意状态
|
||
is_policies_agreed: bool = True
|
||
|
||
# 登录错误信息
|
||
login_error_message: str = ""
|
||
# 再次发送验证码倒计时
|
||
resend_captcha_countdown: int = 0
|
||
# 验证码已发送状态
|
||
is_captcha_sent: bool = False
|
||
# 登录中状态
|
||
is_logging_in: bool = False
|
||
# 设置悬停卡片打开状态
|
||
is_settings_hover_card_open: bool = False
|
||
|
||
# 激活的导航按钮
|
||
activated_nav_button: str = "conversation"
|
||
|
||
@rx.event
|
||
def reset_error_message(self) -> None:
|
||
"""
|
||
重置登录错误信息
|
||
:return: None
|
||
"""
|
||
self.login_error_message = ""
|
||
|
||
@rx.event
|
||
def set_email(self, email: str) -> None:
|
||
"""
|
||
设置邮箱
|
||
:param email: 邮箱
|
||
:return: None
|
||
"""
|
||
self.email = email.strip()
|
||
|
||
def validate_email(self) -> bool:
|
||
"""
|
||
校验邮箱
|
||
:return: 校验是否通过，True 表示通过，False 表示不通过
|
||
"""
|
||
# 邮箱非空校验
|
||
if not self.email:
|
||
self.login_error_message = "请输入邮箱"
|
||
return False
|
||
|
||
# 邮箱格式校验
|
||
"""
|
||
邮箱由用户名@域名组成，其中：
|
||
1）用户名首字符为字母或数字，中间可穿插下划线、短横线、加号或点，符号后为字母或数字；
|
||
2）域名由若干子域名段和顶级域名段组成，各段之间用点分隔，其中：
|
||
-- 子域名段：首字符为字母或数字，中间可穿插下划线或短横线，符号后为字母或数字；
|
||
-- 顶级域名段：至少包含两个字母或数字；
|
||
"""
|
||
if not re.fullmatch(
|
||
r"^[a-zA-Z0-9]+([_\-+.][a-zA-Z0-9_\-+]+)*@[a-zA-Z0-9]+(\-[a-zA-Z0-9\-]+)*(\.[a-zA-Z0-9]+(\-[a-zA-Z0-9\-]+)*)*\.[a-zA-Z0-9]{2,}$",
|
||
self.email,
|
||
):
|
||
self.login_error_message = "邮箱格式不正确，请重新输入"
|
||
return False
|
||
self.login_error_message = ""
|
||
return True
|
||
|
||
@rx.var
|
||
def is_captcha_sending_disabled(self) -> bool:
|
||
"""
|
||
验证码发送禁用状态
|
||
邮箱为空或再次发送验证码倒计时大于0时禁用
|
||
:return: 禁用状态，True 表示禁用，False 表示未禁用
|
||
"""
|
||
return not self.email or self.resend_captcha_countdown > 0
|
||
|
||
@rx.event(background=True)
|
||
async def countdown(self) -> None:
|
||
"""
|
||
倒计时（后台任务）
|
||
:return: None
|
||
"""
|
||
async with self:
|
||
self.resend_captcha_countdown = 59
|
||
|
||
while self.resend_captcha_countdown > 0:
|
||
await sleep(1)
|
||
async with self:
|
||
self.resend_captcha_countdown -= 1
|
||
|
||
@rx.event(background=True)
|
||
async def send_captcha_background(self) -> None:
|
||
"""
|
||
发送验证码（后台任务）
|
||
:return: None
|
||
"""
|
||
async with self:
|
||
# 创建验证码记录并获取验证码
|
||
database_state = await self.get_state(DatabaseState)
|
||
captcha = await database_state.create_captcha_record(email=self.email)
|
||
|
||
# 构建验证码邮件
|
||
message = MIMEText(
|
||
CAPTCHA_HTML_TEMPLATE.format(captcha=captcha),
|
||
"html",
|
||
"utf-8",
|
||
)
|
||
message["Subject"] = Header("验证码", "utf-8").encode()
|
||
message["From"] = "prism@liubiren.cloud"
|
||
message["To"] = self.email
|
||
try:
|
||
async with SMTP(
|
||
hostname="smtp.feishu.cn",
|
||
port=465,
|
||
username="prism@liubiren.cloud",
|
||
password="cx1dM5tBcbyWftpS",
|
||
use_tls=True,
|
||
local_hostname="localhost",
|
||
) as smtp:
|
||
await smtp.send_message(message)
|
||
|
||
except SMTPException:
|
||
pass
|
||
|
||
@rx.event
|
||
async def send_captcha(self) -> AsyncGenerator[EventCallback, None]:
|
||
"""
|
||
发送验证码
|
||
:return: AsyncGenerator[EventCallback, None]
|
||
"""
|
||
# 校验邮箱
|
||
if not self.validate_email():
|
||
return
|
||
|
||
if self.resend_captcha_countdown > 0:
|
||
return
|
||
|
||
self.is_captcha_sent = True
|
||
|
||
# 将倒计时和发送验证码事件添加至后台任务队列
|
||
yield type(self).countdown()
|
||
yield type(self).send_captcha_background()
|
||
|
||
@rx.event
|
||
def set_captcha(self, captcha: str) -> None:
|
||
"""
|
||
设置验证码
|
||
:param captcha: 验证码
|
||
:return: None
|
||
"""
|
||
self.captcha = captcha.strip()
|
||
|
||
@rx.event
|
||
def toggle_policies_agreed(self) -> None:
|
||
"""
|
||
切换协议同意状态
|
||
:return: None
|
||
"""
|
||
self.is_policies_agreed = not self.is_policies_agreed
|
||
|
||
@rx.event
|
||
async def sync_user_id(self):
|
||
"""
|
||
同步用户唯一标识
|
||
"""
|
||
# 同步会话状态中用户唯一标识
|
||
conversation_state = await self.get_state(ConversationState)
|
||
await conversation_state.set_user_id(user_id=self.user_id)
|
||
|
||
@rx.event
|
||
async def login(self) -> None:
|
||
"""
|
||
登录
|
||
:return: None
|
||
"""
|
||
# 验证邮箱
|
||
if not self.validate_email():
|
||
return
|
||
|
||
# 校验验证码非空
|
||
if not self.captcha:
|
||
self.login_error_message = "请输入验证码"
|
||
return
|
||
|
||
# 校验验证码格式
|
||
if not re.fullmatch(r"^\d{6}$", self.captcha):
|
||
self.login_error_message = "验证码格式不正确，请重新输入"
|
||
return
|
||
|
||
# 校验协议同意状态
|
||
if not self.is_policies_agreed:
|
||
self.login_error_message = "请先阅读并同意协议和政策"
|
||
return
|
||
|
||
self.is_logging_in = True
|
||
|
||
# 获取数据库状态
|
||
database_state = await self.get_state(DatabaseState)
|
||
# 核验验证码
|
||
if not await database_state.verify_captcha(
|
||
email=self.email, captcha=self.captcha
|
||
):
|
||
self.is_logging_in = False
|
||
self.login_error_message = "验证码错误"
|
||
return
|
||
# 创建用户记录
|
||
user_id = await database_state.create_user_record(email=self.email)
|
||
|
||
# 就会话状态设置用户唯一标识
|
||
conversation_state = await self.get_state(ConversationState)
|
||
await conversation_state.set_user_id(user_id=user_id)
|
||
|
||
self.user_id = user_id
|
||
|
||
self.email = ""
|
||
self.captcha = ""
|
||
self.is_policies_agreed = True
|
||
self.login_error_message = ""
|
||
self.resend_captcha_countdown = 0
|
||
self.is_captcha_sent = False
|
||
self.is_logging_in = False
|
||
|
||
@rx.event
|
||
def set_settings_hover_card_open(self, is_settings_hover_card_open: bool) -> None:
|
||
"""
|
||
设置设置悬停卡片打开状态
|
||
:param is_settings_hover_card_open: 设置悬停卡片打开状态，True 表示打开，False 表示关闭
|
||
:return: None
|
||
"""
|
||
self.is_settings_hover_card_open = is_settings_hover_card_open
|
||
|
||
@rx.event
|
||
async def logout(self) -> None:
|
||
"""
|
||
退出登录
|
||
:return: None
|
||
"""
|
||
self.is_settings_hover_card_open = False
|
||
|
||
conversation_state = await self.get_state(ConversationState)
|
||
conversation_state.user_id = ""
|
||
conversation_state.conversations = []
|
||
conversation_state.conversation_id = ""
|
||
conversation_state.is_conversation_history_shown = False
|
||
conversation_state.is_conversation_creating = False
|
||
|
||
self.user_id = ""
|
||
|
||
@rx.event
|
||
def switch_activated_nav_button(self, button: str):
|
||
"""
|
||
将指定导航按钮设置为激活的导航按钮
|
||
"""
|
||
self.activated_nav_button = button
|