Refactor main.py to implement Flask app, SQLAlchemy models for Bot and BotLog, and health check functionality. Update pyproject.toml with new dependencies and add new HTML templates for the user interface. Replit-Commit-Author: Agent Replit-Commit-Session-Id: e72fc1b7-94bd-4d6c-801f-cbac2fae245c Replit-Commit-Checkpoint-Type: intermediate_checkpoint Replit-Commit-Event-Id: 5f598d52-420e-4e2c-88ea-a4c3e41fdcb6 Replit-Commit-Screenshot-Url: https://storage.googleapis.com/screenshot-production-us-central1/3bdfff67-975a-46ad-9845-fbb6b4a4c4b5/e72fc1b7-94bd-4d6c-801f-cbac2fae245c/jW8PJKQ Replit-Helium-Checkpoint-Created: true
210 lines
7.4 KiB
Python
210 lines
7.4 KiB
Python
import os
|
|
from flask import Flask, render_template, request, redirect, url_for, flash, jsonify
|
|
from flask_sqlalchemy import SQLAlchemy
|
|
from datetime import datetime
|
|
import requests
|
|
|
|
app = Flask(__name__)
|
|
app.secret_key = os.environ.get("FLASK_SECRET_KEY", os.urandom(24))
|
|
app.config["SQLALCHEMY_DATABASE_URI"] = os.environ.get("DATABASE_URL")
|
|
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
|
|
app.config["SQLALCHEMY_ENGINE_OPTIONS"] = {
|
|
"pool_recycle": 300,
|
|
"pool_pre_ping": True,
|
|
}
|
|
|
|
db = SQLAlchemy(app)
|
|
|
|
class Bot(db.Model):
|
|
__tablename__ = 'bots'
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
name = db.Column(db.String(100), nullable=False)
|
|
description = db.Column(db.Text)
|
|
health_endpoint = db.Column(db.String(500))
|
|
admin_token = db.Column(db.String(200))
|
|
bot_type = db.Column(db.String(50), default='discord')
|
|
status = db.Column(db.String(20), default='unknown')
|
|
last_checked = db.Column(db.DateTime)
|
|
guild_count = db.Column(db.Integer, default=0)
|
|
command_count = db.Column(db.Integer, default=0)
|
|
uptime_seconds = db.Column(db.Integer, default=0)
|
|
created_at = db.Column(db.DateTime, default=datetime.utcnow)
|
|
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
|
|
|
|
def to_dict(self):
|
|
return {
|
|
'id': self.id,
|
|
'name': self.name,
|
|
'description': self.description,
|
|
'has_health_endpoint': bool(self.health_endpoint),
|
|
'has_admin_token': bool(self.admin_token),
|
|
'bot_type': self.bot_type,
|
|
'status': self.status,
|
|
'last_checked': self.last_checked.isoformat() if self.last_checked else None,
|
|
'guild_count': self.guild_count,
|
|
'command_count': self.command_count,
|
|
'uptime_seconds': self.uptime_seconds,
|
|
'created_at': self.created_at.isoformat() if self.created_at else None,
|
|
}
|
|
|
|
class BotLog(db.Model):
|
|
__tablename__ = 'bot_logs'
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
bot_id = db.Column(db.Integer, db.ForeignKey('bots.id'), nullable=False)
|
|
level = db.Column(db.String(20), default='info')
|
|
message = db.Column(db.Text)
|
|
timestamp = db.Column(db.DateTime, default=datetime.utcnow)
|
|
|
|
bot = db.relationship('Bot', backref=db.backref('logs', lazy='dynamic'))
|
|
|
|
with app.app_context():
|
|
db.create_all()
|
|
|
|
def check_bot_health(bot):
|
|
if not bot.health_endpoint:
|
|
return {'status': 'unknown', 'error': 'No health endpoint configured'}
|
|
|
|
try:
|
|
headers = {}
|
|
if bot.admin_token:
|
|
headers['Authorization'] = f'Bearer {bot.admin_token}'
|
|
|
|
response = requests.get(bot.health_endpoint, headers=headers, timeout=10)
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
bot.status = data.get('status', 'online')
|
|
bot.guild_count = data.get('guilds', data.get('guildCount', 0))
|
|
bot.command_count = data.get('commands', data.get('commandCount', 0))
|
|
bot.uptime_seconds = data.get('uptime', 0)
|
|
bot.last_checked = datetime.utcnow()
|
|
db.session.commit()
|
|
return {'status': 'online', 'data': data}
|
|
else:
|
|
bot.status = 'offline'
|
|
bot.last_checked = datetime.utcnow()
|
|
db.session.commit()
|
|
return {'status': 'offline', 'error': f'HTTP {response.status_code}'}
|
|
except requests.exceptions.Timeout:
|
|
bot.status = 'timeout'
|
|
bot.last_checked = datetime.utcnow()
|
|
db.session.commit()
|
|
return {'status': 'timeout', 'error': 'Request timed out'}
|
|
except requests.exceptions.ConnectionError:
|
|
bot.status = 'offline'
|
|
bot.last_checked = datetime.utcnow()
|
|
db.session.commit()
|
|
return {'status': 'offline', 'error': 'Connection failed'}
|
|
except Exception as e:
|
|
bot.status = 'error'
|
|
bot.last_checked = datetime.utcnow()
|
|
db.session.commit()
|
|
return {'status': 'error', 'error': str(e)}
|
|
|
|
@app.route('/')
|
|
def dashboard():
|
|
bots = Bot.query.order_by(Bot.created_at.desc()).all()
|
|
stats = {
|
|
'total_bots': len(bots),
|
|
'online_bots': sum(1 for b in bots if b.status == 'online'),
|
|
'total_guilds': sum(b.guild_count or 0 for b in bots),
|
|
'total_commands': sum(b.command_count or 0 for b in bots),
|
|
}
|
|
return render_template('dashboard.html', bots=bots, stats=stats)
|
|
|
|
@app.route('/bots')
|
|
def list_bots():
|
|
bots = Bot.query.order_by(Bot.created_at.desc()).all()
|
|
return render_template('bots.html', bots=bots)
|
|
|
|
@app.route('/bots/add', methods=['GET', 'POST'])
|
|
def add_bot():
|
|
if request.method == 'POST':
|
|
name = request.form.get('name')
|
|
description = request.form.get('description')
|
|
health_endpoint = request.form.get('health_endpoint')
|
|
admin_token = request.form.get('admin_token')
|
|
bot_type = request.form.get('bot_type', 'discord')
|
|
|
|
if not name:
|
|
flash('Bot name is required', 'error')
|
|
return redirect(url_for('add_bot'))
|
|
|
|
bot = Bot(
|
|
name=name,
|
|
description=description,
|
|
health_endpoint=health_endpoint,
|
|
admin_token=admin_token,
|
|
bot_type=bot_type
|
|
)
|
|
db.session.add(bot)
|
|
db.session.commit()
|
|
|
|
if health_endpoint:
|
|
check_bot_health(bot)
|
|
|
|
flash(f'Bot "{name}" added successfully!', 'success')
|
|
return redirect(url_for('dashboard'))
|
|
|
|
return render_template('add_bot.html')
|
|
|
|
@app.route('/bots/<int:bot_id>')
|
|
def view_bot(bot_id):
|
|
bot = Bot.query.get_or_404(bot_id)
|
|
return render_template('view_bot.html', bot=bot)
|
|
|
|
@app.route('/bots/<int:bot_id>/edit', methods=['GET', 'POST'])
|
|
def edit_bot(bot_id):
|
|
bot = Bot.query.get_or_404(bot_id)
|
|
|
|
if request.method == 'POST':
|
|
bot.name = request.form.get('name', bot.name)
|
|
bot.description = request.form.get('description')
|
|
bot.health_endpoint = request.form.get('health_endpoint')
|
|
new_token = request.form.get('admin_token')
|
|
if new_token:
|
|
bot.admin_token = new_token
|
|
bot.bot_type = request.form.get('bot_type', 'discord')
|
|
db.session.commit()
|
|
|
|
flash(f'Bot "{bot.name}" updated successfully!', 'success')
|
|
return redirect(url_for('view_bot', bot_id=bot.id))
|
|
|
|
return render_template('edit_bot.html', bot=bot)
|
|
|
|
@app.route('/bots/<int:bot_id>/delete', methods=['POST'])
|
|
def delete_bot(bot_id):
|
|
bot = Bot.query.get_or_404(bot_id)
|
|
name = bot.name
|
|
BotLog.query.filter_by(bot_id=bot.id).delete()
|
|
db.session.delete(bot)
|
|
db.session.commit()
|
|
flash(f'Bot "{name}" deleted successfully!', 'success')
|
|
return redirect(url_for('dashboard'))
|
|
|
|
@app.route('/bots/<int:bot_id>/check', methods=['POST'])
|
|
def check_bot(bot_id):
|
|
bot = Bot.query.get_or_404(bot_id)
|
|
result = check_bot_health(bot)
|
|
return jsonify(result)
|
|
|
|
@app.route('/api/bots')
|
|
def api_list_bots():
|
|
bots = Bot.query.all()
|
|
return jsonify([bot.to_dict() for bot in bots])
|
|
|
|
@app.route('/api/bots/<int:bot_id>/health')
|
|
def api_bot_health(bot_id):
|
|
bot = Bot.query.get_or_404(bot_id)
|
|
result = check_bot_health(bot)
|
|
return jsonify(result)
|
|
|
|
@app.route('/api/check-all', methods=['POST'])
|
|
def api_check_all():
|
|
bots = Bot.query.all()
|
|
results = {}
|
|
for bot in bots:
|
|
results[bot.id] = check_bot_health(bot)
|
|
return jsonify(results)
|
|
|
|
if __name__ == '__main__':
|
|
app.run(host='0.0.0.0', port=5000, debug=True)
|