#!/usr/bin/env python3 from flask import Flask, render_template_string, jsonify import sqlite3 import datetime import os app = Flask(__name__) HTML_TEMPLATE = """ 战锤40K行星总督 - 游戏监控

战锤40K行星总督

服务器监控面板

总在线用户

游戏中用户

注册用户总数

当前在线用户:

{% if active_users %} {% for user in active_users %}

{% endfor %} {% else %}

当前没有用户在线

{% endif %}

上次更新时间: {{ refresh_time }}

""" def get_db_connection(): """连接到SQLite数据库""" db_path = os.path.join('data', 'warhammer.db') conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row return conn def get_stats(): """获取游戏统计数据""" try: conn = get_db_connection() # 获取注册用户总数 registered_users = conn.execute('SELECT COUNT(*) as count FROM users').fetchone()['count'] # 获取活跃会话 (假设数据库中有sessions表，根据实际情况调整) # 如果没有sessions表，需要创建或使用其他方式跟踪在线用户 try: cursor = conn.execute(''' SELECT s.*, u.username FROM sessions s JOIN users u ON s.user_id = u.id WHERE s.last_activity > ? ''', (datetime.datetime.now() - datetime.timedelta(minutes=30),)) active_sessions = cursor.fetchall() except sqlite3.OperationalError: # 如果sessions表不存在，尝试备选方案 try: # 尝试从game_states表获取活跃用户 cursor = conn.execute(''' SELECT gs.*, u.username, u.id as user_id FROM game_states gs JOIN users u ON gs.user_id = u.id WHERE gs.updated_at > ? ''', ((datetime.datetime.now() - datetime.timedelta(minutes=30)).isoformat(),)) active_sessions = cursor.fetchall() # 添加状态字段模拟 for session in active_sessions: # 假设最近30分钟内更新的都是在玩游戏的 session_dict = dict(session) session_dict['status'] = 'playing' session = session_dict except: # 备选方案也失败，返回空列表 active_sessions = [] # 计算游戏中的用户数 playing_users = sum(1 for session in active_sessions if session['status'] == 'playing') # 格式化活跃用户列表 active_users = [] for session in active_sessions: user_info = { 'username': session['username'], 'status': session['status'], 'user_id': session['user_id'] } active_users.append(user_info) conn.close() return { 'registered_users': registered_users, 'total_users': len(active_sessions), 'playing_users': playing_users, 'active_users': active_users, 'refresh_time': datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') } except Exception as e: print(f"获取统计数据时出错: {str(e)}") return { 'registered_users': 0, 'total_users': 0, 'playing_users': 0, 'active_users': [], 'refresh_time': datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), 'error': str(e) } @app.route('/monitor') def monitor_page(): """监控页面""" stats = get_stats() return render_template_string(HTML_TEMPLATE, **stats) @app.route('/monitor/api/stats') def api_stats(): """统计数据API""" return jsonify(get_stats()) if __name__ == '__main__': app.run(host='127.0.0.1', port=5050, debug=False)