nvpy-db-utils.py

#!/usr/bin/env python3
# Usage:
#   # Validate local nvPY database.
#   ./nvpy-db-utils.py validate
#
#   # Clear local and remote database.
#   # WARNING: ALL NOTES WILL BE DELETED WITHOUT ANY PROMPT.
#   ./nvpy-db-utils.py delete --remote --all
#   ./nvpy-db-utils.py delete --local --all
#
#   # Download all notes to local.
#   ./nvpy-db-utils.py update --direction download --all
#
#   # Use custom configuration.
#   ./nvpy-db-utils.py --cfg ~/nvpy-backup.cfg update --direction download --all

import json
import traceback
import time
import pathlib
import argparse
import typing

import simplenote  # type:ignore

from nvpy import nvpy


class APIError(Exception):
    pass


class LocalDB:

    def __init__(self, db_path):
        self.db_path = pathlib.Path(db_path)

    def target(self, keys: typing.Iterable[str], is_all: bool) -> typing.Iterable[str]:
        files = self.target_files(keys, is_all=is_all)
        return (self.key_from_path(p) for p in files)

    def target_files(self, keys: typing.Iterable[str], is_all: bool) -> typing.Iterable[pathlib.Path]:
        if is_all:
            # The keys parameter ignores.
            return self.db_path.glob('*.json')
        else:
            return (self.path_from_key(k) for k in keys)

    def key_from_path(self, path: pathlib.Path) -> str:
        return path.with_suffix('').name

    def path_from_key(self, key) -> pathlib.Path:
        return self.db_path / f'{key}.json'

    def update(self, key, note):
        data = json.dumps(note)
        self.path_from_key(key).write_text(data)


class RemoteDB:

    def __init__(self, user, password):
        self.sn = simplenote.Simplenote(user, password)
        self.sn.get_token()

    def targets(self, keys: typing.Iterable[str], is_all: bool) -> typing.Iterable[str]:
        if is_all:
            # The keys parameter ignores.
            notes, status = self.sn.get_note_list(data=False)
            if status != 0:
                # error
                raise APIError(notes)
            keys = (n['key'] for n in notes)
        return keys

    def delete(self, key):
        print('delete', key)
        res, status = self.sn.delete_note(key)
        if status != 0:
            raise APIError(res)

    def get(self, key) -> dict:
        print('get', key)
        note, status = self.sn.get_note(key)
        if status != 0:
            raise APIError(note)
        return note

    def update(self, key: str, note: dict):
        print('update', key)
        note, status = self.sn.update_note(note)
        if status != 0:
            raise APIError(note)


class ValidateCmd:

    def run(self, args, config):
        files = LocalDB(config.db_path).target_files(args.keys, is_all=not args.keys)
        now = time.time()
        is_valid = True
        for file in files:
            with open(file) as f:
                obj = json.load(f)
            try:
                # See https://simplenotepy.readthedocs.io/en/latest/api.html#simperium-api-note-object

                # Optional field for simplenote.
                # Only if we have synced with simplenote once or more times, those fields may exist.
                # TODO: New note format does not require the key field. Please consider removing it.
                if 'key' in obj:
                    assert type(obj['key']) == str
                if 'deleted' in obj:
                    assert type(obj['deleted']) in [bool, int]
                if 'version' in obj:
                    assert type(obj['version']) == int
                if 'systemtags' in obj:
                    assert type(obj['systemtags']) == list
                if 'sharedURL' in obj:
                    assert type(obj['sharedURL']) == str
                if 'publishURL' in obj:
                    assert type(obj['publishURL']) == str

                # Required fields for simplenote.
                assert 'modifydate' in obj and type(obj['modifydate']) in [float, int] and obj['modifydate'] <= now
                assert 'createdate' in obj and type(obj['createdate']) in [float, int] and obj['createdate'] <= now
                assert 'tags' in obj and type(obj['tags']) == list
                assert 'content' in obj and type(obj['content']) == str

                # Optional field for nvpy.
                if 'savedate' in obj:
                    assert type(obj['savedate'] in [float, int])
                    assert float(obj['savedate']) <= now
                # Required field for nvpy.
                assert 'syncdate' in obj and type(obj['syncdate'] in [float, int]) and obj['createdate'] <= now
            except:
                print('{}  Invalid'.format(file))
                print(obj)
                print(traceback.format_exc())
                print('')
                is_valid = False
            else:
                print('{}  OK'.format(file))

        if is_valid:
            print('Done.  All notes are valid!')
            return 0
        else:
            print('Done.  Some notes are broken :-(')
            print('See above log for details.')
            return 1


class DeleteCmd:

    def run(self, args, config):
        if args.local:
            local = LocalDB(config.db_path)
            for t in local.target_files(args.keys, is_all=args.all):
                t.unlink(missing_ok=True)
        if args.remote:
            r = RemoteDB(config.sn_username, config.sn_password)
            for t in r.targets(args.keys, is_all=args.all):
                r.delete(t)


class UpdateCmd:

    def run(self, args, config):
        remote = RemoteDB(config.sn_username, config.sn_password)
        local = LocalDB(config.db_path)
        if args.direction == 'download':
            for key in remote.targets(args.keys, is_all=args.all):
                note = remote.get(key)
                local.update(key, note)
        elif args.direction == 'upload':
            for key in local.target(args.keys, is_all=args.all):
                data = local.path_from_key(key).read_text()
                note = json.loads(data)
                remote.update(key, note)
        else:
            raise RuntimeError('bug: invalid direction', args.direction)


def parse_cmd_line_args():
    p = argparse.ArgumentParser()
    p.add_argument('--cfg', '-c', dest='cfg', type=pathlib.Path, metavar='nvpy.cfg', help='path to config file')
    sp = p.add_subparsers(required=True)

    validate = sp.add_parser('validate')
    validate.set_defaults(clazz=ValidateCmd)
    validate.add_argument('keys', nargs='*')

    delete = sp.add_parser('delete')
    delete.set_defaults(clazz=DeleteCmd)
    delete.add_argument('--all', '-a', action='store_true')
    delete.add_argument('--local', '-l', action='store_true')
    delete.add_argument('--remote', '-r', action='store_true')
    delete.add_argument('keys', nargs='*')

    update = sp.add_parser('update')
    update.set_defaults(clazz=UpdateCmd)
    update.add_argument('--all', '-a', action='store_true')
    update.add_argument('--direction', '-d', choices=('download', 'upload'), required=True)
    update.add_argument('keys', nargs='*')

    return p.parse_args()


def main():
    args = parse_cmd_line_args()
    cfg_files = None
    if args.cfg:
        cfg_files = [args.cfg]
    config = nvpy.Config(nvpy.get_appdir(), cfg_files)
    return args.clazz().run(args, config)


if __name__ == '__main__':
    exit(main())