import os
import sys
import time
import shutil
import platform
import logging
from pathlib import Path
from typing import Dict, Any, Tuple
try:
import psutil
except ImportError:
psutil = None
logger = logging.getLogger("AGYSysMonitor")
def make_progress_bar(percent: float, length: int = 10) -> str:
"""Generates a visual emoji progress bar based on percentage."""
percent = max(0.0, min(100.0, float(percent)))
filled = int(round((percent / 100.0) * length))
filled = max(0, min(length, filled))
if percent < 60:
fill_char = "🟩"
elif percent < 85:
fill_char = "🟨"
else:
fill_char = "🟥"
empty_char = "⬜"
return f"{fill_char * filled}{empty_char * (length - filled)}"
def make_load_bar(load_val: float, cpu_cores: int, length: int = 10) -> Tuple[str, float]:
"""Calculates load percentage based on core count and returns (bar, percent)."""
cores = max(1, cpu_cores)
percent = round((float(load_val) / cores) * 100.0, 1)
bar = make_progress_bar(percent, length)
return bar, percent
def format_bytes(bytes_val: int) -> str:
"""Converts bytes to human readable format (MB, GB, TB)."""
if bytes_val < 1024 * 1024:
return f"{bytes_val / 1024:.1f} KB"
elif bytes_val < 1024 * 1024 * 1024:
return f"{bytes_val / (1024 * 1024):.1f} MB"
elif bytes_val < 1024 * 1024 * 1024 * 1024:
return f"{bytes_val / (1024 * 1024 * 1024):.2f} GB"
else:
return f"{bytes_val / (1024 * 1024 * 1024 * 1024):.2f} TB"
def get_server_uptime_str(is_fa: bool = True) -> str:
"""Returns server uptime string."""
try:
if psutil:
boot_time = psutil.boot_time()
uptime_seconds = int(time.time() - boot_time)
else:
with open("/proc/uptime", "r") as f:
uptime_seconds = int(float(f.readline().split()[0]))
days = uptime_seconds // 86400
hours = (uptime_seconds % 86400) // 3600
minutes = (uptime_seconds % 3600) // 60
if is_fa:
parts = []
if days > 0:
parts.append(f"{days} روز")
if hours > 0 or days > 0:
parts.append(f"{hours} ساعت")
parts.append(f"{minutes} دقیقه")
return " و ".join(parts)
else:
return f"{days}d {hours}h {minutes}m"
except Exception:
return "N/A"
def get_system_hardware_stats() -> Dict[str, Any]:
"""Gathers detailed real-time RAM, CPU, Disk, and process statistics."""
stats: Dict[str, Any] = {}
# 1. RAM & Swap
if psutil:
mem = psutil.virtual_memory()
stats["ram_total"] = mem.total
stats["ram_used"] = mem.used
stats["ram_free"] = mem.available
stats["ram_percent"] = mem.percent
swap = psutil.swap_memory()
stats["swap_total"] = swap.total
stats["swap_used"] = swap.used
stats["swap_free"] = swap.free
stats["swap_percent"] = swap.percent
else:
# Fallback reading /proc/meminfo
try:
meminfo = {}
with open("/proc/meminfo", "r") as f:
for line in f:
parts = line.split(":")
if len(parts) == 2:
key = parts[0].strip()
val = int(parts[1].split()[0]) * 1024 # kB to bytes
meminfo[key] = val
total = meminfo.get("MemTotal", 1)
free = meminfo.get("MemAvailable", meminfo.get("MemFree", 0))
used = total - free
stats["ram_total"] = total
stats["ram_used"] = used
stats["ram_free"] = free
stats["ram_percent"] = round((used / total) * 100, 1)
stats["swap_total"] = meminfo.get("SwapTotal", 0)
stats["swap_free"] = meminfo.get("SwapFree", 0)
stats["swap_used"] = stats["swap_total"] - stats["swap_free"]
stats["swap_percent"] = round((stats["swap_used"] / stats["swap_total"] * 100), 1) if stats["swap_total"] > 0 else 0.0
except Exception as e:
logger.error(f"Error reading meminfo: {e}")
# 2. CPU & Load
if psutil:
stats["cpu_percent"] = psutil.cpu_percent(interval=0.2)
stats["cpu_count_logical"] = psutil.cpu_count(logical=True) or 1
stats["cpu_count_physical"] = psutil.cpu_count(logical=False) or stats["cpu_count_logical"]
else:
stats["cpu_percent"] = 0.0
stats["cpu_count_logical"] = os.cpu_count() or 1
stats["cpu_count_physical"] = stats["cpu_count_logical"]
try:
load_1, load_5, load_15 = os.getloadavg()
stats["load_avg"] = (round(load_1, 2), round(load_5, 2), round(load_15, 2))
except Exception:
stats["load_avg"] = (0.0, 0.0, 0.0)
# 3. Disk Storage
try:
root_usage = shutil.disk_usage("/")
stats["disk_total"] = root_usage.total
stats["disk_used"] = root_usage.used
stats["disk_free"] = root_usage.free
stats["disk_percent"] = round((root_usage.used / root_usage.total) * 100, 1)
except Exception as e:
logger.error(f"Error reading root disk usage: {e}")
# 4. Bot Process Memory / CPU
try:
if psutil:
proc = psutil.Process(os.getpid())
proc_mem = proc.memory_info().rss
stats["bot_mem_rss"] = proc_mem
stats["bot_cpu_percent"] = proc.cpu_percent(interval=None)
else:
stats["bot_mem_rss"] = 0
stats["bot_cpu_percent"] = 0.0
except Exception:
stats["bot_mem_rss"] = 0
stats["bot_cpu_percent"] = 0.0
stats["uptime_fa"] = get_server_uptime_str(is_fa=True)
stats["uptime_en"] = get_server_uptime_str(is_fa=False)
stats["os_info"] = f"{platform.system()} {platform.release()} ({platform.machine()})"
return stats
def render_server_hardware_report(is_fa: bool = True) -> str:
"""Renders formatted HTML hardware status report with progress bars for RAM, CPU, Load Avg, and Disk."""
st = get_system_hardware_stats()
# RAM
ram_pct = st.get("ram_percent", 0.0)
ram_bar = make_progress_bar(ram_pct)
ram_used_str = format_bytes(st.get("ram_used", 0))
ram_total_str = format_bytes(st.get("ram_total", 0))
ram_free_str = format_bytes(st.get("ram_free", 0))
# CPU & Load Average
cpu_pct = st.get("cpu_percent", 0.0)
cpu_bar = make_progress_bar(cpu_pct)
cpu_cores = st.get("cpu_count_logical", 1)
load_1, load_5, load_15 = st.get("load_avg", (0.0, 0.0, 0.0))
load_1_bar, load_1_pct = make_load_bar(load_1, cpu_cores)
load_5_bar, load_5_pct = make_load_bar(load_5, cpu_cores)
load_15_bar, load_15_pct = make_load_bar(load_15, cpu_cores)
# Disk
disk_pct = st.get("disk_percent", 0.0)
disk_bar = make_progress_bar(disk_pct)
disk_used_str = format_bytes(st.get("disk_used", 0))
disk_total_str = format_bytes(st.get("disk_total", 0))
disk_free_str = format_bytes(st.get("disk_free", 0))
# Swap
swap_total = st.get("swap_total", 0)
swap_pct = st.get("swap_percent", 0.0)
swap_bar = make_progress_bar(swap_pct) if swap_total > 0 else ""
bot_mem_str = format_bytes(st.get("bot_mem_rss", 0))
uptime_str = st.get("uptime_fa") if is_fa else st.get("uptime_en")
os_info = st.get("os_info", "Linux")
if is_fa:
lines = [
"🖥 گزارش لحظهای وضعیت سختافزار و سرور",
"",
"🧠 حافظه رم (RAM Usage):",
f" {ram_bar} {ram_pct}%",
f" • مصرف شده: {ram_used_str} از {ram_total_str}",
f" • حافظه آزاد: {ram_free_str}",
]
if swap_total > 0:
lines.extend([
"",
"🔄 حافظه مجازی (Swap):",
f" {swap_bar} {swap_pct}% ({format_bytes(st.get('swap_used', 0))} / {format_bytes(swap_total)})",
])
lines.extend([
"",
"⚡ میانگین بار پردازشی پردازنده (CPU Load Average):",
f" • میانگین بار ۱ دقیقه:",
f" {load_1_bar} {load_1_pct}% (بار: {load_1})",
f" • میانگین بار ۵ دقیقه:",
f" {load_5_bar} {load_5_pct}% (بار: {load_5})",
f" • میانگین بار ۱۵ دقیقه:",
f" {load_15_bar} {load_15_pct}% (بار: {load_15})",
f" • تعداد هستهها: {cpu_cores} Cores / Threads",
"",
"💾 حافظه داخلی و دیسک (Disk Storage):",
f" {disk_bar} {disk_pct}%",
f" • فضای مصرف شده: {disk_used_str} از {disk_total_str}",
f" • فضای خالی باقیمانده: {disk_free_str}",
"",
"⚙️ اطلاعات سیستم و ربات:",
f" • 🤖 رم مصرفی پردازش ربات: {bot_mem_str}",
f" • ⏱ مدت زمان روشن بودن سرور (Uptime): {uptime_str}",
f" • 🐧 سیستمعامل سرور: {os_info}",
])
else:
lines = [
"🖥 Real-Time Server Hardware Status",
"",
"🧠 Memory (RAM Usage):",
f" {ram_bar} {ram_pct}%",
f" • Used: {ram_used_str} / {ram_total_str}",
f" • Available: {ram_free_str}",
]
if swap_total > 0:
lines.extend([
"",
"🔄 Swap Memory:",
f" {swap_bar} {swap_pct}% ({format_bytes(st.get('swap_used', 0))} / {format_bytes(swap_total)})",
])
lines.extend([
"",
"⚡ Processor Load Average (CPU Load):",
f" • 1-Minute Load:",
f" {load_1_bar} {load_1_pct}% (Load: {load_1})",
f" • 5-Minute Load:",
f" {load_5_bar} {load_5_pct}% (Load: {load_5})",
f" • 15-Minute Load:",
f" {load_15_bar} {load_15_pct}% (Load: {load_15})",
f" • Cores: {cpu_cores} Cores / Threads",
"",
"💾 Disk Storage:",
f" {disk_bar} {disk_pct}%",
f" • Used Space: {disk_used_str} / {disk_total_str}",
f" • Free Space: {disk_free_str}",
"",
"⚙️ Process & System Info:",
f" • 🤖 Bot Process RAM: {bot_mem_str}",
f" • ⏱ Server Uptime: {uptime_str}",
f" • 🐧 OS / Kernel: {os_info}",
])
return "\n".join(lines)