46 lines
1.7 KiB
Python
46 lines
1.7 KiB
Python
import logging
|
|
import shutil
|
|
|
|
from celery import shared_task
|
|
|
|
from core.models import UsernameLoginAuthCheck
|
|
from core.services.authentication import create_authentication_log_message
|
|
from core.utils import dtnow
|
|
from notifications.enums import NotificationImportance, NotificationGroup
|
|
from notifications.models import Notification
|
|
from notifications.services.notification_sender import NotificationService
|
|
|
|
_log = logging.getLogger(__name__)
|
|
|
|
PERCENT_10 = 10
|
|
|
|
|
|
def check_for_users_end_of_timeouts():
|
|
""" Function for getting users with blocked access """
|
|
for blocked_user in UsernameLoginAuthCheck.objects.filter(is_username_auth_blocked=True):
|
|
if dtnow() > blocked_user.username_unlock_time:
|
|
""" Resetting access to authentication for user account """
|
|
blocked_user.failed_login_attempts = 0
|
|
blocked_user.is_username_auth_blocked = False
|
|
_log.info(create_authentication_log_message(
|
|
'settings', f'user [{blocked_user.user.username}] has been unlocked due to the end of block timeout'))
|
|
blocked_user.save()
|
|
|
|
|
|
@shared_task
|
|
def check_blocked_users():
|
|
check_for_users_end_of_timeouts()
|
|
|
|
|
|
def get_disk_usage():
|
|
total, used, free = shutil.disk_usage("/")
|
|
if free / total * 100 < PERCENT_10: # send if less than 10% left
|
|
notification = Notification.objects.create(text='Free disk space is running out',
|
|
importance=NotificationImportance.HIGH)
|
|
NotificationService().send(notification=notification, group=NotificationGroup.NOTIFICATION)
|
|
_log.info(f'Send Notification "Free disk space is running out": total: {total}, used: {used}, free: {free}')
|
|
|
|
|
|
@shared_task()
|
|
def get_disk_usage_task():
|
|
get_disk_usage()
|