205 lines
6.4 KiB
Python
205 lines
6.4 KiB
Python
# 认证路由 - 使用字段编码
|
||
from fastapi import APIRouter, Depends, HTTPException, status, Body
|
||
from fastapi.security import OAuth2PasswordRequestForm
|
||
from sqlalchemy.orm import Session
|
||
from app.core.database import get_db
|
||
from app.core.auth import verify_password, create_access_token, get_password_hash, get_current_user
|
||
from app.models.models import User
|
||
from app.schemas.schemas import Token, UserCreate, UserResponse
|
||
|
||
router = APIRouter(prefix="/api/auth", tags=["认证"])
|
||
|
||
|
||
@router.post("/register", response_model=UserResponse)
|
||
def register(user_data: UserCreate, db: Session = Depends(get_db)):
|
||
"""用户注册"""
|
||
# 验证短信验证码
|
||
if user_data.phone and user_data.verifyCode:
|
||
from app.services.sms import verify_code
|
||
if not verify_code(user_data.phone, user_data.verifyCode):
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="验证码错误或已过期"
|
||
)
|
||
else:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="手机号和验证码为必填项"
|
||
)
|
||
|
||
# 验证密码复杂度（至少8位，包含大写、小写、数字）
|
||
if len(user_data.password) < 8:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="密码至少8位"
|
||
)
|
||
import re
|
||
if not re.search(r'[A-Z]', user_data.password):
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="密码必须包含大写字母"
|
||
)
|
||
if not re.search(r'[a-z]', user_data.password):
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="密码必须包含小写字母"
|
||
)
|
||
if not re.search(r'[0-9]', user_data.password):
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="密码必须包含数字"
|
||
)
|
||
|
||
# 检查用户名是否已存在
|
||
existing_user = db.query(User).filter(User.f01_01_name == user_data.f01_01_name).first()
|
||
if existing_user:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="f01_01_name: 用户名已存在"
|
||
)
|
||
|
||
# 检查邮箱是否已存在
|
||
if user_data.email:
|
||
existing_email = db.query(User).filter(User.email == user_data.email).first()
|
||
if existing_email:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="邮箱已被注册"
|
||
)
|
||
|
||
# 创建用户
|
||
import uuid
|
||
hashed_password = get_password_hash(user_data.password)
|
||
user = User(
|
||
f99_90_id=str(uuid.uuid4()),
|
||
f99_91_user_id=str(uuid.uuid4()), # 生成唯一 user_id
|
||
f01_01_name=user_data.f01_01_name,
|
||
email=user_data.email,
|
||
phone=user_data.phone,
|
||
avatar=user_data.avatar,
|
||
address=user_data.address,
|
||
bio=user_data.bio,
|
||
password=hashed_password,
|
||
role="user"
|
||
)
|
||
|
||
db.add(user)
|
||
db.commit()
|
||
db.refresh(user)
|
||
|
||
return user
|
||
|
||
|
||
@router.post("/login", response_model=Token)
|
||
def login(
|
||
form_data: OAuth2PasswordRequestForm = Depends(),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
"""用户登录"""
|
||
# 查找用户
|
||
user = db.query(User).filter(User.f01_01_name == form_data.username).first()
|
||
if not user:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="E00011: 用户名或密码错误",
|
||
headers={"WWW-Authenticate": "Bearer"},
|
||
)
|
||
|
||
# 验证密码
|
||
if not verify_password(form_data.password, user.password):
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="E00011: 用户名或密码错误",
|
||
headers={"WWW-Authenticate": "Bearer"},
|
||
)
|
||
|
||
# 生成 token
|
||
access_token = create_access_token(data={"sub": user.f99_90_id})
|
||
|
||
return {
|
||
"access_token": access_token,
|
||
"token_type": "bearer"
|
||
}
|
||
|
||
|
||
@router.get("/me", response_model=UserResponse)
|
||
def get_current_user_info(
|
||
current_user: User = Depends(lambda: None)
|
||
):
|
||
"""获取当前用户信息"""
|
||
raise HTTPException(
|
||
status_code=status.HTTP_501_NOT_IMPLEMENTED,
|
||
detail="请使用正确的依赖注入"
|
||
)
|
||
|
||
|
||
@router.post("/change-password")
|
||
def change_password(
|
||
old_password: str = Body(...),
|
||
new_password: str = Body(...),
|
||
current_user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
"""修改当前用户密码"""
|
||
from app.core.auth import verify_password, get_password_hash
|
||
|
||
# 在当前session中重新查询用户
|
||
user = db.query(User).filter(User.f99_90_id == current_user.f99_90_id).first()
|
||
if not user:
|
||
raise HTTPException(status_code=404, detail="用户不存在")
|
||
|
||
# 验证旧密码
|
||
if not verify_password(old_password, user.password):
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="当前密码错误"
|
||
)
|
||
|
||
# 更新密码
|
||
user.password = get_password_hash(new_password)
|
||
db.commit()
|
||
|
||
return {"message": "密码修改成功"}
|
||
|
||
|
||
# ============ 短信验证码接口 ============
|
||
|
||
@router.post("/send-verification-code")
|
||
def send_verification_code(
|
||
phone: str = Body(..., min_length=11, max_length=11),
|
||
purpose: str = Body("register") # register | login | reset_password
|
||
):
|
||
"""发送短信验证码"""
|
||
from app.services.sms import send_verification_code as send_sms
|
||
|
||
# 验证手机号格式
|
||
if not phone.startswith("1") or len(phone) != 11:
|
||
return {"success": False, "message": "手机号格式不正确"}
|
||
|
||
result = send_sms(phone)
|
||
|
||
if result["success"]:
|
||
return {
|
||
"success": True,
|
||
"message": f"验证码已发送到 {phone[:3]}****{phone[7:]}",
|
||
"expire": result.get("expire", 300)
|
||
}
|
||
else:
|
||
return result
|
||
|
||
|
||
@router.post("/verify-code")
|
||
def verify_code(
|
||
phone: str = Body(...),
|
||
code: str = Body(..., min_length=6, max_length=6)
|
||
):
|
||
"""验证短信验证码（仅验证，不执行后续操作）"""
|
||
from app.services.sms import verify_code as check_code
|
||
|
||
is_valid = check_code(phone, code)
|
||
|
||
if is_valid:
|
||
return {"success": True, "message": "验证成功"}
|
||
else:
|
||
return {"success": False, "message": "验证码错误或已过期"}
|