用 Python 实现一个可自动识别的文件夹单向同步功能.

写在前面:这是我对类 unison 这样猝发式同步工具的一种复现尝试;因为我平时主要就是在一个文件夹内进行操作,所以算法更为简单,并没有实现针对单个文件进行处理的双向同步。 我如今已经使用了更为高效易用的 FreeFileSync 去进行文件同步和备份操作,该程序已经归档。

为什么我想写这个算法

在日常生活以及学习中,文件夹同步这一操作自然是必不可少的,一个方便使用的文件同步软件可以很好的对这种进行操作。通过我以前的文章可以了解到,我之前一直都在使用 Unison 来进行本地以及 WSL 不同的文件夹间的同步。而后来由于自己的问题,更换电脑后,由于 Unison 麻烦的配置以及自己对于文件夹同步的想法,我还是想尝试自己写一个文件夹同步的算法。

文件夹同步的方式有两种,一种是 rsync 那样的单向同步,一种是 Unison 一样的双向同步(Unison 也是基于 rsync 的);当然我更需要的是双向的,所以这就不由得涉及到了很多的同步问题,最后我还是改成实现一个自动识别主文件夹的单向同步,这同样也可以达到双向同步的效果。

算法细节

算法很简单,总共分为两个层面。首先是同步的工具,我直接使用了 rsync 来进行传输的操作,然后同步两个或多个文件夹即可;第二个部分就是判断哪个文件夹为同步的主文件夹,这一部分使用了文件的时间戳作为指纹保存,在同步之后,如果有与保存的指纹不同,则可判定其做了修改,即新文件夹。

时间戳生成算法

def timeset(path, data: dict, prefix: str):
    source = os.getcwd()
    path_list = os.listdir(path)
    os.chdir(path)
    for n in path_list:
        if os.path.isfile(n):
            if n != 'syncing.json':
                data[os.path.join(prefix, n)] = int(os.path.getmtime(n))
        else:
            timeset(n, data, os.path.join(prefix, n))
    os.chdir(source)
    return data

而如果在数据中没有指纹,则两个文件夹为新文件夹,此时直接取时间戳最大的文件夹。在判断完哪一个文件夹为同步的主文件夹之后,直接进行 rsync 同步即可。

算法源码

当然,为了保证该脚本的可用性,文件地址判定以及信息确认之类的都采用了严谨的写法,从而保证不会出现错误。

def sync_folder_local(*args, main=None):

    # WARNING: This function only works correctly with single-folder editing
    # It would have a bug occured when multi folders were edited

    database = load_json(f'{ os.path.split(os.path.realpath(sys.argv[0]))[0] }/syncing.json')
    token = str(args)

    # if main, then use main mode
    if main:
        temp = timeset(main, {}, '')
        database[token] = timeset(main, {}, '')
        save_json(database, f'{ os.path.split(os.path.realpath(sys.argv[0]))[0] }/syncing.json')

        for directory in args:
            if directory != main:
                os.system(f"sudo rsync -ra --delete --exclude 'syncing.json' '{ main }/' '{ directory }'")
                print((f"| Info |  Syncing '{ main }' to '{ directory }'"))

    # not main mode
    else:
        main = None
        
        if token in database:  # token was recorded in database
            for directory in args:
                if timeset(directory, {}, '') != database[token]:
                    main = directory
                    break

            if main:
                input(f"| Info |  The main directory detected is { main }, proceed? [Enter]")
                temp = timeset(main, {}, '')
                database[token] = timeset(main, {}, '')
                save_json(database, f'{ os.path.split(os.path.realpath(sys.argv[0]))[0] }/syncing.json')

                for directory in args:
                    if directory != main:
                        os.system(f"sudo rsync -ra --delete --exclude 'syncing.json' '{ main }/' '{ directory }'")
                        print(f"| Info |  Syncing '{ main }' to '{ directory }'")
            else:
                print("| Info |  The directories were synced.")

        else:  # token was not recorded in database
            time_token = float('-inf')
            for directory in args:
                time = sorted(timeset(directory, {}, '').items(), key=lambda x : x[1])[-1][1]
                if time > time_token:
                    time_token = time
                    main = directory

            input(f"| Info |  The main directory detected is { main }, proceed? [Enter]")

            temp = timeset(main, {}, '')
            database[token] = timeset(main, {}, '')
            save_json(database, f'{ os.path.split(os.path.realpath(sys.argv[0]))[0] }/syncing.json')
            
            for directory in args:
                if directory != main:
                    os.system(f"sudo rsync -ra --delete --exclude 'syncing.json' '{ main }/' '{ directory }'")
                    print((f"| Info |  Syncing '{ main }/' to '{ directory }'"))