写在前面:这是我对类 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 }'"))