165 lines
5.3 KiB
Python
165 lines
5.3 KiB
Python
"""
|
||
管理员 API：用户管理
|
||
"""
|
||
from fastapi import APIRouter, HTTPException, Depends, status
|
||
from pydantic import BaseModel
|
||
from typing import Optional, List, Any, cast
|
||
from datetime import datetime, timezone, timedelta
|
||
from app.core.deps import get_current_admin
|
||
from app.core.response import success_response
|
||
from app.repositories.sessions import delete_sessions
|
||
from app.repositories.users import get_user_by_id, list_users as list_users_repo, update_user
|
||
from loguru import logger
|
||
|
||
router = APIRouter(prefix="/api/admin", tags=["管理"])
|
||
|
||
|
||
class UserListItem(BaseModel):
|
||
id: str
|
||
phone: str
|
||
username: Optional[str]
|
||
role: str
|
||
is_active: bool
|
||
expires_at: Optional[str]
|
||
created_at: str
|
||
|
||
|
||
class ActivateRequest(BaseModel):
|
||
expires_days: Optional[int] = None # 授权天数，None 表示永久
|
||
|
||
|
||
@router.get("/users")
|
||
async def list_users(admin: dict = Depends(get_current_admin)):
|
||
"""获取所有用户列表"""
|
||
try:
|
||
data = list_users_repo()
|
||
return success_response([
|
||
UserListItem(
|
||
id=u["id"],
|
||
phone=u["phone"],
|
||
username=u.get("username"),
|
||
role=u["role"],
|
||
is_active=u["is_active"],
|
||
expires_at=u.get("expires_at"),
|
||
created_at=u["created_at"]
|
||
).model_dump()
|
||
for u in data
|
||
])
|
||
except Exception as e:
|
||
logger.error(f"获取用户列表失败: {e}")
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="获取用户列表失败"
|
||
)
|
||
|
||
|
||
@router.post("/users/{user_id}/activate")
|
||
async def activate_user(
|
||
user_id: str,
|
||
request: ActivateRequest,
|
||
admin: dict = Depends(get_current_admin)
|
||
):
|
||
"""
|
||
激活用户
|
||
|
||
Args:
|
||
user_id: 用户 ID
|
||
request.expires_days: 授权天数 (None 表示永久)
|
||
"""
|
||
try:
|
||
# 计算过期时间
|
||
expires_at = None
|
||
if request.expires_days:
|
||
expires_at = (datetime.now(timezone.utc) + timedelta(days=request.expires_days)).isoformat()
|
||
|
||
result = update_user(user_id, {
|
||
"is_active": True,
|
||
"role": "user",
|
||
"expires_at": expires_at
|
||
})
|
||
|
||
if not result:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail="用户不存在"
|
||
)
|
||
|
||
logger.info(f"管理员 {admin['phone']} 激活用户 {user_id}, 有效期: {request.expires_days or '永久'} 天")
|
||
|
||
return success_response(message=f"用户已激活，有效期: {request.expires_days or '永久'} 天")
|
||
except HTTPException:
|
||
raise
|
||
except Exception as e:
|
||
logger.error(f"激活用户失败: {e}")
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="激活用户失败"
|
||
)
|
||
|
||
|
||
@router.post("/users/{user_id}/deactivate")
|
||
async def deactivate_user(
|
||
user_id: str,
|
||
admin: dict = Depends(get_current_admin)
|
||
):
|
||
"""停用用户"""
|
||
try:
|
||
# 不能停用管理员
|
||
user = cast(dict[str, Any], get_user_by_id(user_id) or {})
|
||
if user.get("role") == "admin":
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="不能停用管理员账号"
|
||
)
|
||
|
||
update_user(user_id, {"is_active": False})
|
||
delete_sessions(user_id)
|
||
|
||
logger.info(f"管理员 {admin['phone']} 停用用户 {user_id}")
|
||
|
||
return success_response(message="用户已停用")
|
||
except HTTPException:
|
||
raise
|
||
except Exception as e:
|
||
logger.error(f"停用用户失败: {e}")
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="停用用户失败"
|
||
)
|
||
|
||
|
||
@router.post("/users/{user_id}/extend")
|
||
async def extend_user(
|
||
user_id: str,
|
||
request: ActivateRequest,
|
||
admin: dict = Depends(get_current_admin)
|
||
):
|
||
"""延长用户授权期限"""
|
||
try:
|
||
if not request.expires_days:
|
||
# 设为永久
|
||
expires_at = None
|
||
else:
|
||
# 获取当前过期时间
|
||
user = cast(dict[str, Any], get_user_by_id(user_id) or {})
|
||
|
||
if user and user.get("expires_at"):
|
||
current_expires = datetime.fromisoformat(user["expires_at"].replace("Z", "+00:00"))
|
||
base_time = max(current_expires, datetime.now(timezone.utc))
|
||
else:
|
||
base_time = datetime.now(timezone.utc)
|
||
|
||
expires_at = (base_time + timedelta(days=request.expires_days)).isoformat()
|
||
|
||
update_user(user_id, {"expires_at": expires_at})
|
||
|
||
logger.info(f"管理员 {admin['phone']} 延长用户 {user_id} 授权 {request.expires_days or '永久'} 天")
|
||
|
||
return success_response(message=f"授权已延长 {request.expires_days or '永久'} 天")
|
||
except Exception as e:
|
||
logger.error(f"延长授权失败: {e}")
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="延长授权失败"
|
||
)
|