From 40063cfe6bc6d1161911ef9a2e92415d59bee928 Mon Sep 17 00:00:00 2001 From: Menno van Leeuwen Date: Tue, 23 Sep 2025 13:53:29 +0000 Subject: [PATCH] Refactor for consistent string quoting and formatting --- bin/actions/service.py | 22 +++++++++------------- bin/actions/source.py | 19 +++++++++++++++---- bin/actions/timers.py | 21 +++++++++++++++------ bin/actions/update.py | 29 ++++++++++++++++++++--------- 4 files changed, 59 insertions(+), 32 deletions(-) diff --git a/bin/actions/service.py b/bin/actions/service.py index 2877c8f..cff8488 100755 --- a/bin/actions/service.py +++ b/bin/actions/service.py @@ -259,31 +259,25 @@ def get_systemd_timer_status(timer_name): """Check if a systemd timer is active and enabled, and get next run time""" # Check if timer is active (running/waiting) active_result = subprocess.run( - ["sudo", "systemctl", "is-active", timer_name], - capture_output=True, - text=True + ["sudo", "systemctl", "is-active", timer_name], capture_output=True, text=True ) # Check if timer is enabled (will start on boot) enabled_result = subprocess.run( - ["sudo", "systemctl", "is-enabled", timer_name], - capture_output=True, - text=True + ["sudo", "systemctl", "is-enabled", timer_name], capture_output=True, text=True ) # Check corresponding service status - service_name = timer_name.replace('.timer', '.service') + service_name = timer_name.replace(".timer", ".service") service_result = subprocess.run( - ["sudo", "systemctl", "is-active", service_name], - capture_output=True, - text=True + ["sudo", "systemctl", "is-active", service_name], capture_output=True, text=True ) # Get next run time list_result = subprocess.run( ["sudo", "systemctl", "list-timers", timer_name, "--no-legend"], capture_output=True, - text=True + text=True, ) is_active = active_result.returncode == 0 @@ -337,8 +331,10 @@ def cmd_list(args): systemd_timers = ["borg-backup.timer", "borg-local-sync.timer", "dynamic-dns.timer"] for timer in systemd_timers: - is_active, is_enabled, next_run, service_status = get_systemd_timer_status(timer) - service_name = timer.replace('.timer', '') + is_active, is_enabled, next_run, service_status = get_systemd_timer_status( + timer + ) + service_name = timer.replace(".timer", "") if service_status in ["activating", "active"]: # Service is currently running diff --git a/bin/actions/source.py b/bin/actions/source.py index d34327b..a173837 100755 --- a/bin/actions/source.py +++ b/bin/actions/source.py @@ -5,23 +5,33 @@ import sys import subprocess # Add the helpers directory to the path -sys.path.append(os.path.join(os.path.dirname(__file__), '..', 'helpers')) +sys.path.append(os.path.join(os.path.dirname(__file__), "..", "helpers")) from functions import printfe + def get_borg_passphrase(): """Get Borg passphrase from 1Password""" try: result = subprocess.run( - ["op", "item", "get", "Borg Backup", "--vault=Dotfiles", "--fields=password", "--reveal"], + [ + "op", + "item", + "get", + "Borg Backup", + "--vault=Dotfiles", + "--fields=password", + "--reveal", + ], capture_output=True, text=True, - check=True + check=True, ) return result.stdout.strip() except subprocess.CalledProcessError: printfe("red", "Error: Failed to retrieve Borg passphrase from 1Password") return None + def main(): """Generate export commands for Borg environment variables""" args = sys.argv[1:] if len(sys.argv) > 1 else [] @@ -38,7 +48,7 @@ def main(): f'export BORG_CACHE_DIR="/home/menno/.config/borg/cache"', f'export BORG_CONFIG_DIR="/home/menno/.config/borg/config"', f'export BORG_SECURITY_DIR="/home/menno/.config/borg/security"', - f'export BORG_KEYS_DIR="/home/menno/.config/borg/keys"' + f'export BORG_KEYS_DIR="/home/menno/.config/borg/keys"', ] # Check if we're being eval'd (no arguments and stdout is a pipe) @@ -77,5 +87,6 @@ def main(): return 0 + if __name__ == "__main__": sys.exit(main()) diff --git a/bin/actions/timers.py b/bin/actions/timers.py index ec76b36..b48543b 100755 --- a/bin/actions/timers.py +++ b/bin/actions/timers.py @@ -5,18 +5,22 @@ import subprocess import sys # Add the helpers directory to the path -sys.path.append(os.path.join(os.path.dirname(__file__), '..', 'helpers')) +sys.path.append(os.path.join(os.path.dirname(__file__), "..", "helpers")) from functions import printfe + def run_command(cmd, capture_output=True): """Run a command and return the result""" try: - result = subprocess.run(cmd, shell=True, capture_output=capture_output, text=True) + result = subprocess.run( + cmd, shell=True, capture_output=capture_output, text=True + ) return result except Exception as e: printfe("red", f"Error running command: {e}") return None + def show_timer_status(timer_name, system_level=True): """Show concise status for a specific timer""" cmd_prefix = "sudo systemctl" if system_level else "systemctl --user" @@ -24,10 +28,12 @@ def show_timer_status(timer_name, system_level=True): # Get timer status status_cmd = f"{cmd_prefix} is-active {timer_name}" status_result = run_command(status_cmd) - timer_status = "active" if status_result and status_result.returncode == 0 else "inactive" + timer_status = ( + "active" if status_result and status_result.returncode == 0 else "inactive" + ) # Get corresponding service status - service_name = timer_name.replace('.timer', '.service') + service_name = timer_name.replace(".timer", ".service") service_cmd = f"{cmd_prefix} is-active {service_name}" service_result = run_command(service_cmd) service_status = service_result.stdout.strip() if service_result else "unknown" @@ -43,7 +49,7 @@ def show_timer_status(timer_name, system_level=True): next_run = f"{parts[0]} {parts[1]} {parts[2]} ({parts[3]})" # Format output based on service status - service_short = service_name.replace('.service', '') + service_short = service_name.replace(".service", "") if service_status in ["activating", "active"]: # Service is currently running @@ -63,6 +69,7 @@ def show_timer_status(timer_name, system_level=True): printfe(status_color, f"{symbol} {service_short:<12} {status_text}") + def show_examples(): """Show example commands for checking services and logs""" printfe("cyan", "=== Useful Commands ===") @@ -92,6 +99,7 @@ def show_examples(): print(" sudo systemctl list-timers") print() + def main(): """Main timers action""" args = sys.argv[1:] if len(sys.argv) > 1 else [] @@ -103,7 +111,7 @@ def main(): timers = [ ("borg-backup.timer", True), ("borg-local-sync.timer", True), - ("dynamic-dns.timer", True) + ("dynamic-dns.timer", True), ] for timer_name, system_level in timers: @@ -118,5 +126,6 @@ def main(): return 0 + if __name__ == "__main__": sys.exit(main()) diff --git a/bin/actions/update.py b/bin/actions/update.py index af7ff30..de2e936 100755 --- a/bin/actions/update.py +++ b/bin/actions/update.py @@ -230,7 +230,10 @@ def get_sudo_password_from_1password(username, hostname): printfe("red", f"Failed to fetch password from 1Password: {e.stderr.strip()}") return None except FileNotFoundError: - printfe("red", "Error: 'op' command not found. Please ensure 1Password CLI is installed and in your PATH.") + printfe( + "red", + "Error: 'op' command not found. Please ensure 1Password CLI is installed and in your PATH.", + ) return None except Exception as e: printfe("red", f"An unexpected error occurred while fetching password: {e}") @@ -247,13 +250,11 @@ def main(): "--ansible", "-A", action="store_true", help="Upgrade Ansible packages" ) parser.add_argument( - "--ansible-verbose", + "--ansible-verbose", action="store_true", help="Upgrade Ansible packages with verbose output", ) - parser.add_argument( - "--tags", type=str, help="Run only specific Ansible tags" - ) + parser.add_argument("--tags", type=str, help="Run only specific Ansible tags") parser.add_argument( "--full-speed", "-F", action="store_true", help="Use all available cores" ) @@ -262,7 +263,10 @@ def main(): ) parser.add_argument( - "--skip-check", "-s", action="store_true", help="Skip checking for dotfiles updates" + "--skip-check", + "-s", + action="store_true", + help="Skip checking for dotfiles updates", ) args = parser.parse_args() @@ -290,6 +294,7 @@ def main(): # Set cores and jobs based on full-speed flag if args.full_speed: import multiprocessing + cores = jobs = multiprocessing.cpu_count() else: cores = 8 @@ -385,14 +390,20 @@ def main(): sudo_password = None if not os.isatty(sys.stdin.fileno()): - printfe("yellow", "Warning: Not running in an interactive terminal. Cannot fetch password from 1Password.") + printfe( + "yellow", + "Warning: Not running in an interactive terminal. Cannot fetch password from 1Password.", + ) ansible_cmd.append("--ask-become-pass") else: sudo_password = get_sudo_password_from_1password(username, hostname) if sudo_password: ansible_cmd.extend(["--become-pass-file", "-"]) else: - printfe("yellow", "Could not fetch password from 1Password. Falling back to --ask-become-pass.") + printfe( + "yellow", + "Could not fetch password from 1Password. Falling back to --ask-become-pass.", + ) ansible_cmd.append("--ask-become-pass") if args.tags: @@ -406,7 +417,7 @@ def main(): # Execute the Ansible command, passing password via stdin if available if sudo_password: - result = subprocess.run(ansible_cmd, input=sudo_password.encode('utf-8')) + result = subprocess.run(ansible_cmd, input=sudo_password.encode("utf-8")) else: result = subprocess.run(ansible_cmd)