如何用脚本同步配置文件

wen 实用脚本 2026-08-02 2

本文目录导读：

如何用脚本同步配置文件

基础版：使用 rsync 同步
进阶版：带备份和校验的同步
高级版：Git 版本控制同步
守护进程版：实时监控同步
使用方法
推荐使用场景

我来介绍几种同步配置文件的脚本方案,从简单到复杂：

基础版：使用 rsync 同步

简单同步脚本

#!/bin/bash
# sync_config.sh - 基础配置文件同步
# 配置来源和目的地
SOURCE_DIR="/etc/nginx/"
DEST_DIR="/backup/nginx-config/"
LOG_FILE="/var/log/config-sync.log"
# 创建日志函数
log() {
    echo "$(date '+%Y-%m-%d %H:%M:%S') - $1" >> "$LOG_FILE"
}
# 执行同步
log "开始同步配置文件..."
rsync -avz --delete "$SOURCE_DIR" "$DEST_DIR"
if [ $? -eq 0 ]; then
    log "同步成功完成"
else
    log "同步失败！"
    exit 1
fi

进阶版：带备份和校验的同步

#!/bin/bash
# sync_with_backup.sh - 带备份的配置文件同步
set -e  # 出错立即退出
# 配置变量
CONFIG_FILES=(
    "/etc/nginx/nginx.conf"
    "/etc/nginx/sites-available/"
    "/etc/php/8.1/fpm/php.ini"
    "/etc/mysql/my.cnf"
)
BACKUP_ROOT="/var/backups/configs"
BACKUP_DIR="$BACKUP_ROOT/$(date +%Y%m%d_%H%M%S)"
REMOTE_HOST="user@remote-server"
REMOTE_PATH="/backup/configs/"
# 创建备份目录
mkdir -p "$BACKUP_DIR"
# 校验函数
verify_files() {
    for file in "${CONFIG_FILES[@]}"; do
        if [ ! -e "$file" ]; then
            echo "警告: $file 不存在！"
            return 1
        fi
    done
}
# 备份当前配置
backup_configs() {
    echo "开始备份..."
    for file in "${CONFIG_FILES[@]}"; do
        cp -r "$file" "$BACKUP_DIR/"
    done
    tar -czf "$BACKUP_ROOT/configs_$(date +%Y%m%d).tar.gz" "$BACKUP_DIR"
    echo "本地备份完成: $BACKUP_DIR"
}
# 同步到远程
sync_to_remote() {
    echo "同步到远程服务器..."
    rsync -avz --progress \
        --exclude='*.log' \
        --exclude='cache/' \
        -e ssh "$BACKUP_DIR/" "$REMOTE_HOST:$REMOTE_PATH"
}
# 主流程
main() {
    verify_files || exit 1
    backup_configs
    sync_to_remote
    # 保存最近5份备份，清理旧的
    cd "$BACKUP_ROOT" && ls -t *.tar.gz | tail -n +6 | xargs -r rm -f
    echo "同步完成！"
}
main

高级版：Git 版本控制同步

#!/usr/bin/env python3
# sync_config.py - 使用Git进行配置文件版本控制
import os
import subprocess
import shutil
import yaml
from datetime import datetime
from pathlib import Path
class ConfigSync:
    def __init__(self, config_file='sync_config.yaml'):
        self.load_config(config_file)
    def load_config(self, config_file):
        """加载配置文件"""
        with open(config_file, 'r') as f:
            self.config = yaml.safe_load(f)
        self.git_repo = self.config.get('git_repo', '/etc/config-repo')
        self.apps = self.config.get('apps', [])
    def setup_git(self):
        """初始化Git仓库"""
        Path(self.git_repo).mkdir(parents=True, exist_ok=True)
        os.chdir(self.git_repo)
        if not os.path.exists('.git'):
            subprocess.run(['git', 'init'], check=True)
            subprocess.run(['git', 'config', 'user.name', 'Config Sync'], check=True)
            subprocess.run(['git', 'config', 'user.email', 'sync@localhost'], check=True)
    def collect_configs(self):
        """收集所有配置文件到Git仓库"""
        for app in self.apps:
            src = app['source']
            dest = os.path.join(self.git_repo, app['name'])
            # 确保目标目录存在
            Path(dest).mkdir(parents=True, exist_ok=True)
            # 复制配置文件
            if os.path.isdir(src):
                shutil.copytree(src, dest, dirs_exist_ok=True)
            elif os.path.isfile(src):
                shutil.copy2(src, os.path.join(dest, os.path.basename(src)))
            print(f"收集配置: {app['name']} <- {src}")
    def commit_changes(self):
        """提交变更"""
        os.chdir(self.git_repo)
        # 检查是否有变更
        result = subprocess.run(['git', 'status', '--porcelain'], 
                              capture_output=True, text=True)
        if result.stdout.strip():
            # 添加所有变更
            subprocess.run(['git', 'add', '.'], check=True)
            # 提交
            commit_msg = f"更新配置 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
            subprocess.run(['git', 'commit', '-m', commit_msg], check=True)
            print(f"提交成功: {commit_msg}")
            return True
        else:
            print("没有配置变更")
            return False
    def push_to_remote(self):
        """推送到远程仓库"""
        remote_url = self.config.get('remote_url')
        if remote_url:
            os.chdir(self.git_repo)
            subprocess.run(['git', 'push', 'origin', 'main'], check=True)
            print("已推送到远程仓库")
    def restore_config(self, commit_hash=None):
        """从Git恢复配置"""
        os.chdir(self.git_repo)
        if commit_hash:
            subprocess.run(['git', 'checkout', commit_hash, '--', '.'], check=True)
            print(f"已恢复配置到提交 {commit_hash}")
        else:
            subprocess.run(['git', 'checkout', 'HEAD', '--', '.'], check=True)
            print("已恢复到最新配置")
    def run(self, action='sync'):
        """主执行入口"""
        self.setup_git()
        if action == 'sync':
            self.collect_configs()
            if self.commit_changes():
                if self.config.get('auto_push', False):
                    self.push_to_remote()
        elif action == 'restore':
            commit_hash = input("输入要恢复的提交哈希(留空为最新): ").strip() or None
            self.restore_config(commit_hash)
        elif action == 'history':
            os.chdir(self.git_repo)
            result = subprocess.run(['git', 'log', '--oneline', '-10'],
                                  capture_output=True, text=True)
            print("最近的提交:")
            print(result.stdout)
# 示例配置文件 sync_config.yaml
"""
git_repo: /etc/config-repo
remote_url: git@github.com:user/server-configs.git
auto_push: true
apps:
  - name: nginx
    source: /etc/nginx/conf.d/
  - name: php
    source: /etc/php/8.1/fpm/php.ini
  - name: mysql
    source: /etc/mysql/my.cnf
"""
# 命令行入口
if __name__ == '__main__':
    import sys
    sync = ConfigSync()
    if len(sys.argv) > 1:
        action = sys.argv[1]
    else:
        action = 'sync'
    sync.run(action)

守护进程版：实时监控同步

#!/usr/bin/env python3
# file_sync_daemon.py - 文件监控实时同步
import time
import hashlib
import os
import shutil
import json
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class ConfigSyncHandler(FileSystemEventHandler):
    """文件系统事件处理器"""
    def __init__(self, mappings, backup_dir):
        self.mappings = mappings  # [{src: dest}]
        self.backup_dir = backup_dir
        self.pending_changes = {}
        self.last_sync = time.time()
    def on_modified(self, event):
        """文件修改事件"""
        if event.is_directory:
            return
        self.handle_change(event.src_path, 'modified')
    def on_created(self, event):
        """文件创建事件"""
        if event.is_directory:
            return
        self.handle_change(event.src_path, 'created')
    def on_deleted(self, event):
        """文件删除事件"""
        if event.is_directory:
            return
        self.handle_change(event.src_path, 'deleted')
    def handle_change(self, filepath, action):
        """处理文件变更"""
        # 检查是否在监控列表中
        for mapping in self.mappings:
            src_dir = mapping['source']
            if filepath.startswith(src_dir):
                # 计算相对路径
                rel_path = os.path.relpath(filepath, src_dir)
                dest_path = os.path.join(mapping['dest'], rel_path)
                # 记录变更
                self.pending_changes[filepath] = {
                    'action': action,
                    'dest': dest_path
                }
                # 如果有待处理的变更且超过收集时间，执行同步
                current_time = time.time()
                if current_time - self.last_sync > 5:  # 5秒延迟
                    self.sync_pending()
                break
    def sync_pending(self):
        """执行待处理的同步"""
        if not self.pending_changes:
            return
        print(f"同步 {len(self.pending_changes)} 个文件变更...")
        for src, change in self.pending_changes.items():
            try:
                dest = change['dest']
                action = change['action']
                # 创建备份
                self.create_backup(src, dest)
                # 执行同步
                if action == 'deleted':
                    if os.path.exists(dest):
                        os.remove(dest)
                        print(f"删除: {dest}")
                else:
                    # 确保目录存在
                    os.makedirs(os.path.dirname(dest), exist_ok=True)
                    shutil.copy2(src, dest)
                    print(f"同步: {src} -> {dest}")
            except Exception as e:
                print(f"同步失败 {src}: {e}")
                self.log_error(src, str(e))
        self.pending_changes = {}
        self.last_sync = time.time()
    def create_backup(self, src, dest):
        """创建文件备份"""
        timestamp = time.strftime('%Y%m%d_%H%M%S')
        backup_path = os.path.join(self.backup_dir, 
                                  f"backup_{timestamp}_{os.path.basename(dest)}")
        if os.path.exists(dest):
            shutil.copy2(dest, backup_path)
    def log_error(self, src, error_msg):
        """记录错误"""
        log_file = os.path.join(self.backup_dir, 'sync_errors.log')
        with open(log_file, 'a') as f:
            f.write(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {src}: {error_msg}\n")
class ConfigDaemon:
    """配置同步守护进程"""
    def __init__(self, config_file):
        self.load_config(config_file)
    def load_config(self, config_file):
        """加载配置"""
        with open(config_file, 'r') as f:
            self.config = json.load(f)
        self.mappings = self.config.get('mappings', [])
        self.backup_dir = self.config.get('backup_dir', '/var/backups/configs')
        self.log_dir = self.config.get('log_dir', '/var/log/config-sync')
        # 确保目录存在
        Path(self.backup_dir).mkdir(parents=True, exist_ok=True)
        Path(self.log_dir).mkdir(parents=True, exist_ok=True)
    def start(self):
        """启动监控"""
        event_handler = ConfigSyncHandler(self.mappings, self.backup_dir)
        observers = []
        # 为每个源目录创建观察者
        for mapping in self.mappings:
            src_dir = mapping['source']
            observer = Observer()
            observer.schedule(event_handler, src_dir, recursive=True)
            observer.start()
            observers.append(observer)
            print(f"开始监控: {src_dir}")
        try:
            while True:
                time.sleep(1)
        except KeyboardInterrupt:
            for observer in observers:
                observer.stop()
            print("\n停止监控")
        for observer in observers:
            observer.join()
    def run_once(self):
        """执行一次性同步"""
        print("执行一次性同步...")
        for mapping in self.mappings:
            src = mapping['source']
            dest = mapping['dest']
            if os.path.isdir(src):
                shutil.copytree(src, dest, dirs_exist_ok=True)
            elif os.path.isfile(src):
                os.makedirs(os.path.dirname(dest), exist_ok=True)
                shutil.copy2(src, dest)
            print(f"同步: {src} -> {dest}")
# 配置示例 config_sync.json
"""
{
    "mappings": [
        {
            "source": "/etc/nginx/",
            "dest": "/backup/nginx/"
        },
        {
            "source": "/etc/php/8.1/fpm/",
            "dest": "/backup/php-fpm/"
        }
    ],
    "backup_dir": "/var/backups/configs",
    "log_dir": "/var/log/config-sync"
}
"""
if __name__ == '__main__':
    import sys
    daemon = ConfigDaemon('config_sync.json')
    if len(sys.argv) > 1 and sys.argv[1] == 'run':
        daemon.run_once()
    else:
        daemon.start()

使用方法

安装依赖

# 对于Python脚本
pip install watchdog pyyaml
# 对于Shell脚本
sudo apt-get install rsync  # Ubuntu/Debian
sudo yum install rsync      # CentOS/RHEL

创建系统服务（可选）

# /etc/systemd/system/config-sync.service
[Unit]
Description=Config File Sync Service
After=network.target
[Service]
Type=simple
User=root
ExecStart=/usr/local/bin/config-sync.py
Restart=always
RestartSec=10
[Install]
WantedBy=multi-user.target

定时任务（cron）

# 每天凌晨2点同步配置
0 2 * * * /usr/local/bin/sync_config.sh
# 每小时同步并记录日志
0 * * * * /usr/local/bin/sync_config.sh >> /var/log/config-sync.log 2>&1

如何用脚本同步配置文件

基础版：使用 rsync 同步

简单同步脚本

进阶版：带备份和校验的同步

高级版：Git 版本控制同步

守护进程版：实时监控同步

使用方法

安装依赖

创建系统服务（可选）

定时任务（cron）

推荐使用场景