120 lines
3.0 KiB
Python
120 lines
3.0 KiB
Python
|
from pathlib import Path
|
||
|
|
import os
|
||
|
|
|
||
|
|
|
||
|
|
class Spec:
|
||
|
|
def __init__(self, name, destination, limit, title, events=None):
|
||
|
|
self.name = name
|
||
|
|
self.destination = Path(destination)
|
||
|
|
self.limit = limit
|
||
|
|
self.title = title
|
||
|
|
self.events = ["success"] if events is None else events
|
||
|
|
|
||
|
|
def to_dict(self):
|
||
|
|
return {
|
||
|
|
"name": self.name,
|
||
|
|
"destination": str(self.destination),
|
||
|
|
"limit": self.limit,
|
||
|
|
"notify": {
|
||
|
|
"title": self.title,
|
||
|
|
"events": self.events
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
def backup(self):
|
||
|
|
raise NotImplementedError()
|
||
|
|
|
||
|
|
def notify(status_code):
|
||
|
|
if status_code:
|
||
|
|
if "failure" not in self.events:
|
||
|
|
return
|
||
|
|
|
||
|
|
message = "backup for {} failed.".format(self.name)
|
||
|
|
|
||
|
|
else:
|
||
|
|
if "success" not in self.events:
|
||
|
|
return
|
||
|
|
|
||
|
|
message = "backup for {} succeeded.".format(self.name)
|
||
|
|
|
||
|
|
# Read API key from env vars
|
||
|
|
try:
|
||
|
|
key = os.environ["IFTTT_API_KEY"]
|
||
|
|
|
||
|
|
# Don't send notification if there's not API key defined
|
||
|
|
except KeyError:
|
||
|
|
return
|
||
|
|
|
||
|
|
url = "https://maker.ifttt.com/trigger/{}/with/key/{}".format(
|
||
|
|
"phone_notifications",
|
||
|
|
key
|
||
|
|
)
|
||
|
|
|
||
|
|
data = {
|
||
|
|
"value1": self.title,
|
||
|
|
"value2": message
|
||
|
|
}
|
||
|
|
|
||
|
|
requests.post(url, data=data)
|
||
|
|
|
||
|
|
def get_filename(self):
|
||
|
|
return '{}_{}.tar.gz'.format(
|
||
|
|
self.name,
|
||
|
|
datetime.now().strftime('%Y-%m-%d_%H-%M-%S')
|
||
|
|
)
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def from_dict(name, data) -> "Specification":
|
||
|
|
if data.get("volume", False):
|
||
|
|
return VolumeSpec.from_dict(name, data)
|
||
|
|
|
||
|
|
return DirSpec.from_dict(name, data)
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def from_file(path: str):
|
||
|
|
with open(path, 'r') as yaml_file:
|
||
|
|
data = yaml.load(yaml_file, Loader=yaml.Loader)
|
||
|
|
|
||
|
|
return [Spec.from_dict(name, info)
|
||
|
|
for name, info in data["specs"].items()]
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
class DirSpec(Spec):
|
||
|
|
def __init__(self, name, source, destination, limit, title, events=None):
|
||
|
|
super().__init__(name, destination, limit, title, events)
|
||
|
|
|
||
|
|
self.source = Path(source)
|
||
|
|
|
||
|
|
def backup(self):
|
||
|
|
tarballs = self.destination.glob('*.tar.gz')
|
||
|
|
|
||
|
|
# Remove redundant tarballs
|
||
|
|
if len(tarballs) >= self.limit:
|
||
|
|
for path in tarballs[self.limit - 1:]:
|
||
|
|
path.unlink()
|
||
|
|
|
||
|
|
# Create new tarball
|
||
|
|
status_code = os.system(
|
||
|
|
"tar -C '{}' -czf '{}' -- .".format(
|
||
|
|
self.source,
|
||
|
|
self.destination / self.get_filename()
|
||
|
|
)
|
||
|
|
)
|
||
|
|
|
||
|
|
self.notify(status_code)
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def from_dict(name, data):
|
||
|
|
return DirSpec(
|
||
|
|
name,
|
||
|
|
data["source"],
|
||
|
|
data["destination"],
|
||
|
|
data["limit"],
|
||
|
|
data["notify"]["title"],
|
||
|
|
data["notify"]["events"]
|
||
|
|
)
|
||
|
|
|
||
|
|
class VolumeSpec(Spec):
|
||
|
|
pass
|