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)