Spaces:
Running
Running
| 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}") | |
| def home(): | |
| return render_template_string(''' | |
| <!DOCTYPE html> | |
| <html> | |
| <head> | |
| <title>Script Scheduler</title> | |
| <meta http-equiv="refresh" content="10"> | |
| <style> | |
| body { font-family: Arial, sans-serif; padding: 20px; } | |
| .log-box { | |
| background: #000; | |
| color: #0f0; | |
| padding: 15px; | |
| border-radius: 5px; | |
| margin-top: 20px; | |
| white-space: pre-wrap; | |
| } | |
| .timestamp { color: #888; margin-bottom: 10px; } | |
| .error { color: #ff4444; } | |
| </style> | |
| </head> | |
| <body> | |
| <h1>Script Scheduler</h1> | |
| <p>Next run: {{ scheduler.next_run_time.strftime('%Y-%m-%d %H:%M:%S UTC') if scheduler else 'N/A' }}</p> | |
| <h2>Latest Execution Logs</h2> | |
| <div class="log-box"> | |
| {% for log in logs|reverse %} | |
| <div class="timestamp">{{ log.time }}</div> | |
| {% if log.output %} | |
| <div class="output">{{ log.output }}</div> | |
| {% endif %} | |
| {% if log.error %} | |
| <div class="error">{{ log.error }}</div> | |
| {% endif %} | |
| <hr> | |
| {% else %} | |
| <div>No logs available yet</div> | |
| {% endfor %} | |
| </div> | |
| </body> | |
| </html> | |
| ''', scheduler=scheduler, logs=execution_logs) | |
| 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) |