Refactor for consistent string quoting and formatting
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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())
|
||||
|
||||
@@ -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())
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user