Command for real time server stats

Author: Oget565

app/core/__pycache__/bot.cpython-312.pyc

@@ -4,6 +4,7 @@
 
 from app.handlers.basic import register_basic_handlers
 from app.handlers.schedule import send_report_24hr
+from app.handlers.server import server_stats_commands
 from app.services.metrics import wait_for_clock
 from app.services.clock import Clock
 
@@ -12,6 +13,7 @@ async def main():
     dp = Dispatcher()
 
     register_basic_handlers(dp)
+    server_stats_commands(dp)
 
     clock = Clock()
     clock_task = asyncio.create_task(clock.day_cycle_clock())

app/core/bot.py

@@ -0,0 +1,43 @@
+import sqlite3
+import time
+from pathlib import Path
+import asyncio
+from concurrent.futures import ThreadPoolExecutor
+
+DB_PATH = Path(__file__).resolve().parent / 'server_metrics.db'
+DB_PATH.parent.mkdir(parents=True, exist_ok=True)
+
+_executor = ThreadPoolExecutor(max_workers=1)
+
+CREATE_TABLE_SQL = '''
+CREATE TABLE IF NOT EXISTS Metrics (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    timestamp INTEGER NOT NULL,
+    mem_total REAL,
+    mem_used REAL,
+    mem_prcnt REAL,
+    cpu_prcnt REAL,
+    cpu_freq REAL,
+    cpu_temp REAL
+);
+'''
+
+LATEST_ROW_SQL = '''
+SELECT timestamp, mem_total, mem_used, mem_prcnt, cpu_prcnt, cpu_freq, cpu_temp
+FROM Metrics
+ORDER BY timestamp DESC
+LIMIT 1;
+'''
+
+def _read_latest_sync():
+    with sqlite3.connect(DB_PATH) as connection:
+        connection.row_factory = sqlite3.Row
+        cursor = connection.cursor()
+        cursor.execute(CREATE_TABLE_SQL)
+        cursor.execute(LATEST_ROW_SQL)
+        row = cursor.fetchone()
+        return dict(row) if row else None
+
+async def read_latest():
+    loop = asyncio.get_event_loop()
+    return await loop.run_in_executor(_executor, _read_latest_sync)

app/database/readb.py

@@ -0,0 +1,15 @@
+from app.database.readb import read_latest
+from app.filters.owner import IsOwner
+from aiogram import types
+from aiogram.filters import Command
+
+def server_stats_commands(dp):
+    @dp.message(Command("now"), IsOwner())
+    async def stats_now(message: types.Message):
+        metrics = await read_latest()
+        if metrics:
+            await message.answer(
+                f"{metrics}"
+            )
+        else:
+            await message.answer("No metrics found.")