Persist job history server-side so the output banner and per-step last-run status survive completion and are visible to any logged-in user on any device
This commit is contained in:
+143
-18
@@ -13,7 +13,7 @@ import shlex
|
||||
import shutil
|
||||
import threading
|
||||
import json
|
||||
from datetime import datetime
|
||||
from datetime import datetime, timezone
|
||||
import yaml
|
||||
try:
|
||||
import psutil
|
||||
@@ -120,10 +120,85 @@ def auth_status():
|
||||
username = session.get('username')
|
||||
return jsonify({'authenticated': bool(username), 'username': username or ''})
|
||||
|
||||
# ── Store running processes ────────────────────────────────────────────────────
|
||||
# ── Job registry ──────────────────────────────────────────────────────────────
|
||||
# Every run is recorded under /data/jobs (metadata JSON + log file) so any
|
||||
# logged-in user, on any device, can see what ran, when, and how it ended -
|
||||
# even after the job finished or the server restarted.
|
||||
|
||||
JOBS_DIR = '/data/jobs'
|
||||
JOBS_KEEP = 200
|
||||
|
||||
running_processes = {}
|
||||
process_logs = {}
|
||||
jobs = {} # job_id -> metadata dict (see _new_job)
|
||||
process_logs = {} # job_id -> list of log lines
|
||||
jobs_lock = threading.Lock()
|
||||
|
||||
|
||||
def _utcnow():
|
||||
return datetime.now(timezone.utc).isoformat(timespec='milliseconds').replace('+00:00', 'Z')
|
||||
|
||||
|
||||
def _job_meta_path(job_id):
|
||||
return os.path.join(JOBS_DIR, f'{job_id}.json')
|
||||
|
||||
|
||||
def _job_log_path(job_id):
|
||||
return os.path.join(JOBS_DIR, f'{job_id}.log')
|
||||
|
||||
|
||||
def _save_job(job):
|
||||
os.makedirs(JOBS_DIR, exist_ok=True)
|
||||
tmp = _job_meta_path(job['id']) + '.tmp'
|
||||
with open(tmp, 'w') as f:
|
||||
json.dump(job, f)
|
||||
os.replace(tmp, _job_meta_path(job['id']))
|
||||
|
||||
|
||||
def _load_jobs():
|
||||
"""Load job history from disk. Jobs left 'running' by a previous server
|
||||
process can no longer be watched, so they are marked interrupted."""
|
||||
if not os.path.isdir(JOBS_DIR):
|
||||
return
|
||||
for fname in os.listdir(JOBS_DIR):
|
||||
if not fname.endswith('.json'):
|
||||
continue
|
||||
try:
|
||||
with open(os.path.join(JOBS_DIR, fname)) as f:
|
||||
job = json.load(f)
|
||||
except (OSError, ValueError):
|
||||
continue
|
||||
if job.get('status') == 'running':
|
||||
job['status'] = 'interrupted'
|
||||
job['ended'] = job.get('ended') or _utcnow()
|
||||
_save_job(job)
|
||||
jobs[job['id']] = job
|
||||
|
||||
|
||||
def _prune_jobs():
|
||||
"""Keep only the newest JOBS_KEEP jobs on disk and in memory."""
|
||||
finished = sorted((j for j in jobs.values() if j['status'] != 'running'),
|
||||
key=lambda j: j['started'])
|
||||
for job in finished[:max(0, len(jobs) - JOBS_KEEP)]:
|
||||
jobs.pop(job['id'], None)
|
||||
process_logs.pop(job['id'], None)
|
||||
for path in (_job_meta_path(job['id']), _job_log_path(job['id'])):
|
||||
try:
|
||||
os.remove(path)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
def _job_logs(job_id):
|
||||
if job_id in process_logs:
|
||||
return process_logs[job_id]
|
||||
try:
|
||||
with open(_job_log_path(job_id)) as f:
|
||||
return f.readlines()
|
||||
except OSError:
|
||||
return []
|
||||
|
||||
|
||||
_load_jobs()
|
||||
|
||||
|
||||
def _cleanup_legacy_structure():
|
||||
@@ -407,9 +482,31 @@ def run_script():
|
||||
else:
|
||||
return jsonify({'error': 'Invalid script configuration'}), 400
|
||||
|
||||
job_id = f"{script_name}_{county}_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
|
||||
job_id = f"{script_name}_{county}_{datetime.now().strftime('%Y%m%d_%H%M%S_%f')}"
|
||||
job = {
|
||||
'id': job_id,
|
||||
'script': script_name,
|
||||
'county': county,
|
||||
'user': session.get('username'),
|
||||
'started': _utcnow(),
|
||||
'ended': None,
|
||||
'status': 'running',
|
||||
'exit_code': None,
|
||||
}
|
||||
with jobs_lock:
|
||||
jobs[job_id] = job
|
||||
process_logs[job_id] = []
|
||||
_save_job(job)
|
||||
_prune_jobs()
|
||||
|
||||
def run_command():
|
||||
log_file = open(_job_log_path(job_id), 'w')
|
||||
|
||||
def log(line):
|
||||
process_logs[job_id].append(line)
|
||||
log_file.write(line)
|
||||
log_file.flush()
|
||||
|
||||
try:
|
||||
process = subprocess.Popen(
|
||||
cmd,
|
||||
@@ -419,32 +516,62 @@ def run_script():
|
||||
cwd=os.path.dirname(os.path.dirname(__file__))
|
||||
)
|
||||
running_processes[job_id] = process
|
||||
process_logs[job_id] = []
|
||||
|
||||
for line in process.stdout:
|
||||
process_logs[job_id].append(line)
|
||||
log(line)
|
||||
|
||||
process.wait()
|
||||
process_logs[job_id].append(f"\n[Process completed with exit code {process.returncode}]")
|
||||
job['exit_code'] = process.returncode
|
||||
if job.get('cancelled'):
|
||||
job['status'] = 'cancelled'
|
||||
log(f"\n[Job cancelled by {job['cancelled']}]")
|
||||
else:
|
||||
job['status'] = 'success' if process.returncode == 0 else 'failed'
|
||||
log(f"\n[Process completed with exit code {process.returncode}]")
|
||||
|
||||
except Exception as e:
|
||||
process_logs[job_id].append(f"\n[ERROR: {str(e)}]")
|
||||
job['status'] = 'failed'
|
||||
log(f"\n[ERROR: {str(e)}]")
|
||||
finally:
|
||||
if job_id in running_processes:
|
||||
del running_processes[job_id]
|
||||
running_processes.pop(job_id, None)
|
||||
log_file.close()
|
||||
with jobs_lock:
|
||||
job['ended'] = _utcnow()
|
||||
_save_job(job)
|
||||
# Finished logs live on disk; don't hold them in memory forever.
|
||||
process_logs.pop(job_id, None)
|
||||
|
||||
thread = threading.Thread(target=run_command)
|
||||
thread.daemon = True
|
||||
thread.start()
|
||||
|
||||
return jsonify({'job_id': job_id, 'status': 'started'})
|
||||
return jsonify({'job_id': job_id, 'status': 'started', 'job': job})
|
||||
|
||||
@app.route('/api/jobs')
|
||||
@require_auth
|
||||
def list_jobs():
|
||||
"""Most recent job overall, plus the most recent job for each script/county
|
||||
pair, so every device can show the same banner and per-step status."""
|
||||
with jobs_lock:
|
||||
ordered = sorted(jobs.values(), key=lambda j: j['started'])
|
||||
latest_by_step = {}
|
||||
for job in ordered:
|
||||
latest_by_step[f"{job['script']}|{job['county']}"] = job
|
||||
return jsonify({
|
||||
'latest': ordered[-1] if ordered else None,
|
||||
'steps': latest_by_step,
|
||||
'running': [j['id'] for j in ordered if j['status'] == 'running'],
|
||||
})
|
||||
|
||||
@app.route('/api/job-status/<job_id>')
|
||||
@require_auth
|
||||
def job_status(job_id):
|
||||
"""Get status and logs for a job"""
|
||||
is_running = job_id in running_processes
|
||||
logs = process_logs.get(job_id, [])
|
||||
return jsonify({'job_id': job_id, 'running': is_running, 'logs': logs})
|
||||
job = jobs.get(job_id)
|
||||
if not job:
|
||||
return jsonify({'error': 'Job not found'}), 404
|
||||
return jsonify({'job_id': job_id, 'running': job['status'] == 'running',
|
||||
'job': job, 'logs': _job_logs(job_id)})
|
||||
|
||||
@app.route('/api/cancel-job', methods=['POST'])
|
||||
@require_auth
|
||||
@@ -459,10 +586,8 @@ def cancel_job():
|
||||
return jsonify({'error': 'Job not found or already completed'}), 404
|
||||
|
||||
try:
|
||||
process = running_processes[job_id]
|
||||
process.terminate()
|
||||
process_logs[job_id].append('\n[Job cancelled by user]')
|
||||
del running_processes[job_id]
|
||||
jobs[job_id]['cancelled'] = session.get('username') or 'user'
|
||||
running_processes[job_id].terminate()
|
||||
return jsonify({'success': True})
|
||||
except Exception as e:
|
||||
return jsonify({'error': str(e)}), 500
|
||||
|
||||
Reference in New Issue
Block a user