A simple script for adding articles to wallabag

.wallabag-adder.json

{
    "url": "https://app.wallabag.it",
    "username": "",
    "password": "",
    "client_id": "",
    "client_secret": ""
}

README.md

wallabag-add.py

A simple python script for adding articles to your wallabag account. I've been using this script for a couple of years already, so I post it as is. The only extra dependency is the requests library.

How to use

• Put wallabag-add.py wherever you like.
• Put the config (.wallabag-adder.json) in your home directory.
• Populate the config (see details below).
• Run wallabag-add.py 'https://example.com/some-url' to add an article.

The config

wallabag API authentication is a bit convoluted, so here's the gist.

If you just want things to work, put username, password, client_id and client_secret in the config. In this case the script will acquire the refresh_token and access_token and save it in the config. It will also be able to get new tokens when the old ones expire.

If you don't want to put your password in the config, but want to use the script regularly, then you may try putting in only client_id, client_secret and a valid refresh_token. In this case the script will be able to get access_token (and a new refresh_token), but it will stop working if the refresh_token gets expired (I don't know how long these live).

In an extreme case you may put in only access_token, but these last only for an hour (and you don't really need the whole script anyway).

And of course you need to set url pointing to your wallabag instance.

Problems

If things go wrong the script appends the url to ~/.wallabag-adder-failed-urls.txt.

License

GNU GPLv3

wallabag-add.py

#!/usr/bin/env python

import json
import requests
import logging
import sys
import os


def main(args):
    logger = logging.getLogger(__name__)
    logger.addHandler(logging.StreamHandler())
    # logger.setLevel(logging.INFO)
    logger.setLevel(logging.DEBUG)

    logger.debug('args: {}'.format(args[1:]))

    url_to_add = args[1]

    config = load_config(logger)
    if config is None:
        fail(logger, url_to_add)

    config_changed = False

    response = None
    if can_make_add_request(logger, config):
        response, get_token = add_url(logger, config, url_to_add)
    else:
        logger.debug('Not enough parameters to post a URL')
        get_token = True

    if get_token:
        logger.debug('Need to update token')

        retry = False

        if can_make_refresh_token_request(logger, config) and update_token(logger, config, refresh=True):
            logger.info('Token refreshed')
            config_changed = True
            retry = True
            get_token = False

        if get_token and can_make_get_token_request(logger, config) and update_token(logger, config):
            logger.info('Got new token')
            config_changed = True
            retry = True

        if retry:
            response, _ = add_url(logger, config, url_to_add)

    if config_changed:
        store_config(logger, config)

    if not response:
        fail(logger, url_to_add)


def load_config(logger):
    try:
        with open(get_config_path()) as f:
            return json.load(f)
    except Exception:
        logger.error('Failed to load configuration file', exc_info=True)
        return None


def store_config(logger, config):
    try:
        with open(get_config_path(), 'w') as f:
            json.dump(config, f, indent=4)
    except Exception:
        logger.error('Failed to store configuration file', exc_info=True)


def fail(logger, url):
    logger.warning('Exiting')

    if url:
        logger.info('Saving url to file')
        try:
            with open(get_emergency_file_path(), 'a') as f:
                f.write(url + os.linesep)
        except Exception:
            logger.error('Error while saving a URL to file')

    exit(1)


def get_config_path():
    return os.path.join(os.path.expanduser('~'), '.wallabag-adder.json')


def get_emergency_file_path():
    return os.path.join(os.path.expanduser('~'), '.wallabag-adder-failed-urls.txt')


def can_make_add_request(logger, config):
    if not config.get('access_token'):
        return False

    return True


def can_make_refresh_token_request(logger, config):
    if check_value_by_key(logger, config, 'client_id'):
        return False
    if check_value_by_key(logger, config, 'client_secret'):
        return False
    if check_value_by_key(logger, config, 'refresh_token'):
        return False

    return True


def can_make_get_token_request(logger, config):
    if check_value_by_key(logger, config, 'client_id'):
        return False
    if check_value_by_key(logger, config, 'client_secret'):
        return False
    if check_value_by_key(logger, config, 'username'):
        return False
    if check_value_by_key(logger, config, 'password'):
        return False

    return True


def check_value_by_key(logger, config, key):
    if not config.get(key):
        logger.info('No {} in config'.format(key))
        return True


def update_token(logger, config, refresh=False):
    data = {
        'client_id': config['client_id'],
        'client_secret': config['client_secret'],
    }
    if refresh:
        data['grant_type'] = 'refresh_token'
        data['refresh_token'] = config['refresh_token']
    else:
        data['grant_type'] = 'password'
        data['username'] = config['username'],
        data['password'] = config['password']

    try:
        r = requests.post('{}/oauth/v2/token'.format(config['url']), data=data)
    except Exception:
        logger.error('Error while getting token', exc_info=True)
        return False

    if not r.ok:
        return False

    r = r.json()

    config['access_token'] = r['access_token']
    config['refresh_token'] = r['refresh_token']

    return True


def add_url(logger, config, url):
    headers = {
        'Authorization': 'Bearer {}'.format(config['access_token'])
    }
    data = {
        'url': url
    }

    try:
        r = requests.post('{}/api/entries.json'.format(config['url']), headers=headers, data=data)
    except Exception:
        logger.error('Error adding url', exc_info=True)
        return None, False

    logger.debug('Response: {}, {}'.format(r.status_code, r.text))
    if not r.ok:
        # TODO: check auth error handling
        return None, r.status_code == 401

    return r.json(), False


if __name__ == '__main__':
    main(sys.argv)