admin-scripts/python/sync_from_remotes.py

# /// script
# dependencies = [
#   "resend",
# ]
# ///

import socket
import subprocess
from pathlib import Path

import resend


def send_email(program: str, log: str):
    resend.api_key = Path("/usr/local/etc/resend_api_key.txt").read_text().strip("\n")

    match log:
        case "ok":
            subj = f"[{socket.getfqdn()}] {program} OK ✅"
            msg = f"{program} on {socket.getfqdn()} ran successfully!"
        case "err":
            subj = f"[{socket.getfqdn()}] {program} Error ❌"
            msg = f"There was an error running {program} on {socket.getfqdn()}. Please investigate."
        case _:
            subj = ""
            msg = ""

    params: resend.Emails.SendParams = {
        "from": "Admin <admin@hyperreal.coffee>",
        "to": ["hyperreal@moonshadow.dev"],
        "subject": subj,
        "text": msg,
    }

    email = resend.Emails.send(params)
    print(email)


def sync_from_remotes(src: str, dest: str):
    rsync_cmd = ["rsync", "-avz", "--delete", src, dest]

    try:
        subprocess.run(rsync_cmd, check=True, text=True)
        print(f"Successful sync from {src} to {dest}")
    except subprocess.CalledProcessError as e:
        print(f"Error during sync from {src} to {dest}: {e}")
        send_email("sync_from_remotes", "err")
        exit(1)


if __name__ == "__main__":
    remotes = [
        (
            "root@hyperreal.coffee:/srv/borgbackup/hyperreal/",
            "/mnt/borgbackup/hyperreal",
        ),
        ("root@hyperreal.coffee:/home/mastodon/live/public/system/", "/mnt/mastodata"),
    ]

    for remote in remotes:
        sync_from_remotes(remote[0], remote[1])

    send_email("sync_from_remotes", "ok")