Python System Administration: os, subprocess, and Automation

import subprocess

# Simple command - runs and waits
result = subprocess.run(['ls', '-la'])
print(f'Exit code: {result.returncode}')

# Capture output
result = subprocess.run(['ls', '-la'], capture_output=True, text=True)
print(f'Stdout: {result.stdout}')
print(f'Stderr: {result.stderr}')

# Check if command succeeded (raises exception on failure)
subprocess.run(['ls', '/nonexistent'], check=True, capture_output=True)

# Shell=True (use with caution!)
result = subprocess.run('ls -la | grep .py', shell=True, capture_output=True, text=True)

# Environment variables
result = subprocess.run(
    ['echo', '$HOME'],
    shell=True,
    capture_output=True,
    text=True,
    env={'HOME': '/custom/home'}
)

# Timeout
result = subprocess.run(['sleep', '10'], timeout=5, capture_output=True, text=True)

# Working directory
result = subprocess.run(['pwd'], cwd='/tmp', capture_output=True, text=True)
print(f'Current dir: {result.stdout}')

# Real-world examples

# Get disk usage
df = subprocess.run(['df', '-h'], capture_output=True, text=True)
print(df.stdout)

# Check if service is running
def is_service_running(service_name):
    result = subprocess.run(
        ['systemctl', 'is-active', service_name],
        capture_output=True,
        text=True
    )
    return result.stdout.strip() == 'active'

# Get system load
load = subprocess.run(['uptime'], capture_output=True, text=True)
print(f'Load: {load.stdout.strip()}')

# Restart service
def restart_service(service_name):
    try:
        subprocess.run(['sudo', 'systemctl', 'restart', service_name], check=True)
        return True
    except subprocess.CalledProcessError:
        return False

# Get list of running processes
ps = subprocess.run(['ps', 'aux'], capture_output=True, text=True)
for line in ps.stdout.split('\n')[:10]:
    print(line)

# Find files
find = subprocess.run(
    ['find', '/var/log', '-name', '*.log', '-type', 'f', '-mtime', '-1'],
    capture_output=True,
    text=True
)
recent_logs = find.stdout.strip().split('\n')
print(f'Recent logs: {len(recent_logs)}')

import os

# Get environment variables
home = os.environ.get('HOME', '/default/home')
path = os.environ['PATH']

# Check if variable exists
if 'DEBUG' in os.environ:
    debug_mode = os.environ['DEBUG'] == 'true'

# Set environment variable (affects current process only)
os.environ['MY_APP_ENV'] = 'production'

# Get all variables
for key, value in os.environ.items():
    print(f'{key}={value}')

# Safe access with default
database_url = os.environ.get('DATABASE_URL', 'postgresql://localhost/mydb')

# Load .env file
def load_dotenv(filepath='.env'):
    if os.path.exists(filepath):
        with open(filepath, 'r') as f:
            for line in f:
                line = line.strip()
                if line and not line.startswith('#'):
                    key, value = line.split('=', 1)
                    os.environ[key] = value

# Example .env file content:
# DATABASE_URL=postgresql://user:pass@localhost/db
# API_KEY=abc123
# DEBUG=true

import os
import platform
import sys

# Operating system
print(f'OS: {os.name}')           # 'posix', 'nt', 'java'
print(f'Platform: {sys.platform}') # 'linux', 'win32', 'darwin'
print(f'System: {platform.system()}')  # 'Linux', 'Windows', 'Darwin'

# CPU info
print(f'CPU count: {os.cpu_count()}')
print(f'Architecture: {platform.machine()}')

# User info
print(f'User: {os.getlogin()}')
print(f'User ID: {os.getuid()}')
print(f'Home directory: {os.path.expanduser("~")}')

# Process info
print(f'Process ID: {os.getpid()}')
print(f'Parent PID: {os.getppid()}')

# Working directory
print(f'Current directory: {os.getcwd()}')
os.chdir('/tmp')
print(f'New directory: {os.getcwd()}')

# System paths
print(f'Path separator: {os.pathsep}')
print(f'Line separator: {repr(os.linesep)}')

# Environment variables for paths
print(f'PATH: {os.environ.get("PATH")}')

# Check if running as root/sudo
def is_admin():
    if platform.system() == 'Windows':
        import ctypes
        return ctypes.windll.shell32.IsUserAnAdmin() != 0
    else:
        return os.geteuid() == 0

print(f'Is admin: {is_admin()}')

import os
import signal
import psutil  # pip install psutil

# Get current process
process = psutil.Process()
print(f'PID: {process.pid}')
print(f'Name: {process.name()}')
print(f'Memory: {process.memory_info().rss / 1024 / 1024:.2f} MB')
print(f'CPU percent: {process.cpu_percent()}%')

# Get all processes
for proc in psutil.process_iter(['pid', 'name', 'cpu_percent', 'memory_percent']):
    try:
        print(f"{proc.info['pid']:6} {proc.info['name']:20} CPU: {proc.info['cpu_percent']:5}% MEM: {proc.info['memory_percent']:.1f}%")
    except (psutil.NoSuchProcess, psutil.AccessDenied):
        pass

# Kill a process by name
def kill_process_by_name(name):
    for proc in psutil.process_iter(['pid', 'name']):
        if proc.info['name'] == name:
            proc.kill()
            print(f'Killed {name} (PID: {proc.info["pid"]})')
            return True
    return False

# Check if process is running
def is_process_running(name):
    for proc in psutil.process_iter(['name']):
        if proc.info['name'] == name:
            return True
    return False

# Send custom signal
def send_signal(pid, sig):
    os.kill(pid, sig)

# Example: reload nginx (SIGHUP)
# find_nginx_pid() then os.kill(pid, signal.SIGHUP)

#!/usr/bin/env python3
"""
system_health.py - Comprehensive system health monitor
"""

import os
import sys
import subprocess
import json
import psutil
import platform
import argparse
from datetime import datetime
from pathlib import Path

class SystemHealth:
    """Monitor system health and resources"""
    
    def __init__(self, threshold_cpu=80, threshold_memory=80, threshold_disk=85):
        self.threshold_cpu = threshold_cpu
        self.threshold_memory = threshold_memory
        self.threshold_disk = threshold_disk
        self.alerts = []
    
    def get_cpu_info(self):
        """Get CPU information"""
        cpu_percent = psutil.cpu_percent(interval=1)
        cpu_count = psutil.cpu_count()
        cpu_freq = psutil.cpu_freq()
        
        return {
            'usage_percent': cpu_percent,
            'cores_physical': cpu_count,
            'cores_logical': psutil.cpu_count(logical=True),
            'frequency_mhz': round(cpu_freq.current, 2) if cpu_freq else None,
            'per_core': psutil.cpu_percent(interval=1, percpu=True)
        }
    
    def get_memory_info(self):
        """Get memory information"""
        mem = psutil.virtual_memory()
        swap = psutil.swap_memory()
        
        return {
            'total_gb': round(mem.total / (1024**3), 2),
            'available_gb': round(mem.available / (1024**3), 2),
            'used_gb': round(mem.used / (1024**3), 2),
            'used_percent': mem.percent,
            'swap_total_gb': round(swap.total / (1024**3), 2),
            'swap_used_gb': round(swap.used / (1024**3), 2),
            'swap_used_percent': swap.percent if swap.total > 0 else 0
        }
    
    def get_disk_info(self):
        """Get disk information"""
        disks = []
        for partition in psutil.disk_partitions():
            try:
                usage = psutil.disk_usage(partition.mountpoint)
                disks.append({
                    'device': partition.device,
                    'mountpoint': partition.mountpoint,
                    'fstype': partition.fstype,
                    'total_gb': round(usage.total / (1024**3), 2),
                    'used_gb': round(usage.used / (1024**3), 2),
                    'free_gb': round(usage.free / (1024**3), 2),
                    'used_percent': usage.percent
                })
            except PermissionError:
                continue
        
        return disks
    
    def get_network_info(self):
        """Get network information"""
        net = psutil.net_io_counters()
        interfaces = {}
        
        for name, stats in psutil.net_if_stats().items():
            interfaces[name] = {
                'is_up': stats.isup,
                'speed': stats.speed,
                'mtu': stats.mtu
            }
        
        return {
            'bytes_sent_gb': round(net.bytes_sent / (1024**3), 2),
            'bytes_recv_gb': round(net.bytes_recv / (1024**3), 2),
            'packets_sent': net.packets_sent,
            'packets_recv': net.packets_recv,
            'interfaces': interfaces
        }
    
    def get_top_processes(self, n=10):
        """Get top CPU and memory consuming processes"""
        processes = []
        for proc in psutil.process_iter(['pid', 'name', 'cpu_percent', 'memory_percent', 'memory_rss', 'status']):
            try:
                processes.append(proc.info)
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                continue
        
        # Sort by CPU
        by_cpu = sorted(processes, key=lambda x: x.get('cpu_percent', 0), reverse=True)[:n]
        # Sort by memory
        by_memory = sorted(processes, key=lambda x: x.get('memory_percent', 0), reverse=True)[:n]
        
        return {'by_cpu': by_cpu, 'by_memory': by_memory}
    
    def get_system_info(self):
        """Get general system information"""
        boot_time = datetime.fromtimestamp(psutil.boot_time())
        uptime = datetime.now() - boot_time
        
        return {
            'hostname': platform.node(),
            'system': platform.system(),
            'release': platform.release(),
            'version': platform.version(),
            'architecture': platform.machine(),
            'python_version': sys.version,
            'boot_time': boot_time.isoformat(),
            'uptime_days': uptime.days,
            'uptime_hours': uptime.seconds // 3600,
            'load_avg': psutil.getloadavg() if hasattr(psutil, 'getloadavg') else None
        }
    
    def check_thresholds(self, data):
        """Check if any metrics exceed thresholds"""
        alerts = []
        
        # CPU check
        cpu_percent = data['cpu']['usage_percent']
        if cpu_percent > self.threshold_cpu:
            alerts.append(f"⚠️ CPU usage is {cpu_percent}% (threshold: {self.threshold_cpu}%)")
        
        # Memory check
        mem_percent = data['memory']['used_percent']
        if mem_percent > self.threshold_memory:
            alerts.append(f"⚠️ Memory usage is {mem_percent}% (threshold: {self.threshold_memory}%)")
        
        # Disk checks
        for disk in data['disks']:
            if disk['used_percent'] > self.threshold_disk:
                alerts.append(f"⚠️ Disk {disk['device']} at {disk['mountpoint']} is {disk['used_percent']}% full")
        
        return alerts
    
    def generate_report(self):
        """Generate complete health report"""
        report = {
            'timestamp': datetime.now().isoformat(),
            'system': self.get_system_info(),
            'cpu': self.get_cpu_info(),
            'memory': self.get_memory_info(),
            'disks': self.get_disk_info(),
            'network': self.get_network_info(),
            'top_processes': self.get_top_processes(5),
            'alerts': []
        }
        
        report['alerts'] = self.check_thresholds(report)
        
        return report
    
    def print_report(self):
        """Print formatted report to console"""
        report = self.generate_report()
        
        print("\n" + "="*70)
        print(f"📊 SYSTEM HEALTH REPORT - {report['timestamp']}")
        print("="*70)
        
        # System info
        sys_info = report['system']
        print(f"\n🖥️  SYSTEM")
        print(f"   Hostname: {sys_info['hostname']}")
        print(f"   OS: {sys_info['system']} {sys_info['release']}")
        print(f"   Architecture: {sys_info['architecture']}")
        print(f"   Uptime: {sys_info['uptime_days']}d {sys_info['uptime_hours']}h")
        
        # CPU
        cpu = report['cpu']
        print(f"\n💻 CPU")
        print(f"   Usage: {cpu['usage_percent']}%")
        print(f"   Cores: {cpu['cores_physical']} physical, {cpu['cores_logical']} logical")
        if cpu['per_core']:
            cores_str = ' '.join([f"{c}%" for c in cpu['per_core'][:8]])
            print(f"   Per core: {cores_str}...")
        
        # Memory
        mem = report['memory']
        print(f"\n💾 MEMORY")
        print(f"   Used: {mem['used_gb']}GB / {mem['total_gb']}GB ({mem['used_percent']}%)")
        print(f"   Available: {mem['available_gb']}GB")
        if mem['swap_total_gb'] > 0:
            print(f"   Swap: {mem['swap_used_gb']}GB / {mem['swap_total_gb']}GB ({mem['swap_used_percent']}%)")
        
        # Disk
        print(f"\n💿 DISK")
        for disk in report['disks']:
            bar = '█' * int(disk['used_percent'] / 5) + '░' * (20 - int(disk['used_percent'] / 5))
            print(f"   {disk['device']:12} [{bar}] {disk['used_percent']:.0f}%")
            print(f"      {disk['mountpoint']}: {disk['used_gb']}GB / {disk['total_gb']}GB")
        
        # Top processes
        print(f"\n📈 TOP PROCESSES BY CPU")
        for proc in report['top_processes']['by_cpu'][:5]:
            cpu_pct = proc.get('cpu_percent', 0)
            mem_pct = proc.get('memory_percent', 0)
            print(f"   {proc['name']:25} CPU: {cpu_pct:5.1f}%  MEM: {mem_pct:5.1f}% (PID: {proc['pid']})")
        
        # Alerts
        if report['alerts']:
            print(f"\n⚠️  ALERTS")
            for alert in report['alerts']:
                print(f"   {alert}")
        
        print("\n" + "="*70 + "\n")
    
    def save_report(self, filepath='health_report.json'):
        """Save report to JSON file"""
        report = self.generate_report()
        with open(filepath, 'w') as f:
            json.dump(report, f, indent=2)
        print(f"✓ Report saved to {filepath}")


def main():
    parser = argparse.ArgumentParser(description='System Health Monitor')
    parser.add_argument('--save', '-s', action='store_true', help='Save report to file')
    parser.add_argument('--output', '-o', default='health_report.json', help='Output file path')
    parser.add_argument('--threshold-cpu', type=int, default=80, help='CPU alert threshold')
    parser.add_argument('--threshold-mem', type=int, default=80, help='Memory alert threshold')
    parser.add_argument('--threshold-disk', type=int, default=85, help='Disk alert threshold')
    
    args = parser.parse_args()
    
    monitor = SystemHealth(
        threshold_cpu=args.threshold_cpu,
        threshold_memory=args.threshold_mem,
        threshold_disk=args.threshold_disk
    )
    
    monitor.print_report()
    
    if args.save:
        monitor.save_report(args.output)


if __name__ == '__main__':
    main()