default/telegram-agy-bot/sys_monitor.py

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 = [
            "🖥 <b>گزارش لحظه‌ای وضعیت سخت‌افزار و سرور</b>",
            "",
            "🧠 <b>حافظه رم (RAM Usage):</b>",
            f"  {ram_bar} <b>{ram_pct}%</b>",
            f"  • <b>مصرف شده:</b> <code>{ram_used_str}</code> از <code>{ram_total_str}</code>",
            f"  • <b>حافظه آزاد:</b> <code>{ram_free_str}</code>",
        ]
        if swap_total > 0:
            lines.extend([
                "",
                "🔄 <b>حافظه مجازی (Swap):</b>",
                f"  {swap_bar} <b>{swap_pct}%</b> ({format_bytes(st.get('swap_used', 0))} / {format_bytes(swap_total)})",
            ])

        lines.extend([
            "",
            "⚡ <b>میانگین بار پردازشی پردازنده (CPU Load Average):</b>",
            f"  • <b>میانگین بار ۱ دقیقه:</b>",
            f"    {load_1_bar} <b>{load_1_pct}%</b> (بار: <code>{load_1}</code>)",
            f"  • <b>میانگین بار ۵ دقیقه:</b>",
            f"    {load_5_bar} <b>{load_5_pct}%</b> (بار: <code>{load_5}</code>)",
            f"  • <b>میانگین بار ۱۵ دقیقه:</b>",
            f"    {load_15_bar} <b>{load_15_pct}%</b> (بار: <code>{load_15}</code>)",
            f"  • <b>تعداد هسته‌ها:</b> <code>{cpu_cores} Cores / Threads</code>",
            "",
            "💾 <b>حافظه داخلی و دیسک (Disk Storage):</b>",
            f"  {disk_bar} <b>{disk_pct}%</b>",
            f"  • <b>فضای مصرف شده:</b> <code>{disk_used_str}</code> از <code>{disk_total_str}</code>",
            f"  • <b>فضای خالی باقیمانده:</b> <code>{disk_free_str}</code>",
            "",
            "⚙️ <b>اطلاعات سیستم و ربات:</b>",
            f"  • 🤖 <b>رم مصرفی پردازش ربات:</b> <code>{bot_mem_str}</code>",
            f"  • ⏱ <b>مدت زمان روشن بودن سرور (Uptime):</b> <code>{uptime_str}</code>",
            f"  • 🐧 <b>سیستم‌عامل سرور:</b> <code>{os_info}</code>",
        ])
    else:
        lines = [
            "🖥 <b>Real-Time Server Hardware Status</b>",
            "",
            "🧠 <b>Memory (RAM Usage):</b>",
            f"  {ram_bar} <b>{ram_pct}%</b>",
            f"  • <b>Used:</b> <code>{ram_used_str}</code> / <code>{ram_total_str}</code>",
            f"  • <b>Available:</b> <code>{ram_free_str}</code>",
        ]
        if swap_total > 0:
            lines.extend([
                "",
                "🔄 <b>Swap Memory:</b>",
                f"  {swap_bar} <b>{swap_pct}%</b> ({format_bytes(st.get('swap_used', 0))} / {format_bytes(swap_total)})",
            ])

        lines.extend([
            "",
            "⚡ <b>Processor Load Average (CPU Load):</b>",
            f"  • <b>1-Minute Load:</b>",
            f"    {load_1_bar} <b>{load_1_pct}%</b> (Load: <code>{load_1}</code>)",
            f"  • <b>5-Minute Load:</b>",
            f"    {load_5_bar} <b>{load_5_pct}%</b> (Load: <code>{load_5}</code>)",
            f"  • <b>15-Minute Load:</b>",
            f"    {load_15_bar} <b>{load_15_pct}%</b> (Load: <code>{load_15}</code>)",
            f"  • <b>Cores:</b> <code>{cpu_cores} Cores / Threads</code>",
            "",
            "💾 <b>Disk Storage:</b>",
            f"  {disk_bar} <b>{disk_pct}%</b>",
            f"  • <b>Used Space:</b> <code>{disk_used_str}</code> / <code>{disk_total_str}</code>",
            f"  • <b>Free Space:</b> <code>{disk_free_str}</code>",
            "",
            "⚙️ <b>Process & System Info:</b>",
            f"  • 🤖 <b>Bot Process RAM:</b> <code>{bot_mem_str}</code>",
            f"  • ⏱ <b>Server Uptime:</b> <code>{uptime_str}</code>",
            f"  • 🐧 <b>OS / Kernel:</b> <code>{os_info}</code>",
        ])

    return "\n".join(lines)