tools/system_tools.py

"""System and Environment Tools"""
import os
import sys
import time
import platform
import subprocess
from typing import Dict, List, Optional
from config import Config
from utils import Logger

class SystemTools:
    """System monitoring and environment management"""
    
    def __init__(self):
        self.logger = Logger()
    
    def system_monitor(self, metric: str = "all", duration: int = 1) -> Dict:
        """Monitor system resources"""
        try:
            result = {"success": True, "timestamp": time.time()}
            
            if metric in ["cpu", "all"]:
                try:
                    cpu_percent = os.getloadavg()[0] if hasattr(os, "getloadavg") else 0
                    result["cpu"] = {
                        "load_average": cpu_percent,
                        "cores": os.cpu_count()
                    }
                except:
                    pass
            
            if metric in ["memory", "all"]:
                try:
                    with open("/proc/meminfo", "r") as f:
                        meminfo = f.read()
                    values = {}
                    for line in meminfo.split("\n"):
                        if ":" in line:
                            key, val = line.split(":", 1)
                            values[key.strip()] = val.strip().replace(" kB", "")
                    result["memory"] = {
                        "total_kb": int(values.get("MemTotal", 0)),
                        "available_kb": int(values.get("MemAvailable", 0)),
                        "free_kb": int(values.get("MemFree", 0))
                    }
                except:
                    pass
            
            if metric in ["disk", "all"]:
                try:
                    stat = os.statvfs(Config.SANDBOX_PATH)
                    result["disk"] = {
                        "total_gb": round(stat.f_frsize * stat.f_blocks / (1024**3), 2),
                        "free_gb": round(stat.f_frsize * stat.f_bavail / (1024**3), 2),
                        "used_percent": round((1 - stat.f_bavail / stat.f_blocks) * 100, 1)
                    }
                except:
                    pass
            
            if metric == "processes":
                result["processes"] = self._get_processes()
            
            self.logger.log("system_monitor", result)
            return result
            
        except Exception as e:
            return {"success": False, "error": str(e)}
    
    def _get_processes(self, limit: int = 20) -> List[Dict]:
        """Get running processes"""
        try:
            result = subprocess.run(
                "ps aux --sort=-%cpu | head -n " + str(limit + 1),
                shell=True, capture_output=True, text=True, timeout=5
            )
            lines = result.stdout.strip().split("\n")[1:]
            processes = []
            for line in lines:
                parts = line.split()
                if len(parts) >= 11:
                    processes.append({
                        "pid": parts[1],
                        "cpu": parts[2],
                        "mem": parts[3],
                        "command": " ".join(parts[10:])[:50]
                    })
            return processes
        except:
            return []
    
    def network_tools(self, action: str = "ping", host: str = "",
                      port: int = 0, timeout: int = 5) -> Dict:
        """Network diagnostics"""
        try:
            if action == "ping":
                result = subprocess.run(
                    f"ping -c 3 -W {timeout} {host}",
                    shell=True, capture_output=True, text=True, timeout=timeout + 5
                )
                return {
                    "success": result.returncode == 0,
                    "host": host,
                    "output": result.stdout[:1000]
                }
            
            elif action == "curl":
                result = subprocess.run(
                    f"curl -s -o /dev/null -w '%{{http_code}}|%{{time_total}}|%{{size_download}}' {host}",
                    shell=True, capture_output=True, text=True, timeout=timeout
                )
                parts = result.stdout.strip().split("|")
                return {
                    "success": True,
                    "url": host,
                    "status_code": parts[0] if len(parts) > 0 else "unknown",
                    "time_seconds": parts[1] if len(parts) > 1 else "unknown",
                    "size_bytes": parts[2] if len(parts) > 2 else "unknown"
                }
            
            elif action == "port_check":
                import socket
                sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                sock.settimeout(timeout)
                result = sock.connect_ex((host, port))
                sock.close()
                return {
                    "success": True,
                    "host": host,
                    "port": port,
                    "open": result == 0
                }
            
            return {"success": False, "error": f"Unknown action: {action}"}
            
        except Exception as e:
            return {"success": False, "error": str(e)}
    
    def clipboard_tools(self, action: str = "read") -> Dict:
        """Clipboard operations"""
        try:
            if action == "read":
                for cmd in ["xclip -o -selection clipboard", "xsel -b", "pbpaste", "powershell Get-Clipboard"]:
                    result = subprocess.run(
                        cmd, shell=True, capture_output=True, text=True, timeout=5
                    )
                    if result.returncode == 0:
                        return {
                            "success": True,
                            "content": result.stdout[:1000],
                            "method": cmd
                        }
                return {"success": False, "error": "No clipboard access available"}
            
            return {"success": False, "error": f"Unknown action: {action}"}
            
        except Exception as e:
            return {"success": False, "error": str(e)}