-
Notifications
You must be signed in to change notification settings - Fork 0
/
sync_file.py
117 lines (101 loc) · 3.33 KB
/
sync_file.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
# -*- encoding: utf-8 -*-
"""
@File : sync_file.py
@Time : 2024/03/12 15:32:24
@Author : lihao57
@Version : 1.0
@Contact : lihao57@baidu.com
"""
import os
import json
import time
import shutil
import logging
import argparse
import schedule
def sync_file(config_file: str):
"""
synchronize files
Args:
config_file (str): config file
Return:
None
"""
try:
with open(config_file, "r") as f:
configs = json.load(f)
logging.info("configs:")
logging.info(configs)
assert len(configs), "configs is empty"
for item in configs:
if not item.get("sync", False):
continue
src_path = item["src_path"]
dst_path = item["dst_path"]
excludes = item.get("exclude", [])
os.makedirs(dst_path, exist_ok=True)
dst_folder = os.path.join(dst_path, os.path.basename(src_path))
backup_folder = dst_folder + ".bak"
if os.path.exists(dst_folder):
if os.path.exists(backup_folder):
logging.info(
"remove backup folder: {}".format(backup_folder)
)
shutil.rmtree(backup_folder)
shutil.move(dst_folder, backup_folder)
logging.info("move {} to {}".format(dst_folder, backup_folder))
cmd = "rsync -av -e ssh "
for exclude in excludes:
cmd += "--exclude={} ".format(exclude)
cmd += "{} {}".format(src_path, dst_path)
logging.info("cmd: {}".format(cmd))
error_code = os.system(cmd)
logging.info("error_code: {}".format(error_code))
if error_code != 0:
if os.path.exists(backup_folder):
shutil.move(backup_folder, dst_folder, )
logging.info("restore move {} to {}".format(dst_folder, backup_folder))
else:
for exclude in excludes:
src_sub_folder = os.path.join(backup_folder, exclude)
dst_sub_folder = os.path.join(dst_folder, exclude)
if os.path.exists(src_sub_folder):
shutil.move(src_sub_folder, dst_sub_folder)
logging.info(
"move {} to {}".format(src_sub_folder, dst_sub_folder)
)
except Exception as e:
logging.error(e)
if __name__ == "__main__":
# set base logging config
fmt = "[%(asctime)s - %(levelname)s - %(filename)s:%(lineno)s] %(message)s"
logging.basicConfig(format=fmt, level=logging.INFO)
# args
parser = argparse.ArgumentParser()
parser.add_argument(
"-c",
"--config_file",
type=str,
help="config file",
default="config.json",
)
parser.add_argument(
"-t",
"--sync_time",
type=str,
help="sync time",
default="01:00",
)
opt = parser.parse_args()
print(opt)
t1 = time.time()
config_file = opt.config_file
sync_time = opt.sync_time
schedule.every().day.at(sync_time).do(sync_file, config_file)
try:
while True:
schedule.run_pending()
except KeyboardInterrupt:
pass
t2 = time.time()
logging.info("time: {}".format(t2 - t1))