如何写本地文件云端同步脚本

wen 实用脚本 2026-06-28 234

本文目录导读：

如何写本地文件云端同步脚本

基于 rclone 的同步脚本（推荐）
基于 rsync 的同步脚本
支持多种云存储的通用脚本
自动化定时同步
使用 inotify 实时监控同步
配置文件示例
使用建议

我来为您介绍几种常见的本地文件云端同步脚本方案。

基于 rclone 的同步脚本（推荐）

安装 rclone

# macOS
brew install rclone
# Ubuntu/Debian
sudo apt install rclone
# 使用脚本安装
curl https://rclone.org/install.sh | sudo bash

基础同步脚本

#!/bin/bash
# sync_to_cloud.sh
set -e
# 配置
LOCAL_DIR="/path/to/local/directory"
REMOTE_NAME="mycloud"  # rclone 配置的远程名称
REMOTE_PATH="mycloud:backup"
LOG_FILE="/var/log/sync.log"
EXCLUDE_FILE="/path/to/exclude.txt"
# 日志函数
log() {
    echo "$(date '+%Y-%m-%d %H:%M:%S') - $1" >> "$LOG_FILE"
}
# 同步到云端
sync_to_cloud() {
    log "开始同步到云端..."
    rclone sync "$LOCAL_DIR" "$REMOTE_PATH" \
        --verbose \
        --progress \
        --exclude-from "$EXCLUDE_FILE" \
        --delete-excluded \
        --backup-dir "$REMOTE_PATH/_archive" \
        --log-file "$LOG_FILE" 2>&1
    if [ $? -eq 0 ]; then
        log "同步成功完成"
    else
        log "同步失败"
        exit 1
    fi
}
# 从云端同步回本地
sync_from_cloud() {
    log "开始从云端同步..."
    rclone sync "$REMOTE_PATH" "$LOCAL_DIR" \
        --verbose \
        --progress \
        --exclude-from "$EXCLUDE_FILE" \
        --log-file "$LOG_FILE" 2>&1
    if [ $? -eq 0 ]; then
        log "同步成功完成"
    else
        log "同步失败"
        exit 1
    fi
}
# 主函数
main() {
    case "$1" in
        "up")
            sync_to_cloud
            ;;
        "down")
            sync_from_cloud
            ;;
        *)
            echo "使用方法: $0 {up|down}"
            exit 1
            ;;
    esac
}
main "$@"

排除文件规则

# exclude.txt
.git/
node_modules/
*.log
*.tmp
.DS_Store
Thumbs.db

基于 rsync 的同步脚本

#!/bin/bash
# rsync_sync.sh
set -e
# 配置
LOCAL_DIR="/path/to/local"
REMOTE_USER="username"
REMOTE_HOST="remote.server.com"
REMOTE_DIR="/path/to/remote/backup"
SSH_KEY="/path/to/ssh/key"
LOG_FILE="/var/log/rsync_sync.log"
# 检查依赖
check_dependencies() {
    command -v rsync >/dev/null 2>&1 || {
        echo "需要安装 rsync"
        exit 1
    }
}
# 执行同步
sync_directory() {
    local source="$1"
    local destination="$2"
    local direction="$3"
    log "开始 $direction 同步..."
    rsync -avz --progress \
        --delete \
        --partial \
        --append \
        --exclude='.git/' \
        --exclude='node_modules/' \
        --exclude='*.log' \
        --exclude='.DS_Store' \
        -e "ssh -i $SSH_KEY" \
        "$source" "$destination" \
        --log-file="$LOG_FILE"
    if [ $? -eq 0 ]; then
        log "$direction 同步成功"
    else
        log "$direction 同步失败"
        return 1
    fi
}
# 日志函数
log() {
    echo "$(date '+%Y-%m-%d %H:%M:%S') - $1" | tee -a "$LOG_FILE"
}
# 主逻辑
main() {
    check_dependencies
    case "$1" in
        "up")
            sync_directory "$LOCAL_DIR/" "$REMOTE_USER@$REMOTE_HOST:$REMOTE_DIR" "上传"
            ;;
        "down")
            sync_directory "$REMOTE_USER@$REMOTE_HOST:$REMOTE_DIR/" "$LOCAL_DIR" "下载"
            ;;
        *)
            echo "用法: $0 {up|down}"
            exit 1
            ;;
    esac
}
main "$@"

支持多种云存储的通用脚本

#!/usr/bin/env python3
# cloud_sync.py
import os
import sys
import json
import logging
import argparse
from datetime import datetime
from pathlib import Path
# 云存储库（需要安装）
# pip install boto3 google-cloud-storage azure-storage-blob
class CloudSync:
    def __init__(self, config_file='sync_config.json'):
        self.config = self.load_config(config_file)
        self.setup_logging()
    def load_config(self, config_file):
        with open(config_file, 'r') as f:
            return json.load(f)
    def setup_logging(self):
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('sync.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
    def sync_to_s3(self, local_path, bucket, prefix=''):
        """同步到 AWS S3"""
        import boto3
        s3 = boto3.client('s3')
        local = Path(local_path)
        for file in local.rglob('*'):
            if file.is_file():
                s3_key = f"{prefix}/{file.relative_to(local)}"
                try:
                    s3.upload_file(str(file), bucket, s3_key)
                    self.logger.info(f"上传: {file} -> s3://{bucket}/{s3_key}")
                except Exception as e:
                    self.logger.error(f"上传失败 {file}: {e}")
    def sync_from_s3(self, local_path, bucket, prefix=''):
        """从 AWS S3 同步"""
        import boto3
        s3 = boto3.client('s3')
        local = Path(local_path)
        local.mkdir(parents=True, exist_ok=True)
        response = s3.list_objects_v2(Bucket=bucket, Prefix=prefix)
        if 'Contents' in response:
            for obj in response['Contents']:
                key = obj['Key']
                rel_path = key.replace(prefix, '').lstrip('/')
                dest_file = local / rel_path
                dest_file.parent.mkdir(parents=True, exist_ok=True)
                try:
                    s3.download_file(bucket, key, str(dest_file))
                    self.logger.info(f"下载: s3://{bucket}/{key} -> {dest_file}")
                except Exception as e:
                    self.logger.error(f"下载失败 {key}: {e}")
    def sync_to_gcs(self, local_path, bucket, prefix=''):
        """同步到 Google Cloud Storage"""
        from google.cloud import storage
        client = storage.Client()
        bucket = client.bucket(bucket)
        local = Path(local_path)
        for file in local.rglob('*'):
            if file.is_file():
                blob_name = f"{prefix}/{file.relative_to(local)}"
                blob = bucket.blob(blob_name)
                try:
                    blob.upload_from_filename(str(file))
                    self.logger.info(f"上传: {file} -> gs://{bucket.name}/{blob_name}")
                except Exception as e:
                    self.logger.error(f"上传失败 {file}: {e}")
    def sync_to_azure(self, local_path, container_name, prefix=''):
        """同步到 Azure Blob Storage"""
        from azure.storage.blob import BlobServiceClient
        conn_str = self.config.get('azure_connection_string')
        service = BlobServiceClient.from_connection_string(conn_str)
        container = service.get_container_client(container_name)
        local = Path(local_path)
        for file in local.rglob('*'):
            if file.is_file():
                blob_name = f"{prefix}/{file.relative_to(local)}"
                try:
                    with open(file, "rb") as data:
                        container.upload_blob(name=blob_name, data=data, overwrite=True)
                    self.logger.info(f"上传: {file} -> {blob_name}")
                except Exception as e:
                    self.logger.error(f"上传失败 {file}: {e}")
def main():
    parser = argparse.ArgumentParser(description='文件云同步工具')
    parser.add_argument('action', choices=['up', 'down'], help='同步方向')
    parser.add_argument('--provider', default='s3', choices=['s3', 'gcs', 'azure'], help='云服务提供商')
    parser.add_argument('--config', default='sync_config.json', help='配置文件路径')
    args = parser.parse_args()
    # 配置文件示例
    config = {
        "local_path": "/path/to/local/directory",
        "s3_bucket": "my-bucket",
        "gcs_bucket": "my-bucket",
        "azure_container": "my-container",
        "prefix": "backup",
        "azure_connection_string": ""
    }
    sync = CloudSync()
    if args.provider == 's3':
        if args.action == 'up':
            sync.sync_to_s3(config['local_path'], config['s3_bucket'], config['prefix'])
        else:
            sync.sync_from_s3(config['local_path'], config['s3_bucket'], config['prefix'])
    elif args.provider == 'gcs':
        if args.action == 'up':
            sync.sync_to_gcs(config['local_path'], config['gcs_bucket'], config['prefix'])
    elif args.provider == 'azure':
        if args.action == 'up':
            sync.sync_to_azure(config['local_path'], config['azure_container'], config['prefix'])
if __name__ == "__main__":
    main()

自动化定时同步

使用 crontab（Linux/macOS）

# 编辑 crontab
crontab -e
# 每天凌晨2点同步
0 2 * * * /path/to/sync_script.sh up >> /var/log/sync_cron.log 2>&1
# 每6小时同步一次
0 */6 * * * /path/to/sync_script.sh up

使用 systemd 定时器（Linux）

# /etc/systemd/system/cloud-sync.service
[Unit]
Description=Cloud Sync Service
[Service]
Type=oneshot
ExecStart=/path/to/sync_script.sh up
User=your_username
[Install]
WantedBy=multi-user.target

# /etc/systemd/system/cloud-sync.timer
[Unit]
Description=Run cloud sync every day
[Timer]
OnCalendar=daily
Persistent=true
[Install]
WantedBy=timers.target

使用 inotify 实时监控同步

#!/bin/bash
# realtime_sync.sh - 使用 inotify 实时监控文件变化
set -e
# 配置
WATCH_DIR="/path/to/watch"
REMOTE_NAME="mycloud"
REMOTE_PATH="mycloud:backup"
RCLONE_OPTS="--verbose --progress"
# 检查依赖
check_dependencies() {
    command -v inotifywait >/dev/null 2>&1 || {
        echo "请安装 inotify-tools: sudo apt install inotify-tools"
        exit 1
    }
    command -v rclone >/dev/null 2>&1 || {
        echo "请安装 rclone"
        exit 1
    }
}
# 同步变动
sync_changes() {
    local changed_file="$1"
    local event="$2"
    echo "$(date) - 检测到 $event: $changed_file"
    rclone sync "$WATCH_DIR" "$REMOTE_PATH" $RCLONE_OPTS
}
# 主监控循环
main() {
    check_dependencies
    echo "开始监控 $WATCH_DIR 的文件变化..."
    inotifywait -m -r -e modify,create,delete,move "$WATCH_DIR" --format '%w%f %e' |
    while read file event; do
        # 忽略临时文件和某些目录
        case "$file" in
            *.swp|*.swx|*.tmp|*.part)
                continue
                ;;
            *)
                sync_changes "$file" "$event"
                ;;
        esac
    done
}
main

配置文件示例

// sync_config.json
{
    "local_path": "/home/user/documents",
    "cloud_providers": {
        "s3": {
            "bucket": "my-backup-bucket",
            "region": "us-east-1",
            "prefix": "documents"
        },
        "gcs": {
            "bucket": "my-backup-bucket",
            "project": "my-project",
            "prefix": "documents"
        }
    },
    "exclude_patterns": [
        ".git/",
        "node_modules/",
        "*.tmp",
        "*.log"
    ],
    "sync_options": {
        "delete": true,
        "backup": true,
        "backup_dir": "_archive"
    }
}