from flask import Flask, render_template_string, jsonify
from apscheduler.schedulers.background import BackgroundScheduler
import subprocess
import threading
import pytz
import logging
from datetime import datetime
# Initialize Flask app
app = Flask(__name__)
# Setup logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
# Execution logs
execution_logs = []
MAX_LOG_ENTRIES = 50
log_lock = threading.Lock() # Prevents race conditions when modifying logs
def add_log_entry(entry):
"""Safely add log entries while maintaining MAX_LOG_ENTRIES limit."""
with log_lock:
execution_logs.append(entry)
if len(execution_logs) > MAX_LOG_ENTRIES:
execution_logs.pop(0)
def get_ist_time():
"""Get the current time in IST (Indian Standard Time)."""
utc_now = datetime.utcnow()
ist_timezone = pytz.timezone("Asia/Kolkata")
return utc_now.replace(tzinfo=pytz.utc).astimezone(ist_timezone)
def run_cli_script():
"""Runs cli.py and streams logs in real-time to both UI and terminal."""
timestamp = get_ist_time().strftime("%Y-%m-%d %H:%M:%S IST")
try:
process = subprocess.Popen(
["python", "cli.py"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
stdout, stderr = process.communicate() # Wait for process to complete
if stdout:
add_log_entry({'time': timestamp, 'output': stdout, 'error': ''})
logging.info(stdout.strip())
if stderr:
add_log_entry({'time': timestamp, 'output': '', 'error': stderr})
logging.error(stderr.strip())
except Exception as e:
error_msg = str(e)
add_log_entry({'time': timestamp, 'output': '', 'error': error_msg})
logging.error(f"Error: {error_msg}")
def start_initial_run():
"""Runs the CLI script immediately upon startup in a separate thread."""
threading.Thread(target=run_cli_script, daemon=True).start()
# Initialize scheduler
scheduler = BackgroundScheduler(daemon=True)
scheduler.add_job(
run_cli_script,
'interval',
hours=3,
id='main_job',
next_run_time=datetime.now()
)
scheduler.start()
# Ensure script runs once on startup
start_initial_run()
@app.route('/')
def home():
"""Main UI displaying logs and next run time."""
job = scheduler.get_job('main_job')
next_run = get_ist_time().strftime('%Y-%m-%d %H:%M:%S IST') if job else 'N/A'
return render_template_string('''
Script Scheduler
Script Scheduler
Next run: {{ next_run }}
Latest Execution Logs
Trigger Manual Run
Check Scheduler Status
''', next_run=next_run)
@app.route('/logs')
def logs():
"""Returns logs as JSON for AJAX polling."""
return jsonify({'logs': execution_logs})
@app.route('/force-run')
def force_run():
"""Manually trigger the script execution."""
threading.Thread(target=run_cli_script, daemon=True).start()
logging.info("Manual script execution triggered")
return "Script executed manually", 200
@app.route('/run-check')
def run_check():
"""Check if the scheduler is still running and restart if necessary."""
if not scheduler.running:
logging.warning("Scheduler was stopped! Restarting...")
scheduler.start()
return "Scheduler restarted", 200
return "Scheduler is running", 200
if __name__ == '__main__':
app.run(host='0.0.0.0', port=7860)