Linux下Python自动化脚本编写到运维工具开发详解

目录

• 一、引言
• 二、基础脚本编写：实用示例与核心模块
• 2.1 需求说明
• 2.2 完整脚本代码
• 2.3 核心模块解析
• 2.4 脚本使用方法
• 三、进阶工具开发：脚本封装与功能强化
• 3.1 强化功能说明
• 3.2 进阶工具代码片段
• 3.3 核心强化点解析
• 四、实际案例：自动检测并重启宕机服务
• 4.1 需求分析
• 4.2 设计思路
• 4.3 实现过程
• 4.4 测试方法
• 五、案例流程图
• 六、总结与扩展建议
• 6.1 总结
• 6.2 扩展建议

在 Linux 运维场景中，重复性工作（如日志清理、服务监控、批量部署）占据了运维人员大量时间，而 Python 凭借其简洁的语法、丰富的内置库与第三方生态，成为 Linux 自动化运维的首选语言。相较于 Shell 脚本，Python 具备更强的逻辑处理能力、跨平台兼容性和可维护性，能轻松应对复杂的自动化需求；同时，Python 与 Linux 系统底层交互顺畅，可直接调用系统命令、操作文件目录、管理进程服务，大幅提升运维效率。

Python 在 Linux 自动化运维中的典型应用场景包括：日志收集与分析（快速提取关键信息、定位问题）、服务状态监控（实时检测服务可用性，自动告警）、批量任务执行（批量部署软件、配置服务器）、配置文件管理（统一修改、备份配置）以及系统资源统计（CPU、内存、磁盘使用率采集）等。本文将从基础脚本编写入手，逐步讲解如何封装进阶工具，并通过实际案例演示完整开发流程，帮助读者快速掌握 Linux 下 Python 自动化的核心技能。

基础脚本是自动化运维的基石，通常聚焦于单一简单任务。本节以“自动清理旧日志文件”为例，编写可直接运行的 Python 脚本，同时讲解 os、subprocess、argparse 等核心模块的使用方法。

清理指定目录下指定后缀（如 .log）、超过指定天数的旧日志文件，支持显示清理过程与跳过空目录，避免误删重要文件。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
自动清理 Linux 下的旧日志文件
功能：删除指定目录下超过指定天数的指定后缀文件
依赖：Python 3.6+（无需额外第三方库）
"""
import os
import argparse
import time
from datetime import datetime, timedelta

def parse_args():
    """解析命令行参数"""
    parser = argparse.ArgumentParser(description="Linux 旧日志文件清理工具")
    # 必选参数：日志目录
    parser.add_argument("-d", "--dir", required=True, help="日志文件所在目录（必填）")
    # 可选参数：文件后缀，默认 .log
    parser.add_argument("-s", "--suffix", default=".log", help="需要清理的文件后缀（默认：.log）")
    # 可选参数：保留天数，默认 7 天
    parser.add_argument("-t", "--days", type=int, default=7, help="保留文件的天数（默认：7 天，超过则删除）")
    return parser.parse_args()

def clean_old_files(log_dir, file_suffix, keep_days):
    """
    清理旧文件核心逻辑
    :param log_dir: 日志目录路径
    :param file_suffix: 目标文件后缀
    :param keep_days: 保留天数
    """
    # 校验目录是否存在
    if not os.path.isdir(log_dir):
        print(f"错误：目录 {log_dir} 不存在，请检查路径是否正确")
        return
    # 计算过期时间阈值（当前时间 - 保留天数）
    expire_time = datetime.now() - timedelta(days=keep_days)
    # 遍历目录下所有文件
    for root, dirs, files in os.walk(log_dir):
        for file in files:
            # 筛选指定后缀的文件
            if file.endswith(file_suffix):
                file_path = os.path.join(root, file)
                # 获取文件最后修改时间（Linux 时间戳转 datetime）
                mtime = os.path.getmtime(file_path)
                mtime_datetime = datetime.fromtimestamp(mtime)
                # 判断文件是否过期
                if mtime_datetime < expire_time:
                    try:
                        # 删除文件
                        os.remove(file_path)
                        print(f"已删除过期文件：{file_path}")
                    except Exception as e:
                        print(f"删除文件 {file_path} 失败：{str(e)}")
    print("清理任务执行完成！")

if __name__ == "__main__":
    # 解析参数
    args = parse_args()
    # 执行清理逻辑
    clean_old_files(args.dir, args.suffix, args.days)

os 模块：与 Linux 系统交互的核心模块，本文中用于判断目录是否存在（os.path.isdir）、拼接文件路径（os.path.join）、获取文件修改时间（os.path.getmtime）、删除文件（os.remove）以及遍历目录（os.walk）。该模块无需额外安装，是 Python 内置核心库，能直接操作系统文件与目录。

argparse 模块：命令行参数解析模块，用于接收用户输入的目录、后缀、保留天数等参数，支持必选/可选参数配置、参数说明提示，让脚本更灵活通用。通过 argparse.ArgumentParser 创建解析器，add_argument 定义参数规则，最终通过 parse_args() 解析参数。

datetime/timedelta 模块：用于时间计算，本文中通过 datetime.now() 获取当前时间，减去 timedelta(days=keep_days) 得到文件保留的时间阈值，再与文件最后修改时间对比，判断文件是否过期。

保存脚本为 clean_old_logs.py，赋予执行权限：chmod +x clean_old_logs.py；

基本使用（清理 /var/log 目录下超过 7 天的 .log 文件）：./clean_old_logs.py -d /var/log；

自定义参数（清理 /data/logs 目录下超过 3 天的 .log.1 后缀文件）：./clean_old_logs.py -d /data/logs -s .log.1 -t 3；

查看帮助：./clean_old_logs.py -h。

基础脚本仅能满足简单需求，在实际运维中，工具需具备更强的健壮性、可维护性和可追溯性。本节将基础日志清理脚本封装为专业命令行工具，新增错误处理、日志记录、日志级别控制等功能，提升工具的生产可用性。

完善错误处理：捕获目录权限不足、文件被占用等异常，避免程序崩溃；

日志记录：将清理过程、错误信息写入日志文件，便于后续追溯；

日志级别控制：支持 DEBUG/INFO/WARNING/ERROR 级别，灵活控制日志输出详细程度；

dry-run 模式：模拟清理操作，不实际删除文件，用于测试验证。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Linux 旧日志文件清理工具（进阶版）
功能：支持错误处理、日志记录、dry-run 模拟、日志级别控制
依赖：Python 3.6+
"""
import os
import argparse
import time
import logging
from datetime import datetime, timedelta

def setup_logging(log_level, log_file=None):
    """配置日志系统"""
    # 日志级别映射
    level_map = {
        "DEBUG": logging.DEBUG,
        "INFO": logging.INFO,
        "WARNING": logging.WARNING,
        "ERROR": logging.ERROR
    }
    level = level_map.get(log_level.upper(), logging.INFO)
    
    # 日志格式
    formatter = logging.Formatter(
        "%(asctime)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s"
    )
    
    # 日志处理器（控制台 + 文件）
    handlers = [logging.StreamHandler()]
    if log_file:
        file_handler = logging.FileHandler(log_file, encoding="utf-8")
        file_handler.setFormatter(formatter)
        handlers.append(file_handler)
    
    # 配置根日志
    logging.basicConfig(level=level, handlers=handlers)
    return logging.getLogger(__name__)

def parse_args():
    """解析命令行参数（强化版）"""
    parser = argparse.ArgumentParser(description="Linux 旧日志文件清理工具（进阶版）")
    parser.add_argument("-d", "--dir", required=True, help="日志文件所在目录（必填）")
    parser.add_argument("-s", "--suffix", default=".log", help="需要清理的文件后缀（默认：.log）")
    parser.add_argument("-t", "--days", type=int, default=7, help="保留文件的天数（默认：7 天）")
    parser.add_argument("--dry-run", action="store_true", help="模拟清理，不实际删除文件")
    parser.add_argument("--log-level", default="INFO", choices=["DEBUG", "INFO", "WARNING", "ERROR"], help="日志级别（默认：INFO）")
    parser.add_argument("--log-file", help="日志文件路径（默认仅输出到控制台）")
    return parser.parse_args()

def clean_old_files(log_dir, file_suffix, keep_days, dry_run, logger):
    """核心清理逻辑（带日志与错误处理）"""
    try:
        if not os.path.isdir(log_dir):
            logger.error(f"目录 {log_dir} 不存在，清理任务终止")
            return
        # 校验目录权限（是否可读取）
        if not os.access(log_dir, os.R_OK):
            logger.error(f"无权限访问目录 {log_dir}，请检查权限配置")
            return
        
        expire_time = datetime.now() - timedelta(days=keep_days)
        logger.info(f"开始执行清理任务：目录={log_dir}，后缀={file_suffix}，保留天数={keep_days}")
        logger.debug(f"过期时间阈值：{expire_time.strftime('%Y-%m-%d %H:%M:%S')}")
        
        deleted_count = 0
        for root, dirs, files in os.walk(log_dir):
            logger.debug(f"遍历目录：{root}，包含文件数：{len(files)}")
            for file in files:
                if file.endswith(file_suffix):
                    file_path = os.path.join(root, file)
                    try:
                        mtime = os.path.getmtime(file_path)
                        mtime_datetime = datetime.fromtimestamp(mtime)
                        if mtime_datetime < expire_time:
                            if dry_run:
                                logger.info(f"[Dry-Run] 拟删除过期文件：{file_path}（最后修改时间：{mtime_datetime.strftime('%Y-%m-%d %H:%M:%S')}）")
                            else:
                                os.remove(file_path)
                                logger.info(f"已删除过期文件：{file_path}（最后修改时间：{mtime_datetime.strftime('%Y-%m-%d %H:%M:%S')}）")
                                deleted_count += 1
                    except PermissionError:
                        logger.warning(f"无权限操作文件：{file_path}，跳过该文件")
                    except Exception as e:
                        logger.error(f"处理文件 {file_path} 时出错：{str(e)}", exc_info=True)
        
        logger.info(f"清理任务执行完成！共处理过期文件 {deleted_count} 个（{'模拟模式，未实际删除' if dry_run else '已实际删除'}）")
    except Exception as e:
        logger.critical(f"清理任务意外终止：{str(e)}", exc_info=True)

if __name__ == "__main__":
    args = parse_args()
    # 初始化日志
    logger = setup_logging(args.log_level, args.log_file)
    # 执行清理
    clean_old_files(args.dir, args.suffix, args.days, args.dry_run, logger)

日志系统（logging 模块）：Python 内置日志模块，支持多级别日志输出、多处理器（控制台 + 文件）、自定义日志格式。通过 setup_logging 函数统一配置，既能在控制台实时查看结果，又能将日志写入文件便于追溯，错误信息还会记录堆栈轨迹（exc_info=True），快速定位问题。

完善错误处理：新增目录权限校验（os.access），捕获 PermissionError、文件操作异常等特定异常，避免单一文件处理失败导致整个脚本终止；同时通过 logger 记录不同级别错误，而非简单 print，更符合生产工具规范。

dry-run 模式：通过 –dry-run 参数启用模拟清理，仅打印拟删除文件信息，不实际执行删除操作，可在正式清理前验证参数是否正确，避免误删风险，这是运维工具的常用核心功能。

本节以“自动检测并重启宕机服务”为例，完整演示从需求分析、设计思路到代码实现、测试验证的全流程，帮助读者将前文所学知识落地应用。

针对 Linux 系统中的指定服务（如 nginx、mysql），实现以下功能：

• 定期检测服务运行状态，支持指定服务名称、检测间隔；
• 若服务宕机，自动尝试重启，记录重启过程与结果；
• 重启失败时，记录错误日志并退出（避免无限重启）；
• 支持后台运行，日志写入指定文件，便于长期监控。

服务状态检测：通过 Linux 系统命令 systemctl is-active 服务名 或 ps aux | grep 服务名 判断服务是否运行，本文采用 systemctl（适用于 systemd 系统，主流 Linux 发行版均支持）；

命令执行：使用 subprocess 模块调用系统命令，捕获输出结果与返回码，判断命令执行成功与否；

循环检测：通过 while 循环实现定期检测，间隔时间由用户指定，支持手动终止（Ctrl+C）；

异常处理：捕获命令执行失败、服务重启失败等异常，记录日志；

日志与参数：集成 logging 模块记录全程日志，通过 argparse 支持自定义参数。

完整代码

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
服务状态监控与自动重启工具
功能：定期检测服务状态，宕机时自动重启，记录全程日志
依赖：Python 3.6+，适用于 systemd 系统（CentOS 7+、Ubuntu 16.04+ 等）
"""
import os
import sys
import time
import logging
import argparse
import subprocess
from signal import signal, SIGINT

def setup_logging(log_level, log_file):
    """配置日志系统"""
    level_map = {"DEBUG": logging.DEBUG, "INFO": logging.INFO, "WARNING": logging.WARNING, "ERROR": logging.ERROR}
    level = level_map.get(log_level.upper(), logging.INFO)
    formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
    handlers = [logging.StreamHandler()]
    if log_file:
        file_handler = logging.FileHandler(log_file, encoding="utf-8")
        file_handler.setFormatter(formatter)
        handlers.append(file_handler)
    logging.basicConfig(level=level, handlers=handlers)
    return logging.getLogger(__name__)

def parse_args():
    """解析命令行参数"""
    parser = argparse.ArgumentParser(description="服务状态监控与自动重启工具")
    parser.add_argument("-s", "--service", required=True, help="需要监控的服务名称（必填，如 nginx、mysql）")
    parser.add_argument("-i", "--interval", type=int, default=60, help="检测间隔（秒，默认：60 秒）")
    parser.add_argument("--log-level", default="INFO", choices=["DEBUG", "INFO", "WARNING", "ERROR"], help="日志级别（默认：INFO）")
    parser.add_argument("--log-file", help="日志文件路径（默认仅输出到控制台）")
    return parser.parse_args()

def execute_command(cmd):
    """
    执行系统命令并返回结果
    :param cmd: 命令字符串
    :return: (return_code, stdout, stderr) 命令返回码、标准输出、标准错误
    """
    try:
        result = subprocess.run(
            cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding="utf-8", timeout=10
        )
        return (result.returncode, result.stdout.strip(), result.stderr.strip())
    except subprocess.TimeoutExpired:
        return (-1, "", "命令执行超时")
    except Exception as e:
        return (-2, "", f"命令执行异常：{str(e)}")

def check_service_status(service_name, logger):
    """检查服务运行状态"""
    cmd = f"systemctl is-active {service_name}"
    ret_code, stdout, stderr = execute_command(cmd)
    if ret_code == 0:
        if stdout == "active":
            logger.debug(f"服务 {service_name} 运行正常（状态：{stdout}）")
            return True
        else:
            logger.warning(f"服务 {service_name} 状态异常（状态：{stdout}）")
            return False
    else:
        logger.error(f"检测服务 {service_name} 状态失败：{stderr}（返回码：{ret_code}）")
        return False

def restart_service(service_name, logger):
    """重启服务并返回结果"""
    logger.warning(f"开始尝试重启服务 {service_name}")
    cmd = f"systemctl restart {service_name}"
    # 重启命令需要 root 权限，检测当前用户是否为 root
    if os.geteuid() != 0:
        logger.error("重启服务需要 root 权限，请使用 sudo 运行脚本")
        return False
    ret_code, stdout, stderr = execute_command(cmd)
    if ret_code == 0:
        logger.info(f"服务 {service_name} 重启命令执行成功，等待 5 秒后验证状态")
        time.sleep(5)
        # 验证重启后状态
        if check_service_status(service_name, logger):
            logger.info(f"服务 {service_name} 重启成功并正常运行")
            return True
        else:
            logger.error(f"服务 {service_name} 重启命令执行成功，但状态仍异常")
            return False
    else:
        logger.error(f"服务 {service_name} 重启失败：{stderr}（返回码：{ret_code}）")
        return False

def signal_handler(signum, frame, logger):
    """处理 Ctrl+C 终止信号"""
    logger.info("收到终止信号，正在退出监控程序...")
    sys.exit(0)

def start_monitor(service_name, interval, logger):
    """启动服务监控循环"""
    logger.info(f"服务监控程序启动：监控服务={service_name}，检测间隔={interval}秒")
    # 注册 Ctrl+C 信号处理
    signal(SIGINT, lambda signum, frame: signal_handler(signum, frame, logger))
    while True:
        try:
            if not check_service_status(service_name, logger):
                # 服务异常，尝试重启
                if not restart_service(service_name, logger):
                    logger.critical(f"服务 {service_name} 重启失败，监控程序退出")
                    break
            # 等待下一次检测
            time.sleep(interval)
        except Exception as e:
            logger.error(f"监控循环异常：{str(e)}", exc_info=True)
            time.sleep(interval)

if __name__ == "__main__":
    args = parse_args()
    logger = setup_logging(args.log_level, args.log_file)
    # 启动监控
    start_monitor(args.service, args.interval, logger)

关键模块补充解析

subprocess 模块：用于调用 Linux 系统命令，相较于 os.system 更强大，能捕获命令返回码、标准输出和标准错误。本文中通过 subprocess.run 执行 systemctl 命令，设置 shell=True 支持命令字符串格式，timeout 避免命令卡死，encoding 指定输出编码为 utf-8，便于中文处理。

信号处理（signal 模块）：注册 SIGINT 信号（对应 Ctrl+C）的处理函数，使程序在被手动终止时能优雅退出，记录终止日志，避免强制终止导致的资源泄露。

权限校验（os.geteuid）：重启服务需要 root 权限，通过 os.geteuid() == 0 判断当前用户是否为 root，若不是则提示并退出，避免因权限不足导致重启失败。

环境准备

以 CentOS 8 系统、nginx 服务为例，确保系统已安装 nginx 并启用 systemd 管理：yum install -y nginx && systemctl enable --now nginx。

功能测试

• 保存脚本为 service_monitor.py，赋予执行权限：chmod +x service_monitor.py；
• 正常监控测试：以 root 身份运行 ./service_monitor.py -s nginx -i 30 --log-file /var/log/service_monitor.log，观察日志是否正常记录服务运行状态；
• 宕机重启测试：手动停止 nginx 服务 systemctl stop nginx，等待 30 秒后查看日志，验证程序是否自动重启 nginx，且重启后状态正常；
• 重启失败测试：模拟服务无法重启（如修改 nginx 配置文件导致启动失败），查看日志是否记录重启失败信息并退出程序；
• 权限测试：以普通用户运行脚本，验证是否提示需要 root 权限；
• 终止测试：按下 Ctrl+C，验证程序是否优雅退出并记录终止日志。

以下使用 Mermaid 语法绘制“自动检测并重启宕机服务”案例的逻辑流程图，清晰展示程序执行流程：

Python 在 Linux 自动化运维中具备“轻量、高效、易扩展”的核心优势，从简单的日志清理脚本，到功能完善的服务监控工具，仅需依托内置模块即可快速实现。本文通过“基础脚本-进阶封装-实际案例”的递进式讲解，展示了 Python 自动化的开发流程：先通过核心模块实现核心功能，再通过错误处理、日志记录、参数解析强化工具健壮性，最后结合实际需求落地案例，形成可直接应用于生产环境的运维工具。

相较于传统手动运维或 Shell 脚本，Python 自动化工具更易维护、可扩展性更强，能有效减少重复性工作，降低人为操作失误，提升运维效率与系统稳定性。

• 结合自动化运维工具链：将 Python 脚本与 Ansible、SaltStack 等工具结合，实现大规模服务器集群的自动化管理。例如，通过 Ansible 批量分发 Python 监控脚本，收集各节点的监控数据并汇总分析。
• 集成 systemd 服务：将本文开发的服务监控工具注册为 systemd 服务，设置开机自启、自动重启，实现工具的长期稳定运行。例如，创建 /etc/systemd/system/service-monitor.service 配置文件，定义服务启动参数与日志路径。
• 新增告警功能：扩展案例工具，在服务重启失败时添加邮件、钉钉、企业微信告警，及时通知运维人员处理。可借助 smtplib 模块发送邮件，或通过第三方 API 调用钉钉/企业微信机器人。
• 数据持久化与可视化：将监控数据（服务状态、重启次数、异常信息）存入数据库（如 SQLite、MySQL），结合 Flask/Django 开发简单的 Web 界面，实现监控数据可视化与历史查询。
• 引入配置文件：对于复杂工具，可通过 YAML/JSON 配置文件管理参数（如服务列表、检测间隔、日志配置），替代命令行参数，提升工具的灵活性与可维护性，可使用 PyYAML 模块解析 YAML 配置。

通过持续学习与实践，结合 Python 丰富的生态与 Linux 系统特性，可开发出更贴合实际运维需求的自动化工具，逐步实现运维工作的“自动化、智能化、可视化”。

以上就是Linux下Python自动化脚本编写到运维工具开发详解的详细内容，更多关于Python自动化脚本的资料请关注风君子博客其它相关文章！

您可能感兴趣的文章:

• Python解放双手的15个超实用的自动化办公脚本(附代码)
• Python中实现文件批量处理的自动化脚本
• Python自动化办公的10个脚本合集(含源码)
• 10个让你效率翻倍的Python自动化脚本分享
• Python自动化读取txt文件数据的8个实用脚本
• Windows自动化重启python脚本的方法详解
• 10个Python自动化办公的脚本分享