mirror of
https://github.com/eduardogsilva/routerfleet.git
synced 2025-06-21 01:25:41 +02:00
214 lines
9 KiB
Python
214 lines
9 KiB
Python
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}')
|