urban-investment-research/Mods/User/Router.py

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session

from Utils.EmailUtils import email_check
from Utils.VerifyCodeUtils import EmailVerifyType, PhoneVerifyType
from . import Schemas
from . import Crud
from Context.common import common_db, email_code_verify, auth_util, phone_verify_code

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)


@router.post("/get_phone_verify_code", tags=["用户接口"], summary="获取短信验证码")
def get_phone_verify_code(body: Schemas.GetPhoneVerifyCodeReq, db=Depends(common_db.get_db)):
    phone = body.phone
    user = Crud.get_user_info_by_phone(db, phone)
    if not user:
        raise HTTPException(detail="手机号未录入系统", status_code=303)
    try:
        phone_verify_code.send_code(phone, PhoneVerifyType.login)
    except Exception as e:
        print(e)
        raise HTTPException(detail="验证码发送失败", status_code=403)
    return {"msg": "验证码已发送至手机,请查看", "state": 1}


@router.post('/login_by_phone', tags=["用户接口"], summary='手机号登录')
def login_by_phone(req: Schemas.LoginByPhoneReq,
                   db: Session = Depends(common_db.get_db)):
    phone = req.phone.replace(" ", "")
    checked = phone_verify_code.check_code(phone, req.code, PhoneVerifyType.login)
    if not checked:
        raise HTTPException(detail="验证码错误", status_code=303)
    user_data = Crud.get_user_info_by_phone(db, phone)
    if not user_data:
        raise HTTPException(detail="手机号未录入系统", status_code=303)
    auth_data = Crud.get_user_auth(db, user_data['email'])
    user_data["auth_data"] = auth_data
    token_data = auth_util.tokenDataModel(**user_data).dict()
    token = auth_util.create_token(token_data)
    return {'msg': "成功", 'state': 1, 'data': {'token': token}}