Day 11 | Python 调用 Shell

#!/usr/bin/env python3
"""
Day 11 - subprocess 基础示例
演示如何安全地执行系统命令
"""

import subprocess

# ========== 1. 基本命令执行 ==========
print("=== 基本命令执行 ===")

# 定义要执行的命令（使用列表形式，更安全）
cmd = ["ls", "-l", "/tmp"]

try:
    # 执行命令
    # capture_output=True: 捕获标准输出和错误输出
    # text=True: 将字节流自动解码为字符串
    # check=True: 如果命令失败（返回非 0），自动抛出异常
    result = subprocess.run(
        cmd,
        capture_output=True,
        text=True,
        check=True
    )
    
    print("命令执行成功！")
    print(f"返回码: {result.returncode}")  # 0 表示成功
    print(f"标准输出:\n{result.stdout}")
    
except subprocess.CalledProcessError as e:
    # 命令执行失败
    print(f"❌ 命令执行失败！")
    print(f"返回码: {e.returncode}")
    print(f"错误信息: {e.stderr}")

# ========== 2. 不捕获输出的命令 ==========
print("\n=== 不捕获输出（直接打印到终端）===")

# 如果不需要处理输出，可以不捕获
result = subprocess.run(["echo", "Hello from Python!"])
# 输出会直接打印到终端

# ========== 3. 带超时的命令 ==========
print("\n=== 带超时的命令 ===")

try:
    # timeout=5: 5 秒后如果命令未完成，会抛出 TimeoutExpired 异常
    result = subprocess.run(
        ["sleep", "2"],  # 模拟一个需要 2 秒的命令
        timeout=5,
        capture_output=True,
        text=True
    )
    print("✅ 命令在超时前完成")
except subprocess.TimeoutExpired:
    print("❌ 命令执行超时")

# ========== 4. 获取命令返回码 ==========
print("\n=== 获取命令返回码 ===")

# 不设置 check=True，即使失败也不会抛出异常
result = subprocess.run(["ls", "/nonexistent"], capture_output=True, text=True)

if result.returncode == 0:
    print("✅ 命令执行成功")
else:
    print(f"❌ 命令执行失败，返回码: {result.returncode}")
    print(f"错误信息: {result.stderr}")

# ========== 5. 执行 Shell 命令（谨慎使用）==========
print("\n=== 执行 Shell 命令（谨慎使用）===")

# 只有在确实需要 Shell 特性（如管道、重定向）时才使用 shell=True
# 注意：使用 shell=True 时，应该传递字符串而不是列表
result = subprocess.run(
    "echo $HOME | wc -c",  # 使用 Shell 管道
    shell=True,
    capture_output=True,
    text=True
)
print(f"HOME 路径长度: {result.stdout.strip()}")

#!/usr/bin/env python3
"""
Day 11 - 磁盘监控脚本（Python 版本）
将 Shell 脚本重构为 Python，演示 subprocess 的实际应用
"""

import subprocess
import re

def get_disk_usage(mount_point: str = "/") -> int:
    """
    获取指定挂载点的磁盘使用率
    
    参数:
        mount_point: 挂载点路径（默认 "/"）
    
    返回:
        磁盘使用率（百分比，0-100）
    """
    try:
        # 执行 df -h 命令
        # -h: 人类可读格式（KB, MB, GB）
        result = subprocess.run(
            ["df", "-h", mount_point],
            capture_output=True,
            text=True,
            check=True
        )
        
        # 解析输出
        # 输出格式示例:
        # Filesystem      Size  Used Avail Use% Mounted on
        # /dev/sda1        50G   25G   25G  50% /
        lines = result.stdout.strip().split("\n")
        
        if len(lines) < 2:
            raise ValueError("无法解析 df 命令输出")
        
        # 第二行是数据行（第一行是表头）
        data_line = lines[1]
        
        # 分割行，获取各个字段
        # 使用正则表达式提取使用率（去掉 % 符号）
        match = re.search(r'(\d+)%', data_line)
        if match:
            usage_percent = int(match.group(1))
            return usage_percent
        else:
            raise ValueError("无法从输出中提取使用率")
            
    except subprocess.CalledProcessError as e:
        print(f"❌ 执行 df 命令失败: {e}")
        return -1
    except Exception as e:
        print(f"❌ 解析磁盘使用率失败: {e}")
        return -1

def check_disk_health(threshold: int = 80) -> bool:
    """
    检查磁盘健康状态
    
    参数:
        threshold: 告警阈值（默认 80%）
    
    返回:
        True 表示健康，False 表示需要告警
    """
    usage = get_disk_usage()
    
    if usage < 0:
        print("❌ 无法获取磁盘使用率")
        return False
    
    print(f"当前磁盘使用率: {usage}%")
    
    if usage > threshold:
        print(f"🚨 告警: 磁盘使用率超过 {threshold}%！")
        return False
    else:
        print("✅ 磁盘空间充足")
        return True

def get_memory_usage() -> dict:
    """
    获取内存使用情况
    
    返回:
        包含内存信息的字典
    """
    try:
        # 执行 free -m 命令（以 MB 为单位）
        result = subprocess.run(
            ["free", "-m"],
            capture_output=True,
            text=True,
            check=True
        )
        
        # 解析输出
        # 输出格式示例:
        #               total        used        free      shared  buff/cache   available
        # Mem:          8192        4096        2048         256        2048        4096
        lines = result.stdout.strip().split("\n")
        mem_line = lines[1]  # 第二行是内存信息
        
        # 提取数字
        numbers = [int(x) for x in mem_line.split()[1:]]
        total, used, free, shared, buff_cache, available = numbers
        
        return {
            "total": total,
            "used": used,
            "free": free,
            "available": available,
            "usage_percent": round((used / total) * 100, 2)
        }
        
    except Exception as e:
        print(f"❌ 获取内存信息失败: {e}")
        return {}

if __name__ == "__main__":
    print("=" * 50)
    print("系统资源监控")
    print("=" * 50)
    
    # 检查磁盘
    print("\n--- 磁盘检查 ---")
    check_disk_health(threshold=80)
    
    # 检查内存
    print("\n--- 内存检查 ---")
    mem_info = get_memory_usage()
    if mem_info:
        print(f"总内存: {mem_info['total']} MB")
        print(f"已使用: {mem_info['used']} MB")
        print(f"可用: {mem_info['available']} MB")
        print(f"使用率: {mem_info['usage_percent']}%")