1
import subprocess
2

3
# 等价于 Shell 的 ps aux | grep nginx
4
result = subprocess.run(
5
    "ps aux | grep nginx",
6
    shell=True,
7
    capture_output=True,
8
    text=True
9
)
10
print(result.stdout)

1
import subprocess
2

3
# 调用已有的备份脚本，传入参数
4
script_path = "./backup.sh"
5
result = subprocess.run(
6
    ["bash", script_path, "/data", "30"],
7
    capture_output=True,
8
    text=True
9
)
10

11
if result.returncode == 0:
12
    print("备份脚本执行成功")
13
    # 可以进一步解析脚本输出，记录到日志系统
14
else:
15
    print("备份脚本执行失败：", result.stderr)

1
import subprocess
2

3
# 超时控制：如果命令超过5秒未完成，自动杀死
4
try:
5
    result = subprocess.run(
6
        ["sleep", "10"],
7
        capture_output=True,
8
        text=True,
9
        timeout=5
10
    )
11
except subprocess.TimeoutExpired:
12
    print("命令执行超时，已强制终止")
13

14
# 处理大量输出：使用流式读取，避免内存溢出
15
proc = subprocess.Popen(
16
    ["find", "/", "-type", "f"],
17
    stdout=subprocess.PIPE,
18
    stderr=subprocess.PIPE,
19
    text=True
20
)
21

22
# 逐行处理输出
23
for line in proc.stdout:
24
    if "important" in line:
25
        print(line.strip())
26

27
proc.wait()

1
import psutil

1
import psutil
2
import time
3

4
def check_system_health():
5
    """系统健康检查，返回告警列表"""
6
    alerts = []
7

8
    # 检查CPU使用率
9
    cpu_percent = psutil.cpu_percent(interval=1)
10
    if cpu_percent > 80:
11
        alerts.append(f"⚠️ CPU使用率过高：{cpu_percent}%")
12

13
    # 检查内存使用率
14
    mem = psutil.virtual_memory()
15
    if mem.percent > 85:
16
        alerts.append(f"⚠️ 内存使用率过高：{mem.percent}%")
17

18
    # 检查磁盘使用率
19
    for part in psutil.disk_partitions():
20
        usage = psutil.disk_usage(part.mountpoint)
21
        if usage.percent > 90:
22
            alerts.append(f"⚠️ 磁盘{part.mountpoint}使用率过高：{usage.percent}%")
23

24
    # 检查是否有僵尸进程
25
    zombie_count = len([p for p in psutil.process_iter() if p.status() == psutil.STATUS_ZOMBIE])
26
    if zombie_count > 0:
27
        alerts.append(f"⚠️ 检测到{zombie_count}个僵尸进程")
28

29
    return alerts
30

31
# 定时检查
32
while True:
33
    alerts = check_system_health()
34
    if alerts:
35
        for alert in alerts:
36
            print(alert)
37
    else:
38
        print("✅ 系统状态正常")
39

40
    time.sleep(60)  # 每分钟检查一次

1
import requests

1
import requests
2

3
# 创建会话对象，复用连接，提高性能
4
session = requests.Session()
5

6
# 设置通用headers
7
session.headers.update({
8
    "User-Agent": "Python-Monitor/1.0",
9
    "X-Custom-Header": "value"
10
})
11

12
# 多个请求复用连接
13
urls = [
14
    "http://api1.example.com/status",
15
    "http://api2.example.com/status",
16
    "http://api3.example.com/status"
17
]
18

19
for url in urls:
20
    try:
21
        resp = session.get(url, timeout=5)
22
        print(f"{url}: {resp.status_code}")
23
    except Exception as e:
24
        print(f"{url}: 请求失败 - {e}")
25

26
session.close()  # 关闭会话

1
import paramiko

1
def ssh_exec_cmd(host, port, username, password, cmd):
2
    """
3
    远程执行命令，返回 (是否成功, 输出内容, 错误内容)
4
    """
5
    ssh = paramiko.SSHClient()
6
    # 自动添加主机密钥，避免首次连接提示
7
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
8

9
    try:
10
        # 建立连接
11
        ssh.connect(hostname=host, port=port, username=username, password=password, timeout=10)
12
        # 执行命令
13
        stdin, stdout, stderr = ssh.exec_command(cmd)
14
        # 获取输出
15
        out = stdout.read().decode('utf-8')
16
        err = stderr.read().decode('utf-8')
17
        # 获取返回码
18
        exit_code = stdout.channel.recv_exit_status()
19

20
        return exit_code == 0, out, err
21
    except Exception as e:
22
        return False, "", str(e)
23
    finally:
24
        ssh.close()
25

26
# 调用示例
27
success, output, error = ssh_exec_cmd(
28
    host="192.168.1.100",
29
    port=22,
30
    username="root",
31
    password="your_password",
32
    cmd="df -h /"
33
)
34

35
if success:
36
    print("执行成功：", output)
37
else:
38
    print("执行失败：", error)

1
def sftp_upload(host, port, username, password, local_file, remote_path):
2
    """上传本地文件到远程服务器"""
3
    transport = paramiko.Transport((host, port))
4
    try:
5
        transport.connect(username=username, password=password)
6
        sftp = paramiko.SFTPClient.from_transport(transport)
7
        sftp.put(local_file, remote_path)
8
        sftp.close()
9
        print(f"文件{local_file}上传到{host}:{remote_path}成功")
10
        return True
11
    except Exception as e:
12
        print(f"上传失败：{e}")
13
        return False
14
    finally:
15
        transport.close()
16

17
def sftp_download(host, port, username, password, remote_file, local_path):
18
    """从远程服务器下载文件到本地"""
19
    transport = paramiko.Transport((host, port))
20
    try:
21
        transport.connect(username=username, password=password)
22
        sftp = paramiko.SFTPClient.from_transport(transport)
23
        sftp.get(remote_file, local_path)
24
        sftp.close()
25
        print(f"文件{remote_file}下载到{local_path}成功")
26
        return True
27
    except Exception as e:
28
        print(f"下载失败：{e}")
29
        return False
30
    finally:
31
        transport.close()

1
from concurrent.futures import ThreadPoolExecutor, as_completed
2
import time
3

4
def batch_execute(hosts, cmd, max_workers=5):
5
    """
6
    并发批量执行命令，提高效率
7
    hosts: [{"host": "IP", "port": 22, "user": "root", "pwd": "xxx"}, ...]
8
    """
9
    results = {}
10

11
    def execute_one(server):
12
        success, out, err = ssh_exec_cmd(
13
            server["host"],
14
            server["port"],
15
            server["user"],
16
            server["pwd"],
17
            cmd
18
        )
19
        return server["host"], success, out, err
20

21
    # 使用线程池并发执行
22
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
23
        futures = {executor.submit(execute_one, srv): srv for srv in hosts}
24

25
        for future in as_completed(futures):
26
            host, success, out, err = future.result()
27
            results[host] = (success, out, err)
28

29
            # 实时输出进度
30
            if success:
31
                print(f"✅ {host} 执行成功")
32
            else:
33
                print(f"❌ {host} 执行失败：{err[:50]}")
34

35
    return results
36

37
# 使用示例
38
hosts = [
39
    {"host": "192.168.1.101", "port": 22, "user": "root", "pwd": "xxx"},
40
    {"host": "192.168.1.102", "port": 22, "user": "root", "pwd": "xxx"},
41
    {"host": "192.168.1.103", "port": 22, "user": "root", "pwd": "xxx"},
42
]
43

44
cmd = "uptime"
45
results = batch_execute(hosts, cmd, max_workers=3)
46

47
# 统计结果
48
success_count = sum(1 for _, (ok, _, _) in results.items() if ok)
49
print(f"\n执行完成：{success_count}/{len(hosts)}台服务器成功")

1
import logging
2

3
# 日志基础配置
4
logging.basicConfig(
5
    level=logging.INFO,  # 设置最低输出级别，低于该级别的日志不显示
6
    format="%(asctime)s [%(levelname)s] %(message)s",  # 日志格式
7
    datefmt="%Y-%m-%d %H:%M:%S"  # 时间格式
8
)
9

10
# 输出不同级别的日志
11
logging.debug("这是调试信息")
12
logging.info("任务开始执行")
13
logging.warning("磁盘使用率超过80%")
14
logging.error("备份执行失败")
15
logging.critical("系统严重故障")

1
import logging
2
from logging.handlers import RotatingFileHandler
3

4
logger = logging.getLogger()
5
logger.setLevel(logging.INFO)
6

7
# 日志格式
8
formatter = logging.Formatter(
9
    "%(asctime)s [%(levelname)s] %(name)s - %(message)s",
10
    datefmt="%Y-%m-%d %H:%M:%S"
11
)
12

13
# 1. 终端输出处理器
14
stream_handler = logging.StreamHandler()
15
stream_handler.setFormatter(formatter)
16
logger.addHandler(stream_handler)
17

18
# 2. 文件输出处理器（自动轮转，单文件最大10MB，保留5个备份）
19
file_handler = RotatingFileHandler(
20
    "run.log",
21
    maxBytes=10*1024*1024,
22
    backupCount=5,
23
    encoding="utf-8"
24
)
25
file_handler.setFormatter(formatter)
26
logger.addHandler(file_handler)
27

28
# 使用
29
logger.info("服务启动成功")
30
logger.error("数据库连接失败")

1
import logging
2

3
def get_logger(name):
4
    """获取特定模块的日志记录器"""
5
    logger = logging.getLogger(name)
6
    if not logger.handlers:
7
        handler = logging.FileHandler(f"logs/{name}.log")
8
        formatter = logging.Formatter(
9
            "%(asctime)s [%(levelname)s] %(message)s",
10
            datefmt="%Y-%m-%d %H:%M:%S"
11
        )
12
        handler.setFormatter(formatter)
13
        logger.addHandler(handler)
14
        logger.setLevel(logging.INFO)
15
    return logger
16

17
# 在不同模块中使用
18
# module_backup.py
19
logger_backup = get_logger("backup")
20
logger_backup.info("备份开始")
21

22
# module_monitor.py
23
logger_monitor = get_logger("monitor")
24
logger_monitor.error("CPU告警")

1
import json
2

3
# JSON字符串转Python字典
4
json_str = '{"code": 200, "data": {"status": "running", "cpu": 45}}'
5
data = json.loads(json_str)
6

7
# 直接按键取值，无需正则截取
8
print("状态码：", data["code"])
9
print("服务状态：", data["data"]["status"])
10
print("CPU使用率：", data["data"]["cpu"])
11

12
# Python字典转JSON字符串
13
result = {"success": True, "message": "执行完成"}
14
json_result = json.dumps(result, ensure_ascii=False, indent=2)
15
print(json_result)
16

17
# 安全处理不存在的键
18
data = {"code": 200}
19
cpu = data.get("cpu", "未知")  # 键不存在时返回默认值
20
print("CPU：", cpu)

1
import json
2
import logging
3

4
logger = logging.getLogger(__name__)
5

6
def load_config(config_file):
7
    """加载JSON配置文件，支持容错"""
8
    try:
9
        with open(config_file, "r", encoding="utf-8") as f:
10
            config = json.load(f)
11

12
        # 配置校验
13
        required_keys = ["servers", "timeout", "retry"]
14
        for key in required_keys:
15
            if key not in config:
16
                raise ValueError(f"配置文件缺少必须字段：{key}")
17

18
        logger.info(f"配置文件加载成功，包含{len(config['servers'])}个服务器")
19
        return config
20
    except json.JSONDecodeError as e:
21
        logger.error(f"配置文件格式错误：{e}")
22
        return None
23
    except Exception as e:
24
        logger.error(f"加载配置文件失败：{e}")
25
        return None
26

27
# 配置文件 config.json
28
"""
29
{
30
  "servers": [
31
    {"host": "192.168.1.100", "port": 22},
32
    {"host": "192.168.1.101", "port": 22}
33
  ],
34
  "timeout": 30,
35
  "retry": 3
36
}
37
"""
38

39
config = load_config("config.json")
40
if config:
41
    for server in config["servers"]:
42
        print(f"连接到{server['host']}:{server['port']}")

1
import psutil
2
import requests
3
import subprocess
4
import logging
5
from logging.handlers import RotatingFileHandler
6
import time
7
from datetime import datetime
8

9
# ============= 日志配置 =============
10
logger = logging.getLogger()
11
logger.setLevel(logging.INFO)
12

13
formatter = logging.Formatter(
14
    "%(asctime)s [%(levelname)s] %(message)s",
15
    datefmt="%Y-%m-%d %H:%M:%S"
16
)
17

18
stream_handler = logging.StreamHandler()
19
stream_handler.setFormatter(formatter)
20
logger.addHandler(stream_handler)
21

22
file_handler = RotatingFileHandler(
23
    "monitor.log",
24
    maxBytes=10*1024*1024,
25
    backupCount=5,
26
    encoding="utf-8"
27
)
28
file_handler.setFormatter(formatter)
29
logger.addHandler(file_handler)
30

31
# ============= 告警推送 =============
32
def send_alert(title, content, webhook_url):
33
    """推送告警到企业微信"""
34
    try:
35
        data = {
36
            "msgtype": "markdown",
37
            "markdown": {
38
                "content": f"**{title}**\n\n{content}"
39
            }
40
        }
41
        requests.post(webhook_url, json=data, timeout=5)
42
        logger.info(f"告警已推送：{title}")
43
    except Exception as e:
44
        logger.error(f"告警推送失败：{e}")
45

46
# ============= 系统检查 =============
47
def check_system_health(thresholds, webhook_url):
48
    """综合系统检查，超过阈值触发告警"""
49
    alerts = []
50

51
    # 检查CPU
52
    cpu_percent = psutil.cpu_percent(interval=1)
53
    if cpu_percent > thresholds["cpu"]:
54
        msg = f"CPU使用率告警：{cpu_percent}% (阈值：{thresholds['cpu']}%)"
55
        alerts.append(("CPU告警", msg))
56
        logger.warning(msg)
57

58
    # 检查内存
59
    mem = psutil.virtual_memory()
60
    if mem.percent > thresholds["memory"]:
61
        msg = f"内存使用率告警：{mem.percent}% (阈值：{thresholds['memory']}%)"
62
        alerts.append(("内存告警", msg))
63
        logger.warning(msg)
64

65
    # 检查磁盘
66
    for part in psutil.disk_partitions():
67
        usage = psutil.disk_usage(part.mountpoint)
68
        if usage.percent > thresholds["disk"]:
69
            msg = f"磁盘告警 {part.mountpoint}：{usage.percent}% (阈值：{thresholds['disk']}%)"
70
            alerts.append(("磁盘告警", msg))
71
            logger.warning(msg)
72

73
    # 检查进程
74
    try:
75
        # 检查关键服务是否运行
76
        result = subprocess.run(
77
            ["pgrep", "-a", "nginx"],
78
            capture_output=True,
79
            text=True
80
        )
81
        if result.returncode != 0:
82
            msg = "Nginx服务未运行"
83
            alerts.append(("服务告警", msg))
84
            logger.error(msg)
85
    except Exception as e:
86
        logger.error(f"进程检查异常：{e}")
87

88
    # 推送告警
89
    for title, msg in alerts:
90
        send_alert(title, msg, webhook_url)
91

92
    return len(alerts) > 0
93

94
# ============= 定时监控 =============
95
def monitor_loop(thresholds, webhook_url, interval=60):
96
    """启动监控循环"""
97
    logger.info("监控系统启动")
98

99
    check_count = 0
100
    alert_count = 0
101

102
    try:
103
        while True:
104
            check_count += 1
105

106
            if check_system_health(thresholds, webhook_url):
107
                alert_count += 1
108

109
            # 每小时打印一次统计
110
            if check_count % 60 == 0:
111
                logger.info(f"检查统计：总检查{check_count}次，告警{alert_count}次")
112

113
            time.sleep(interval)
114
    except KeyboardInterrupt:
115
        logger.info("监控系统已停止")
116
    except Exception as e:
117
        logger.critical(f"监控系统异常退出：{e}")
118

119
# ============= 启动程序 =============
120
if __name__ == "__main__":
121
    # 监控阈值配置
122
    thresholds = {
123
        "cpu": 80,      # CPU超过80%告警
124
        "memory": 85,   # 内存超过85%告警
125
        "disk": 90      # 磁盘超过90%告警
126
    }
127

128
    # 企业微信webhook地址（需要替换为实际地址）
129
    webhook_url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx"
130

131
    # 启动监控
132
    monitor_loop(thresholds, webhook_url, interval=60)