from flask import Flask, render_template_string from apscheduler.schedulers.background import BackgroundScheduler import subprocess from datetime import datetime import logging import sys import time app = Flask(__name__) # Configure basic logging logging.basicConfig(stream=sys.stdout, level=logging.INFO) app.logger.addHandler(logging.StreamHandler(sys.stdout)) # Global log storage execution_logs = [] MAX_LOG_ENTRIES = 20 def run_cli_script(): """Execute the script and store logs with timestamp""" timestamp = datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S UTC") log_entry = { 'time': timestamp, 'output': '', 'error': '' } try: app.logger.info(f"Starting script execution at {timestamp}") result = subprocess.run( ["python", "cli.py"], capture_output=True, text=True, timeout=300 # 5 minutes timeout ) log_entry['output'] = result.stdout if result.stderr: log_entry['error'] = result.stderr app.logger.info("Script execution completed") except Exception as e: error_msg = f"Execution failed: {str(e)}" log_entry['error'] = error_msg app.logger.error(error_msg) finally: # Maintain log history execution_logs.append(log_entry) if len(execution_logs) > MAX_LOG_ENTRIES: execution_logs.pop(0) # Initialize scheduler def init_scheduler(): scheduler = BackgroundScheduler() scheduler.add_job(run_cli_script, 'interval', hours=3) scheduler.start() # Initial immediate run for testing run_cli_script() return scheduler # Start scheduler when app starts try: scheduler = init_scheduler() except Exception as e: app.logger.error(f"Failed to initialize scheduler: {e}") @app.route('/') def home(): return render_template_string(''' Script Scheduler

Script Scheduler

Next run: {{ scheduler.next_run_time.strftime('%Y-%m-%d %H:%M:%S UTC') if scheduler else 'N/A' }}

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 %}
''', scheduler=scheduler, logs=execution_logs) @app.route('/force-run') def force_run(): """Manual trigger endpoint for testing""" run_cli_script() return "Script executed manually", 200 if __name__ == '__main__': app.run(host='0.0.0.0', port=7860, debug=True)