SenSu/services/web_panel/utils/system_info.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os
import time
import platform
import logging
from typing import Dict, Any, Tuple

logger = logging.getLogger(__name__)


class SystemInfoCollector:
    def __init__(self):
        self.is_android = 'ANDROID_ROOT' in os.environ or os.path.exists('/system/bin/getprop')
        self.psutil = None
        if not self.is_android:
            try:
                import psutil
                self.psutil = psutil
            except ImportError as e:
                logger.debug(f"psutil not available: {e}")
        else:
            logger.info("🤖 Android 平台识别，启用原生采集")

        # State for speed calculation (rx_bytes, tx_bytes, timestamp)
        self._prev_net: Tuple[int, int, float] | None = None

    # ── Raw counters (used by both total and speed) ──

    def _read_net_bytes(self) -> Tuple[int, int]:
        """Read raw (rx_bytes, tx_bytes) since boot from system"""
        try:
            if self.psutil:
                io = self.psutil.net_io_counters()
                return (io.bytes_recv, io.bytes_sent)
        except Exception:
            pass

        # Android / Linux: parse /proc/net/dev
        try:
            rx = tx = 0
            with open('/proc/net/dev', 'r') as f:
                for line in f:
                    if ':' not in line or 'lo' in line:
                        continue
                    parts = line.split(':')[1].split()
                    rx += int(parts[0])
                    if len(parts) > 8:
                        tx += int(parts[8])
            return (rx, tx)
        except Exception as e:
            logger.warning(f"Network read failed: {e}")
            return (0, 0)

    # ── Public collectors ──

    def get_all(self) -> Dict[str, Any]:
        return {
            "platform": {
                "system": platform.system(),
                "machine": platform.machine(),
                "release": platform.release(),
                "python": platform.python_version(),
            },
            "cpu": self._get_cpu(),
            "memory": self._get_memory(),
            "process": self._get_process(),
            "network": self._get_network(),
            "net_speed": self._get_network_speed()
        }

    def _get_cpu(self) -> Dict[str, Any]:
        if self.psutil:
            return {
                "percent": self.psutil.cpu_percent(interval=0.1),
                "cores": self.psutil.cpu_count(),
                "load_avg": os.getloadavg() if hasattr(os, 'getloadavg') else [0, 0, 0]
            }
        try:
            load = os.getloadavg() if hasattr(os, "getloadavg") else [0.0, 0.0, 0.0]
            cores = os.cpu_count() or 1
            percent = min(100.0, (load[0] / cores) * 100)
            return {"percent": round(percent, 1), "cores": cores, "load_avg": load}
        except Exception as e:
            logger.error(f"CPU info error: {e}", exc_info=True)
            return {"percent": 0, "cores": 0, "load_avg": [0, 0, 0]}

    def _get_memory(self) -> Dict[str, Any]:
        if self.psutil:
            m = self.psutil.virtual_memory()
            return {"total_gb": round(m.total / 1073741824, 1),
                    "used_gb": round(m.used / 1073741824, 1),
                    "percent": m.percent}
        try:
            mem = {}
            with open('/proc/meminfo') as f:
                for line in f:
                    parts = line.split()
                    if len(parts) >= 2:
                        mem[parts[0].rstrip(':')] = int(parts[1]) * 1024
            t = mem.get('MemTotal', 1)
            a = mem.get('MemAvailable', mem.get('MemFree', 0))
            return {"total_gb": round(t / 1073741824, 1),
                    "used_gb": round((t - a) / 1073741824, 1),
                    "percent": round(((t - a) / t) * 100, 1)}
        except Exception as e:
            logger.warning(f"Memory info failed: {e}")
            return {"total_gb": 0, "used_gb": 0, "percent": 0}

    def _get_process(self) -> Dict[str, Any]:
        """Current process info (PID + RSS memory)"""
        pid = os.getpid()
        mem_mb = 0
        try:
            if self.psutil:
                import psutil
                mem_mb = round(psutil.Process(pid).memory_info().rss / 1048576, 1)
            else:
                # Parse /proc/self/status for VmRSS
                with open('/proc/self/status', 'r') as f:
                    for line in f:
                        if line.startswith('VmRSS:'):
                            mem_mb = round(int(line.split()[1]) / 1024, 1)
                            break
        except Exception as e:
            logger.debug(f"Process info failed: {e}")
        return {"pid": pid, "memory_mb": mem_mb}

    def _get_network(self) -> Dict[str, Any]:
        """Cumulative RX/TX since boot (MB)"""
        try:
            rx, tx = self._read_net_bytes()
            return {"rx_mb": round(rx / 1048576, 1),
                    "tx_mb": round(tx / 1048576, 1)}
        except Exception as e:
            logger.warning(f"Network info failed: {e}")
            return {"rx_mb": 0, "tx_mb": 0}

    def _get_network_speed(self) -> Dict[str, Any]:
        """Real-time RX/TX speed (bytes/sec), computed from delta between calls"""
        now = time.time()
        rx, tx = self._read_net_bytes()

        if self._prev_net:
            prx, ptx, pt = self._prev_net
            dt = now - pt
            speed_rx = (rx - prx) / dt if dt > 0 and rx >= prx else 0
            speed_tx = (tx - ptx) / dt if dt > 0 and tx >= ptx else 0
        else:
            speed_rx = speed_tx = 0

        self._prev_net = (rx, tx, now)
        return {"rx_bytes_sec": round(speed_rx, 1),
                "tx_bytes_sec": round(speed_tx, 1)}