#!/usr/bin/env python3
"""
Ingest cron jobs into the Brain knowledge base.
Domains:
- Cron jobs: definitions, schedules, output history
"""
import json
import os
import sys
from datetime import datetime
from pathlib import Path
# Load brain system
sys.path.insert(0, os.path.expanduser("~/.hermes/brain"))
from brain import init_brain, log_event
init_brain()
stats = {'ingested': 0, 'errors': 0}
def ingest(name, event_type, searchable_text, tags, source, context=None):
"""Helper to ingest with error tracking."""
try:
ctx = {'_searchable_text': searchable_text, '_source': source}
if context:
ctx.update(context)
log_event(event_type=event_type, context=ctx, tags=tags)
stats['ingested'] += 1
print(f" ✓ {name}")
except Exception as e:
stats['errors'] += 1
print(f" ✗ {name}: {e}")
def ingest_cron_jobs():
"""Ingest cron job definitions and recent output."""
print("\n=== Ingesting Cron Jobs ===")
cron_dir = Path.home() / '.hermes' / 'cron'
jobs_file = cron_dir / 'jobs.json'
if not jobs_file.exists():
print(" No jobs.json found")
return
with open(jobs_file) as f:
jobs_data = json.load(f)
jobs = jobs_data.get('jobs', [])
print(f"Found {len(jobs)} cron jobs")
for job in jobs:
job_id = job.get('id', 'unknown')
job_name = job.get('name', 'Unnamed Job')
schedule = job.get('schedule_display', 'unknown')
status = job.get('state', 'unknown')
enabled = job.get('enabled', False)
created = job.get('created_at', '')
last_run = job.get('last_run_at', '')
next_run = job.get('next_run_at', '')
runs = job.get('repeat', {})
completed = runs.get('completed', 0)
deliver = job.get('deliver', '')
origin = job.get('origin', {})
prompt = job.get('prompt', '')
searchable = (
f"Cron Job: {job_name}\n"
f"ID: {job_id}\n"
f"Schedule: {schedule}\n"
f"Status: {status} (enabled: {enabled})\n"
f"Created: {created}\n"
f"Last Run: {last_run}\n"
f"Next Run: {next_run}\n"
f"Total Runs: {completed}\n"
f"Delivery: {deliver}\n"
f"Origin: {origin.get('platform', '')}:{origin.get('chat_name', '')}\n"
f"\nPrompt:\n{prompt[:500]}"
)
ingest(
name=f"Cron Job: {job_name}",
event_type="cron_job",
searchable_text=searchable,
tags=['cron', 'scheduled', 'automation'],
source=f"cron/jobs.json/{job_id}",
context={
'job_id': job_id,
'job_name': job_name,
'schedule': schedule,
'status': status,
'enabled': enabled,
'completed_runs': completed,
'deliver': deliver,
}
)
# Ingest recent cron outputs (last 10 per job)
print("\nIngesting recent cron outputs...")
output_dir = cron_dir / 'output'
if output_dir.exists():
job_dirs = sorted(d for d in output_dir.iterdir() if d.is_dir())
print(f"Found {len(job_dirs)} job output directories")
for job_dir in job_dirs:
outputs = sorted(job_dir.glob("*.md"), reverse=True)[:10]
for output_file in outputs:
content = output_file.read_text()[:3000]
ingest(
name=f"Cron Output: {output_file.stem}",
event_type="cron_output",
searchable_text=content,
tags=['cron', 'output', 'history'],
source=f"cron/output/{job_dir.name}/{output_file.name}",
context={
'job_id': job_dir.name,
'timestamp': output_file.stem,
'size_bytes': output_file.stat().st_size,
}
)
def run_verification():
"""Test queries across cron domain."""
print("\n=== Verification Queries ===")
from brain import query_events
test_queries = [
"cron job schedule",
"memory maintenance backup",
"backup manager",
"seedvault",
"scheduled automation",
]
for query in test_queries:
results = query_events(full_text_search=query, limit=3)
count = len(results)
status = "✓" if count > 0 else "✗"
if results:
top = results[0]
detail = f"({top.get('_source', '?')})"
else:
detail = ""
print(f" {status} '{query}': {count} matches {detail}")
if __name__ == '__main__':
print("=== Brain Ingestion: Cron Jobs ===")
try:
ingest_cron_jobs()
except Exception as e:
print(f"Fatal error ingesting cron jobs: {e}")
print(f"\n=== Ingestion Complete ===")
print(f"Total ingested: {stats['ingested']}")
print(f"Errors: {stats['errors']}")
run_verification()