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}