from fastapi import APIRouter, Depends, HTTPException
|
from sqlalchemy.orm import Session
|
|
from app.auth import create_access_token, get_current_user, hash_password, verify_password
|
from app.database import get_db
|
from app.login_guard import clear_login_lock, ensure_login_allowed, record_login_failure
|
from app.models import User
|
from app.schemas import ChangePasswordRequest, LoginRequest, TokenResponse, UserMe
|
|
router = APIRouter(prefix="/api/auth", tags=["auth"])
|
|
|
@router.post("/login", response_model=TokenResponse)
|
def login(body: LoginRequest, db: Session = Depends(get_db)):
|
ensure_login_allowed(db, body.username)
|
user = db.query(User).filter(User.username == body.username).first()
|
if not user or not user.is_active or not verify_password(body.password, user.password_hash):
|
record_login_failure(db, body.username)
|
raise HTTPException(401, detail="用户名或密码错误")
|
clear_login_lock(db, body.username)
|
token = create_access_token(user.id, user.role)
|
return TokenResponse(access_token=token)
|
|
|
@router.get("/me", response_model=UserMe)
|
def me(user: User = Depends(get_current_user)):
|
return user
|
|
|
@router.post("/change-password")
|
def change_password(
|
body: ChangePasswordRequest,
|
db: Session = Depends(get_db),
|
user: User = Depends(get_current_user),
|
):
|
if body.old_password == body.new_password:
|
raise HTTPException(400, detail="新密码不能与旧密码相同")
|
if not verify_password(body.old_password, user.password_hash):
|
raise HTTPException(400, detail="原密码不正确")
|
user.password_hash = hash_password(body.new_password)
|
db.commit()
|
return {"ok": True}
|
