81972f1328
本次提交完成了用户认证模块的注册扩展： 1. 新增阿里云短信和异步邮件发送工具类 2. 新增短信/邮箱相关配置类与环境变量支持 3. 添加短信验证码、待注册数据的Redis缓存工具方法 4. 扩展用户登录逻辑，支持用户名/邮箱/手机号多方式登录 5. 实现手机短信注册和邮箱验证注册完整流程 6. 更新权限初始化数据与系统配置项 7. 补充相关Pydantic请求响应模型 8. 新增依赖包并完善requirements.txt
538 lines
18 KiB
Python
538 lines
18 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, Query, status, Request
|
||
from fastapi.responses import Response, HTMLResponse
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from sqlalchemy import select, and_, or_
|
||
from fastcrud import FastCRUD
|
||
from datetime import timedelta, datetime
|
||
from uuid import uuid4
|
||
import json
|
||
import redis.asyncio as redis
|
||
|
||
from database import get_db
|
||
from models import User, SystemSetting
|
||
from schemas import (
|
||
UserCreate,
|
||
UserResponse,
|
||
UserLogin,
|
||
Token,
|
||
UserCreateRequest,
|
||
UserUpdate,
|
||
PhoneSendCodeRequest,
|
||
PhoneSendCodeResponse,
|
||
PhoneRegisterRequest,
|
||
EmailRegisterInitRequest,
|
||
EmailRegisterInitResponse,
|
||
RegisterSuccessResponse,
|
||
)
|
||
from utils.password import verify_password, get_password_hash
|
||
from utils.jwt import create_access_token, decode_token
|
||
from config import app_settings
|
||
from utils.redis_client import (
|
||
get_redis, set_token_in_redis, delete_token_from_redis,
|
||
save_sms_code, get_and_delete_sms_code,
|
||
save_pending_registration, get_and_delete_pending_registration,
|
||
)
|
||
from utils.captcha import create_captcha, validate_captcha
|
||
from middleware import get_current_user
|
||
from utils.like_filter import ilike_contains
|
||
from utils.log_utils import get_client_ip, get_user_agent, save_login_log, create_operation_log
|
||
from utils.sms import generate_sms_code, send_sms_code
|
||
from utils.email import send_verification_email
|
||
|
||
router = APIRouter(prefix="/users", tags=["users"])
|
||
|
||
# 创建FastCRUD实例
|
||
user_crud = FastCRUD(User, UserResponse)
|
||
|
||
|
||
@router.get("/captcha")
|
||
async def get_captcha(redis_conn: redis.Redis = Depends(get_redis)):
|
||
"""获取验证码图片"""
|
||
# 创建验证码
|
||
captcha_id, image_bytes = await create_captcha(redis_conn)
|
||
|
||
# 返回验证码图片，同时在响应头中返回验证码ID
|
||
response = Response(content=image_bytes, media_type="image/png")
|
||
response.headers["X-Captcha-ID"] = captcha_id
|
||
|
||
return response
|
||
|
||
|
||
@router.post("/login", response_model=Token)
|
||
async def login(
|
||
user_data: UserLogin,
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
redis_conn: redis.Redis = Depends(get_redis),
|
||
):
|
||
"""用户登录"""
|
||
ip = get_client_ip(request)
|
||
ua = get_user_agent(request)
|
||
|
||
# 验证验证码
|
||
is_captcha_valid = await validate_captcha(
|
||
redis_conn, user_data.captcha_id, user_data.captcha_code
|
||
)
|
||
if not is_captcha_valid:
|
||
await save_login_log(db, user_data.username, ip, ua, "failure", "验证码错误或已过期")
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="验证码错误或已过期",
|
||
headers={"WWW-Authenticate": "Bearer"},
|
||
)
|
||
|
||
# 查找用户（支持用户名/邮箱/手机号）
|
||
login_id = user_data.username
|
||
result = await db.execute(
|
||
select(User).filter(
|
||
or_(
|
||
User.username == login_id,
|
||
User.email == login_id,
|
||
User.phone == login_id,
|
||
)
|
||
)
|
||
)
|
||
user = result.scalar_one_or_none()
|
||
|
||
# 验证用户是否存在且密码正确
|
||
if not user or not verify_password(user_data.password, user.password_hash):
|
||
await save_login_log(db, login_id, ip, ua, "failure", "用户名或密码错误")
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="用户名或密码错误",
|
||
headers={"WWW-Authenticate": "Bearer"},
|
||
)
|
||
|
||
# 创建访问令牌
|
||
access_token_expires = timedelta(minutes=app_settings.access_token_expire_minutes)
|
||
access_token = create_access_token(
|
||
data={"sub": str(user.id), "username": user.username},
|
||
expires_delta=access_token_expires,
|
||
)
|
||
|
||
# 将令牌存储到Redis
|
||
expire_seconds = int(access_token_expires.total_seconds())
|
||
await set_token_in_redis(redis_conn, user.id, access_token, expire_seconds)
|
||
|
||
# 更新用户最后登录时间
|
||
user.lastlogintime = datetime.utcnow()
|
||
await db.commit()
|
||
|
||
# 记录登录成功日志
|
||
await save_login_log(db, user.username, ip, ua, "success", None)
|
||
|
||
return {"access_token": access_token, "token_type": "bearer"}
|
||
|
||
|
||
@router.post("/", response_model=UserResponse)
|
||
async def create_user(
|
||
user_data: UserCreateRequest,
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
):
|
||
"""创建用户"""
|
||
# 检查用户名是否已存在
|
||
result = await db.execute(select(User).filter(User.username == user_data.username))
|
||
existing_user = result.scalar_one_or_none()
|
||
if existing_user:
|
||
raise HTTPException(status_code=400, detail="用户名已存在")
|
||
|
||
# 生成密码哈希值
|
||
hashed_password = get_password_hash(user_data.password)
|
||
|
||
# 准备创建数据
|
||
create_data = user_data.model_dump()
|
||
create_data["password_hash"] = hashed_password
|
||
del create_data["password"] # 删除明文密码
|
||
# 转换为UserCreate模型
|
||
user_data_create = UserCreate(**create_data)
|
||
|
||
# 使用FastCRUD创建用户
|
||
created_user = await user_crud.create(
|
||
db, user_data_create, schema_to_select=UserResponse, return_as_model=True
|
||
)
|
||
|
||
# 记录操作日志
|
||
await create_operation_log(
|
||
db, current_user.id, current_user.username, "create", "user",
|
||
created_user.id, created_user.username,
|
||
{"username": user_data.username, "nickname": user_data.nickname},
|
||
get_client_ip(request), get_user_agent(request),
|
||
)
|
||
await db.commit()
|
||
|
||
return created_user
|
||
|
||
|
||
@router.post("/search", response_model=list[UserResponse])
|
||
async def search_users(
|
||
query: dict = {},
|
||
request: Request = None,
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
):
|
||
"""分页列表用户（POST方式）；支持登录名、昵称、邮箱、手机的模糊查询与组合查询（AND）"""
|
||
stmt = select(User)
|
||
filters = []
|
||
for col, key in (
|
||
(User.username, "username"),
|
||
(User.nickname, "nickname"),
|
||
(User.email, "email"),
|
||
(User.phone, "phone"),
|
||
):
|
||
raw = query.get(key)
|
||
cond = ilike_contains(col, raw)
|
||
if cond is not None:
|
||
filters.append(cond)
|
||
if filters:
|
||
stmt = stmt.where(and_(*filters))
|
||
result = await db.execute(stmt)
|
||
rows = result.scalars().all()
|
||
|
||
# 记录查询日志
|
||
await create_operation_log(
|
||
db, current_user.id, current_user.username, "search", "user",
|
||
None, None, {"filters": {k: v for k, v in query.items() if v}},
|
||
get_client_ip(request), get_user_agent(request),
|
||
)
|
||
await db.commit()
|
||
|
||
return [UserResponse.model_validate(u) for u in rows]
|
||
|
||
|
||
@router.get("/me", response_model=UserResponse)
|
||
async def get_current_user_info(
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
):
|
||
"""获取当前用户信息（须声明在 /{user_id} 之前，否则会被当成 id）"""
|
||
# 记录查询日志
|
||
await create_operation_log(
|
||
db, current_user.id, current_user.username, "search", "user",
|
||
current_user.id, current_user.username, None,
|
||
get_client_ip(request), get_user_agent(request),
|
||
)
|
||
await db.commit()
|
||
return current_user
|
||
|
||
|
||
@router.get("/{user_id}", response_model=UserResponse)
|
||
async def get_user(
|
||
user_id: int,
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
):
|
||
"""获取单个用户"""
|
||
user = await user_crud.get(db, id=user_id)
|
||
if not user:
|
||
raise HTTPException(status_code=404, detail="用户不存在")
|
||
|
||
# 记录查询日志
|
||
await create_operation_log(
|
||
db, current_user.id, current_user.username, "search", "user",
|
||
user_id, user.username, None,
|
||
get_client_ip(request), get_user_agent(request),
|
||
)
|
||
await db.commit()
|
||
|
||
return user
|
||
|
||
|
||
@router.put("/{user_id}", response_model=UserResponse)
|
||
async def update_user(
|
||
user_id: int,
|
||
user_data: UserUpdate,
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
):
|
||
"""更新用户"""
|
||
# 获取原用户数据用于记录变更
|
||
result = await db.execute(select(User).filter(User.id == user_id))
|
||
old_user = result.scalar_one_or_none()
|
||
if not old_user:
|
||
raise HTTPException(status_code=404, detail="用户不存在")
|
||
|
||
# 检查用户名是否被其他用户使用
|
||
result = await db.execute(
|
||
select(User).filter(User.username == user_data.username, User.id != user_id)
|
||
)
|
||
existing_user = result.scalar_one_or_none()
|
||
if existing_user:
|
||
raise HTTPException(status_code=400, detail="用户名已存在")
|
||
|
||
# 生成密码哈希值
|
||
hashed_password = (
|
||
get_password_hash(user_data.password) if user_data.password else None
|
||
)
|
||
|
||
# 准备更新数据
|
||
update_data = user_data.model_dump()
|
||
if hashed_password:
|
||
update_data["password_hash"] = hashed_password
|
||
del update_data["password"] # 删除明文密码
|
||
|
||
# 使用FastCRUD更新用户
|
||
updated_user = await user_crud.update(
|
||
db, update_data, id=user_id, schema_to_select=UserResponse, return_as_model=True
|
||
)
|
||
if not updated_user:
|
||
raise HTTPException(status_code=404, detail="用户不存在")
|
||
|
||
# 记录操作日志（计算变更字段）
|
||
changed = {}
|
||
for field in ["username", "nickname", "email", "phone", "role_id"]:
|
||
old_val = getattr(old_user, field, None)
|
||
new_val = update_data.get(field)
|
||
if old_val != new_val:
|
||
changed[field] = {"old": old_val, "new": new_val}
|
||
await create_operation_log(
|
||
db, current_user.id, current_user.username, "update", "user",
|
||
user_id, old_user.username, changed if changed else None,
|
||
get_client_ip(request), get_user_agent(request),
|
||
)
|
||
await db.commit()
|
||
|
||
return updated_user
|
||
|
||
|
||
@router.delete("/{user_id}")
|
||
async def delete_user(
|
||
user_id: int,
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
):
|
||
"""删除用户"""
|
||
# 检查用户是否存在
|
||
result = await db.execute(select(User).filter(User.id == user_id))
|
||
existing_user = result.scalar_one_or_none()
|
||
if not existing_user:
|
||
raise HTTPException(status_code=404, detail="用户不存在")
|
||
elif existing_user.id == current_user.id:
|
||
raise HTTPException(status_code=400, detail="不能删除当前登录用户")
|
||
elif existing_user.role_id == 1:
|
||
raise HTTPException(status_code=400, detail="不能删除管理员用户")
|
||
else:
|
||
# 记录操作日志（删除前记录被删用户信息）
|
||
await create_operation_log(
|
||
db, current_user.id, current_user.username, "delete", "user",
|
||
user_id, existing_user.username,
|
||
{"username": existing_user.username, "nickname": existing_user.nickname},
|
||
get_client_ip(request), get_user_agent(request),
|
||
)
|
||
# 使用FastCRUD删除用户
|
||
deleted = await user_crud.delete(db, id=user_id)
|
||
return {"message": "用户删除成功"}
|
||
|
||
|
||
# ========================================
|
||
# 用户注册
|
||
# ========================================
|
||
|
||
|
||
@router.post("/register/send-code", response_model=PhoneSendCodeResponse)
|
||
async def send_register_sms_code(
|
||
req: PhoneSendCodeRequest,
|
||
db: AsyncSession = Depends(get_db),
|
||
redis_conn: redis.Redis = Depends(get_redis),
|
||
):
|
||
"""发送手机注册验证码"""
|
||
# 检查短信注册开关
|
||
result = await db.execute(select(SystemSetting).filter(SystemSetting.key == "启用手机短信注册"))
|
||
sms_setting = result.scalar_one_or_none()
|
||
if not sms_setting or sms_setting.value != "true":
|
||
raise HTTPException(status_code=403, detail="手机短信注册功能未启用")
|
||
|
||
# 检查手机号是否已被注册
|
||
result = await db.execute(select(User).filter(User.phone == req.phone))
|
||
if result.scalar_one_or_none():
|
||
raise HTTPException(status_code=400, detail="该手机号已被注册")
|
||
|
||
# 频率限制：60秒内不能重复发送
|
||
ttl = await redis_conn.ttl(f"sms_code:{req.phone}")
|
||
if ttl > 240:
|
||
raise HTTPException(status_code=429, detail="验证码已发送，请60秒后再试")
|
||
|
||
# 生成验证码并存入 Redis
|
||
code = generate_sms_code()
|
||
await save_sms_code(redis_conn, req.phone, code)
|
||
|
||
# 发送短信
|
||
await send_sms_code(req.phone, code)
|
||
|
||
return PhoneSendCodeResponse()
|
||
|
||
|
||
@router.post("/register/phone", response_model=RegisterSuccessResponse)
|
||
async def register_by_phone(
|
||
req: PhoneRegisterRequest,
|
||
db: AsyncSession = Depends(get_db),
|
||
redis_conn: redis.Redis = Depends(get_redis),
|
||
):
|
||
"""通过手机号注册"""
|
||
# 检查短信注册开关
|
||
result = await db.execute(select(SystemSetting).filter(SystemSetting.key == "启用手机短信注册"))
|
||
sms_setting = result.scalar_one_or_none()
|
||
if not sms_setting or sms_setting.value != "true":
|
||
raise HTTPException(status_code=403, detail="手机短信注册功能未启用")
|
||
|
||
# 检查用户名是否已存在
|
||
result = await db.execute(select(User).filter(User.username == req.username))
|
||
if result.scalar_one_or_none():
|
||
raise HTTPException(status_code=400, detail="用户名已存在")
|
||
|
||
# 检查手机号是否已被注册
|
||
result = await db.execute(select(User).filter(User.phone == req.phone))
|
||
if result.scalar_one_or_none():
|
||
raise HTTPException(status_code=400, detail="该手机号已被注册")
|
||
|
||
# 验证短信验证码
|
||
stored_code = await get_and_delete_sms_code(redis_conn, req.phone)
|
||
if stored_code is None:
|
||
raise HTTPException(status_code=400, detail="验证码已过期，请重新发送")
|
||
if stored_code != req.code:
|
||
raise HTTPException(status_code=400, detail="验证码错误")
|
||
|
||
# 创建用户
|
||
hashed_pw = get_password_hash(req.password)
|
||
new_user = User(
|
||
username=req.username,
|
||
nickname=req.nickname,
|
||
phone=req.phone,
|
||
password_hash=hashed_pw,
|
||
role_id=2,
|
||
)
|
||
db.add(new_user)
|
||
await db.commit()
|
||
await db.refresh(new_user)
|
||
|
||
return RegisterSuccessResponse(
|
||
id=new_user.id, username=new_user.username, nickname=new_user.nickname
|
||
)
|
||
|
||
|
||
@router.post("/register/email/init", response_model=EmailRegisterInitResponse)
|
||
async def init_email_registration(
|
||
req: EmailRegisterInitRequest,
|
||
db: AsyncSession = Depends(get_db),
|
||
redis_conn: redis.Redis = Depends(get_redis),
|
||
):
|
||
"""初始化邮箱注册（发送验证邮件）"""
|
||
# 检查邮箱注册开关
|
||
result = await db.execute(select(SystemSetting).filter(SystemSetting.key == "启用邮箱注册"))
|
||
email_setting = result.scalar_one_or_none()
|
||
if not email_setting or email_setting.value != "true":
|
||
raise HTTPException(status_code=403, detail="邮箱注册功能未启用")
|
||
|
||
# 检查用户名是否已存在
|
||
result = await db.execute(select(User).filter(User.username == req.username))
|
||
if result.scalar_one_or_none():
|
||
raise HTTPException(status_code=400, detail="用户名已存在")
|
||
|
||
# 检查邮箱是否已被注册
|
||
result = await db.execute(select(User).filter(User.email == req.email))
|
||
if result.scalar_one_or_none():
|
||
raise HTTPException(status_code=400, detail="该邮箱已被注册")
|
||
|
||
# 哈希密码后存储待注册数据
|
||
hashed_pw = get_password_hash(req.password)
|
||
token_id = str(uuid4())
|
||
pending_data = json.dumps({
|
||
"username": req.username,
|
||
"nickname": req.nickname,
|
||
"email": req.email,
|
||
"password_hash": hashed_pw,
|
||
})
|
||
await save_pending_registration(redis_conn, token_id, pending_data)
|
||
|
||
# 生成 JWT 验证令牌
|
||
jwt_token = create_access_token(
|
||
data={"sub": token_id, "type": "email_verify", "email": req.email},
|
||
expires_delta=timedelta(minutes=30),
|
||
)
|
||
|
||
# 发送验证邮件
|
||
await send_verification_email(req.email, jwt_token)
|
||
|
||
return EmailRegisterInitResponse()
|
||
|
||
|
||
@router.get("/verify-email")
|
||
async def verify_email(
|
||
token: str = Query(..., description="验证令牌"),
|
||
db: AsyncSession = Depends(get_db),
|
||
redis_conn: redis.Redis = Depends(get_redis),
|
||
):
|
||
"""验证邮箱并完成注册"""
|
||
# 检查邮箱注册开关
|
||
result = await db.execute(select(SystemSetting).filter(SystemSetting.key == "启用邮箱注册"))
|
||
email_setting = result.scalar_one_or_none()
|
||
if not email_setting or email_setting.value != "true":
|
||
return HTMLResponse("<h2>邮箱注册功能未启用</h2>", status_code=403)
|
||
|
||
# 解析 JWT
|
||
payload = decode_token(token)
|
||
if payload is None:
|
||
return HTMLResponse("<h2>验证链接无效或已过期</h2>", status_code=400)
|
||
if payload.get("type") != "email_verify":
|
||
return HTMLResponse("<h2>无效的验证链接</h2>", status_code=400)
|
||
|
||
token_id = payload.get("sub")
|
||
pending_json = await get_and_delete_pending_registration(redis_conn, token_id)
|
||
if pending_json is None:
|
||
return HTMLResponse("<h2>验证链接已过期或已使用</h2>", status_code=400)
|
||
|
||
data = json.loads(pending_json)
|
||
|
||
# 再次检查用户名和邮箱唯一性
|
||
result = await db.execute(select(User).filter(User.username == data["username"]))
|
||
if result.scalar_one_or_none():
|
||
return HTMLResponse("<h2>该用户名已被注册</h2>", status_code=400)
|
||
|
||
result = await db.execute(select(User).filter(User.email == data["email"]))
|
||
if result.scalar_one_or_none():
|
||
return HTMLResponse("<h2>该邮箱已被注册</h2>", status_code=400)
|
||
|
||
# 创建用户
|
||
new_user = User(
|
||
username=data["username"],
|
||
nickname=data["nickname"],
|
||
email=data["email"],
|
||
password_hash=data["password_hash"],
|
||
role_id=2,
|
||
)
|
||
db.add(new_user)
|
||
await db.commit()
|
||
await db.refresh(new_user)
|
||
|
||
return HTMLResponse(f"""<!DOCTYPE html>
|
||
<html><head><meta charset="utf-8"></head>
|
||
<body style="font-family:Arial,sans-serif;text-align:center;padding:60px 20px;">
|
||
<h1 style="color:#67c23a;">✅ 注册成功</h1>
|
||
<p>欢迎 <b>{new_user.username}</b>，您的账号已创建完成。</p>
|
||
<p>请返回应用使用 <b>用户名、邮箱或手机号</b> 登录。</p>
|
||
</body></html>""")
|
||
|
||
|
||
@router.post("/logout")
|
||
async def logout(
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
redis_conn: redis.Redis = Depends(get_redis),
|
||
):
|
||
"""用户登出"""
|
||
await delete_token_from_redis(redis_conn, current_user.id)
|
||
|
||
# 记录登出日志
|
||
await save_login_log(
|
||
db, current_user.username,
|
||
get_client_ip(request), get_user_agent(request),
|
||
"success", None,
|
||
)
|
||
return {"message": "登出成功"}
|