This repository has been archived by the owner on Mar 21, 2023. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathbackuproll.py
executable file
·318 lines (267 loc) · 11.6 KB
/
backuproll.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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Backup roll script for one or more Minecraft servers
Usage:
backuproll [options] [<world>]
Options:
-h, --help Print this message and exit.
--all Apply the action to all configured worlds. This is the default.
--config=<config> Path to the config file [default: /opt/wurstmineberg/config/backuproll.json].
--version Print version info and exit.
--verbose Print things.
--simulate Don't do any destructive operation, implies --verbose
"""
import sys
import datetime
import os
import pathlib
import contextlib
import json
import subprocess
from docopt import docopt
__version__ = '0.1'
DEFAULT_CONFIG = {
"backupcommand": "/opt/wurstmineberg/bin/minecraft backup",
"backupfolder": "/opt/wurstmineberg/backup/worlds/",
"worlds": {
"testworld": {
"keep": {
"recent": 6,
"daily": 10,
"weekly": 4,
"monthly": 6
}
}
},
"pidfile": "/var/local/wurstmineberg/backuproll.pid"
}
class BackupFile:
def __init__(self, basedir, filename, prefix, suffix, dateformat):
self.basedir = basedir
self.filename = filename
self.prefix = prefix
self.suffix = suffix
self.dateformat = dateformat
self.path = os.path.join(basedir, filename)
@property
def datetime(self):
datestr = self.filename[len(self.prefix):-len(self.suffix)]
return datetime.datetime.strptime(datestr, self.dateformat)
def __repr__(self):
return "<Backup from date '{}'>".format(self.datetime, self.filename)
class BackupRoll:
def __init__(self, backupdir, prefix, suffix, dateformat, keepdict, simulate=False, verbose=False):
self.backupdir = backupdir
self.prefix = prefix
self.suffix = suffix
self.dateformat = dateformat
self.keepdict = keepdict
self.simulate = simulate
self.verbose = verbose
@property
def dailydir(self):
return os.path.join(self.backupdir, 'daily')
@property
def weeklydir(self):
return os.path.join(self.backupdir, 'weekly')
@property
def monthlydir(self):
return os.path.join(self.backupdir, 'monthly')
def sorted_backups(self, backups):
return sorted(backups, key=lambda b: b.datetime)
def select_promote_daily_backup(self, backups, date):
"""Selects the backup to promote as daily backup for the calendar day given
This selects the latest backup earlier than 13:00 if possible"""
backups = [ b for b in backups if b.datetime.date() == date ]
selected_backup = None
if len(backups) >= 1:
selected_backup = backups[0]
for backup in backups:
if backup.datetime.hour < 13:
selected_backup = backup
return selected_backup
def select_promote_weekly_backup(self, backups, date):
"""Selects the backup to promote as weekly backup for the first day of the calendar week
the given day is in"""
weeknumber = date.isocalendar()[1]
backups = [ b for b in backups if b.datetime.isocalendar()[1] == weeknumber ]
selected_backup = None
if len(backups) >= 1:
selected_backup = backups[0]
return selected_backup
def select_promote_monthly_backup(self, backups, date):
"""Selects the backup to promote as monthly backup for the first day of the month
the given day is in"""
month = date.month
backups = [ b for b in backups if b.datetime.month == month ]
selected_backup = None
if len(backups) >= 1:
selected_backup = backups[0]
return selected_backup
def should_promote_daily_backup(self, date):
now = datetime.datetime.utcnow()
if now.date() > date or now.hour >= 12:
# If it is already 13:00 or a later date a backup should be promoted if none exists
return not self.get_backup_daily_for_date(date) and self.keepdict['daily'] > 0
return False
def should_promote_weekly_backup(self, date):
return not self.get_backup_weekly_for_date(date) and self.keepdict['weekly'] > 0
def should_promote_monthly_backup(self, date):
return not self.get_backup_monthly_for_date(date) and self.keepdict['monthly'] > 0
def list_backups_to_delete(self):
recents = self.list_backups_recent()
daily = self.list_backups_daily()
weekly = self.list_backups_weekly()
monthly = self.list_backups_monthly()
recentkeep = self.keepdict['recent']
dailykeep = self.keepdict['daily']
weeklykeep = self.keepdict['weekly']
monthlykeep = self.keepdict['monthly']
return recents[:-recentkeep] + daily[:-dailykeep] + weekly[:-weeklykeep] + monthly[:-monthlykeep]
def promote_backup_to_dir(self, backup, directory):
if self.verbose:
print("Promoting {} to dir: {}".format(backup, directory))
if not self.simulate:
try:
os.makedirs(directory)
except FileExistsError:
pass
os.link(backup.path, os.path.join(directory, backup.filename))
def promote_backups(self):
date = datetime.datetime.utcnow().date()
if self.should_promote_daily_backup(date):
if self.verbose:
print("We should promote a daily backup")
backup_to_promote = self.select_promote_daily_backup(self.list_backups_recent(), date)
if backup_to_promote:
self.promote_backup_to_dir(backup_to_promote, self.dailydir)
elif self.verbose:
print("Can't find a daily backup to promote. Try later.")
if self.should_promote_weekly_backup(date):
if self.verbose:
print("We should promote a weekly backup")
backup_to_promote = self.select_promote_weekly_backup(self.list_backups_daily(), date)
if backup_to_promote:
self.promote_backup_to_dir(backup_to_promote, self.weeklydir)
elif self.verbose:
print("Can't find a weekly backup to promote. Try later.")
if self.should_promote_monthly_backup(date):
if self.verbose:
print("We should promote a monthly backup")
backup_to_promote = self.select_promote_monthly_backup(self.list_backups_daily(), date)
if backup_to_promote:
self.promote_backup_to_dir(backup_to_promote, self.monthlydir)
elif self.verbose:
print("Can't find a monthly backup to promote. Try later.")
def delete_backup(self, backup):
if self.verbose:
print("Deleting {}".format(backup))
if not self.simulate:
os.remove(backup.path)
def cleanup_backups(self):
to_delete = self.list_backups_to_delete()
for backup in to_delete:
self.delete_backup(backup)
def list_backups_from(self, folder):
try:
files = [ f for f in os.listdir(folder) if os.path.isfile(os.path.join(folder, f)) and
f.startswith(self.prefix) and f.endswith(self.suffix) ]
backups = [ BackupFile(folder, f, self.prefix, self.suffix, self.dateformat) for f in files ]
return self.sorted_backups(backups)
except FileNotFoundError:
return []
def list_backups_recent(self):
return self.list_backups_from(self.backupdir)
def list_backups_daily(self):
return self.list_backups_from(self.dailydir)
def list_backups_weekly(self):
return self.list_backups_from(self.weeklydir)
def list_backups_monthly(self):
return self.list_backups_from(self.monthlydir)
def get_backup_daily_for_date(self, date):
for backup in self.list_backups_daily():
if backup.datetime.date() == date:
return backup
def get_backup_weekly_for_date(self, date):
weeknumber = date.isocalendar()[1]
for backup in self.list_backups_weekly():
if backup.datetime.isocalendar()[1] == weeknumber:
return backup
def get_backup_monthly_for_date(self, date):
for backup in self.list_backups_monthly():
if backup.datetime.month == date.month:
return backup
def get_default_backuproll(world, simulate=False):
return BackupRoll('/opt/wurstmineberg/backup/{}'.format(world), '{}_'.format(world), '.tar.gz', '%Y-%m-%d_%Hh%M', None, simulate=simulate)
class BackupRunner:
def __init__(self, command, simulate=False, verbose=False):
self.command = command
self.simulate = simulate
self.verbose = verbose
def run_blocking(self):
if self.verbose:
print("Running command '{}'".format(self.command))
if not self.simulate:
out = None if self.verbose else subprocess.PIPE
return subprocess.run(self.command, stdout=out, stderr=out, shell=True, universal_newlines=True)
def do_backuproll(worlds, backupcommand, has_world_prefix=True, extension='tar.gz', dateformat='%Y-%m-%d_%Hh%M', simulate=False, verbose=False):
for world in worlds:
command = backupcommand + ' ' + world
runner = BackupRunner(command, simulate, verbose)
ret = runner.run_blocking()
if ret is not None and ret.returncode != 0:
print("Backup failed! Not running backuproll!", file=sys.stderr)
if ret.stdout is not None:
print('stdout:')
print(ret.stdout)
if ret.stderr is not None:
print('stderr:')
print(ret.stderr)
exit(1)
prefix = world + '_' if has_world_prefix else ''
keepdict = CONFIG['worlds'][world]['keep']
roll = BackupRoll(str(CONFIG['backupfolder'] / world), prefix, '.' + extension, dateformat, keepdict, simulate=simulate, verbose=verbose)
roll.promote_backups()
roll.cleanup_backups()
if __name__ == "__main__":
arguments = docopt(__doc__, version='Minecraft backup roll ' + __version__)
CONFIG_FILE = pathlib.Path(arguments['--config'])
CONFIG = DEFAULT_CONFIG.copy()
with contextlib.suppress(FileNotFoundError):
with CONFIG_FILE.open() as config_file:
CONFIG.update(json.load(config_file))
CONFIG['backupfolder'] = pathlib.Path(CONFIG['backupfolder'])
backupcommand = CONFIG['backupcommand']
selected_worlds = CONFIG['worlds'].keys()
if arguments['<world>'] and not arguments['--all']:
selected_worlds = [arguments['<world>']]
if len(selected_worlds) == 0:
print("No world selected and none found in the config file. Exiting.")
exit(1)
verbose = False
if arguments['--verbose']:
verbose = True
simulate = False
if arguments['--simulate']:
print("Simulating backuproll: No real action will be performed")
simulate = True
verbose = True
pid_filename = CONFIG['pidfile']
if os.path.isfile(pid_filename):
with open(pid_filename, 'r') as pidfile:
try:
pid = int(pidfile.read())
except ValueError:
pid = None
if pid:
try:
os.kill(pid, 0)
print("Another backuproll process is still running. Terminating.", file=sys.stderr)
exit(1)
except ProcessLookupError:
pass
mypid = os.getpid()
with open(pid_filename, "w+") as pidfile:
pidfile.write(str(mypid))
do_backuproll(selected_worlds, backupcommand, simulate=simulate, verbose=verbose)
os.remove(pid_filename)