| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768 |
- """
- 管理员认证接口
- """
- from fastapi import APIRouter, Depends, HTTPException, Request
- from sqlalchemy.orm import Session
- from pydantic import BaseModel
- from app.database import get_db
- from app.models.admin import Admin
- from app.core.security import verify_password, create_access_token
- from app.core.logger import logger
- from datetime import datetime
- router = APIRouter()
- class LoginRequest(BaseModel):
- """登录请求"""
- username: str
- password: str
- @router.post("/login")
- async def login(
- req: LoginRequest,
- request: Request,
- db: Session = Depends(get_db)
- ):
- """管理员登录"""
- # 查找管理员
- admin = db.query(Admin).filter(Admin.username == req.username).first()
- if not admin:
- raise HTTPException(status_code=401, detail="用户名或密码错误")
-
- # 验证密码
- if not verify_password(req.password, admin.password):
- raise HTTPException(status_code=401, detail="用户名或密码错误")
-
- # 检查状态
- if admin.status != 1:
- raise HTTPException(status_code=403, detail="账号已被禁用")
-
- # 更新登录信息
- client_ip = request.client.host if request.client else None
- admin.last_login_time = datetime.now()
- admin.last_login_ip = client_ip
- admin.updated_at = datetime.now()
- db.commit()
-
- # 生成令牌
- token = create_access_token({"sub": admin.username})
-
- logger.info(f"管理员登录成功: {req.username}")
- return {
- "code": 200,
- "msg": "登录成功",
- "data": {
- "token": token,
- "admin": {
- "id": admin.id,
- "username": admin.username,
- "role": admin.role
- }
- }
- }
- @router.post("/logout")
- async def logout():
- """退出登录"""
- return {"code": 200, "msg": "退出成功"}
|
