Server backup with notification (rsync + rclone + ntfy)

	[[job]] # External drive backup
	src = "/home/bazizi/.config/"
	dst = "/mnt/seagate/.config/"

	[[job]] # Dropbox backup
	src = "/home/bazizi/.config"
	dst = "dbox:.config/"

	#!/usr/bin/env python3

	import argparse
	import os
	import subprocess
	import tomllib
	from typing import List, Dict
	import requests
	import docker

	NTFY_URL = "https://"
	RCLONE_REMOTES = ("dbox:",)
	EXCLUDES_OPTION = "--exclude-from=/path/to/rsync-excludes.txt"
	CONFIG_FILE = "/path/to/backup_jobs.toml"

	def load_jobs(filepath: str) -> List[Dict[str, str]]:
	"""Load backup locations from an external TOML file."""
	if not os.path.exists(filepath):
	print(f"❌ Configuration file not found: {filepath}")
	return []
	try:
	with open(filepath, "rb") as f:
	data = tomllib.load(f)
	# Expects an array of tables under the key [[job]]
	return data.get("job", [])
	except Exception as e:
	print(f"❌ Failed to parse TOML configuration file: {e}")
	return []

	def run_job(src: str, dst: str, dry_run: bool = False) -> tuple[bool, str]:
	if any(dst.startswith(r) for r in RCLONE_REMOTES):
	cmd = ["rclone", "copy", "-v", EXCLUDES_OPTION, src, dst]
	else:
	cmd = [
	"rsync", "-avz", "--mkpath", "--no-links",
	EXCLUDES_OPTION,
	src, dst,
	]

	cmd_str = " ".join(cmd)
	print("Running:", cmd_str)

	if dry_run:
	return True, f"[DRY RUN] Would execute: {cmd_str}"

	try:
	result = subprocess.run(cmd, capture_output=True, text=True, check=True)
	return True, result.stdout + result.stderr
	except subprocess.CalledProcessError as e:
	return False, (e.stdout or "") + (e.stderr or "") + f"\nExit code: {e.returncode}"

	def notify(title: str, body: str, dry_run: bool = False) -> None:
	print(f"\n--- NOTIFICATION ---\nTitle: {title}\n{body}\n--------------------")

	if dry_run:
	print("ℹ️ [DRY RUN] Notification skipped.")
	return

	try:
	requests.post(
	NTFY_URL,
	data=f"From/To job\n\n```\n{body}\n```",
	headers={"Title": title, "Markdown": "yes"},
	timeout=10
	)
	except Exception as e:
	print(f"⚠️ Failed to send ntfy notification: {e}")

	def main() -> None:
	parser = argparse.ArgumentParser(description="Automated Docker-aware backup script.")
	parser.add_argument("--dry-run", action="store_true", help="Simulate backup jobs and docker actions without changing anything.")
	args = parser.parse_args()

	locations = load_jobs(CONFIG_FILE)
	if not locations:
	print("⚠️ No backup locations loaded. Exiting.")
	return

	try:
	client = docker.from_env()
	except Exception as e:
	print(f"❌ Error connecting to Docker daemon. Is Docker running? Details: {e}")
	return

	try:
	running_containers = client.containers.list(filters={"status": "running"})
	except Exception as e:
	print(f"❌ Failed to list containers: {e}")
	return

	container_records = []
	print(f"📋 Found {len(running_containers)} running container(s):")
	for container in running_containers:
	print(f" • {container.name} ({container.id[:12]})")
	container_records.append({
	"id": container.id,
	"name": container.name,
	"container_obj": container
	})

	prefix_msg = "🔍 [DRY RUN] " if args.dry_run else ""
	docker_output = f"{prefix_msg}🛑 Stopping containers..."

	# Stop containers
	for item in container_records:
	try:
	docker_output += f"\n Stopping {item['name']}..."
	if not args.dry_run:
	item['container_obj'].stop(timeout=10)
	except Exception as e:
	docker_output += f" ❌ Failed: {e}"

	try:
	# Run backup jobs
	for config in locations:
	ok, output = run_job(config["src"], config["dst"], dry_run=args.dry_run)
	status = "OK" if ok else "FAILED"
	print(output)
	docker_output += f"\nBackup {status}: {config['src']} -> {config['dst']}"
	finally:
	# Guaranteed Restart of containers
	docker_output += f"\n\n{prefix_msg}🔄 Restarting containers..."
	for item in container_records:
	try:
	docker_output += f"\n Starting {item['name']}..."
	if not args.dry_run:
	item['container_obj'].start()
	except Exception as e:
	docker_output += f" ❌ Failed: {e}"

	docker_output += f"\n\n{prefix_msg}✅ Process completed successfully!"
	notify("Backup Summary", docker_output, dry_run=args.dry_run)

	if __name__ == "__main__":
	main()

参与讨论