pip install psutil

import psutil

# 完整内存信息
mem = psutil.virtual_memory()
print(f"总量: {mem.total / (1024**3):.2f} GB")
print(f"可用: {mem.available / (1024**3):.2f} GB")
print(f"已用: {mem.used / (1024**3):.2f} GB")
print(f"使用率: {mem.percent}%")

import psutil

# 磁盘使用情况
disk = psutil.disk_usage('/')
print(f"总量: {disk.total / (1024**3):.2f} GB")
print(f"已用: {disk.used / (1024**3):.2f} GB")
print(f"可用: {disk.free / (1024**3):.2f} GB")
print(f"使用率: {disk.percent}%")

# 磁盘 I/O 统计
io = psutil.disk_io_counters()
print(f"读取: {io.read_bytes / (1024**2):.2f} MB")
print(f"写入: {io.write_bytes / (1024**2):.2f} MB")

import psutil

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

# 网络连接
connections = psutil.net_connections()
for conn in connections:
    if conn.status == 'ESTABLISHED':
        print(f"连接: {conn.laddr} -> {conn.raddr}")

import psutil

# 获取所有进程
for proc in psutil.process_iter(['pid', 'name', 'cpu_percent', 'memory_percent']):
    if proc.info['cpu_percent'] > 10:  # CPU 使用率 > 10%
        print(f"高 CPU 进程: {proc.info['name']} (PID: {proc.info['pid']})")
        print(f"  CPU: {proc.info['cpu_percent']}%")
        print(f"  内存: {proc.info['memory_percent']:.2f}%")

from dataclasses import dataclass
from typing import Callable, Any
from enum import Enum

class Severity(Enum):
    INFO = "info"
    WARNING = "warning"
    CRITICAL = "critical"

@dataclass
class AlertRule:
    """告警规则定义"""
    name: str
    metric: str           # 监控指标名称
    condition: str        # 条件: "gt", "lt", "eq", "contains"
    threshold: float      # 阈值
    severity: Severity    # 告警级别
    duration: int = 0     # 持续时间（秒），0 表示立即触发
    description: str = "" # 规则描述

class RuleEngine:
    """告警规则引擎"""
    
    def __init__(self):
        self.rules: list[AlertRule] = []
        self.alert_history: list[dict] = []
    
    def add_rule(self, rule: AlertRule):
        """添加告警规则"""
        self.rules.append(rule)
    
    def evaluate(self, rule: AlertRule, value: Any) -> bool:
        """评估规则是否触发"""
        if rule.condition == "gt":
            return value > rule.threshold
        elif rule.condition == "lt":
            return value < rule.threshold
        elif rule.condition == "eq":
            return value == rule.threshold
        elif rule.condition == "gte":
            return value >= rule.threshold
        elif rule.condition == "lte":
            return value <= rule.threshold
        return False
    
    def check_all_rules(self, metrics: dict) -> list[dict]:
        """检查所有规则"""
        alerts = []
        for rule in self.rules:
            value = metrics.get(rule.metric)
            if value is not None and self.evaluate(rule, value):
                alert = {
                    "rule": rule.name,
                    "metric": rule.metric,
                    "value": value,
                    "threshold": rule.threshold,
                    "severity": rule.severity.value,
                    "description": rule.description
                }
                alerts.append(alert)
                self.alert_history.append(alert)
        return alerts

import json
import urllib.request
from abc import ABC, abstractmethod

class Notifier(ABC):
    """通知基类"""
    
    @abstractmethod
    def send(self, alert: dict) -> bool:
        """发送告警通知"""
        pass

class DingTalkNotifier(Notifier):
    """钉钉通知"""
    
    def __init__(self, webhook_url: str):
        self.webhook_url = webhook_url
    
    def send(self, alert: dict) -> bool:
        """发送钉钉消息"""
        severity_emoji = {
            "info": "ℹ️",
            "warning": "⚠️",
            "critical": "🚨"
        }
        
        message = {
            "msgtype": "markdown",
            "markdown": {
                "title": f"{severity_emoji.get(alert['severity'], '📢')} 服务器告警",
                "text": f"""
### {alert['rule']}

- **指标**: {alert['metric']}
- **当前值**: {alert['value']}
- **阈值**: {alert['threshold']}
- **级别**: {alert['severity']}
- **描述**: {alert['description']}
                """
            }
        }
        
        try:
            data = json.dumps(message).encode('utf-8')
            req = urllib.request.Request(
                self.webhook_url,
                data=data,
                headers={'Content-Type': 'application/json'}
            )
            urllib.request.urlopen(req)
            return True
        except Exception as e:
            print(f"钉钉通知失败: {e}")
            return False

class ConsoleNotifier(Notifier):
    """控制台输出（开发测试用）"""
    
    def send(self, alert: dict) -> bool:
        print(f"\n{'='*50}")
        print(f"🚨 告警: {alert['rule']}")
        print(f"   指标: {alert['metric']} = {alert['value']}")
        print(f"   阈值: {alert['threshold']}")
        print(f"   级别: {alert['severity']}")
        print(f"{'='*50}\n")
        return True

import psutil
import time
import json
import sqlite3
from datetime import datetime, timedelta
from pathlib import Path
from typing import Optional
import threading
from queue import Queue

class MetricsCollector:
    """系统指标采集器"""
    
    def __init__(self):
        self._prev_net_io = None
    
    def collect_cpu(self) -> dict:
        """采集 CPU 指标"""
        return {
            "usage_percent": psutil.cpu_percent(interval=1),
            "count": psutil.cpu_count(),
            "freq": psutil.cpu_freq()._asdict() if psutil.cpu_freq() else None
        }
    
    def collect_memory(self) -> dict:
        """采集内存指标"""
        mem = psutil.virtual_memory()
        return {
            "total": mem.total,
            "available": mem.available,
            "used": mem.used,
            "percent": mem.percent
        }
    
    def collect_disk(self, path: str = '/') -> dict:
        """采集磁盘指标"""
        disk = psutil.disk_usage(path)
        io = psutil.disk_io_counters()
        return {
            "total": disk.total,
            "used": disk.used,
            "free": disk.free,
            "percent": disk.percent,
            "read_bytes": io.read_bytes if io else 0,
            "write_bytes": io.write_bytes if io else 0
        }
    
    def collect_network(self) -> dict:
        """采集网络指标"""
        io = psutil.net_io_counters()
        
        # 计算速率（需要保存上次的值）
        sent_rate = 0
        recv_rate = 0
        if self._prev_net_io:
            sent_rate = (io.bytes_sent - self._prev_net_io.bytes_sent)
            recv_rate = (io.bytes_recv - self._prev_net_io.bytes_recv)
        
        self._prev_net_io = io
        
        return {
            "bytes_sent": io.bytes_sent,
            "bytes_recv": io.bytes_recv,
            "sent_rate": sent_rate,
            "recv_rate": recv_rate,
            "packets_sent": io.packets_sent,
            "packets_recv": io.packets_recv
        }
    
    def collect_all(self) -> dict:
        """采集所有指标"""
        return {
            "timestamp": datetime.now().isoformat(),
            "cpu": self.collect_cpu(),
            "memory": self.collect_memory(),
            "disk": self.collect_disk(),
            "network": self.collect_network()
        }


class MetricsStore:
    """指标存储（SQLite）"""
    
    def __init__(self, db_path: str = "metrics.db"):
        self.db_path = db_path
        self._init_db()
    
    def _init_db(self):
        """初始化数据库"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS metrics (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                timestamp TEXT NOT NULL,
                data JSON NOT NULL,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        conn.commit()
        conn.close()
    
    def save(self, metrics: dict):
        """保存指标数据"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute(
            "INSERT INTO metrics (timestamp, data) VALUES (?, ?)",
            (metrics["timestamp"], json.dumps(metrics))
        )
        conn.commit()
        conn.close()
    
    def query(self, start_time: str, end_time: str) -> list:
        """查询历史指标"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute(
            "SELECT data FROM metrics WHERE timestamp BETWEEN ? AND ?",
            (start_time, end_time)
        )
        results = [json.loads(row[0]) for row in cursor.fetchall()]
        conn.close()
        return results

class MonitoringService:
    """监控服务主类"""
    
    def __init__(self, config: dict):
        self.config = config
        self.collector = MetricsCollector()
        self.store = MetricsStore(config.get("db_path", "metrics.db"))
        self.rule_engine = RuleEngine()
        self.notifiers = []
        self.running = False
        
        # 初始化规则
        self._setup_rules()
        
        # 初始化通知器
        self._setup_notifiers()
    
    def _setup_rules(self):
        """设置告警规则"""
        rules = [
            AlertRule(
                name="CPU使用率过高",
                metric="cpu.usage_percent",
                condition="gt",
                threshold=90,
                severity=Severity.CRITICAL,
                description="CPU使用率超过90%，持续5分钟可能影响服务"
            ),
            AlertRule(
                name="内存使用率过高",
                metric="memory.percent",
                condition="gt",
                threshold=85,
                severity=Severity.WARNING,
                description="内存使用率超过85%，建议清理或扩容"
            ),
            AlertRule(
                name="磁盘空间不足",
                metric="disk.percent",
                condition="gt",
                threshold=90,
                severity=Severity.CRITICAL,
                description="磁盘使用率超过90%，需要清理空间"
            ),
            AlertRule(
                name="磁盘空间警告",
                metric="disk.percent",
                condition="gt",
                threshold=80,
                severity=Severity.WARNING,
                description="磁盘使用率超过80%，请关注"
            )
        ]
        
        for rule in rules:
            self.rule_engine.add_rule(rule)
    
    def _setup_notifiers(self):
        """设置通知渠道"""
        # 默认使用控制台输出
        self.notifiers.append(ConsoleNotifier())
        
        # 如果配置了钉钉 webhook，添加钉钉通知
        webhook_url = self.config.get("dingtalk_webhook")
        if webhook_url:
            self.notifiers.append(DingTalkNotifier(webhook_url))
    
    def _get_nested_value(self, data: dict, key: str):
        """获取嵌套字典的值，如 'cpu.usage_percent'"""
        keys = key.split('.')
        value = data
        for k in keys:
            if isinstance(value, dict):
                value = value.get(k)
            else:
                return None
        return value
    
    def check_alerts(self, metrics: dict):
        """检查告警"""
        # 将嵌套的 metrics 转换为扁平格式用于规则匹配
        flat_metrics = {}
        for key in ["cpu.usage_percent", "memory.percent", "disk.percent"]:
            value = self._get_nested_value(metrics, key)
            if value is not None:
                flat_metrics[key] = value
        
        alerts = self.rule_engine.check_all_rules(flat_metrics)
        
        for alert in alerts:
            for notifier in self.notifiers:
                notifier.send(alert)
    
    def run_once(self):
        """执行一次采集和检查"""
        try:
            # 采集指标
            metrics = self.collector.collect_all()
            
            # 存储指标
            self.store.save(metrics)
            
            # 检查告警
            self.check_alerts(metrics)
            
            return metrics
        except Exception as e:
            print(f"采集失败: {e}")
            return None
    
    def run(self, interval: int = 60):
        """持续运行监控"""
        self.running = True
        print(f"🚀 监控服务启动，采集间隔: {interval}秒")
        
        while self.running:
            metrics = self.run_once()
            if metrics:
                print(f"✅ 采集完成: {metrics['timestamp']}")
            
            time.sleep(interval)
    
    def stop(self):
        """停止监控"""
        self.running = False
        print("🛑 监控服务停止")

class InspectionReport:
    """服务器巡检报告"""
    
    def __init__(self, store: MetricsStore):
        self.store = store
    
    def generate(self, hours: int = 24) -> str:
        """生成巡检报告"""
        end_time = datetime.now()
        start_time = end_time - timedelta(hours=hours)
        
        metrics_list = self.store.query(
            start_time.isoformat(),
            end_time.isoformat()
        )
        
        if not metrics_list:
            return "⚠️ 无数据可生成报告"
        
        # 统计各项指标
        report = []
        report.append("=" * 60)
        report.append(f"📊 服务器巡检报告")
        report.append(f"时间范围: {start_time.strftime('%Y-%m-%d %H:%M')} ~ {end_time.strftime('%Y-%m-%d %H:%M')}")
        report.append(f"数据点数: {len(metrics_list)}")
        report.append("=" * 60)
        
        # CPU 统计
        cpu_values = [m['cpu']['usage_percent'] for m in metrics_list]
        report.append(f"\n🖥️  CPU 统计:")
        report.append(f"  平均使用率: {sum(cpu_values)/len(cpu_values):.1f}%")
        report.append(f"  最大使用率: {max(cpu_values):.1f}%")
        report.append(f"  最小使用率: {min(cpu_values):.1f}%")
        
        # 内存统计
        mem_values = [m['memory']['percent'] for m in metrics_list]
        report.append(f"\n💾 内存统计:")
        report.append(f"  平均使用率: {sum(mem_values)/len(mem_values):.1f}%")
        report.append(f"  最大使用率: {max(mem_values):.1f}%")
        
        # 磁盘统计
        disk_values = [m['disk']['percent'] for m in metrics_list]
        report.append(f"\n💿 磁盘统计:")
        report.append(f"  当前使用率: {disk_values[-1]:.1f}%")
        report.append(f"  最大使用率: {max(disk_values):.1f}%")
        
        # 告警统计
        alerts = []
        for metrics in metrics_list:
            for key in ["cpu.usage_percent", "memory.percent", "disk.percent"]:
                value = self._get_nested_value(metrics, key)
                if value and value > 90:
                    alerts.append({"time": metrics['timestamp'], "metric": key, "value": value})
        
        report.append(f"\n🚨 告警统计:")
        report.append(f"  总告警数: {len(alerts)}")
        if alerts:
            for alert in alerts[:5]:  # 只显示前5条
                report.append(f"  - {alert['time']}: {alert['metric']} = {alert['value']:.1f}%")
        
        report.append("\n" + "=" * 60)
        
        return "\n".join(report)
    
    def _get_nested_value(self, data: dict, key: str):
        """获取嵌套字典的值"""
        keys = key.split('.')
        value = data
        for k in keys:
            if isinstance(value, dict):
                value = value.get(k)
            else:
                return None
        return value

def main():
    # 配置
    config = {
        "db_path": "server_metrics.db",
        "dingtalk_webhook": None  # 设置钉钉 webhook URL
    }
    
    # 创建监控服务
    service = MonitoringService(config)
    
    # 执行一次巡检
    print("🔍 执行服务器巡检...")
    metrics = service.run_once()
    
    if metrics:
        print("\n📊 当前系统状态:")
        print(f"  CPU: {metrics['cpu']['usage_percent']}%")
        print(f"  内存: {metrics['memory']['percent']}%")
        print(f"  磁盘: {metrics['disk']['percent']}%")
    
    # 生成巡检报告
    report_generator = InspectionReport(service.store)
    report = report_generator.generate(hours=1)
    print(report)
    
    # 持续监控（每60秒采集一次）
    # service.run(interval=60)

if __name__ == "__main__":
    main()

from collections import defaultdict
from datetime import datetime, timedelta

class AlertDeduplicator:
    """告警去重器"""
    
    def __init__(self, cooldown_seconds: int = 300):
        self.cooldown = timedelta(seconds=cooldown_seconds)
        self.last_alerts = {}  # rule_name -> last_alert_time
    
    def should_alert(self, rule_name: str) -> bool:
        """判断是否应该发送告警"""
        now = datetime.now()
        last_time = self.last_alerts.get(rule_name)
        
        if last_time is None or (now - last_time) > self.cooldown:
            self.last_alerts[rule_name] = now
            return True
        
        return False

class AlertAggregator:
    """告警聚合器"""
    
    def __init__(self, window_seconds: int = 60):
        self.window = timedelta(seconds=window_seconds)
        self.pending_alerts = []
    
    def add_alert(self, alert: dict):
        """添加告警到聚合窗口"""
        self.pending_alerts.append({
            **alert,
            "timestamp": datetime.now()
        })
    
    def get_aggregated(self) -> list:
        """获取聚合后的告警"""
        now = datetime.now()
        
        # 清理过期告警
        self.pending_alerts = [
            a for a in self.pending_alerts
            if (now - a["timestamp"]) < self.window
        ]
        
        # 按规则分组
        groups = defaultdict(list)
        for alert in self.pending_alerts:
            groups[alert["rule"]].append(alert)
        
        # 聚合
        aggregated = []
        for rule_name, alerts in groups.items():
            aggregated.append({
                "rule": rule_name,
                "count": len(alerts),
                "first_occurrence": min(a["timestamp"] for a in alerts),
                "last_occurrence": max(a["timestamp"] for a in alerts),
                "severity": alerts[0]["severity"]
            })
        
        return aggregated