Spaces:
Runtime error
Runtime error
File size: 3,700 Bytes
5c6037b e96a62a 5c6037b 3a2ce2a e96a62a 3a2ce2a 5c6037b e96a62a 3a2ce2a 5c6037b 3a2ce2a 5c6037b e96a62a 3a2ce2a 5c6037b 3a2ce2a 5c6037b 3a2ce2a 5c6037b 3a2ce2a 5c6037b 3a2ce2a 5c6037b 3a2ce2a 5c6037b 3a2ce2a e96a62a 3a2ce2a e96a62a 5c6037b 3a2ce2a 5c6037b 3a2ce2a 5c6037b 3a2ce2a 5c6037b 3a2ce2a 5c6037b 3a2ce2a 5c6037b 3a2ce2a 5c6037b 3a2ce2a e96a62a 3a2ce2a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 |
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('''
<!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)
@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) |