307 lines
8.9 KiB
Python
307 lines
8.9 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, Optional
|
||
|
||
from aiosmtplib import SMTP, SMTPException
|
||
import reflex as rx
|
||
from reflex.event import EventCallback
|
||
|
||
from application.states.task import TaskState
|
||
from application.states.database import DatabaseState
|
||
|
||
# 验证码HTML模板
|
||
CAPTCHA_HTML_TEMPLATE = """
|
||
<p>验证码：<strong>{captcha_code}</strong></p>
|
||
"""
|
||
|
||
|
||
class AuthState(rx.State):
|
||
"""
|
||
认证状态
|
||
"""
|
||
|
||
# 用户唯一标识（通过本地存储同步）
|
||
user_id: str = rx.LocalStorage("user_id", sync=True)
|
||
|
||
# 邮箱
|
||
email: str = ""
|
||
# 验证码
|
||
captcha_code: str = ""
|
||
# 验证码已发送
|
||
is_captcha_code_sent: bool = False
|
||
# 再次发送验证码倒计时
|
||
resend_captcha_code_countdown: int = 0
|
||
# 协议同意状态
|
||
is_policies_agreed: bool = True
|
||
# 错误信息
|
||
error: str = ""
|
||
# 登录中状态
|
||
is_logging_in: bool = False
|
||
|
||
# 选中的导航按钮，默认为创建任务
|
||
selected_nav_button: str = "create_task"
|
||
# 设置悬停卡片打开状态
|
||
is_settings_hover_card_open: bool = False
|
||
|
||
# 数据库状态
|
||
_db_state: Optional[DatabaseState] = None
|
||
|
||
async def get_db_state(self) -> DatabaseState:
|
||
"""
|
||
获取当前数据库状态
|
||
:return: 当前数据库状态
|
||
"""
|
||
# 若数据库状态为空则获取数据库状态，否则直接返回当前数据库状态
|
||
if not self._db_state:
|
||
self._db_state = await self.get_state(DatabaseState)
|
||
return self._db_state
|
||
|
||
@rx.event
|
||
def reset_error(self) -> None:
|
||
"""
|
||
重置错误信息
|
||
:return: None
|
||
"""
|
||
self.error = ""
|
||
|
||
@rx.event
|
||
def set_email(self, email: str) -> None:
|
||
"""
|
||
设置邮箱
|
||
:param email: 邮箱
|
||
:return: None
|
||
"""
|
||
self.email = email.strip()
|
||
|
||
def validate_email(self) -> bool:
|
||
"""
|
||
校验邮箱
|
||
:return: 校验是否通过
|
||
"""
|
||
# 邮箱非空校验
|
||
if not self.email:
|
||
self.error = "请输入邮箱"
|
||
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.error = "邮箱格式不正确，请重新输入"
|
||
return False
|
||
|
||
self.error = ""
|
||
return True
|
||
|
||
@rx.var
|
||
def is_captcha_code_sending_button_disabled(self) -> bool:
|
||
"""
|
||
验证码发送按钮不可点击状态
|
||
邮箱为空或再次发送验证码倒计时大于0时禁用
|
||
:return: bool
|
||
"""
|
||
return not self.email or self.resend_captcha_code_countdown > 0
|
||
|
||
@rx.event(background=True)
|
||
async def countdown_background(self) -> None:
|
||
"""
|
||
倒计时（后台任务）
|
||
:return: None
|
||
"""
|
||
async with self:
|
||
self.resend_captcha_code_countdown = 59
|
||
|
||
while self.resend_captcha_code_countdown > 0:
|
||
await sleep(1)
|
||
async with self:
|
||
self.resend_captcha_code_countdown -= 1
|
||
|
||
@rx.event(background=True)
|
||
async def send_captcha_code_background(self) -> None:
|
||
"""
|
||
发送验证码（后台任务）
|
||
:return: None
|
||
"""
|
||
|
||
async with self:
|
||
# 获取当前数据库状态
|
||
db_state = await self.get_db_state()
|
||
# 创建验证记录
|
||
captcha_code = await db_state.create_captcha_record(email=self.email)
|
||
|
||
# 构建验证码邮件
|
||
message = MIMEText(
|
||
CAPTCHA_HTML_TEMPLATE.format(captcha_code=captcha_code),
|
||
"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_code(self) -> AsyncGenerator[EventCallback, None]:
|
||
"""
|
||
发送验证码
|
||
:return: AsyncGenerator[EventCallback, None]
|
||
"""
|
||
# 校验邮箱
|
||
if not self.validate_email():
|
||
return
|
||
|
||
if self.resend_captcha_code_countdown > 0:
|
||
return
|
||
|
||
# 标记验证码已发送
|
||
self.is_captcha_code_sent = True
|
||
|
||
# 同时触发发送验证码和倒计时后台事件
|
||
yield type(self).send_captcha_code_background()
|
||
yield type(self).countdown_background()
|
||
|
||
@rx.event
|
||
def set_captcha_code(self, captcha_code: str) -> None:
|
||
"""
|
||
设置验证码
|
||
:param captcha_code: 验证码
|
||
:return: None
|
||
"""
|
||
self.captcha_code = captcha_code.strip()
|
||
|
||
@rx.event
|
||
def toggle_policies_agreed(self) -> None:
|
||
"""
|
||
切换协议同意状态
|
||
:return: None
|
||
"""
|
||
self.is_policies_agreed = not self.is_policies_agreed
|
||
|
||
@rx.event
|
||
async def resume_task_state(self):
|
||
"""
|
||
恢复任务状态
|
||
"""
|
||
# 获取任务状态
|
||
task_state = await self.get_state(TaskState)
|
||
# 恢复任务状态
|
||
await task_state.resume_task_state(user_id=self.user_id)
|
||
|
||
@rx.event
|
||
async def login(self) -> None:
|
||
"""
|
||
登录
|
||
:return: None
|
||
"""
|
||
# 验证邮箱
|
||
if not self.validate_email():
|
||
return
|
||
|
||
# 校验验证码非空
|
||
if not self.captcha_code:
|
||
self.error = "请输入验证码"
|
||
return
|
||
|
||
# 校验验证码格式
|
||
if not re.fullmatch(r"^\d{6}$", self.captcha_code):
|
||
self.error = "验证码格式不正确，请重新输入"
|
||
return
|
||
|
||
# 校验协议同意状态
|
||
if not self.is_policies_agreed:
|
||
self.error = "请先阅读并同意协议和政策"
|
||
return
|
||
|
||
self.is_logging_in = True
|
||
|
||
# 获取当前数据库状态
|
||
db_state = await self.get_db_state()
|
||
|
||
# 核验验证码
|
||
if not await db_state.verify_captcha_code(
|
||
email=self.email, captcha_code=self.captcha_code
|
||
):
|
||
self.is_logging_in = False
|
||
self.error = "验证码错误"
|
||
return
|
||
|
||
# 创建用户记录
|
||
user_id = await db_state.create_user_record(email=self.email)
|
||
|
||
# 获取任务状态
|
||
task_state = await self.get_state(TaskState)
|
||
# 恢复当前用户的任务状态
|
||
await task_state.resume_task_state(user_id=user_id)
|
||
|
||
self.user_id = user_id
|
||
|
||
self.email = ""
|
||
self.captcha_code = ""
|
||
self.is_captcha_code_sent = False
|
||
self.resend_captcha_code_countdown = 0
|
||
self.is_policies_agreed = True
|
||
self.error = ""
|
||
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 set_selected_nav_button(self, nav_button: str):
|
||
"""
|
||
设置选中导航按钮
|
||
"""
|
||
self.selected_nav_button = nav_button
|
||
|
||
if nav_button == "create_task":
|
||
# 获取任务状态
|
||
task_state = await self.get_state(TaskState)
|
||
# 将空设置为当前任务唯一标识
|
||
task_state.set_current_task_id("")
|
||
|
||
@rx.event
|
||
async def logout(self) -> None:
|
||
"""
|
||
退出登录
|
||
:return: None
|
||
"""
|
||
self.is_settings_hover_card_open = False # 设置设置悬停卡片关闭
|
||
task_state = await self.get_state(TaskState)
|
||
task_state.user_id = ""
|
||
task_state.tasks = {}
|
||
task_state.actived_task_id = ""
|
||
task_state.is_task_history_shown = False
|
||
self.user_id = ""
|