Add admin panel
This commit is contained in:
@@ -1,4 +1,5 @@
|
||||
from typing import Annotated
|
||||
from datetime import datetime
|
||||
|
||||
from fastapi import Depends, HTTPException, status, Header
|
||||
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
|
||||
@@ -8,7 +9,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from app.core.config import settings
|
||||
from app.core.database import get_db
|
||||
from app.core.security import decode_access_token
|
||||
from app.models import User, Participant, Marathon, UserRole, ParticipantRole
|
||||
from app.models import User, Participant, Marathon, UserRole, ParticipantRole, AdminLog, AdminActionType
|
||||
|
||||
security = HTTPBearer()
|
||||
|
||||
@@ -43,6 +44,50 @@ async def get_current_user(
|
||||
detail="User not found",
|
||||
)
|
||||
|
||||
# Check if user is banned
|
||||
if user.is_banned:
|
||||
# Auto-unban if ban expired
|
||||
if user.banned_until and datetime.utcnow() > user.banned_until:
|
||||
# Save ban info for logging before clearing
|
||||
old_ban_reason = user.ban_reason
|
||||
old_banned_until = user.banned_until.isoformat() if user.banned_until else None
|
||||
|
||||
user.is_banned = False
|
||||
user.banned_at = None
|
||||
user.banned_until = None
|
||||
user.banned_by_id = None
|
||||
user.ban_reason = None
|
||||
|
||||
# Log system auto-unban action
|
||||
log = AdminLog(
|
||||
admin_id=None, # System action, no admin
|
||||
action=AdminActionType.USER_AUTO_UNBAN.value,
|
||||
target_type="user",
|
||||
target_id=user.id,
|
||||
details={
|
||||
"nickname": user.nickname,
|
||||
"reason": old_ban_reason,
|
||||
"banned_until": old_banned_until,
|
||||
"system": True,
|
||||
},
|
||||
ip_address=None,
|
||||
)
|
||||
db.add(log)
|
||||
|
||||
await db.commit()
|
||||
await db.refresh(user)
|
||||
else:
|
||||
# Still banned - return ban info in error
|
||||
ban_info = {
|
||||
"banned_at": user.banned_at.isoformat() if user.banned_at else None,
|
||||
"banned_until": user.banned_until.isoformat() if user.banned_until else None,
|
||||
"reason": user.ban_reason,
|
||||
}
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail=ban_info,
|
||||
)
|
||||
|
||||
return user
|
||||
|
||||
|
||||
@@ -56,6 +101,21 @@ def require_admin(user: User) -> User:
|
||||
return user
|
||||
|
||||
|
||||
def require_admin_with_2fa(user: User) -> User:
|
||||
"""Check if user is admin with Telegram linked (2FA enabled)"""
|
||||
if not user.is_admin:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Admin access required",
|
||||
)
|
||||
if not user.telegram_id:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Для доступа к админ-панели необходимо привязать Telegram в профиле",
|
||||
)
|
||||
return user
|
||||
|
||||
|
||||
async def get_participant(
|
||||
db: AsyncSession,
|
||||
user_id: int,
|
||||
|
||||
Reference in New Issue
Block a user