我要评论在 linux 运维场景中，重复性工作（如日志清理、服务监控、批量部署）占据了运维人员大量时间，而 python 凭借其简洁的语法、丰富的内置库与第三方生态，成为 linux 自动化运维的首选语言。相较于 shell 脚本，python 具备更强的逻辑处理能力、跨平台兼容性和可维护性，能轻松应对复杂的自动化需求；同时，python 与 linux 系统底层交互顺畅，可直接调用系统命令、操作文件目录、管理进程服务，大幅提升运维效率。
一、引言
python 在 linux 自动化运维中的典型应用场景包括：日志收集与分析（快速提取关键信息、定位问题）、服务状态监控（实时检测服务可用性，自动告警）、批量任务执行（批量部署软件、配置服务器）、配置文件管理（统一修改、备份配置）以及系统资源统计（cpu、内存、磁盘使用率采集）等。本文将从基础脚本编写入手，逐步讲解如何封装进阶工具，并通过实际案例演示完整开发流程，帮助读者快速掌握 linux 下 python 自动化的核心技能。
二、基础脚本编写：实用示例与核心模块
基础脚本是自动化运维的基石，通常聚焦于单一简单任务。本节以“自动清理旧日志文件”为例，编写可直接运行的 python 脚本，同时讲解 os、subprocess、argparse 等核心模块的使用方法。
2.1 需求说明
清理指定目录下指定后缀（如 .log）、超过指定天数的旧日志文件，支持显示清理过程与跳过空目录，避免误删重要文件。
2.2 完整脚本代码
#!/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)
2.3 核心模块解析
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) 得到文件保留的时间阈值，再与文件最后修改时间对比，判断文件是否过期。
2.4 脚本使用方法
保存脚本为 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。
三、进阶工具开发：脚本封装与功能强化
基础脚本仅能满足简单需求，在实际运维中，工具需具备更强的健壮性、可维护性和可追溯性。本节将基础日志清理脚本封装为专业命令行工具，新增错误处理、日志记录、日志级别控制等功能，提升工具的生产可用性。
3.1 强化功能说明
完善错误处理：捕获目录权限不足、文件被占用等异常，避免程序崩溃；
日志记录：将清理过程、错误信息写入日志文件，便于后续追溯；
日志级别控制：支持 debug/info/warning/error 级别，灵活控制日志输出详细程度；
dry-run 模式：模拟清理操作，不实际删除文件，用于测试验证。
3.2 进阶工具代码片段
#!/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)
3.3 核心强化点解析
日志系统（logging 模块）：python 内置日志模块，支持多级别日志输出、多处理器（控制台 + 文件）、自定义日志格式。通过 setup_logging 函数统一配置，既能在控制台实时查看结果，又能将日志写入文件便于追溯，错误信息还会记录堆栈轨迹（exc_info=true），快速定位问题。
完善错误处理：新增目录权限校验（os.access），捕获 permissionerror、文件操作异常等特定异常，避免单一文件处理失败导致整个脚本终止；同时通过 logger 记录不同级别错误，而非简单 print，更符合生产工具规范。
dry-run 模式：通过 --dry-run 参数启用模拟清理，仅打印拟删除文件信息，不实际执行删除操作，可在正式清理前验证参数是否正确，避免误删风险，这是运维工具的常用核心功能。
四、实际案例：自动检测并重启宕机服务
本节以“自动检测并重启宕机服务”为例，完整演示从需求分析、设计思路到代码实现、测试验证的全流程，帮助读者将前文所学知识落地应用。
4.1 需求分析
针对 linux 系统中的指定服务（如 nginx、mysql），实现以下功能：
- 定期检测服务运行状态，支持指定服务名称、检测间隔；
- 若服务宕机，自动尝试重启，记录重启过程与结果；
- 重启失败时，记录错误日志并退出（避免无限重启）；
- 支持后台运行，日志写入指定文件，便于长期监控。
4.2 设计思路
服务状态检测：通过 linux 系统命令 systemctl is-active 服务名 或 ps aux | grep 服务名 判断服务是否运行，本文采用 systemctl（适用于 systemd 系统，主流 linux 发行版均支持）；
命令执行：使用 subprocess 模块调用系统命令，捕获输出结果与返回码，判断命令执行成功与否；
循环检测：通过 while 循环实现定期检测，间隔时间由用户指定，支持手动终止（ctrl+c）；
异常处理：捕获命令执行失败、服务重启失败等异常，记录日志；
日志与参数：集成 logging 模块记录全程日志，通过 argparse 支持自定义参数。
4.3 实现过程
完整代码
#!/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，若不是则提示并退出，避免因权限不足导致重启失败。
4.4 测试方法
环境准备
以 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 语法绘制“自动检测并重启宕机服务”案例的逻辑流程图，清晰展示程序执行流程：
六、总结与扩展建议
6.1 总结
python 在 linux 自动化运维中具备“轻量、高效、易扩展”的核心优势，从简单的日志清理脚本，到功能完善的服务监控工具，仅需依托内置模块即可快速实现。本文通过“基础脚本-进阶封装-实际案例”的递进式讲解，展示了 python 自动化的开发流程：先通过核心模块实现核心功能，再通过错误处理、日志记录、参数解析强化工具健壮性，最后结合实际需求落地案例，形成可直接应用于生产环境的运维工具。
相较于传统手动运维或 shell 脚本，python 自动化工具更易维护、可扩展性更强，能有效减少重复性工作，降低人为操作失误，提升运维效率与系统稳定性。
6.2 扩展建议
- 结合自动化运维工具链：将 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自动化脚本的资料请关注代码网其它相关文章！
发表评论