av/pet-project-server
1
0
Fork 0
Code Pull Requests Actions Activity

Backup apps from /mnt/applications

Browse Source
This commit is contained in:
Anton Vakhrushev
2025-12-07 17:16:43 +03:00
parent 306d4bf8d0
commit 8dfd061991
1 changed files with 39 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

79
files/backups/backup-all.py
View File

@@ -14,6 +14,8 @@ from pathlib import Path
from typing import List, Tuple, Optional from typing import List, Tuple, Optional
import requests import requests
import configparser import configparser
import itertools
# Configure logging # Configure logging
logging.basicConfig( logging.basicConfig(
@@ -45,32 +47,31 @@ class BackupManager:
self.warnings = [] self.warnings = []
self.successful_backups = [] self.successful_backups = []
def get_home_directories(self) -> List[Tuple[str, str]]: def get_application_directories(self) -> List[Tuple[str, str]]:
"""Get all home directories and their owners""" """Get all home directories and their owners"""
home_dirs = [] app_dirs = []
applications_path = Path("/mnt/applications")
home_path = Path("/home") home_path = Path("/home")
if not home_path.exists(): source_dirs = itertools.chain(applications_path.iterdir(), home_path.iterdir())
logger.error("/home directory does not exist")
return home_dirs
for user_dir in home_path.iterdir(): for app_dir in source_dirs:
if user_dir.is_dir(): if app_dir.is_dir():
try: try:
# Get the owner of the directory # Get the owner of the directory
stat_info = user_dir.stat() stat_info = app_dir.stat()
owner = pwd.getpwuid(stat_info.st_uid).pw_name owner = pwd.getpwuid(stat_info.st_uid).pw_name
home_dirs.append((str(user_dir), owner)) app_dirs.append((str(app_dir), owner))
except (KeyError, OSError) as e: except (KeyError, OSError) as e:
logger.warning(f"Could not get owner for {user_dir}: {e}") logger.warning(f"Could not get owner for {app_dir}: {e}")
return home_dirs return app_dirs
def find_backup_script(self, home_dir: str) -> Optional[str]: def find_backup_script(self, app_dir: str) -> Optional[str]:
"""Find backup script in user's home directory""" """Find backup script in user's home directory"""
possible_scripts = [ possible_scripts = [
os.path.join(home_dir, "backup.sh"), os.path.join(app_dir, "backup.sh"),
os.path.join(home_dir, "backup"), os.path.join(app_dir, "backup"),
] ]
for script_path in possible_scripts: for script_path in possible_scripts:
@@ -85,16 +86,20 @@ class BackupManager:
return None return None
def run_user_backup(self, script_path: str, username: str) -> bool: def run_app_backup(self, script_path: str, app_dir: str, username: str) -> bool:
"""Run backup script as the specified user""" """Run backup script as the specified user"""
try: try:
logger.info(f"Running backup script {script_path} as user {username}") logger.info(f"Running backup script {script_path} (user {username})")
# Use su to run the script as the user # Use su to run the script as the user
cmd = ["su", "--login", username, "--command", script_path] cmd = ["su", "--login", username, "--command", script_path]
result = subprocess.run( result = subprocess.run(
cmd, capture_output=True, text=True, timeout=3600 # 1 hour timeout cmd,
cwd=app_dir,
capture_output=True,
text=True,
timeout=3600, # 1 hour timeout
) )
if result.returncode == 0: if result.returncode == 0:
@@ -102,31 +107,31 @@ class BackupManager:
self.successful_backups.append(username) self.successful_backups.append(username)
return True return True
else: else:
error_msg = f"Backup script for {username} failed with return code {result.returncode}" error_msg = f"Backup script {script_path} failed with return code {result.returncode}"
if result.stderr: if result.stderr:
error_msg += f": {result.stderr}" error_msg += f": {result.stderr}"
logger.error(error_msg) logger.error(error_msg)
self.errors.append(f"User {username}: {error_msg}") self.errors.append(f"App {username}: {error_msg}")
return False return False
except subprocess.TimeoutExpired: except subprocess.TimeoutExpired:
error_msg = f"Backup script for {username} timed out" error_msg = f"Backup script {script_path} timed out"
logger.error(error_msg) logger.error(error_msg)
self.errors.append(f"User {username}: {error_msg}") self.errors.append(f"App {username}: {error_msg}")
return False return False
except Exception as e: except Exception as e:
error_msg = f"Failed to run backup script for {username}: {str(e)}" error_msg = f"Failed to run backup script {script_path}: {str(e)}"
logger.error(error_msg) logger.error(error_msg)
self.errors.append(f"User {username}: {error_msg}") self.errors.append(f"App {username}: {error_msg}")
return False return False
def get_backup_directories(self) -> List[str]: def get_backup_directories(self) -> List[str]:
"""Get all backup directories that exist""" """Get all backup directories that exist"""
backup_dirs = [] backup_dirs = []
home_dirs = self.get_home_directories() app_dirs = self.get_application_directories()
for home_dir, _ in home_dirs: for app_dir, _ in app_dirs:
backup_path = os.path.join(home_dir, "backups") backup_path = os.path.join(app_dir, "backups")
if os.path.exists(backup_path) and os.path.isdir(backup_path): if os.path.exists(backup_path) and os.path.isdir(backup_path):
backup_dirs.append(backup_path) backup_dirs.append(backup_path)
@@ -260,26 +265,25 @@ class BackupManager:
logger.info("Starting backup process") logger.info("Starting backup process")
# Get all home directories # Get all home directories
home_dirs = self.get_home_directories() app_dirs = self.get_application_directories()
logger.info(f"Found {len(home_dirs)} home directories") logger.info(f"Found {len(app_dirs)} application directories")
# Process each user's backup # Process each user's backup
for home_dir, username in home_dirs: for app_dir, username in app_dirs:
logger.info(f"Processing backup for user: {username} ({home_dir})") logger.info(f"Processing backup for app: {app_dir} (user {username})")
# Find backup script # Find backup script
backup_script = self.find_backup_script(home_dir) backup_script = self.find_backup_script(app_dir)
if backup_script is None: if backup_script is None:
warning_msg = ( warning_msg = (
f"No backup script found for user {username} in {home_dir}" f"No backup script found for app: {app_dir} (user {username})"
) )
logger.warning(warning_msg) logger.warning(warning_msg)
self.warnings.append(warning_msg) self.warnings.append(warning_msg)
continue continue
# Run backup script self.run_app_backup(backup_script, app_dir, username)
self.run_user_backup(backup_script, username)
# Get backup directories # Get backup directories
backup_dirs = self.get_backup_directories() backup_dirs = self.get_backup_directories()
@@ -308,16 +312,11 @@ class BackupManager:
def main(): def main():
"""Main entry point"""
try: try:
backup_manager = BackupManager() backup_manager = BackupManager()
success = backup_manager.run_backup_process() success = backup_manager.run_backup_process()
if not success:
if success:
sys.exit(0)
else:
sys.exit(1) sys.exit(1)
except KeyboardInterrupt: except KeyboardInterrupt:
logger.info("Backup process interrupted by user") logger.info("Backup process interrupted by user")
sys.exit(130) sys.exit(130)