AeThex-Bot-Master/main.py
sirpiglr 3d4e07b965 Add bot management dashboard with health checks and status monitoring
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
2025-12-06 21:56:18 +00:00

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)