如何编写文件同步脚本

wen 实用脚本 2

本文目录导读:

如何编写文件同步脚本

  1. 基本用法
  2. 双向同步
  3. 包含隐藏文件

我来帮你编写一个跨平台的文件同步脚本,这里提供多个版本供选择:

Python 版本(推荐,跨平台)

基础版文件同步脚本

#!/usr/bin/env python3
"""
文件同步脚本 - 支持单向和双向同步
"""
import os
import shutil
import hashlib
import argparse
from pathlib import Path
from datetime import datetime
import json
class FileSync:
    def __init__(self, source, target, mode='one-way', ignore_hidden=True):
        """
        初始化同步器
        Args:
            source: 源目录
            target: 目标目录
            mode: 'one-way' 单向同步,'two-way' 双向同步
            ignore_hidden: 是否忽略隐藏文件
        """
        self.source = Path(source)
        self.target = Path(target)
        self.mode = mode
        self.ignore_hidden = ignore_hidden
        self.log = []
    def get_file_hash(self, filepath):
        """计算文件MD5哈希值"""
        hash_md5 = hashlib.md5()
        with open(filepath, "rb") as f:
            for chunk in iter(lambda: f.read(4096), b""):
                hash_md5.update(chunk)
        return hash_md5.hexdigest()
    def should_ignore(self, path):
        """检查是否应该忽略该路径"""
        if self.ignore_hidden and path.name.startswith('.'):
            return True
        # 添加其他忽略规则
        ignore_patterns = ['__pycache__', '.git', '.svn', 'node_modules']
        for pattern in ignore_patterns:
            if pattern in path.parts:
                return True
        return False
    def relative_path(self, path):
        """获取相对路径"""
        if self.mode == 'one-way':
            return path.relative_to(self.source)
        else:
            # 双向同步时,找出相对于源或目标的路径
            try:
                return path.relative_to(self.source)
            except ValueError:
                return path.relative_to(self.target)
    def sync_file(self, src_file, dst_file):
        """同步单个文件"""
        if not dst_file.exists():
            self.copy_file(src_file, dst_file)
        else:
            # 比较文件是否相同
            if self.get_file_hash(src_file) != self.get_file_hash(dst_file):
                self.copy_file(src_file, dst_file)
    def copy_file(self, src, dst):
        """复制文件"""
        try:
            dst.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(src, dst)
            self.log.append(f"复制: {src} -> {dst}")
            print(f"复制: {src.name} -> {dst}")
        except Exception as e:
            print(f"错误: 无法复制 {src}: {e}")
    def delete_file(self, filepath):
        """删除文件"""
        try:
            filepath.unlink()
            self.log.append(f"删除: {filepath}")
            print(f"删除: {filepath}")
        except Exception as e:
            print(f"错误: 无法删除 {filepath}: {e}")
    def sync_one_way(self):
        """单向同步:源目录同步到目标目录"""
        print(f"开始单向同步: {self.source} -> {self.target}")
        # 复制源目录中的文件到目标目录
        for root, dirs, files in os.walk(self.source):
            root_path = Path(root)
            # 过滤隐藏目录
            if self.ignore_hidden:
                dirs[:] = [d for d in dirs if not d.startswith('.')]
            for file in files:
                src_file = root_path / file
                if self.should_ignore(src_file):
                    continue
                rel_path = src_file.relative_to(self.source)
                dst_file = self.target / rel_path
                self.sync_file(src_file, dst_file)
        # 删除目标目录中源目录不存在的文件
        for root, dirs, files in os.walk(self.target):
            root_path = Path(root)
            if self.ignore_hidden:
                dirs[:] = [d for d in dirs if not d.startswith('.')]
            for file in files:
                dst_file = root_path / file
                if self.should_ignore(dst_file):
                    continue
                rel_path = dst_file.relative_to(self.target)
                src_file = self.source / rel_path
                if not src_file.exists():
                    self.delete_file(dst_file)
    def sync_two_way(self):
        """双向同步:两边互相同步"""
        print(f"开始双向同步: {self.source} <-> {self.target}")
        # 从源到目标
        for root, dirs, files in os.walk(self.source):
            root_path = Path(root)
            if self.ignore_hidden:
                dirs[:] = [d for d in dirs if not d.startswith('.')]
            for file in files:
                src_file = root_path / file
                if self.should_ignore(src_file):
                    continue
                rel_path = src_file.relative_to(self.source)
                dst_file = self.target / rel_path
                self.sync_file(src_file, dst_file)
        # 从目标到源
        for root, dirs, files in os.walk(self.target):
            root_path = Path(root)
            if self.ignore_hidden:
                dirs[:] = [d for d in dirs if not d.startswith('.')]
            for file in files:
                dst_file = root_path / file
                if self.should_ignore(dst_file):
                    continue
                rel_path = dst_file.relative_to(self.target)
                src_file = self.source / rel_path
                if not src_file.exists():
                    self.sync_file(dst_file, src_file)
                else:
                    # 如果两边都有,保留最新的
                    dst_mtime = dst_file.stat().st_mtime
                    src_mtime = src_file.stat().st_mtime
                    if dst_mtime > src_mtime:
                        self.sync_file(dst_file, src_file)
                    elif src_mtime > dst_mtime:
                        self.sync_file(src_file, dst_file)
    def run(self):
        """执行同步"""
        if not self.source.exists():
            raise ValueError(f"源目录不存在: {self.source}")
        self.target.mkdir(parents=True, exist_ok=True)
        if self.mode == 'one-way':
            self.sync_one_way()
        elif self.mode == 'two-way':
            self.sync_two_way()
        else:
            raise ValueError(f"无效的同步模式: {self.mode}")
        # 输出同步日志
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        log_file = f"sync_log_{timestamp}.json"
        with open(log_file, 'w', encoding='utf-8') as f:
            json.dump(self.log, f, indent=2, ensure_ascii=False)
        print(f"同步完成!日志保存至: {log_file}")
def main():
    parser = argparse.ArgumentParser(description='文件同步工具')
    parser.add_argument('source', help='源目录路径')
    parser.add_argument('target', help='目标目录路径')
    parser.add_argument('--mode', choices=['one-way', 'two-way'], 
                       default='one-way', help='同步模式')
    parser.add_argument('--show-hidden', action='store_true', 
                       help='包含隐藏文件')
    parser.add_argument('--dry-run', action='store_true', 
                       help='只显示将要进行的操作,不实际执行')
    args = parser.parse_args()
    # 创建同步器并执行
    syncer = FileSync(
        source=args.source,
        target=args.target,
        mode=args.mode,
        ignore_hidden=not args.show_hidden
    )
    try:
        syncer.run()
    except Exception as e:
        print(f"同步失败: {e}")
        exit(1)
if __name__ == "__main__":
    main()

Bash 版本(Linux/macOS)

#!/bin/bash
# sync.sh - 文件同步脚本
# 颜色定义
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
NC='\033[0m'
# 配置
SOURCE_DIR="${1:-./source}"  # 源目录
TARGET_DIR="${2:-./target}"  # 目标目录
MODE="${3:-one-way}"        # 同步模式:one-way 或 two-way
# 检查目录
check_dirs() {
    if [ ! -d "$SOURCE_DIR" ]; then
        echo -e "${RED}错误: 源目录不存在: $SOURCE_DIR${NC}"
        exit 1
    fi
    if [ ! -d "$TARGET_DIR" ]; then
        echo "创建目标目录: $TARGET_DIR"
        mkdir -p "$TARGET_DIR"
    fi
}
# 同步单个文件
sync_file() {
    local src_file="$1"
    local dst_file="$2"
    if [ ! -f "$src_file" ]; then
        return
    fi
    # 如果目标文件不存在,直接复制
    if [ ! -f "$dst_file" ]; then
        cp -p "$src_file" "$dst_file"
        echo -e "${GREEN}复制: $(basename "$src_file")${NC}"
        return
    fi
    # 比较文件是否相同
    if ! cmp -s "$src_file" "$dst_file"; then
        cp -p "$src_file" "$dst_file"
        echo -e "${YELLOW}更新: $(basename "$src_file")${NC}"
    fi
}
# 单向同步
sync_one_way() {
    echo "开始单向同步: $SOURCE_DIR -> $TARGET_DIR"
    # 同步文件
    find "$SOURCE_DIR" -type f | while read -r src_file; do
        # 获取相对路径
        rel_path="${src_file#$SOURCE_DIR/}"
        dst_file="$TARGET_DIR/$rel_path"
        # 确保目标目录存在
        mkdir -p "$(dirname "$dst_file")"
        sync_file "$src_file" "$dst_file"
    done
    # 删除目标中多余的文件
    find "$TARGET_DIR" -type f | while read -r dst_file; do
        rel_path="${dst_file#$TARGET_DIR/}"
        src_file="$SOURCE_DIR/$rel_path"
        if [ ! -f "$src_file" ]; then
            rm "$dst_file"
            echo -e "${RED}删除: ${rel_path}${NC}"
        fi
    done
}
# 双向同步
sync_two_way() {
    echo "开始双向同步: $SOURCE_DIR <-> $TARGET_DIR"
    # 从源到目标
    find "$SOURCE_DIR" -type f | while read -r src_file; do
        rel_path="${src_file#$SOURCE_DIR/}"
        dst_file="$TARGET_DIR/$rel_path"
        mkdir -p "$(dirname "$dst_file")"
        sync_file "$src_file" "$dst_file"
    done
    # 从目标到源
    find "$TARGET_DIR" -type f | while read -r dst_file; do
        rel_path="${dst_file#$TARGET_DIR/}"
        src_file="$SOURCE_DIR/$rel_path"
        if [ -f "$src_file" ]; then
            # 比较修改时间,保留最新的
            if [ "$dst_file" -nt "$src_file" ]; then
                cp -p "$dst_file" "$src_file"
                echo -e "${YELLOW}更新(反向): $rel_path${NC}"
            fi
        else
            # 目标文件在源中不存在,复制过去
            mkdir -p "$(dirname "$src_file")"
            cp -p "$dst_file" "$src_file"
            echo -e "${GREEN}复制(反向): $rel_path${NC}"
        fi
    done
}
# 主函数
main() {
    check_dirs
    case "$MODE" in
        "one-way")
            sync_one_way
            ;;
        "two-way")
            sync_two_way
            ;;
        *)
            echo -e "${RED}错误: 无效的同步模式: $MODE${NC}"
            echo "可用模式: one-way, two-way"
            exit 1
            ;;
    esac
    echo -e "${GREEN}同步完成!${NC}"
}
main

实时监控版(基于Python watchdog)

#!/usr/bin/env python3
"""
实时文件同步监控脚本
"""
import time
import shutil
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class SyncHandler(FileSystemEventHandler):
    def __init__(self, source, target):
        self.source = Path(source)
        self.target = Path(target)
    def get_target_path(self, src_path):
        """获取相对于源目录的目标路径"""
        src_path = Path(src_path)
        rel_path = src_path.relative_to(self.source)
        return self.target / rel_path
    def on_modified(self, event):
        if not event.is_directory:
            src_file = Path(event.src_path)
            dst_file = self.get_target_path(src_file)
            dst_file.parent.mkdir(parents=True, exist_ok=True)
            try:
                shutil.copy2(src_file, dst_file)
                print(f"已同步修改: {event.src_path}")
            except Exception as e:
                print(f"同步失败: {e}")
    def on_created(self, event):
        if not event.is_directory:
            src_file = Path(event.src_path)
            dst_file = self.get_target_path(src_file)
            dst_file.parent.mkdir(parents=True, exist_ok=True)
            try:
                shutil.copy2(src_file, dst_file)
                print(f"已同步新增: {event.src_path}")
            except Exception as e:
                print(f"同步失败: {e}")
    def on_deleted(self, event):
        if not event.is_directory:
            src_file = Path(event.src_path)
            dst_file = self.get_target_path(src_file)
            if dst_file.exists():
                try:
                    dst_file.unlink()
                    print(f"已同步删除: {event.src_path}")
                except Exception as e:
                    print(f"删除失败: {e}")
def main():
    source = input("请输入源目录路径: ")
    target = input("请输入目标目录路径: ")
    source_path = Path(source)
    if not source_path.exists():
        print("源目录不存在!")
        return
    Path(target).mkdir(parents=True, exist_ok=True)
    event_handler = SyncHandler(source, target)
    observer = Observer()
    observer.schedule(event_handler, source, recursive=True)
    observer.start()
    print(f"开始实时监控: {source} -> {target}")
    print("按 Ctrl+C 停止监控")
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
        print("\n停止监控")
    observer.join()
if __name__ == "__main__":
    main()

使用说明

  1. Python版本
    # 安装依赖
    pip install watchdog

基本用法

python file_sync.py /path/to/source /path/to/target

双向同步

python file_sync.py /path/to/source /path/to/target --mode two-way

包含隐藏文件

python file_sync.py /path/to/source /path/to/target --show-hidden


2. **Bash版本**:
```bash
# 赋予执行权限
chmod +x sync.sh
# 基本用法
./sync.sh /path/to/source /path/to/target
# 双向同步
./sync.sh /path/to/source /path/to/target two-way

功能特性

  • ✅ 单向/双向同步
  • ✅ 文件校验(MD5)
  • ✅ 自动创建目录
  • ✅ 忽略隐藏文件
  • ✅ 同步日志记录
  • ✅ 错误处理
  • ✅ 实时监控(可选)

需要我帮你调整特定功能吗?比如增加排除规则、压缩备份或Web界面等。

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