📚 课程目录第4章配置文件批量修改工具

4. 配置文件批量修改工具

任务说明

开发一个Python命令行工具，读取服务器清单与配置修改规则，通过SSH批量登录多台服务器，对目标配置文件（如nginx.conf）中的指定参数进行精确替换，并在操作前后输出变更摘要以便核对。

环境依赖

Python 3.8+
安装依赖：pip install paramiko

核心实现

import paramiko
import re
import sys
from dataclasses import dataclass
from typing import List, Dict

@dataclass
class Server:
    host: str
    username: str
    password: str
    port: int = 22

@dataclass
class ConfigChange:
    file_path: str
    old_pattern: str
    new_value: str

def load_servers(file_path: str) -> List[Server]:
    servers = []
    with open(file_path, 'r') as f:
        for line in f:
            line = line.strip()
            if not line or line.startswith('#'):
                continue
            parts = line.split()
            if len(parts) < 3:
                continue
            host, user, pwd = parts[0], parts[1], parts[2]
            port = int(parts[3]) if len(parts) > 3 else 22
            servers.append(Server(host, user, pwd, port))
    return servers

def load_changes(file_path: str) -> List[ConfigChange]:
    changes = []
    with open(file_path, 'r') as f:
        for line in f:
            line = line.strip()
            if not line or line.startswith('#'):
                continue
            parts = line.split('|')
            if len(parts) != 3:
                continue
            changes.append(ConfigChange(parts[0].strip(), parts[1].strip(), parts[2].strip()))
    return changes

def modify_config(ssh_client, change: ConfigChange) -> tuple:
    """返回 (是否修改, 修改前内容, 修改后内容)"""
    sftp = ssh_client.open_sftp()
    try:
        with sftp.open(change.file_path, 'r') as f:
            content = f.read().decode('utf-8')
        
        pattern = re.compile(change.old_pattern)
        matches = pattern.findall(content)
        if not matches:
            return (False, None, None)
        
        new_content = pattern.sub(change.new_value, content)
        if new_content == content:
            return (False, None, None)
        
        with sftp.open(change.file_path, 'w') as f:
            f.write(new_content.encode('utf-8'))
        
        return (True, content, new_content)
    finally:
        sftp.close()

def process_server(server: Server, changes: List[ConfigChange]) -> Dict:
    result = {'host': server.host, 'changes': []}
    try:
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh.connect(server.host, port=server.port, username=server.username, 
                   password=server.password, timeout=10)
        
        for change in changes:
            modified, old_content, new_content = modify_config(ssh, change)
            if modified:
                # 提取变更行用于展示
                old_lines = [l for l in old_content.split('\n') if change.old_pattern in l]
                new_lines = [l for l in new_content.split('\n') if change.old_pattern in l]
                result['changes'].append({
                    'file': change.file_path,
                    'pattern': change.old_pattern,
                    'old_lines': old_lines[:3],
                    'new_lines': new_lines[:3]
                })
            else:
                result['changes'].append({
                    'file': change.file_path,
                    'pattern': change.old_pattern,
                    'status': '未匹配或无需修改'
                })
        
        ssh.close()
        result['status'] = 'success'
    except Exception as e:
        result['status'] = f'连接失败: {str(e)}'
    return result

def main():
    if len(sys.argv) != 3:
        print("用法: python config_batch_modifier.py servers.txt changes.txt")
        sys.exit(1)
    
    servers = load_servers(sys.argv[1])
    changes = load_changes(sys.argv[2])
    
    if not servers or not changes:
        print("服务器清单或修改规则为空，请检查输入文件")
        sys.exit(1)
    
    print(f"共 {len(servers)} 台服务器，{len(changes)} 条修改规则\n")
    
    for server in servers:
        result = process_server(server, changes)
        print(f"=== {result['host']} [{result['status']}] ===")
        for c in result['changes']:
            print(f"  文件: {c['file']}")
            print(f"  规则: {c['pattern']} → {changes[result['changes'].index(c)].new_value}")
            if 'old_lines' in c:
                print(f"  修改前: {c['old_lines']}")
                print(f"  修改后: {c['new_lines']}")
            else:
                print(f"  状态: {c['status']}")
        print()

if __name__ == '__main__':
    main()

输入文件格式

servers.txt（每行一台服务器，支持#注释）：

192.168.1.10 root password123
192.168.1.11 root password456 22
# 192.168.1.12 root password789

changes.txt（每行一条规则，格式：文件路径|正则模式|替换值）：

/etc/nginx/nginx.conf|worker_processes\s+\d+|worker_processes 8
/etc/nginx/nginx.conf|keepalive_timeout\s+\d+|keepalive_timeout 65

运行方式

python config_batch_modifier.py servers.txt changes.txt

验证方法

在测试服务器上先手动执行 nginx -t 确认配置语法正确。
运行工具后，观察输出中每台服务器的状态是否为 success。
检查输出中的“修改前/修改后”行内容，确认替换符合预期。
登录任意一台服务器执行 grep -n "worker_processes" /etc/nginx/nginx.conf，确认值已更新为规则中指定的新值。
若某条规则显示“未匹配或无需修改”，说明原配置中不存在该模式或值已相同，属正常情况。

注意事项

正则模式中的特殊字符需按Python正则语法转义，例如 \s 匹配空白。
工具使用密码认证，生产环境建议改用密钥认证（修改 connect 参数即可）。
修改前建议先备份原配置文件，可在 modify_config 中增加备份逻辑。