""" Backup system views for admin to backup the entire system """ import os import zipfile import tempfile from datetime import datetime from django.shortcuts import render, redirect from django.contrib.auth.decorators import login_required from django.contrib import messages from django.http import HttpResponse, FileResponse from django.conf import settings from django.db import connection from django.core.management import call_command from django.utils import timezone from users.decorators import admin_required import sys @login_required @admin_required def backup_system(request): """Create a complete backup of the system""" if request.method == 'POST': try: # Create temporary directory for backup backup_dir = tempfile.mkdtemp() timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') backup_filename = f'branch_system_backup_{timestamp}.zip' backup_path = os.path.join(backup_dir, backup_filename) with zipfile.ZipFile(backup_path, 'w', zipfile.ZIP_DEFLATED) as zipf: # 1. Database backup db_backup_path = os.path.join(backup_dir, 'database_backup.sql') try: # Use Django's dumpdata command with open(db_backup_path, 'w', encoding='utf-8') as f: call_command('dumpdata', '--natural-foreign', '--natural-primary', stdout=f, verbosity=0) zipf.write(db_backup_path, 'database_backup.json') os.remove(db_backup_path) except Exception as e: # Fallback: try SQL dump if using MySQL/PostgreSQL try: db_name = settings.DATABASES['default']['NAME'] db_user = settings.DATABASES['default'].get('USER', '') db_password = settings.DATABASES['default'].get('PASSWORD', '') db_host = settings.DATABASES['default'].get('HOST', 'localhost') # Create SQL dump import subprocess if 'mysql' in settings.DATABASES['default']['ENGINE']: subprocess.run([ 'mysqldump', '-u', db_user, f'-p{db_password}', '-h', db_host, db_name ], stdout=open(db_backup_path, 'w'), stderr=subprocess.PIPE, check=False) elif 'postgresql' in settings.DATABASES['default']['ENGINE']: subprocess.run([ 'pg_dump', '-U', db_user, '-h', db_host, db_name ], stdout=open(db_backup_path, 'w'), stderr=subprocess.PIPE, check=False) if os.path.exists(db_backup_path): zipf.write(db_backup_path, 'database_backup.sql') os.remove(db_backup_path) except: pass # 2. Media files backup media_root = getattr(settings, 'MEDIA_ROOT', None) if media_root and os.path.exists(media_root): for root, dirs, files in os.walk(media_root): for file in files: file_path = os.path.join(root, file) arcname = os.path.relpath(file_path, media_root) zipf.write(file_path, f'media/{arcname}') # 3. Static files backup (if needed) static_root = getattr(settings, 'STATIC_ROOT', None) if static_root and os.path.exists(static_root): for root, dirs, files in os.walk(static_root): for file in files: file_path = os.path.join(root, file) arcname = os.path.relpath(file_path, static_root) zipf.write(file_path, f'static/{arcname}') # 4. Settings and configuration settings_info = { 'backup_date': timezone.now().isoformat(), 'django_version': settings.DJANGO_VERSION if hasattr(settings, 'DJANGO_VERSION') else 'Unknown', 'python_version': sys.version, 'database_engine': settings.DATABASES['default']['ENGINE'], 'database_name': settings.DATABASES['default']['NAME'], } import json settings_path = os.path.join(backup_dir, 'backup_info.json') with open(settings_path, 'w') as f: json.dump(settings_info, f, indent=2) zipf.write(settings_path, 'backup_info.json') os.remove(settings_path) # Optionally save to backup directory for later download backup_storage_dir = os.path.join(settings.MEDIA_ROOT, 'backups') if hasattr(settings, 'MEDIA_ROOT') else None if backup_storage_dir: os.makedirs(backup_storage_dir, exist_ok=True) storage_path = os.path.join(backup_storage_dir, backup_filename) import shutil shutil.copy2(backup_path, storage_path) # Read file content before returning with open(backup_path, 'rb') as f: file_content = f.read() # Clean up temporary directory try: os.remove(backup_path) os.rmdir(backup_dir) except: pass # Return the backup file as download response = HttpResponse(file_content, content_type='application/zip') response['Content-Disposition'] = f'attachment; filename="{backup_filename}"' response['Content-Length'] = len(file_content) messages.success(request, f'Backup created successfully: {backup_filename}') return response except Exception as e: messages.error(request, f'Error creating backup: {str(e)}') return redirect('utils:backup_system') # Get backup history (from media/backups if exists) backup_history = [] backup_dir = os.path.join(settings.MEDIA_ROOT, 'backups') if hasattr(settings, 'MEDIA_ROOT') else None if backup_dir and os.path.exists(backup_dir): try: for file in os.listdir(backup_dir): if file.endswith('.zip') and 'backup' in file.lower(): file_path = os.path.join(backup_dir, file) if os.path.isfile(file_path): file_stat = os.stat(file_path) backup_history.append({ 'filename': file, 'size': file_stat.st_size, 'created': datetime.fromtimestamp(file_stat.st_mtime), 'size_mb': round(file_stat.st_size / (1024 * 1024), 2) }) except: pass backup_history.sort(key=lambda x: x['created'], reverse=True) context = { 'backup_history': backup_history[:10], # Last 10 backups } return render(request, 'utils/backup_system.html', context) @login_required @admin_required def download_backup(request, filename): """Download a specific backup file""" backup_dir = os.path.join(settings.MEDIA_ROOT, 'backups') if hasattr(settings, 'MEDIA_ROOT') else None if not backup_dir or not os.path.exists(backup_dir): messages.error(request, 'Backup directory not found') return redirect('utils:backup_system') file_path = os.path.join(backup_dir, filename) if not os.path.exists(file_path) or not filename.endswith('.zip'): messages.error(request, 'Backup file not found') return redirect('utils:backup_system') with open(file_path, 'rb') as f: file_content = f.read() response = HttpResponse(file_content, content_type='application/zip') response['Content-Disposition'] = f'attachment; filename="{filename}"' response['Content-Length'] = len(file_content) return response