本文目录导读：
我来分享几个Python运维脚本封装的实际案例,从简单到复杂，帮助你理解如何构建可复用的运维工具。
基础封装：使用函数和参数
案例：系统健康检查脚本
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
系统健康检查工具 - 基础封装版本
"""
import os
import sys
import json
import psutil
import argparse
from datetime import datetime
from pathlib import Path
class SystemHealthChecker:
"""系统健康检查器"""
def __init__(self, threshold_cpu=80, threshold_memory=90, threshold_disk=85):
self.thresholds = {
'cpu': threshold_cpu,
'memory': threshold_memory,
'disk': threshold_disk
}
self.results = {}
def check_cpu(self):
"""检查CPU使用率"""
cpu_percent = psutil.cpu_percent(interval=1)
status = 'OK' if cpu_percent < self.thresholds['cpu'] else 'WARNING'
self.results['cpu'] = {
'value': f"{cpu_percent}%",
'status': status,
'threshold': f"{self.thresholds['cpu']}%"
}
return self.results['cpu']
def check_memory(self):
"""检查内存使用率"""
memory = psutil.virtual_memory()
mem_percent = memory.percent
status = 'OK' if mem_percent < self.thresholds['memory'] else 'WARNING'
self.results['memory'] = {
'value': f"{mem_percent}%",
'status': status,
'threshold': f"{self.thresholds['memory']}%",
'details': {
'total': self._format_bytes(memory.total),
'used': self._format_bytes(memory.used),
'available': self._format_bytes(memory.available)
}
}
return self.results['memory']
def check_disk(self):
"""检查磁盘使用率"""
disk = psutil.disk_usage('/')
disk_percent = disk.percent
status = 'OK' if disk_percent < self.thresholds['disk'] else 'WARNING'
self.results['disk'] = {
'value': f"{disk_percent}%",
'status': status,
'threshold': f"{self.thresholds['disk']}%",
'details': {
'total': self._format_bytes(disk.total),
'used': self._format_bytes(disk.used),
'free': self._format_bytes(disk.free)
}
}
return self.results['disk']
def check_all(self):
"""执行所有检查"""
self.check_cpu()
self.check_memory()
self.check_disk()
return self.results
def get_summary(self):
"""获取健康检查摘要"""
if not self.results:
return "未执行检查"
summary = []
for component, result in self.results.items():
icon = "✅" if result['status'] == 'OK' else "⚠️"
summary.append(f"{icon} {component.upper()}: {result['value']} ({result['status']})")
return "\n".join(summary)
def to_json(self):
"""转换为JSON格式"""
return json.dumps({
'timestamp': datetime.now().isoformat(),
'hostname': os.uname().nodename,
'results': self.results
}, indent=2, ensure_ascii=False)
@staticmethod
def _format_bytes(bytes_value):
"""格式化字节大小"""
for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
if bytes_value < 1024:
return f"{bytes_value:.2f} {unit}"
bytes_value /= 1024
return f"{bytes_value:.2f} PB"
def main():
"""主函数"""
parser = argparse.ArgumentParser(
description='系统健康检查工具',
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
示例:
%(prog)s # 执行默认检查
%(prog)s --cpu 90 --memory 85 # 自定义阈值
%(prog)s --output json # JSON格式输出
%(prog)s --check cpu,memory # 只检查CPU和内存
"""
)
parser.add_argument('--cpu', type=int, default=80, help='CPU使用率阈值(默认: 80%%)')
parser.add_argument('--memory', type=int, default=90, help='内存使用率阈值(默认: 90%%)')
parser.add_argument('--disk', type=int, default=85, help='磁盘使用率阈值(默认: 85%%)')
parser.add_argument('--output', choices=['text', 'json'], default='text', help='输出格式')
parser.add_argument('--check', type=str, help='指定检查项(逗号分隔,如: cpu,memory)')
args = parser.parse_args()
# 创建检查器
checker = SystemHealthChecker(
threshold_cpu=args.cpu,
threshold_memory=args.memory,
threshold_disk=args.disk
)
# 执行检查
if args.check:
check_items = [item.strip().lower() for item in args.check.split(',')]
for item in check_items:
if hasattr(checker, f'check_{item}'):
getattr(checker, f'check_{item}')()
else:
checker.check_all()
# 输出结果
if args.output == 'json':
print(checker.to_json())
else:
print(f"\n系统健康检查报告 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print("=" * 50)
print(checker.get_summary())
# 检查是否有警告
warnings = [r for r in checker.results.values() if r['status'] == 'WARNING']
if warnings:
sys.exit(1)
if __name__ == '__main__':
main()
模块化封装：使用面向对象设计
案例：定时任务管理工具
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
运维定时任务管理 - 模块化封装
"""
import os
import sys
import json
import yaml
import logging
import schedule
import time
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from datetime import datetime
from abc import ABC, abstractmethod
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
from pathlib import Path
# 日志配置
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('ops_scheduler.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
@dataclass
class TaskConfig:
"""任务配置数据类"""
name: str
command: str
schedule: str # cron表达式或schedule表达式
enabled: bool = True
timeout: int = 300
retry_count: int = 3
retry_delay: int = 60
notify_on_failure: bool = True
notify_on_success: bool = False
tags: List[str] = field(default_factory=list)
env_vars: Dict[str, str] = field(default_factory=dict)
class TaskResult:
"""任务执行结果"""
def __init__(self, task_name: str, success: bool, output: str, duration: float):
self.task_name = task_name
self.success = success
self.output = output
self.duration = duration
self.timestamp = datetime.now()
def to_dict(self):
return {
'task_name': self.task_name,
'success': self.success,
'output': self.output[:500] if self.output else '', # 截断长输出
'duration': f"{self.duration:.2f}s",
'timestamp': self.timestamp.isoformat()
}
class TaskExecutor(ABC):
"""任务执行器抽象基类"""
@abstractmethod
def execute(self, task: TaskConfig) -> TaskResult:
"""执行任务"""
pass
class ShellTaskExecutor(TaskExecutor):
"""Shell命令执行器"""
def __init__(self):
import subprocess
def execute(self, task: TaskConfig) -> TaskResult:
"""执行Shell命令"""
import subprocess
import time
start_time = time.time()
try:
env = os.environ.copy()
env.update(task.env_vars)
result = subprocess.run(
task.command,
shell=True,
capture_output=True,
text=True,
timeout=task.timeout,
env=env
)
duration = time.time() - start_time
success = result.returncode == 0
output = result.stdout if success else result.stderr
logger.info(f"任务 {task.name} 执行{'成功' if success else '失败'}: {duration:.2f}s")
return TaskResult(task.name, success, output, duration)
except subprocess.TimeoutExpired:
duration = time.time() - start_time
logger.error(f"任务 {task.name} 超时 ({task.timeout}s)")
return TaskResult(task.name, False, "任务执行超时", duration)
except Exception as e:
duration = time.time() - start_time
logger.error(f"任务 {task.name} 执行异常: {str(e)}")
return TaskResult(task.name, False, str(e), duration)
class PythonTaskExecutor(TaskExecutor):
"""Python函数执行器"""
def __init__(self):
self.functions: Dict[str, Callable] = {}
def register_function(self, name: str, func: Callable):
"""注册Python函数"""
self.functions[name] = func
def execute(self, task: TaskConfig) -> TaskResult:
"""执行Python函数"""
import time
start_time = time.time()
try:
func = self.functions.get(task.command)
if not func:
raise ValueError(f"未注册的函数: {task.command}")
# 解析参数
import ast
func_args = {}
if 'args' in task.env_vars:
func_args = ast.literal_eval(task.env_vars['args'])
result = func(**func_args)
duration = time.time() - start_time
return TaskResult(task.name, True, str(result), duration)
except Exception as e:
duration = time.time() - start_time
logger.error(f"Python任务 {task.name} 执行失败: {str(e)}")
return TaskResult(task.name, False, str(e), duration)
class TaskScheduler:
"""任务调度器"""
def __init__(self):
self.executors: Dict[str, TaskExecutor] = {
'shell': ShellTaskExecutor(),
'python': PythonTaskExecutor()
}
self.tasks: Dict[str, TaskConfig] = {}
self.results: List[TaskResult] = []
self.running = False
def load_config(self, config_path: str):
"""从配置文件加载任务"""
config_path = Path(config_path)
if not config_path.exists():
raise FileNotFoundError(f"配置文件不存在: {config_path}")
with open(config_path, 'r', encoding='utf-8') as f:
if config_path.suffix == '.yaml':
config_data = yaml.safe_load(f)
elif config_path.suffix == '.json':
config_data = json.load(f)
else:
raise ValueError("不支持的配置文件格式")
# 创建任务配置
for task_dict in config_data.get('tasks', []):
task = TaskConfig(**task_dict)
self.tasks[task.name] = task
logger.info(f"已加载 {len(self.tasks)} 个任务配置")
def add_task(self, task: TaskConfig):
"""添加任务"""
self.tasks[task.name] = task
logger.info(f"添加任务: {task.name}")
def remove_task(self, task_name: str):
"""移除任务"""
if task_name in self.tasks:
del self.tasks[task_name]
logger.info(f"移除任务: {task_name}")
def schedule_task(self, task: TaskConfig):
"""调度单个任务"""
schedule_time = task.schedule
# 解析调度时间
parts = schedule_time.split()
if len(parts) == 2:
# "every 1 hour"
interval, unit = parts
if unit in ['minutes', 'minute', 'min']:
schedule.every(int(interval)).minutes.do(self.run_task, task)
elif unit in ['hours', 'hour']:
schedule.every(int(interval)).hours.do(self.run_task, task)
elif unit in ['days', 'day']:
schedule.every(int(interval)).days.do(self.run_task, task)
elif len(parts) == 3:
# "at 10:30" 或 "every monday"
if parts[0] == 'at':
schedule.every().day.at(parts[1]).do(self.run_task, task)
elif 'every' in parts:
day = parts[1]
getattr(schedule.every(), day).do(self.run_task, task)
logger.info(f"任务 {task.name} 已调度: {task.schedule}")
def run_task(self, task: TaskConfig) -> TaskResult:
"""执行单个任务"""
if not task.enabled:
logger.info(f"任务 {task.name} 已禁用，跳过执行")
return None
# 选择执行器
executor = self.executors['shell'] # 默认使用shell
if task.command.startswith('python:'):
executor = self.executors['python']
task.command = task.command.replace('python:', '')
# 执行任务
result = None
for attempt in range(task.retry_count):
result = executor.execute(task)
if result.success:
break
logger.warning(f"任务 {task.name} 第 {attempt + 1} 次尝试失败")
if attempt < task.retry_count - 1:
time.sleep(task.retry_delay)
# 保存结果
self.results.append(result)
if len(self.results) > 1000: # 保留最近1000条
self.results.pop(0)
# 通知
if not result.success and task.notify_on_failure:
self._send_notification(result)
return result
def run_all_tasks_now(self):
"""立即运行所有任务"""
logger.info("开始执行所有任务...")
results = []
for task in self.tasks.values():
result = self.run_task(task)
if result:
results.append(result)
logger.info("所有任务执行完成")
return results
def start(self):
"""启动调度器"""
if self.running:
logger.warning("调度器已在运行")
return
self.running = True
# 调度所有任务
for task in self.tasks.values():
self.schedule_task(task)
logger.info("调度器已启动")
# 运行调度循环
while self.running:
schedule.run_pending()
time.sleep(1)
def stop(self):
"""停止调度器"""
self.running = False
logger.info("调度器已停止")
def _send_notification(self, result: TaskResult):
"""发送通知"""
# 邮件通知示例
subject = f"[运维通知] 任务 {result.task_name} 执行失败"
body = f"""
任务名称: {result.task_name}
执行时间: {result.timestamp}
执行状态: 失败
错误信息: {result.output}
"""
logger.info(f"发送通知: {subject}")
# 实际邮件发送代码...
def get_statistics(self) -> Dict:
"""获取统计信息"""
if not self.results:
return {"message": "暂无执行记录"}
total = len(self.results)
success = sum(1 for r in self.results if r.success)
failed = total - success
avg_duration = sum(r.duration for r in self.results) / total
return {
"total_executions": total,
"success_count": success,
"failed_count": failed,
"success_rate": f"{success/total*100:.1f}%",
"average_duration": f"{avg_duration:.2f}s"
}
class TaskSchedulerCLI:
"""任务调度器命令行接口"""
def __init__(self):
self.scheduler = TaskScheduler()
def run(self):
"""运行CLI"""
import argparse
parser = argparse.ArgumentParser(
description='运维任务调度工具',
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
示例:
%(prog)s start -c tasks.yaml # 启动调度器
%(prog)s run-now -c tasks.yaml # 立即执行所有任务
%(prog)s status # 查看调度器状态
%(prog)s stats # 查看执行统计
%(prog)s add --name backup --command "tar -czf backup.tar.gz /data" --schedule "every 1 day"
%(prog)s list # 列出所有任务
"""
)
subparsers = parser.add_subparsers(dest='command', help='可用命令')
# start命令
start_parser = subparsers.add_parser('start', help='启动调度器')
start_parser.add_argument('-c', '--config', required=True, help='配置文件路径')
start_parser.add_argument('-d', '--daemon', action='store_true', help='以守护进程方式运行')
# run-now命令
run_parser = subparsers.add_parser('run-now', help='立即执行所有任务')
run_parser.add_argument('-c', '--config', required=True, help='配置文件路径')
# status命令
subparsers.add_parser('status', help='查看调度器状态')
# stats命令
subparsers.add_parser('stats', help='查看执行统计')
# add命令
add_parser = subparsers.add_parser('add', help='添加任务')
add_parser.add_argument('--name', required=True, help='任务名称')
add_parser.add_argument('--command', required=True, help='执行命令')
add_parser.add_argument('--schedule', required=True, help='调度时间')
add_parser.add_argument('--enabled', action='store_true', default=True, help='启用任务')
# list命令
subparsers.add_parser('list', help='列出所有任务')
# remove命令
remove_parser = subparsers.add_parser('remove', help='移除任务')
remove_parser.add_argument('--name', required=True, help='任务名称')
args = parser.parse_args()
if args.command == 'start':
self._handle_start(args)
elif args.command == 'run-now':
self._handle_run_now(args)
elif args.command == 'status':
self._handle_status()
elif args.command == 'stats':
self._handle_stats()
elif args.command == 'add':
self._handle_add(args)
elif args.command == 'list':
self._handle_list()
elif args.command == 'remove':
self._handle_remove(args)
else:
parser.print_help()
def _handle_start(self, args):
"""处理start命令"""
try:
self.scheduler.load_config(args.config)
print(f"调度器配置已加载: {args.config}")
print("启动调度器...(按Ctrl+C停止)")
self.scheduler.start()
except KeyboardInterrupt:
print("\n正在停止调度器...")
self.scheduler.stop()
except Exception as e:
print(f"启动失败: {str(e)}")
def _handle_run_now(self, args):
"""处理run-now命令"""
try:
self.scheduler.load_config(args.config)
results = self.scheduler.run_all_tasks_now()
print(f"\n执行结果 ({len(results)} 个任务):")
for result in results:
status = "成功" if result.success else "失败"
print(f" [{status}] {result.task_name} ({result.duration:.2f}s)")
except Exception as e:
print(f"执行失败: {str(e)}")
def _handle_status(self):
"""处理status命令"""
if self.scheduler.running:
print("调度器状态: 运行中")
print(f"已调度任务数: {len(self.scheduler.tasks)}")
else:
print("调度器状态: 已停止")
def _handle_stats(self):
"""处理stats命令"""
stats = self.scheduler.get_statistics()
print("\n执行统计:")
for key, value in stats.items():
print(f" {key}: {value}")
def _handle_add(self, args):
"""处理add命令"""
task = TaskConfig(
name=args.name,
command=args.command,
schedule=args.schedule,
enabled=args.enabled
)
self.scheduler.add_task(task)
print(f"任务 '{args.name}' 已添加")
def _handle_list(self):
"""处理list命令"""
if not self.scheduler.tasks:
print("暂无任务")
return
print(f"\n已配置任务 ({len(self.scheduler.tasks)}):")
for task in self.scheduler.tasks.values():
status = "启用" if task.enabled else "禁用"
print(f" [{status}] {task.name}")
print(f" 命令: {task.command}")
print(f" 调度: {task.schedule}")
def _handle_remove(self, args):
"""处理remove命令"""
self.scheduler.remove_task(args.name)
print(f"任务 '{args.name}' 已移除")
# YAML配置文件示例
"""
tasks:
- name: "备份数据库"
command: "pg_dump -U postgres mydb > /backups/mydb_$(date +%Y%m%d).sql"
schedule: "every 1 day"
timeout: 600
retry_count: 2
notify_on_failure: true
tags: ["backup", "database"]
- name: "清理日志"
command: "find /var/log -name '*.log' -mtime +30 -delete"
schedule: "every 1 week"
timeout: 300
notify_on_failure: true
tags: ["cleanup", "logs"]
- name: "健康检查"
command: "python:health_check"
schedule: "every 5 minutes"
timeout: 30
tags: ["monitoring"]
"""
def main():
"""主入口"""
import logging as log_module
# 配置日志
log_module.basicConfig(
level=log_module.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
log_module.FileHandler('ops_scheduler.log'),
log_module.StreamHandler()
]
)
cli = TaskSchedulerCLI()
cli.run()
if __name__ == '__main__':
main()
插件化封装：使用插件架构
案例：可扩展的监控插件系统
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
可扩展的监控插件系统 - 插件化封装
"""
import os
import sys
import json
import importlib
import inspect
import pkgutil
from typing import Dict, List, Type, Any
from datetime import datetime
from abc import ABC, abstractmethod
from pathlib import Path
class PluginBase(ABC):
"""插件基类"""
@property
@abstractmethod
def name(self) -> str:
"""插件名称"""
pass
@property
def version(self) -> str:
"""插件版本"""
return "1.0.0"
@property
def description(self) -> str:
"""插件描述"""
return ""
@abstractmethod
def check(self) -> Dict[str, Any]:
"""执行检查"""
pass
def initialize(self):
"""初始化插件"""
pass
def cleanup(self):
"""清理资源"""
pass
class PluginManager:
"""插件管理器"""
def __init__(self, plugin_dirs: List[str] = None):
self.plugin_dirs = plugin_dirs or []
self.plugins: Dict[str, PluginBase] = {}
self.loaded = False
def discover_plugins(self):
"""发现并加载插件"""
self.plugins.clear()
# 搜索插件目录
for plugin_dir in self.plugin_dirs:
self._load_plugins_from_directory(plugin_dir)
# 搜索内置插件
self._load_plugins_from_package()
self.loaded = True
print(f"发现 {len(self.plugins)} 个插件")
def _load_plugins_from_directory(self, directory: str):
"""从目录加载插件"""
directory = Path(directory)
if not directory.exists():
return
# 添加目录到Python路径
sys.path.insert(0, str(directory.parent))
# 扫描Python文件
for file_path in directory.glob("*.py"):
if file_path.name.startswith('_'):
continue
module_name = file_path.stem
self._load_plugin_from_module(module_name, directory.parent)
def _load_plugins_from_package(self):
"""从内置包加载插件"""
# 这里假设内置插件在plugins包中
try:
import plugins as builtin_plugins
package_path = Path(builtin_plugins.__file__).parent
for importer, name, ispkg in pkgutil.iter_modules([str(package_path)]):
if not ispkg:
self._load_plugin_from_module(name, package_path)
except ImportError:
pass
def _load_plugin_from_module(self, module_name: str, package_path: Path):
"""从模块加载插件"""
try:
spec = importlib.util.spec_from_file_location(
module_name,
str(package_path / f"{module_name}.py")
)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
# 查找插件类
for name, obj in inspect.getmembers(module):
if (inspect.isclass(obj) and
issubclass(obj, PluginBase) and
obj != PluginBase and
not inspect.isabstract(obj)):
plugin = obj()
if plugin.name not in self.plugins:
self.plugins[plugin.name] = plugin
print(f"加载插件: {plugin.name} v{plugin.version}")
else:
print(f"插件已存在: {plugin.name}")
except Exception as e:
print(f"加载插件 {module_name} 失败: {str(e)}")
def get_plugin(self, name: str) -> PluginBase:
"""获取插件实例"""
return self.plugins.get(name)
def get_all_plugins(self) -> Dict[str, PluginBase]:
"""获取所有插件"""
return self.plugins
def initialize_all(self):
"""初始化所有插件"""
for name, plugin in self.plugins.items():
try:
plugin.initialize()
print(f"初始化插件: {name}")
except Exception as e:
print(f"初始化插件 {name} 失败: {str(e)}")
def cleanup_all(self):
"""清理所有插件"""
for name, plugin in self.plugins.items():
try:
plugin.cleanup()
print(f"清理插件: {name}")
except Exception as e:
print(f"清理插件 {name} 失败: {str(e)}")
class MonitoringSystem:
"""监控系统主类"""
def __init__(self):
self.plugin_manager = PluginManager()
self.data_collectors: Dict[str, Any] = {}
self.alerts: List[Dict] = []
def load_plugins(self, plugin_dirs: List[str] = None):
"""加载插件"""
self.plugin_manager.plugin_dirs = plugin_dirs or []
self.plugin_manager.discover_plugins()
self.plugin_manager.initialize_all()
def collect_data(self) -> Dict[str, Any]:
"""收集监控数据"""
data = {
'timestamp': datetime.now().isoformat(),
'metrics': {}
}
for name, plugin in self.plugin_manager.get_all_plugins().items():
try:
result = plugin.check()
data['metrics'][name] = result
except Exception as e:
data['metrics'][name] = {
'error': str(e),
'status': 'error'
}
return data
def check_alerts(self, data: Dict[str, Any]):
"""检查告警"""
self.alerts = []
for name, metrics in data['metrics'].items():
if 'error' in metrics:
self.alerts.append({
'source': name,
'type': 'error',
'message': metrics['error'],
'timestamp': data['timestamp']
})
elif metrics.get('status') == 'warning':
self.alerts.append({
'source': name,
'type': 'warning',
'message': metrics.get('message', ''),
'value': metrics.get('value'),
'threshold': metrics.get('threshold'),
'timestamp': data['timestamp']
})
return self.alerts
def run_once(self) -> Dict[str, Any]:
"""运行一次监控检查"""
print("开始监控检查...")
data = self.collect_data()
alerts = self.check_alerts(data)
if alerts:
print(f"发现 {len(alerts)} 个告警")
for alert in alerts:
print(f" [{alert['type'].upper()}] {alert['source']}: {alert['message']}")
else:
print("所有检查通过")
return data
# 示例插件实现
# plugin_example.py
"""
from monitoring_system import PluginBase
class CPUMonitor(PluginBase):
@property
def name(self):
return "cpu_monitor"
@property
def version(self):
return "1.0.0"
def check(self):
# 实际中应该使用psutil
return {
'status': 'ok',
'value': 45.5,
'unit': '%',
'threshold': 80
}
class DiskMonitor(PluginBase):
@property
def name(self):
return "disk_monitor"
def check(self):
return {
'status': 'warning',
'value': 85.2,
'unit': '%',
'threshold': 80,
'message': '磁盘使用率超过阈值'
}
"""
def main():
"""主函数"""
import argparse
parser = argparse.ArgumentParser(
description='可扩展的监控系统',
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
示例:
%(prog)s --plugin-dir ./plugins # 使用插件目录
%(prog)s --list-plugins # 列出所有插件
%(prog)s --run-once # 运行一次检查
%(prog)s --output json # JSON格式输出
"""
)
parser.add_argument('--plugin-dir', action='append', help='插件目录(可多次指定)')
parser.add_argument('--list-plugins', action='store_true', help='列出所有插件')
parser.add_argument('--run-once', action='store_true', help='运行一次监控检查')
parser.add_argument('--output', choices=['text', 'json'], default='text', help='输出格式')
args = parser.parse_args()
# 创建监控系统
monitor = MonitoringSystem()
# 加载插件
plugin_dirs = args.plugin_dir or ['./plugins']
monitor.load_plugins(plugin_dirs)
if args.list_plugins:
print(f"\n已加载的插件 ({len(monitor.plugin_manager.get_all_plugins())}):")
for name, plugin in monitor.plugin_manager.get_all_plugins().items():
print(f" - {name} (v{plugin.version})")
if plugin.description:
print(f" 描述: {plugin.description}")
return
if args.run_once:
data = monitor.run_once()
if args.output == 'json':
print(json.dumps(data, indent=2, ensure_ascii=False))
else:
print(f"\n监控结果 ({data['timestamp']}):")
for name, metrics in data['metrics'].items():
status = metrics.get('status', 'unknown')
icon = "✅" if status == 'ok' else "⚠️" if status == 'warning' else "❌"
print(f" {icon} {name}: {status}")
if 'value' in metrics:
print(f" 值: {metrics['value']} {metrics.get('unit', '')}")
# 清理
monitor.plugin_manager.cleanup_all()
if __name__ == '__main__':
main()
实用运维脚本封装技巧
1 配置管理
# config_manager.py
import json
import yaml
from pathlib import Path
from typing import Any, Dict, Optional
class ConfigManager:
"""配置管理器"""
def __init__(self, config_path: str = None):
self.config_path = Path(config_path or 'config.yaml')
self.config: Dict[str, Any] = {}
self.load()
def load(self):
"""加载配置"""
if not self.config_path.exists():
raise FileNotFoundError(f"配置文件不存在: {self.config_path}")
try:
with open(self.config_path, 'r') as f:
if self.config_path.suffix == '.yaml':
self.config = yaml.safe_load(f)
elif self.config_path.suffix == '.json':
self.config = json.load(f)
else:
# 尝试多种格式
try:
self.config = yaml.safe_load(f)
except:
f.seek(0)
self.config = json.load(f)
except Exception as e:
raise ValueError(f"配置文件解析失败: {e}")
def get(self, key: str, default: Any = None) -> Any:
"""获取配置值"""
keys = key.split('.')
value = self.config
for k in keys:
if isinstance(value, dict):
value = value.get(k)
if value is None:
return default
else:
return default
return value
def set(self, key: str, value: Any):
"""设置配置值"""
keys = key.split('.')
config = self.config
for k in keys[:-1]:
if k not in config:
config[k] = {}
config = config[k]
config[keys[-1]] = value
def save(self, path: str = None):
"""保存配置"""
save_path = Path(path) if path else self.config_path
with open(save_path, 'w') as f:
if save_path.suffix == '.yaml':
yaml.dump(self.config, f, default_flow_style=False)
else:
json.dump(self.config, f, indent=2)
def to_dict(self) -> Dict:
"""转换为字典"""
return self.config.copy()
2 日志管理
# log_manager.py
import logging
import sys
from pathlib import Path
from logging.handlers import RotatingFileHandler
class LogManager:
"""日志管理器"""
def __init__(self, name: str = 'ops_tool', log_dir: str =
