61 lines
2.7 KiB
Python
61 lines
2.7 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
|
|
from Utils.EmailUtils import email_check
|
|
from Utils.VerifyCodeUtils import EmailVerifyType
|
|
from . import Schemas
|
|
from . import Crud
|
|
from Context.common import common_db, email_code_verify, auth_util
|
|
|
|
router = APIRouter(prefix="/api/user")
|
|
|
|
|
|
@router.post('/login_by_email', summary='邮箱登录', tags=['用户接口'], response_model=Schemas.LoginByEmailRes)
|
|
def login_by_email(req: Schemas.LoginByEmailReq,
|
|
db: Session = Depends(common_db.get_db)):
|
|
checked = email_code_verify.check_code(req.email, req.code, EmailVerifyType.login)
|
|
if not checked:
|
|
raise HTTPException(detail="邮箱验证码错误", status_code=303)
|
|
user = Crud.get_user_info(db, req.email)
|
|
if not user:
|
|
user = Crud.create_user(db, req.email)
|
|
auth_data = Crud.get_user_auth(db, req.email)
|
|
user_data = user.to_dict()
|
|
user_data["auth_data"] = auth_data
|
|
token_data = auth_util.tokenDataModel(**user_data)
|
|
token = auth_util.create_token(token_data)
|
|
return Schemas.LoginByEmailRes(token=token)
|
|
|
|
|
|
@router.post("/get_email_verify_code", tags=["用户接口"], summary="获取邮箱验证码")
|
|
def get_email_verify_code(body: Schemas.GetEmailVerifyCodeReq):
|
|
try:
|
|
email = body.email
|
|
if not email_check(email):
|
|
raise HTTPException(detail="错误的邮箱格式", status_code=403)
|
|
email_code_verify.send_code(email, EmailVerifyType.login)
|
|
except Exception as e:
|
|
print(e)
|
|
raise HTTPException(detail="邮箱验证码发送失败", status_code=403)
|
|
return {'state': 1, "msg": "邮箱验证码已发送至邮箱,请查看"}
|
|
|
|
|
|
@router.post("/refresh_token", summary="重新获取token", tags=['用户接口'], response_model=Schemas.RefreshTokenRes)
|
|
def refresh_token(token_data: auth_util.tokenDataModel = Depends(auth_util.token_data_depend),
|
|
db: Session = Depends(common_db.get_db)):
|
|
user_data = Crud.get_full_user_info(db, token_data.email)
|
|
token_data = auth_util.tokenDataModel(**user_data)
|
|
token = auth_util.create_token(token_data)
|
|
return {'token': token}
|
|
|
|
|
|
@router.post('/get_user_info', summary='获取用户信息', tags=["用户接口"], response_model=Schemas.GetUserInfoRes)
|
|
def get_user_info(token_data: auth_util.tokenDataModel = Depends(auth_util.token_data_depend),
|
|
db: Session = Depends(common_db.get_db)):
|
|
user_data = Crud.get_full_user_info(db, token_data.email)
|
|
if not user_data:
|
|
raise HTTPException(detail="用户未注册", status_code=403)
|
|
if not user_data['email']:
|
|
raise HTTPException(detail="用户未绑定邮箱", status_code=403)
|
|
return Schemas.GetUserInfoRes(**user_data)
|