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

#!/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)

#!/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)

#!/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)