别再手动巡检了：用 Python 把服务器监控和告警自动化起来

pip install psutil requests schedule

from datetime import datetime

import psutil

print("== CPU ==")
print("逻辑核心数:", psutil.cpu_count())
print("物理核心数:", psutil.cpu_count(logical=False))
print(f"CPU 使用率(采样1秒): {psutil.cpu_percent(interval=1)}%")

print("\n== 内存 ==")
mem = psutil.virtual_memory()
print(f"总计 {mem.total / 1024**3:.1f} GB，已用 {mem.used / 1024**3:.1f} GB，使用率 {mem.percent}%")

print("\n== 磁盘 ==")
for part in psutil.disk_partitions():
    if "cdrom" in part.opts or not part.fstype:
        continue
    u = psutil.disk_usage(part.mountpoint)
    print(f"{part.device:<10} 总 {u.total / 1024**3:7.1f} GB，已用 {u.percent}%")

print("\n== 开机时间 ==")
boot = datetime.fromtimestamp(psutil.boot_time())
print("本次开机于:", boot.strftime("%Y-%m-%d %H:%M:%S"))

print("\n== 内存占用 Top 5 进程 ==")
for p in sorted(
    psutil.process_iter(["name", "memory_percent"]),
    key=lambda x: x.info["memory_percent"] or 0,
    reverse=True,
)[:5]:
    print(f"  {p.info['name'] or '?':<30} {p.info['memory_percent'] or 0:.1f}%")

== CPU ==
逻辑核心数: 16
物理核心数: 8
CPU 使用率(采样1秒): 11.0%

== 内存 ==
总计 15.3 GB，已用 10.2 GB，使用率 66.5%

== 磁盘 ==
C:\        总   200.0 GB，已用 92.2%
D:\        总   276.6 GB，已用 70.4%
E:\        总   952.7 GB，已用 5.6%

== 开机时间 ==
本次开机于: 2026-09-27 16:20:48

== 内存占用 Top 5 进程 ==
  Qoder.exe                      5.2%
  MemCompression                 5.0%
  QQ.exe                         2.5%
  Qoder.exe                      2.5%
  AutoClaw2.exe                  2.0%

import logging
import logging.handlers
import os

os.makedirs("logs", exist_ok=True)

logger = logging.getLogger("rot_demo")
logger.setLevel(logging.INFO)
handler = logging.handlers.RotatingFileHandler(
    "logs/app.log", maxBytes=10_000, backupCount=3, encoding="utf-8"
)
handler.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s"))
logger.addHandler(handler)

line = "订单 10086 处理成功，耗时 0.42 秒 " + "x" * 400  # 约 460 字节一行
for i in range(40):
    logger.info("第 %d 条: %s", i, line)

print("写完 40 条日志，logs 目录里现在有：")
for f in sorted(os.listdir("logs")):
    print(f"  {f:<12} {os.path.getsize(os.path.join('logs', f)):>6} 字节")

写完 40 条日志，logs 目录里现在有：
  app.log        9780 字节
  app.log.1      9770 字节

import time

import schedule


def heartbeat():
    print(time.strftime("%H:%M:%S"), "心跳：监控进程还活着")


schedule.every(2).seconds.do(heartbeat)

# 以下写法真实可用：
# schedule.every().hour.do(heartbeat)                 # 每小时
# schedule.every().day.at("09:00").do(heartbeat)      # 每天 09:00
# schedule.every().monday.at("08:30").do(heartbeat)   # 每周一 08:30

while True:
    schedule.run_pending()
    time.sleep(0.5)

20:39:46 心跳：监控进程还活着
20:39:48 心跳：监控进程还活着
20:39:50 心跳：监控进程还活着

import argparse
import logging
import logging.handlers
import os
import time
from datetime import datetime

import psutil
import requests

# ---------- 配置：默认值 + 环境变量覆盖 ----------
CPU_THRESHOLD = float(os.environ.get("MONITOR_CPU_THRESHOLD", "80"))
MEM_THRESHOLD = float(os.environ.get("MONITOR_MEM_THRESHOLD", "85"))
DISK_THRESHOLD = float(os.environ.get("MONITOR_DISK_THRESHOLD", "90"))
WEBHOOK_URL = os.environ.get("MONITOR_WEBHOOK_URL", "")
# -------------------------------------------------

LOG_DIR = "logs"
os.makedirs(LOG_DIR, exist_ok=True)

logger = logging.getLogger("monitor")
logger.setLevel(logging.INFO)

_file_handler = logging.handlers.RotatingFileHandler(
    os.path.join(LOG_DIR, "monitor.log"),
    maxBytes=5 * 1024 * 1024,
    backupCount=3,
    encoding="utf-8",
)
_file_handler.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s"))
logger.addHandler(_file_handler)

_console_handler = logging.StreamHandler()
_console_handler.setFormatter(logging.Formatter("%(message)s"))
logger.addHandler(_console_handler)


def notify(text: str) -> None:
    """把告警发到群机器人；未配置 WEBHOOK_URL 时只写日志。"""
    logger.warning(text)
    if not WEBHOOK_URL:
        print("(未配置 MONITOR_WEBHOOK_URL，以上告警只写入了日志)")
        return
    try:
        resp = requests.post(
            WEBHOOK_URL,
            json={"msgtype": "text", "text": {"content": text}},
            timeout=10,  # 网络请求必须给超时，否则告警卡住会拖垮整个监控
        )
        resp.raise_for_status()
    except requests.RequestException as exc:
        # 告警发不出去也要留下痕迹，但不能让监控本身挂掉
        logger.error("告警发送失败: %s", exc)


def check_cpu() -> str | None:
    usage = psutil.cpu_percent(interval=1)
    print(f"CPU 使用率: {usage:.1f}%")
    if usage >= CPU_THRESHOLD:
        return f"- CPU 使用率 {usage:.1f}%，超过阈值 {CPU_THRESHOLD}%"
    return None


def check_memory() -> str | None:
    mem = psutil.virtual_memory()
    print(
        f"内存使用率: {mem.percent:.1f}%"
        f"（已用 {mem.used / 1024**3:.1f} GB / 共 {mem.total / 1024**3:.1f} GB）"
    )
    if mem.percent >= MEM_THRESHOLD:
        return f"- 内存使用率 {mem.percent:.1f}%，超过阈值 {MEM_THRESHOLD}%"
    return None


def check_disks() -> list[str]:
    alerts = []
    for part in psutil.disk_partitions():
        if "cdrom" in part.opts or not part.fstype:
            continue
        usage = psutil.disk_usage(part.mountpoint)
        print(f"磁盘 {part.device:<8} 已用 {usage.percent:.1f}%")
        if usage.percent >= DISK_THRESHOLD:
            alerts.append(f"- 磁盘 {part.device} 已用 {usage.percent:.1f}%，超过阈值 {DISK_THRESHOLD}%")
    return alerts


def top_by_memory(n: int = 5) -> list[dict]:
    procs = []
    for p in psutil.process_iter(["pid", "name", "memory_percent"]):
        procs.append(p.info)
    procs.sort(key=lambda x: x["memory_percent"] or 0, reverse=True)
    return procs[:n]


def run_once() -> None:
    print(f"== 检查开始 {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} ==")
    alerts = [a for a in (check_cpu(), check_memory()) if a]
    alerts.extend(check_disks())

    if alerts:
        lines = [f"[告警] {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"]
        lines.extend(alerts)
        lines.append("当前内存占用最高的进程:")
        for info in top_by_memory():
            mp = info["memory_percent"] or 0
            lines.append(f"  {info['name'] or '?'} (pid {info['pid']}, {mp:.1f}%)")
        notify("\n".join(lines))
    else:
        logger.info("一切正常，未触发告警")
    print("== 检查结束 ==\n")


def main() -> None:
    parser = argparse.ArgumentParser(description="简单实用的主机监控告警脚本")
    parser.add_argument(
        "--loop", type=int, default=0, metavar="秒",
        help="每 N 秒检查一轮；不传则只检查一次",
    )
    args = parser.parse_args()

    if args.loop <= 0:
        run_once()
        return

    import schedule  # 只在常驻模式下需要

    schedule.every(args.loop).seconds.do(run_once)
    print(f"已进入常驻模式：每 {args.loop} 秒检查一轮，Ctrl+C 退出")
    while True:
        schedule.run_pending()
        time.sleep(1)


if __name__ == "__main__":
    main()

== 检查开始 2026-09-28 20:39:51 ==
CPU 使用率: 10.4%
内存使用率: 66.7%（已用 10.2 GB / 共 15.3 GB）
磁盘 C:\      已用 92.2%
磁盘 D:\      已用 70.4%
磁盘 E:\      已用 5.6%
(未配置 MONITOR_WEBHOOK_URL，以上告警只写入了日志)
== 检查结束 ==

[告警] 2026-09-28 20:39:52
- 磁盘 C:\ 已用 92.2%，超过阈值 90.0%
当前内存占用最高的进程:
  Qoder.exe (pid 19292, 5.3%)
  MemCompression (pid 4016, 5.0%)
  QQ.exe (pid 272, 2.5%)
  Qoder.exe (pid 3044, 2.5%)
  AutoClaw2.exe (pid 5428, 2.0%)

$env:MONITOR_CPU_THRESHOLD='1'; $env:MONITOR_MEM_THRESHOLD='1'; python monitor.py

[告警] 2026-09-28 20:39:55
- CPU 使用率 10.6%，超过阈值 1.0%
- 内存使用率 66.6%，超过阈值 1.0%
- 磁盘 C:\ 已用 92.2%，超过阈值 90.0%
当前内存占用最高的进程:
  Qoder.exe (pid 19292, 5.2%)
  ...