Last active
January 13, 2020 08:32
-
-
Save tevino/3ae37b0aa1f70979422f928186220870 to your computer and use it in GitHub Desktop.
A BitBar plugin that runs duplicacy backup hourly
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python3 | |
# <bitbar.title>Duplicacy Scheduler</bitbar.title> | |
# <bitbar.version>v1.0</bitbar.version> | |
# <bitbar.author>Tevin</bitbar.author> | |
# <bitbar.author.github>tevino</bitbar.author.github> | |
# <bitbar.desc>Run duplicacy backup every hour.</bitbar.desc> | |
# <bitbar.dependencies>duplicacy,python3</bitbar.dependencies> | |
import os | |
import json | |
import subprocess | |
from pathlib import Path | |
from datetime import datetime | |
PATH_DUPLICACY = Path('duplicacy').expanduser() | |
PATH_DUPLICACY_PREFERENCE = Path('~/.duplicacy/preferences').expanduser() | |
PATH_LOG_FILE = Path('~/.duplicacy/duplicacy.log').expanduser() | |
PATH_RET_FILE = Path('~/.duplicacy/duplicacy.ret').expanduser() | |
PID_FILE = '/tmp/duplicacy.pid' | |
FLAG_STOPPED = Path('tmp/FLAG_STOP_BACKUP') | |
FLAG_BACKUP_NOW = Path('/tmp/FLAG_BACKUP_NOW') | |
SUMMARY_START = 'Backup for ' | |
def is_alive() -> bool: | |
pid = -1 | |
try: | |
with open(PID_FILE) as f: | |
pid = int(f.read()) | |
except FileNotFoundError: | |
return False | |
if pid > 0: | |
try: | |
os.kill(pid, 0) | |
except OSError: | |
# No such process or operation not permitted | |
pass | |
else: | |
# process exists | |
return True | |
return False | |
REVISION_START = 'Last backup at revision ' | |
def get_progress() -> str: | |
percent = '0%' | |
revision = 1 | |
with open(PATH_LOG_FILE) as f: | |
for line in f: | |
if line.startswith(REVISION_START): | |
# [4] is the digit after REVISION_START | |
revision = int(line.split(' ')[4]) + 1 | |
elif line.endswith('%\n'): | |
percent = line.split(' ')[-1] | |
return '[%s] %s' % (revision, percent.strip()) | |
def render_summary() -> str: | |
summary = '' | |
try: | |
si = os.stat(PATH_RET_FILE) | |
mtime = datetime.fromtimestamp(si.st_mtime).strftime("%Y-%m-%d %H:%M") | |
summary += 'Last Completed Backup: %s | color=green \n' % mtime | |
with open(PATH_LOG_FILE) as f: | |
log = f.read() | |
except FileNotFoundError: | |
log = '' | |
lines = log.split('\n') | |
for i, line in enumerate(lines): | |
if line.startswith(SUMMARY_START): | |
summary += '\n'.join(lines[i:]) | |
break | |
return summary | |
COLORED_KEYWORDS = { | |
'yellow': ('error', 'warning'), | |
'green': ('backup for',), | |
} | |
def render_text(s: str) -> str: | |
return "\n".join([_render_line(line) for line in s.split("\n")]) | |
def _render_line(line: str) -> str: | |
for color, keywords in COLORED_KEYWORDS.items(): | |
for k in keywords: | |
if k in line.lower(): | |
line += ' | color=%s ' % color | |
break | |
return line | |
def get_ret_code() -> int: | |
try: | |
with open(PATH_RET_FILE) as f: | |
return int(f.read()) | |
except FileNotFoundError: | |
return -1 | |
def render_title() -> None: | |
ret_code = get_ret_code() | |
if is_alive(): | |
return ":open_file_folder:%s | color=gray" % get_progress() | |
elif not Path(PATH_RET_FILE).exists(): | |
return "Waiting for First Backup" | |
elif ret_code != 0: | |
return ":warning: Last Backup Error(%d) | color=yellow" % ret_code | |
else: | |
# success | |
return ":floppy_disk: | color=green" | |
def render_menu() -> str: | |
menus = [] | |
menus.append(":wavy_dash: Backup Now | " | |
"bash=/usr/bin/touch param1=%s " | |
"refresh=true color=green terminal=false" % FLAG_BACKUP_NOW) | |
if FLAG_STOPPED.exists(): | |
menus.append(":heavy_check_mark: Start Scheduling | " | |
"bash=/bin/rm param1=-f param2=%s " | |
"refresh=true color=green terminal=false" % FLAG_STOPPED) | |
else: | |
menus.append(":heavy_multiplication_x: Stop Scheduling | " | |
"bash=/usr/bin/touch param1=%s " | |
"refresh=true color=red terminal=false" % FLAG_STOPPED) | |
menus.append(":information_source: Open Log File | " | |
"href=file://%s" % PATH_LOG_FILE) | |
return "\n".join(menus) | |
def report_state() -> None: | |
print('\n'.join([ | |
render_title(), | |
"---", | |
render_menu(), | |
"---", | |
render_text(render_summary()), | |
])) | |
def is_time_to_backup() -> bool: | |
# backup every hour, at 42th minute | |
return datetime.now().minute == 42 | |
def is_stopped() -> bool: | |
return FLAG_STOPPED.exists() | |
def get_repository() -> str: | |
with open(PATH_DUPLICACY_PREFERENCE) as f: | |
pref = json.load(f) | |
return pref[0]['repository'] | |
return "" | |
MSG_SET_REPO = 'Please set "repository" in the preferences file of duplicacy' | |
ADDITIONAL_PATH = '/usr/local/bin' | |
def start_backup_if_necessary() -> None: | |
if is_stopped(): | |
return | |
if FLAG_BACKUP_NOW.exists() or is_time_to_backup(): | |
try: | |
os.remove(FLAG_BACKUP_NOW) | |
except FileNotFoundError: | |
pass | |
# TODO: add support for multiple repository | |
repo = get_repository() | |
if not repo: | |
print(MSG_SET_REPO) | |
return | |
env = os.environ.copy() | |
env['PATH'] = os.getenv("PATH") + ':' + ADDITIONAL_PATH | |
cmd = '%s backup -stats > "%s" 2>&1; echo $? > "%s"' \ | |
% (PATH_DUPLICACY, PATH_LOG_FILE, PATH_RET_FILE) | |
proc = subprocess.Popen(cmd, | |
stdin=subprocess.DEVNULL, | |
stdout=subprocess.DEVNULL, | |
stderr=subprocess.DEVNULL, | |
env=env, | |
cwd=repo, | |
shell=True, | |
start_new_session=True) | |
with open(PID_FILE, 'w+') as f: | |
f.write(str(proc.pid)) | |
if __name__ == '__main__': | |
if not is_alive(): | |
start_backup_if_necessary() | |
report_state() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment