IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

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

wen 实用脚本 26

本文目录导读：

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

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

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

基于 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"
    }
}

使用建议

  1. 首次使用：先在测试目录测试脚本功能
  2. 错误处理：添加合适的错误处理和重试机制
  3. 日志记录：详细记录同步过程，便于排查问题
  4. 增量同步：使用 --checksum 或 --size-only 避免全量同步
  5. 带宽控制：使用 --bwlimit 限制带宽使用
  6. 加密传输：配置 SSH 或使用 HTTPS 确保传输安全

选择哪种方案取决于您的具体需求：

  • 简单同步：使用 rclone
  • Linux 服务器：使用 rsync
  • 多平台支持：使用 Python 脚本
  • 实时同步：使用 inotify + rclone

需要我详细解释某个特定方案或功能吗？

上一篇如何编写回收站管理脚本

下一篇脚本如何恢复回收站文件

相关文章

抱歉，评论功能暂时关闭!