写了个Python脚本监控服务器，再也不怕半夜被告警吵醒

import psutil

# CPU使用率（取1秒的采样）
cpu = psutil.cpu_percent(interval=1)

# 内存
mem = psutil.virtual_memory()
print(f"内存使用率: {mem.percent}%")
print(f"可用内存: {mem.available / 1024**3:.1f}GB")

# 磁盘
disk = psutil.disk_usage('/')
print(f"磁盘使用率: {disk.percent}%")

# 网络
net = psutil.net_io_counters()
print(f"累计发送: {net.bytes_sent / 1024**2:.1f}MB")
print(f"累计接收: {net.bytes_recv / 1024**2:.1f}MB")

# 进程级别的监控也行（注意：第一次调用cpu_percent返回0，需调两次）
psutil.process_iter(['pid', 'name', 'cpu_percent'])  # 第一次，预热
time.sleep(1)
for proc in psutil.process_iter(['pid', 'name', 'cpu_percent', 'memory_percent']):
    if proc.info['cpu_percent'] andproc.info['cpu_percent'] >50:
        print(f"高CPU进程: {proc.info['name']} (PID:{proc.info['pid']})")

if cpu>80:
    send_alert("CPU过高！")

# config.py — 监控配置集中管理
THRESHOLDS = {
    'cpu': {
        'warning': 80,       # 提醒级别，不紧急
        'critical': 95,      # 严重级别，需要关注
        'sustained_minutes': 5  # 连续超过阈值N分钟才算真的有问题
    },
    'memory': {
        'warning': 85,
        'critical': 95,
        'sustained_minutes': 3
    },
    'disk': {
        'warning': 85,
        'critical': 95,
        'sustained_minutes': 0  # 磁盘不需要持续判断，超了就是超了
    }
}

# 告警级别定义
ALERT_LEVELS = {
    'info': '仅记录日志',
    'warning': '发群消息，不@人',
    'critical': '发群消息，@值班人'
}

import time
from collections import defaultdict

class AlertStateManager:
    """管理每个指标的告警状态"""

    def __init__(self):
        # 记录每个指标首次超标的时间
        self.first_breach = {}
        # 记录已经告警过的指标，避免重复
        self.alerted = set()

    def check(self, metric_name, value, threshold, sustained_seconds=300):
        """
        检查指标是否需要告警
        返回: None（不告警）, 'warning'（首次告警）, 'repeat'（重复告警）
        """
        key = f"{metric_name}:{threshold}"

        if value>threshold:
            now = time.time()

            if key not in self.first_breach:
                # 第一次超标，记下来，先不报警
                self.first_breach[key] = now
                return None

            # 已经超标一段时间了，判断是否达到持续时长
            duration = now-self.first_breach[key]
            if duration>= sustained_seconds:
                if key not in self.alerted:
                    self.alerted.add(key)
                    return 'warning'
                return 'repeat'  # 持续超标但已经报过了

            return None  # 还没到持续时长，继续观察

        else:
            # 指标恢复正常，清除记录
            if key in self.first_breach:
                del self.first_breach[key]
            if key in self.alerted:
                self.alerted.discard(key)
                return 'recovered'  # 可选：发一条恢复通知
            return None

from collections import defaultdict
import threading

class AlertAggregator:
    """在一个时间窗口内收集告警，合并发送"""

    def __init__(self, window_seconds=60):
        self.window = window_seconds
        self.pending = defaultdict(list)
        self.timer = None
        self.lock = threading.Lock()

    def add(self, host, metric, value, level):
        with self.lock:
            self.pending[host].append({
                'metric': metric,
                'value': value,
                'level': level
            })
            # 每次来新告警都重置定时器（标准debounce策略）
            if self.timer is not None:
                self.timer.cancel()
            self.timer = threading.Timer(self.window, self._flush)
            self.timer.start()

    def _flush(self):
        """把攒了一分钟的告警合并成一条消息发出去"""
        with self.lock:
            if not self.pending:
                self.timer = None
                return

            lines = ["## ⚠️ 服务器告警汇总\n"]
            for host, alerts in self.pending.items():
                lines.append(f"**{host}**：")
                for a in alerts:
                    emoji = "🔴"ifa['level'] == 'critical'else"🟡"
                    lines.append(f"> {emoji} {a['metric']}：{a['value']}")
                lines.append("")

            message = "\n".join(lines)
            send_to_wechat(message)  # 你之前写的通知函数

            self.pending.clear()
            self.timer = None

from datetime import datetime

class QuietHoursManager:
    def __init__(self, quiet_start=23, quiet_end=8):
        self.quiet_start = quiet_start
        self.quiet_end = quiet_end

    def is_quiet_now(self):
        hour = datetime.now().hour
        return hour>= self.quiet_startorhour<self.quiet_end

    def should_alert(self, level):
        """
        静默期间的告警策略：
        - critical 级别：照常告警，该叫就叫
        - warning 级别：只记日志，等早上统一汇报
        - info 级别：直接忽略
        """
        if not self.is_quiet_now():
            returnTrue  # 非静默期，全部告警

        if level == 'critical':
            returnTrue   # 严重问题，半夜也得叫起来

        # warning 和 info 在静默期只记日志
        log_to_file(f"[静默期-抑制] {level} 级别告警已记录，待早上汇报")
        return False

def morning_summary():
    """每天早上8点发送夜间告警汇总"""
    overnight_alerts = get_alerts_since(hours_ago=9)

    if not overnight_alerts:
        send_to_wechat("## ☀️ 早间简报\n> 昨晚一切正常，无告警记录。")
    else:
        lines = [f"## 🌅 夜间告警汇总（共{len(overnight_alerts)}条）\n"]
        for alert in overnight_alerts:
            lines.append(f"> [{alert['time']}] {alert['host']} - {alert['metric']}: {alert['value']}")
        send_to_wechat("\n".join(lines))

import psutil

def check_top_processes(top_n=5):
    """找出占资源最多的进程"""
    processes = []
    for proc in psutil.process_iter(['pid', 'name', 'cpu_percent', 'memory_percent', 'status']):
        try:
            info = proc.info
            if info['cpu_percent'] andinfo['cpu_percent'] >20:
                processes.append(info)
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            continue

    # 按CPU排序，取前N
    processes.sort(key=lambda x: x.get('cpu_percent', 0), reverse=True)
    return processes[:top_n]

def monitor_specific_process(process_name, max_memory_mb=2048):
    """监控特定进程的内存使用，防止内存泄漏"""
    for procinpsutil.process_iter(['pid', 'name', 'memory_info']):
        try:
            if proc.info['name'] == process_name:
                mem_mb = proc.info['memory_info'].rss/1024/1024
                if mem_mb>max_memory_mb:
                    return {
                        'alert': True,
                        'message': f"进程 {process_name}(PID:{proc.info['pid']}) 内存占用 {mem_mb:.0f}MB，超过阈值 {max_memory_mb}MB"
                    }
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            continue
    return {'alert': False}

import time
import psutil

# 初始化各模块
alert_state = AlertStateManager()
aggregator = AlertAggregator(window_seconds=60)
quiet_hours = QuietHoursManager(quiet_start=23, quiet_end=8)

def check_and_alert(metric_name, value, config):
    """统一的告警检查逻辑"""
    host = get_hostname()

    # 先检查critical级别
    status = alert_state.check(
        f'{metric_name}_critical', value,
        config['critical'],
        sustained_seconds=config['sustained_minutes'] *60
    )
    if status == 'warning':
        if quiet_hours.should_alert('critical'):
            aggregator.add(host, metric_name, f'{value}%（严重）', 'critical')
        return  # critical已触发，不用再检查warning

    # 再检查warning级别
    status = alert_state.check(
        f'{metric_name}_warning', value,
        config['warning'],
        sustained_seconds=config['sustained_minutes'] *60
    )
    if status == 'warning':
        if quiet_hours.should_alert('warning'):
            aggregator.add(host, metric_name, f'{value}%', 'warning')

def run_monitor(interval=60):
    """主监控循环"""
    print("监控已启动，按 Ctrl+C 停止...")

    whileTrue:
        try:
            cpu = psutil.cpu_percent(interval=1)
            mem = psutil.virtual_memory().percent
            disk = psutil.disk_usage('/').percent

            check_and_alert('CPU', cpu, THRESHOLDS['cpu'])
            check_and_alert('内存', mem, THRESHOLDS['memory'])
            # 磁盘sustained_minutes=0，超了就报
            check_and_alert('磁盘', disk, THRESHOLDS['disk'])

            # 检查关键进程
            java_status = monitor_specific_process('java', max_memory_mb=3072)
            if java_status.get('alert'):
                if quiet_hours.should_alert('warning'):
                    aggregator.add(get_hostname(), 'Java进程',
                                   java_status['message'], 'warning')

        except Exceptionase:
            # 监控脚本自身出错也要通知，不能悄悄挂了
            send_to_wechat("监控脚本异常: {e}")

        time.sleep(interval)

# 启动时预热一次（interval=1 会阻塞1秒，自带采样）
psutil.cpu_percent(interval=1)
# 从第二次调用开始才是有效数据

from datetime import datetime
import pytz

def get_local_hour():
    tz = pytz.timezone('Asia/Shanghai')
    return datetime.now(tz).hour

指标采集（每分钟）
    ↓
阈值判断（warning / critical）
    ↓
持续时长检查（连续超标N分钟？）
    ↓ 是                    ↓ 否
告警聚合器            继续观察（计时器清零）
（60秒窗口内合并）
    ↓
静默期判断（当前是夜间？）
    ↓ 是                    ↓ 否
只推送critical         正常推送
warning记日志
    ↓
早上8点发送夜间汇总