diff --git a/src/helpers.py b/src/helpers.py index 16dfa10..5ff7e3d 100644 --- a/src/helpers.py +++ b/src/helpers.py @@ -5,6 +5,11 @@ from datetime import datetime from subprocess import run import os import typer +from typing import Optional + +# Constants +JSON_TARGETS = Path(__file__).parent / "targets.json" +EXCLUDES_PATH = Path(__file__).parent / "excludes.txt" # Default exclude patterns for backups default_excludes = [ @@ -48,102 +53,191 @@ default_excludes = [ def ask_repokey() -> None: + """ + Prompt user for Borg repository key and set it as environment variable. + """ repokey = typer.prompt("Enter repository key: ", hide_input=True) os.environ["BORG_PASSPHRASE"] = repokey def get_targets() -> list[BackupItem]: - JSON_TARGETS = Path(__file__).parent / "targets.json" - with open(JSON_TARGETS, "r") as f: - targets = json.load(f) - return [BackupItem(**target) for target in targets] + """ + Load and parse backup targets from the JSON configuration file. + + Returns: + List of BackupItem objects representing the targets. + Empty list if the file doesn't exist or contains invalid data. + """ + try: + with open(JSON_TARGETS, "r") as f: + targets = json.load(f) + return [BackupItem(**target) for target in targets] + except (json.JSONDecodeError, FileNotFoundError) as e: + typer.echo(f"Error loading targets: {e}") + return [] -def add_target(target: BackupItem) -> None: - # Get current targets - targets = get_targets() +def add_target(target: BackupItem) -> bool: + """ + Add a new backup target to the configuration. + + Args: + target: BackupItem object to add + + Returns: + bool: True if successful, False otherwise + """ + try: + # Get current targets + targets = get_targets() - # Add new target - targets.append(target) + # Add new target + targets.append(target) - # Save targets - write_targets(targets) + # Save targets + write_targets(targets) + return True + except Exception as e: + typer.echo(f"Error adding target: {e}") + return False -def write_targets(targets: list[BackupItem]) -> None: - """Save targets to JSON file""" - JSON_TARGETS = Path(__file__).parent / "targets.json" - with open(JSON_TARGETS, "w") as f: - json.dump( - [target.json_serializable for target in targets], - f, - ensure_ascii=False, - indent=2, - ) +def write_targets(targets: list[BackupItem]) -> bool: + """ + Save targets to JSON file. + + Args: + targets: List of BackupItem objects to save + + Returns: + bool: True if successful, False otherwise + """ + try: + with open(JSON_TARGETS, "w") as f: + json.dump( + [target.json_serializable for target in targets], + f, + ensure_ascii=False, + indent=2, + ) + return True + except Exception as e: + typer.echo(f"Error writing targets: {e}") + return False -def set_targets(targets: list[BackupItem]) -> None: - JSON_TARGETS = Path(__file__).parent / "targets.json" - with open(JSON_TARGETS, "w") as f: - json.dump( - [target.json_serializable for target in targets], - f, - ensure_ascii=False, - indent=2, - ) +# The set_targets function has been removed as it was redundant with write_targets -def prune_backups(targets: list[BackupItem]) -> None: +def prune_backups(targets: list[BackupItem], daily: int = 1, weekly: int = 2, + monthly: int = 3, yearly: int = 1) -> list[dict]: + """ + Prune old backups according to retention policy and compact the repository. + + Args: + targets: List of backup targets to prune + daily: Number of daily backups to keep + weekly: Number of weekly backups to keep + monthly: Number of monthly backups to keep + yearly: Number of yearly backups to keep + + Returns: + List of dictionaries containing results for each target + """ + results = [] + for target in targets: if not target.enabled: continue + + result = {"target": target.path, "success": False, "message": ""} + + try: + cmd = [ + "borg", + "prune", + "--keep-daily", str(daily), + "--keep-weekly", str(weekly), + "--keep-monthly", str(monthly), + "--keep-yearly", str(yearly), + target.target, + ] + print("Pruning target:", target.path) + prune_result = run(cmd, capture_output=True, text=True) + + if prune_result.returncode != 0: + result["message"] = f"Prune failed: {prune_result.stderr}" + results.append(result) + continue + + print("Pruned target:", target.path) + print("Compacting cache") + compact_result = run(["borg", "compact", target.target], capture_output=True, text=True) + + if compact_result.returncode != 0: + result["message"] = f"Compact failed: {compact_result.stderr}" + results.append(result) + continue + + print("Compacted cache") + result["success"] = True + result["message"] = "Pruned and compacted successfully" + + except Exception as e: + result["message"] = f"Error during pruning: {str(e)}" + + results.append(result) + + return results + + +def run_backup(target: BackupItem) -> Optional[ResultInfo]: + """ + Execute backup for the specified target using borg. + + Args: + target: The backup target configuration + + Returns: + ResultInfo object with backup statistics or None if backup failed + + Raises: + Exception: If the backup process fails + """ + date_str = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") + + try: + # Build borg command with base options cmd = [ "borg", - "prune", - "--keep-daily", - "1", - "--keep-weekly", - "2", - "--keep-monthly", - "3", - "--keep-yearly", - "1", - target.target, + "create", + "-C", + f"{target.compression.value},{target.comp_ratio}", + "--exclude-from", + EXCLUDES_PATH, + "--stats", + "--json", + f"{target.target}::{date_str}", + target.path, ] - print("Pruning target:", target.path) - run(cmd, capture_output=True) - print("Pruned target:", target.path) - print("Compacting cache") - run(["borg", "compact", target.target], capture_output=True) - print("Compacted cache") + # Add exclude option only if there are excludes defined + if target.excludes: # This is true if the list is not empty + cmd.insert(3, "--exclude") + cmd.insert(4, ",".join(target.excludes)) + + print("Running target:", target.path) + result = run(cmd, capture_output=True) + + if result.returncode != 0: + print("Result:", result.stderr.decode("utf-8")) + raise Exception(result.stderr.decode("utf-8")) + + result_info = ResultInfo(**json.loads(result.stdout.decode("utf-8"))) + prune_backups([target]) -def run_backup(target: BackupItem) -> None: - date_str = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") - # Build borg command with base options - cmd = [ - "borg", - "create", - "-C", - f"{target.compression.value},{target.comp_ratio}", - "--exclude-from", - "/home/esa/bin/backupper/src/excludes.txt", - "--stats", - "--json", - f"{target.target}::{date_str}", - target.path, - ] - - # Add exclude option only if there are excludes defined - if target.excludes: # This is true if the list is not empty - cmd.insert(3, "--exclude") - cmd.insert(4, ",".join(target.excludes)) - print("Running target:", target.path) - result = run(cmd, capture_output=True) - if result.returncode != 0: - print("Result:", result.stderr.decode("utf-8")) - raise Exception(result.stderr.decode("utf-8")) - result_info = ResultInfo(**json.loads(result.stdout.decode("utf-8"))) - prune_backups([target]) - - return result_info + return result_info + + except Exception as e: + print(f"Backup failed: {str(e)}") + return None