代码展示：

"""
磁盘空间监控与自动清理脚本
功能：监控指定目录空间使用率，超阈值时按预设规则清理历史文件
作者：运维团队
版本：v2.1
"""
import shutil
import psutil
import requests
from pathlib import Path
import time

class DiskMonitor:
    def __init__(self, cleanup_rules=None):
        """
        初始化监控器
        :param cleanup_rules: 清理规则字典，格式示例：
            {
                '/var/log': {
                    'extensions': ['.log'],  # 按后缀清理
                    'retention_days': 7      # 保留最近7天
                },
                '/tmp': {
                    'max_age_hours': 48      # 最大保留48小时
                }
            }
        """
        # 设置默认清理规则
        self.cleanup_rules = cleanup_rules or {
            '/var/log': {'extensions': ['.log'], 'retention_days': 7},
            '/tmp': {'max_age_hours': 72}
        }

    def analyze_disk(self, path='/'):
        """
        分析磁盘使用情况
        :param path: 挂载点路径
        :return: 包含使用情况的字典
        """
        try:
            usage = shutil.disk_usage(path)
            return {
                'total': usage.total,         # 总容量（字节）
                'used': usage.used,           # 已用容量
                'free': usage.free,           # 剩余容量
                'percent_used': (usage.used / usage.total) * 100  # 使用率百分比
            }
        except FileNotFoundError:
            print(f"路径 {path} 不存在")
            return None

    def auto_cleanup(self, path):
        """
        执行自动清理操作
        :param path: 需要清理的目录路径
        :return: 清理的文件数量
        """
        cleared = 0
        rules = self.cleanup_rules.get(path, {})
        
        # 按后缀名和保留天数清理
        if 'extensions' in rules:
            current_time = time.time()
            for ext in rules['extensions']:
                # 使用rglob递归遍历所有匹配文件
                for f in Path(path).rglob(f'*{ext}'):
                    # 计算文件修改时间是否超过保留天数
                    file_mtime = f.stat().st_mtime
                    if (current_time - file_mtime) > rules['retention_days'] * 86400:
                        try:
                            f.unlink()  # 删除文件
                            cleared += 1
                        except PermissionError:
                            print(f"无权限删除文件: {f}")

        # 按文件存在时间清理（如/tmp目录）
        if 'max_age_hours' in rules:
            max_age = rules['max_age_hours'] * 3600  # 转换为秒
            current_time = time.time()
            for f in Path(path).iterdir():
                if f.is_file():
                    file_age = current_time - f.stat().st_mtime
                    if file_age > max_age:
                        try:
                            f.unlink()
                            cleared += 1
                        except IsADirectoryError:
                            continue  # 跳过目录

        return cleared

    def check_and_alert(self, threshold=85):
        """
        主检查逻辑
        :param threshold: 触发清理的使用率阈值（百分比）
        """
        for part in psutil.disk_partitions():
            mountpoint = part.mountpoint
            # 仅监控配置了规则的目录
            if mountpoint in self.cleanup_rules:
                usage = self.analyze_disk(mountpoint)
                if not usage:
                    continue
                
                print(f"正在检查 {mountpoint}，当前使用率 {usage['percent_used']:.1f}%")
                
                if usage['percent_used'] > threshold:
                    cleared = self.auto_cleanup(mountpoint)
                    alert_msg = (
                        f"[紧急清理] {mountpoint} 使用率 {usage['percent_used']:.1f}%\n"
                        f"已自动清理 {cleared} 个文件，清理后剩余空间：{usage['free']/1024**3:.2f}GB"
                    )
                    self.send_alert(alert_msg)

    def send_alert(self, message):
        """
        发送报警到企业微信
        :param message: 报警消息内容
        """
        webhook_url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=your_key"
        try:
            resp = requests.post(
                webhook_url,
                json={
                    "msgtype": "text",
                    "text": {"content": message}
                },
                timeout=5
            )
            print(f"报警发送状态：{resp.status_code}")
        except requests.exceptions.RequestException as e:
            print(f"报警发送失败: {str(e)}")

if __name__ == '__main__':
    # 实例化监控器并启动检查
    monitor = DiskMonitor()
    monitor.check_and_alert(threshold=85)  # 设置阈值为85%

代码说明：

1. 类结构设计

2. 关键函数详解

3. 运维实践经验

运行说明

通过以上详细注释和说明，可以快速理解：

• 脚本的整体架构设计
• 各功能模块的实现原理
• 生产环境注意事项
• 调试和扩展方法