A BitBar plugin that runs duplicacy backup hourly

duplicacy.1s.py

#!/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()