Python运维脚本开发入门到高级技巧-源码库

Python运维脚本开发：从新手到高手的实战指南

作为一名在运维领域摸爬滚打多年的工程师，我深知自动化脚本对于提升工作效率的重要性。今天我想和大家分享Python在运维脚本开发中的完整知识体系，从最基础的文件操作到高级的并发处理，这些都是我在实际工作中总结出来的宝贵经验。

一、环境准备与基础工具选择

在开始编写运维脚本之前，我建议先搭建一个舒适的开发环境。我个人偏好使用Virtualenv创建隔离的Python环境，这样可以避免不同项目间的依赖冲突。

# 创建虚拟环境
python -m venv ops_scripts
# 激活环境
source ops_scripts/bin/activate
# 安装常用运维库
pip install psutil requests paramiko

踩坑提示：记得在脚本开头添加shebang行，指定Python解释器路径，这样脚本就能直接执行了。

二、文件与目录操作实战

文件操作是运维脚本中最常见的需求。让我分享一个实用的日志文件清理脚本：

#!/usr/bin/env python3
import os
import glob
from datetime import datetime, timedelta

def clean_old_logs(log_dir, days_to_keep=30):
    """清理指定天数前的日志文件"""
    cutoff_date = datetime.now() - timedelta(days=days_to_keep)
    
    for log_file in glob.glob(os.path.join(log_dir, "*.log")):
        file_mtime = datetime.fromtimestamp(os.path.getmtime(log_file))
        if file_mtime < cutoff_date:
            print(f"删除旧日志: {log_file}")
            os.remove(log_file)

if __name__ == "__main__":
    clean_old_logs("/var/log/myapp")

三、系统监控与进程管理

使用psutil库可以轻松获取系统状态信息。下面是我常用的系统监控脚本：

import psutil
import time

def system_monitor():
    while True:
        # CPU使用率
        cpu_percent = psutil.cpu_percent(interval=1)
        
        # 内存使用情况
        memory = psutil.virtual_memory()
        
        # 磁盘使用率
        disk = psutil.disk_usage('/')
        
        print(f"CPU: {cpu_percent}% | "
              f"内存: {memory.percent}% | "
              f"磁盘: {disk.percent}%")
        
        time.sleep(60)

# 检查特定进程是否存在
def check_process(process_name):
    for proc in psutil.process_iter(['name']):
        if proc.info['name'] == process_name:
            return True
    return False

四、远程服务器管理进阶

使用Paramiko进行SSH连接时，我建议使用密钥认证而非密码，这样更安全：

import paramiko
from io import StringIO

def remote_command_execute(host, username, key_content, command):
    """通过SSH密钥执行远程命令"""
    try:
        # 创建SSH客户端
        client = paramiko.SSHClient()
        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        
        # 从字符串加载私钥
        private_key = paramiko.RSAKey.from_private_key(StringIO(key_content))
        
        # 连接服务器
        client.connect(hostname=host, username=username, pkey=private_key)
        
        # 执行命令
        stdin, stdout, stderr = client.exec_command(command)
        
        # 获取输出
        output = stdout.read().decode()
        error = stderr.read().decode()
        
        return output, error
        
    except Exception as e:
        return None, str(e)
    finally:
        client.close()

五、错误处理与日志记录最佳实践

在运维脚本中，完善的错误处理至关重要。这是我总结的错误处理模板：

import logging
import sys
from functools import wraps

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('ops_script.log'),
        logging.StreamHandler(sys.stdout)
    ]
)

def handle_exceptions(func):
    """异常处理装饰器"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            logging.error(f"函数 {func.__name__} 执行失败: {str(e)}")
            # 可以根据异常类型进行不同的处理
            if isinstance(e, FileNotFoundError):
                logging.error("文件未找到，请检查路径")
            elif isinstance(e, PermissionError):
                logging.error("权限不足，请检查用户权限")
            return None
    return wrapper

六、高级技巧：并发执行与性能优化

当需要同时管理多台服务器时，并发执行能大幅提升效率。使用concurrent.futures模块是个不错的选择：

from concurrent.futures import ThreadPoolExecutor, as_completed

def batch_remote_commands(servers, command):
    """批量在多台服务器上执行命令"""
    results = {}
    
    with ThreadPoolExecutor(max_workers=10) as executor:
        # 提交所有任务
        future_to_server = {
            executor.submit(remote_command_execute, server['host'], 
                           server['username'], server['key'], command): server
            for server in servers
        }
        
        # 收集结果
        for future in as_completed(future_to_server):
            server = future_to_server[future]
            try:
                output, error = future.result()
                results[server['host']] = {'output': output, 'error': error}
            except Exception as e:
                results[server['host']] = {'output': None, 'error': str(e)}
    
    return results

七、实战案例：完整的应用部署脚本

最后，让我分享一个完整的应用部署脚本，结合了前面提到的所有技巧：

#!/usr/bin/env python3
import os
import sys
import logging
from datetime import datetime

@handle_exceptions
def deploy_application():
    """完整的应用部署流程"""
    logging.info("开始应用部署流程")
    
    # 1. 备份当前版本
    backup_existing_app()
    
    # 2. 下载新版本
    download_new_version()
    
    # 3. 停止服务
    stop_services()
    
    # 4. 更新文件
    update_application_files()
    
    # 5. 启动服务
    start_services()
    
    # 6. 健康检查
    if health_check():
        logging.info("应用部署成功完成")
    else:
        logging.error("部署后健康检查失败，开始回滚")
        rollback_deployment()

def backup_existing_app():
    """备份现有应用"""
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    backup_dir = f"/backups/app_{timestamp}"
    os.makedirs(backup_dir, exist_ok=True)
    
    # 使用rsync备份应用目录
    os.system(f"rsync -av /opt/myapp/ {backup_dir}/")
    logging.info(f"应用已备份至: {backup_dir}")

# 其他函数实现...

通过这个完整的教程，相信你已经掌握了Python运维脚本开发的核心技能。记住，好的运维脚本不仅要功能完善，还要有良好的错误处理和日志记录。在实际工作中，我建议先从简单的脚本开始，逐步增加复杂度，这样能更好地掌握每个概念。如果你在实践过程中遇到问题，欢迎在评论区交流讨论！