107 lines
3.2 KiB
Python
107 lines
3.2 KiB
Python
from fastapi import APIRouter, HTTPException, Depends
|
|
from sqlalchemy.orm import Session
|
|
from app.models.user import UserLogin, UserInfo, VerifyCodeRequest, UserDB
|
|
from app.models.database import get_db
|
|
import random
|
|
import string
|
|
import redis
|
|
from app.core.config import settings
|
|
from unisdk.sms import UniSMS
|
|
from unisdk.exception import UniException
|
|
from datetime import timedelta
|
|
from app.core.security import create_access_token
|
|
from app.core.response import success_response, error_response
|
|
|
|
router = APIRouter()
|
|
|
|
# Redis 连接
|
|
redis_client = redis.Redis(
|
|
host=settings.REDIS_HOST,
|
|
port=settings.REDIS_PORT,
|
|
db=settings.REDIS_DB,
|
|
password=settings.REDIS_PASSWORD,
|
|
decode_responses=True
|
|
)
|
|
|
|
# 初始化短信客户端
|
|
client = UniSMS(settings.UNI_APP_ID)
|
|
|
|
@router.post("/send-code")
|
|
async def send_verify_code(request: VerifyCodeRequest):
|
|
"""发送验证码"""
|
|
phone = request.phone
|
|
code = ''.join(random.choices(string.digits, k=6))
|
|
|
|
try:
|
|
# 发送短信
|
|
res = client.send({
|
|
"to": phone,
|
|
"signature": settings.UNI_SMS_SIGN,
|
|
"templateId": settings.UNI_SMS_TEMPLATE_ID,
|
|
"templateData": {
|
|
"code": code
|
|
}
|
|
})
|
|
|
|
if res.code != "0":
|
|
return error_response(message=f"短信发送失败: {res.message}")
|
|
|
|
# 存储验证码到 Redis
|
|
redis_client.setex(
|
|
f"verify_code:{phone}",
|
|
settings.VERIFICATION_CODE_EXPIRE_SECONDS,
|
|
code
|
|
)
|
|
|
|
return success_response(message="验证码已发送")
|
|
|
|
except UniException as e:
|
|
return error_response(message=f"发送验证码失败: {str(e)}")
|
|
|
|
@router.post("/login")
|
|
async def login(user_login: UserLogin, db: Session = Depends(get_db)):
|
|
"""用户登录"""
|
|
phone = user_login.phone
|
|
verify_code = user_login.verify_code
|
|
|
|
# 验证验证码
|
|
stored_code = redis_client.get(f"verify_code:{phone}")
|
|
if not stored_code or stored_code != verify_code:
|
|
return error_response(message="验证码错误或已过期")
|
|
|
|
redis_client.delete(f"verify_code:{phone}")
|
|
|
|
# 查找或创建用户
|
|
user = db.query(UserDB).filter(UserDB.phone == phone).first()
|
|
if not user:
|
|
user = UserDB(
|
|
username=f"user_{phone[-4:]}",
|
|
phone=phone
|
|
)
|
|
db.add(user)
|
|
db.commit()
|
|
db.refresh(user)
|
|
|
|
# 创建访问令牌
|
|
access_token = create_access_token(
|
|
data={"sub": user.phone},
|
|
expires_delta=timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
|
|
)
|
|
|
|
return success_response(
|
|
message="登录成功",
|
|
data={
|
|
"user": UserInfo.model_validate(user),
|
|
"access_token": access_token,
|
|
"token_type": "bearer"
|
|
}
|
|
)
|
|
|
|
@router.get("/info")
|
|
async def get_user_info(phone: str, db: Session = Depends(get_db)):
|
|
"""获取用户信息"""
|
|
user = db.query(UserDB).filter(UserDB.phone == phone).first()
|
|
if not user:
|
|
return error_response(code=404, message="用户不存在")
|
|
|
|
return success_response(data=UserInfo.model_validate(user)) |