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("
欢迎 {new_user.username},您的账号已创建完成。
请返回应用使用 用户名、邮箱或手机号 登录。
""") @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": "登出成功"}