136 lines
4.0 KiB
Python
136 lines
4.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Cron Job Inspector - Lists and displays Hermes cron jobs with their status.
|
|
Usage: python3 cron_job_inspector.py [--json-path PATH] [--status]
|
|
"""
|
|
import json
|
|
import sys
|
|
import argparse
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
|
|
def load_jobs(path: str = "/root/.hermes/cron/jobs.json") -> list:
|
|
"""Load cron jobs from JSON file."""
|
|
try:
|
|
with open(path, 'r') as f:
|
|
data = json.load(f)
|
|
return data.get('jobs', [])
|
|
except FileNotFoundError:
|
|
print(f"Error: Jobs file not found at {path}", file=sys.stderr)
|
|
return []
|
|
except json.JSONDecodeError as e:
|
|
print(f"Error: Invalid JSON in jobs file: {e}", file=sys.stderr)
|
|
return []
|
|
|
|
|
|
def format_schedule(job: dict) -> str:
|
|
"""Format schedule info for display."""
|
|
schedule = job.get('schedule', {})
|
|
if isinstance(schedule, dict):
|
|
return schedule.get('display', str(schedule))
|
|
return str(schedule)
|
|
|
|
|
|
def format_status(job: dict) -> str:
|
|
"""Format status info for display."""
|
|
state = job.get('state', 'unknown')
|
|
enabled = job.get('enabled', False)
|
|
enabled_str = 'ENABLED' if enabled else 'DISABLED'
|
|
|
|
if state == 'paused':
|
|
return f"{enabled_str} (paused)"
|
|
|
|
last_error = job.get('last_error')
|
|
if last_error:
|
|
return f"{enabled_str} (last error: {last_error[:60]}...)"
|
|
|
|
return enabled_str
|
|
|
|
|
|
def format_last_run(job: dict) -> str:
|
|
"""Format last run info."""
|
|
last_run = job.get('last_run_at')
|
|
status = job.get('last_status', 'unknown')
|
|
|
|
if last_run:
|
|
try:
|
|
dt = datetime.fromisoformat(last_run.replace('Z', '+00:00'))
|
|
return f"{dt.strftime('%Y-%m-%d %H:%M:%S')} ({status})"
|
|
except:
|
|
return str(last_run)
|
|
return "never"
|
|
|
|
|
|
def format_repeat(job: dict) -> str:
|
|
"""Format repeat count if applicable."""
|
|
repeat = job.get('repeat', {})
|
|
if isinstance(repeat, dict):
|
|
completed = repeat.get('completed', 0)
|
|
total = repeat.get('times')
|
|
if total:
|
|
return f"{completed}/{total} runs"
|
|
return f"{completed} runs"
|
|
return ""
|
|
|
|
|
|
def display_jobs(jobs: list, show_json: bool = False):
|
|
"""Display jobs in formatted output."""
|
|
if not jobs:
|
|
print("No cron jobs found.")
|
|
return
|
|
|
|
for job in jobs:
|
|
job_id = job.get('id', 'unknown')[:12]
|
|
name = job.get('name', 'unnamed')
|
|
schedule = format_schedule(job)
|
|
status = format_status(job)
|
|
last_run = format_last_run(job)
|
|
repeat = format_repeat(job)
|
|
deliver = job.get('deliver', '')
|
|
model = job.get('model', '')
|
|
script = job.get('script', '')
|
|
|
|
print(f"\n{'='*60}")
|
|
print(f"ID: {job_id}")
|
|
print(f"Name: {name}")
|
|
print(f"Schedule: {schedule}")
|
|
print(f"Status: {status}")
|
|
if deliver:
|
|
print(f"Deliver: {deliver}")
|
|
if model:
|
|
print(f"Model: {model}")
|
|
if last_run != "never":
|
|
print(f"Last run: {last_run}")
|
|
if repeat:
|
|
print(f"Runs: {repeat}")
|
|
if script:
|
|
print(f"Script: {script}")
|
|
|
|
if show_json:
|
|
print(f"\nFull data: {json.dumps(job, indent=2)}")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description='List Hermes cron jobs')
|
|
parser.add_argument('--json-path', default='/root/.hermes/cron/jobs.json',
|
|
help='Path to jobs.json file')
|
|
parser.add_argument('--status', action='store_true',
|
|
help='Show only job IDs and status (quick view)')
|
|
parser.add_argument('--json', action='store_true',
|
|
help='Show full JSON for each job')
|
|
args = parser.parse_args()
|
|
|
|
jobs = load_jobs(args.json_path)
|
|
|
|
if args.status:
|
|
for job in jobs:
|
|
name = job.get('name', 'unnamed')
|
|
status = format_status(job)
|
|
print(f"{name}: {status}")
|
|
else:
|
|
display_jobs(jobs, show_json=args.json)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main() |