from flask import Flask, render_template_string from apscheduler.schedulers.background import BackgroundScheduler import subprocess import threading from datetime import datetime app = Flask(__name__) execution_logs = [] MAX_LOG_ENTRIES = 20 def run_cli_script(): """Runs cli.py and streams the output in real-time.""" timestamp = datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S UTC") log_entry = {'time': timestamp, 'output': '', 'error': ''} try: process = subprocess.Popen( ["python", "cli.py"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1 # Ensures line buffering (real-time output) ) # Stream stdout in real-time for line in process.stdout: log_entry['output'] += line print(line, end="") # Also print to terminal for debugging # Capture errors in real-time for line in process.stderr: log_entry['error'] += line print(line, end="") # Also print to terminal for debugging except Exception as e: log_entry['error'] = str(e) finally: execution_logs.append(log_entry) if len(execution_logs) > MAX_LOG_ENTRIES: execution_logs.pop(0) # Start script in a separate thread to avoid blocking Flask def start_initial_run(): 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() # Run the script asynchronously to prevent blocking start_initial_run() @app.route('/') def home(): """Main UI displaying logs and next run time.""" job = scheduler.get_job('main_job') next_run = job.next_run_time.strftime('%Y-%m-%d %H:%M:%S UTC') if job else 'N/A' return render_template_string(''' Script Scheduler

Script Scheduler

Next run: {{ next_run }}

Latest Execution Logs

{% for log in logs|reverse %}
{{ log.time }}
{% if log.output %}
{{ log.output }}
{% endif %} {% if log.error %}
{{ log.error }}
{% endif %}
{% else %}
No logs available yet
{% endfor %}

Trigger Manual Run

Check Scheduler Status

''', next_run=next_run, logs=execution_logs) @app.route('/force-run') def force_run(): """Manually trigger the script execution.""" threading.Thread(target=run_cli_script, daemon=True).start() return "Script executed manually", 200 @app.route('/run-check') def run_check(): """Check if the scheduler is still running.""" if not scheduler.running: print("Scheduler was stopped! Restarting...") scheduler.start() return "Scheduler is running", 200 if __name__ == '__main__': app.run(host='0.0.0.0', port=7860)