eduardogsilva.routerfleet/backup/views.py

from django.contrib.auth.decorators import login_required
from django.http import JsonResponse, HttpResponse, FileResponse
from django.shortcuts import render, get_object_or_404, redirect, Http404
from django.contrib import messages

from routerfleet_tools.models import WebadminSettings
from routerlib.backup_functions import perform_backup
from .models import BackupProfile
from .forms import BackupProfileForm
from router_manager.models import Router, BackupSchedule
from backup_data.models import RouterBackup
import difflib
import unicodedata
from routerlib.functions import gen_backup_name, get_router_backup_file_extension
from django.conf import settings
from user_manager.models import UserAcl
from django.utils import timezone


@login_required()
def view_backup_profile_list(request):
    default_backup_profile, _ = BackupProfile.objects.get_or_create(name='default')
    context = {
        'backup_profile_list': BackupProfile.objects.all().order_by('name'),
        'page_title': 'Backup Profiles'
    }
    return render(request, 'backup/backup_profile_list.html', context)


@login_required()
def view_manage_backup_profile(request):
    if not UserAcl.objects.filter(user=request.user).filter(user_level__gte=40).exists():
        return render(request, 'access_denied.html', {'page_title': 'Access Denied'})
    if request.GET.get('uuid'):
        backup_profile = get_object_or_404(BackupProfile, uuid=request.GET.get('uuid'))
        if request.GET.get('action') == 'delete':
            if request.GET.get('confirmation') == 'delete':
                if backup_profile.name == 'default':
                    messages.warning(request, 'Backup profile not deleted|Default profile cannot be deleted')
                    return redirect('backup_profile_list')
                else:
                    if Router.objects.filter(backup_profile=backup_profile).exists():
                        messages.warning(request, 'Backup profile in use|Backup profile is in use and cannot be deleted')
                        return redirect('backup_profile_list')
                    else:
                        backup_profile.delete()
                        messages.success(request, 'Backup profile deleted successfully')
                return redirect('backup_profile_list')
            else:
                messages.warning(request, 'Backup profile not deleted|Invalid confirmation')
                return redirect('backup_profile_list')
    else:
        backup_profile = None

    form = BackupProfileForm(request.POST or None, instance=backup_profile)
    if form.is_valid():
        form.instance.profile_error_information = ''
        form.save()
        BackupSchedule.objects.filter(router__backup_profile=form.instance).delete()
        messages.success(request, 'Backup Profile saved successfully')
        return redirect('backup_profile_list')

    context = {
        'form': form,
        'page_title': 'Manage Backup Profile',
        'instance': backup_profile
    }
    return render(request, 'backup/backup_profile_form.html', context=context)


@login_required()
def view_backup_list(request):
    backup_list = RouterBackup.objects.all().order_by('-created')
    if request.GET.get('type') == 'queue':
        backup_list = backup_list.filter(error=False, success=False).order_by('next_retry')
        view_type = 'queue'
    elif request.GET.get('type') == 'errors':
        backup_list = backup_list.filter(error=True).order_by('-created')
        view_type = 'errors'
    else:
        backup_list = backup_list.filter(success=True).order_by('-created')
        view_type = 'success'

    context = {
        'backup_list': backup_list,
        'page_title': 'Backup List',
        'view_type': view_type
    }
    return render(request, 'backup/backup_list.html', context)


@login_required()
def view_backup_details(request):
    if not UserAcl.objects.filter(user=request.user).filter(user_level__gte=20).exists():
        return render(request, 'access_denied.html', {'page_title': 'Access Denied'})
    backup = get_object_or_404(RouterBackup, uuid=request.GET.get('uuid'))
    if request.GET.get('action') == 'anticipate':
        if not backup.success and not backup.error:
            backup.next_retry = timezone.now()
            backup.save()
            messages.success(request, 'Backup task anticipated|Backup task will be retried shortly')
        else:
            messages.warning(request, 'Backup task not anticipated|Task is already completed')
        return redirect(f'/backup/backup_details/?uuid={backup.uuid}')
    hash_list = [backup.backup_text_hash]
    backup_list = []
    for backup_item in RouterBackup.objects.filter(router=backup.router, success=True).order_by('-created'):
        if backup_item.backup_text_hash and backup_item.backup_text_hash not in hash_list:
            hash_list.append(backup_item.backup_text_hash)
            backup_list.append(backup_item)
    webadmin_settings, _ = WebadminSettings.objects.get_or_create(name='webadmin_settings')
    context = {
        'backup': backup,
        'backup_list': backup_list,
        'page_title': 'Backup Details',
        'webadmin_settings': webadmin_settings,
        'now': timezone.now(),
        '5_minutes_ago': timezone.now() - timezone.timedelta(minutes=5),
    }
    return render(request, 'backup/backup_details.html', context)


def normalize_text(text):
    text = unicodedata.normalize('NFC', text)
    text = text.replace('\r\n', '\n')
    text = text.replace('\r', '')
    text = '\n'.join([line.rstrip() for line in text.splitlines()])
    return text


@login_required()
def view_compare_backups(request):
    if not UserAcl.objects.filter(user=request.user).filter(user_level__gte=20).exists():
        return render(request, 'access_denied.html', {'page_title': 'Access Denied'})
    backup1 = get_object_or_404(RouterBackup, uuid=request.GET.get('uuid'))
    backup2 = get_object_or_404(RouterBackup, uuid=request.GET.get('compare_uuid'))
    if request.GET.get('display') == 'all':
        show_lines = 100000
        show_all = True
    else:
        show_lines = 3
        show_all = False

    diff = difflib.unified_diff(normalize_text(backup1.backup_text).splitlines(keepends=True),
                                normalize_text(backup2.backup_text).splitlines(keepends=True),
                                fromfile=backup1.backup_text_hash[:16] + '...',
                                tofile=backup2.backup_text_hash[:16] + '...',
                                lineterm='', n=show_lines)
    diff_str = '\n'.join(list(diff))

    context = {
        'backup1': backup1,
        'backup2': backup2,
        'diff_str': diff_str,
        'page_title': 'Compare Backups',
        'show_all': show_all
    }

    return render(request, 'backup/compare_backups.html', context)


def view_debug_run_backups(request):
    data = {
        'backup_count': 0,
    }
    if settings.DEBUG:
        for backup in RouterBackup.objects.filter(success=False, error=False):
            data['backup_count'] += 1
            perform_backup(backup)
    else:
        data['error'] = 'Debug mode is not enabled'
    return JsonResponse(data)


@login_required()
def view_backup_download(request):
    if not UserAcl.objects.filter(user=request.user).filter(user_level__gte=20).exists():
        return render(request, 'access_denied.html', {'page_title': 'Access Denied'})
    backup = get_object_or_404(RouterBackup, uuid=request.GET.get('uuid'))
    if request.GET.get('type') == 'text':
        response = HttpResponse(backup.backup_text, content_type='text/plain')
        if backup.backup_text_filename:
            filename = backup.backup_text_filename
        else:
            filename = gen_backup_name(backup)
            filename += f'.missing_backup_name.{get_router_backup_file_extension(backup.router.router_type)["text"]}'
        print(filename)
        response['Content-Disposition'] = f'attachment; filename={filename}'
        return response
    elif request.GET.get('type') == 'binary':
        response = FileResponse(backup.backup_binary, as_attachment=True)
        return response
    else:
        raise Http404


@login_required()
def view_backup_delete(request):
    if not UserAcl.objects.filter(user=request.user).filter(user_level__gte=30).exists():
        return render(request, 'access_denied.html', {'page_title': 'Access Denied'})
    backup = get_object_or_404(RouterBackup, uuid=request.GET.get('uuid'))
    router = backup.router
    redirect_url = f'/router/details/?uuid={backup.router.uuid}'
    if request.GET.get('confirmation') == f'delete{backup.id}':
        backup.delete()
        messages.success(request, 'Backup deleted successfully')
        if router.routerstatus.backup_lock:
            if not RouterBackup.objects.filter(router=router, success=False, error=False).exists():
                router.routerstatus.backup_lock = None
                router.routerstatus.save()
        return redirect(redirect_url)
    else:
        messages.warning(request, 'Backup not deleted|Invalid confirmation')
        return redirect(f'/backup/backup_details/?uuid={backup.uuid}')