Python command interpreter interface to NewsBlur API

newsblur-cl.py

#!/usr/bin/env python

"""Command interpreter interface to NewsBlur API.

Prerequisite: A checkout of the NewsBlur Python API library.

The upstream version is here:

    https://github.com/samuelclay/NewsBlur.git

But for the moment you need to use my fork, which contains an API bug
workaround and support for the move_feed_to_folder API call:

    https://github.com/ngoffee/NewsBlur.git

Quick start:

    Download this file to <somefolder>/newsblur-cl.py, then:

    $ cd <somefolder>
    $ git clone https://github.com/ngoffee/NewsBlur.git
    $ export PYTHONPATH=$PYTHONPATH:$PWD/NewsBlur
    $ ./newsblur-cl.py
"""

from cmd import Cmd
from collections import defaultdict
from getpass import getpass

import os
import os.path
import readline
import shlex
import sys

import api.newsblur as newsblur


class NewsblurCmd(Cmd):

    def __init__(self):
        Cmd.__init__(self)
        self._api          = newsblur.API()
        self._cached_feeds = None

    def _fetch_feeds(self):
        print "fetching feeds ..."
        self._cached_feeds = \
            self._api.feeds(include_favicons=False, flat=True)

    @property
    def _feeds(self):
        if not self._cached_feeds:
            self._fetch_feeds()
        return self._cached_feeds

    def do_login(self, line):
        configfile = os.path.expanduser('~/.newsblur-cl')
        if os.path.exists(configfile):
            f = open(configfile)
            username = f.next().strip()
            password = f.next().strip()
            f.close()
        else:
            username = raw_input('username: ')
            password = getpass('password: ')
            fd = os.open(configfile, os.O_WRONLY | os.O_CREAT, 0600)
            f  = os.fdopen(fd, 'w')
            print >> f, username
            print >> f, password
            f.close()
            print 'saved username and password in %s' % configfile
        print 'logging in as %s' % username
        print self._api.login(username, password)

    def do_raw_feeds(self, line):
        print self._feeds

    def do_feeds(self, line):
        for feed_id, feed in self._feeds['feeds'].iteritems():
            print '%-8s: %s (%s)' \
                % (feed_id, feed['feed_title'], feed['feed_link'])

    def _find_feed(self, text, multiple=False):
        # if text is already a feed_id, make sure it exists and return
        # it unchanged
        try:
            int(text)
            if not self._feeds['feeds'].has_key(text):
                print "no such feed_id: %s" % text
                return None
            return [text] if multiple else text
        except ValueError:
            pass
        # now look for feeds whose feed_link or feed_title match text
        results = []
        for feed_id, feed in self._feeds['feeds'].iteritems():
            if text.lower() in feed['feed_title'].lower() \
                    or text.lower() in feed['feed_link'].lower():
                results.append(feed_id)
        if multiple:
            return results
        elif not results:
            print "'%s': not found" % text
            return None
        elif len(results) > 1:
            print "'%s': not unique" % text
            return None
        else:
            return results[0]

    def do_find_feeds(self, line):
        """usage: find_feed <feed_id or search string>"""
        args = shlex.split(line)
        if len(args) != 1:
            self.do_help('find_feed')
            return
        for feed_id in self._find_feed(args[0], multiple=True):
            feed = self._feeds['feeds'][feed_id]
            print '%-8s: %s (%s)' \
                % (feed_id, feed['feed_title'], feed['feed_link'])

    def _find_folder(self, text, multiple=False):
        results = []
        for folder in self._feeds['flat_folders']:
            if not multiple and text.lower() == folder.lower():
                # exact match
                return [folder] if multiple else folder
            if text.lower() in folder.lower():
                results.append(folder)
        if multiple:
            return results
        elif not results:
            print "'%s': not found" % text
            return None
        elif len(results) > 1:
            print "'%s': not unique" % text
            return None
        else:
            return results[0]

    def do_find_folders(self, line):
        """usage: find_folder <search string>"""
        args = shlex.split(line)
        if len(args) != 1:
            self.do_help('find_folder')
            return
        for folder in self._find_folder(args[0], multiple=True):
            print "'%s'" % folder

    def do_raw_feed(self, line):
        """usage: raw_feed <feed_id or search string>

Show the raw JSON of the given feed."""
        args = shlex.split(line)
        if len(args) != 1:
            self.do_help('raw_feed')
            return
        feed_id = self._find_feed(args[0])
        if feed_id is None:
            return
        feed = self._feeds['feeds'][feed_id]
        print feed

    def do_feed(self, line):
        """usage: feed <feed_id or search string>

List details of the given feed."""
        args = shlex.split(line)
        if len(args) != 1:
            self.do_help('feed')
            return
        feed_id = self._find_feed(args[0])
        if feed_id is None:
            return
        feed = self._feeds['feeds'][feed_id]
        print '%-8s: %s (%s)' \
            % (feed_id, feed['feed_title'], feed['feed_link'])

    def do_folders(self, line):
        """list all folders"""
        for folder in self._feeds['flat_folders']:
            print "'%s'" % folder

    def do_raw_folder(self, line):
        """usage: raw_folder <name or search string>"""
        args = shlex.split(line)
        if len(args) != 1:
            self.do_help('folder')
            return
        folder = self._find_folder(args[0])
        if folder is None:
            return
        feed_ids = self._feeds['flat_folders'][folder]
        print feed_ids

    def do_folder(self, line):
        """usage: folder <name or search string>

List all feeds in the given folder."""
        args = shlex.split(line)
        if len(args) != 1:
            self.do_help('folder')
            return
        folder = self._find_folder(args[0])
        if folder is None:
            return
        feed_ids = self._feeds['flat_folders'][folder]
        feed_ids = map(str, feed_ids)
        print "'%s':" % folder
        for feed_id in feed_ids:
            feed    = self._feeds['feeds'][feed_id]
            print '    %-8s: %s (%s)' \
                % (feed_id, feed['feed_title'], feed['feed_link'])

    def _find_feed_folder(self, feed_id):
        if not self._feeds['feeds'].has_key(feed_id):
            print "no such feed_id: %s" % feed_id
            return
        for folder, feed_ids in self._feeds['flat_folders'].iteritems():
            feed_ids = map(str, feed_ids)
            if feed_id in feed_ids:
                return folder
        return ''

    def do_find_feed_folder(self, line):
        """find_feed_folder <feed_id or search string>"""
        args = shlex.split(line)
        if len(args) != 1:
            self.do_help('find_feed_folder')
            return
        feed_id = self._find_feed(args[0])
        if feed_id is None:
            return
        print "feed_id = %s" % repr(feed_id)
        folder = self._find_feed_folder(feed_id)
        if folder is None:
            return
        print "'%s'" % folder

    def _add_folder(self, folder):
        if self._feeds['flat_folders'].has_key(folder):
            print "folder already exists: '%s'" % folder
            return
        print self._api.add_folder(folder)

    def do_add_folder(self, line):
        """usage: add_folder <name>"""
        args = shlex.split(line)
        if len(args) != 1:
            self.do_help('add_folder')
            return
        folder = args[0]
        self._add_folder(folder)
        self._fetch_feeds()

    def _move_feed(self, feed_id, in_folder, to_folder, refetch=True):
        feed_title = self._feeds['feeds'][feed_id]['feed_title']
        feed_link  = self._feeds['feeds'][feed_id]['feed_link']
        print "%s: %s/%s (%s) -> %s/" \
            % (feed_id, in_folder, feed_title, feed_link, to_folder)
        self._api.move_feed_to_folder(feed_id, in_folder, to_folder)
        if refetch:
            self._fetch_feeds()

    def do_move_feed(self, line):
        """usage: move_feed <feed_id or search string> <folder or search string>"""
        args = shlex.split(line)
        if len(args) != 2:
            self.do_help('move_feed')
            return
        feed_id   = self._find_feed(args[0])
        to_folder = self._find_folder(args[1])
        if feed_id is None or to_folder is None:
            return
        in_folder = self._find_feed_folder(feed_id)
        self._move_feed(feed_id, in_folder, to_folder)

    def do_move_all(self, line):
        """usage: move_feed <folder_id or search string> <folder or search string>"""
        args = shlex.split(line)
        if len(args) != 2:
            self.do_help('move_all')
            return
        in_folder = self._find_folder(args[0])
        to_folder = self._find_folder(args[1])
        if in_folder is None or to_folder is None:
            return
        for feed_id in map(str, self._feeds['flat_folders'][in_folder]):
            self._move_feed(feed_id, in_folder, to_folder, refetch=False)
        self._fetch_feeds()

    def _get_orphans(self):
        for feed_id in self._feeds['feeds']:
            if self._find_feed_folder(feed_id) == '':
                yield feed_id, self._feeds['feeds'][feed_id]

    def do_orphans(self, line):
        for feed_id, feed in self._get_orphans():
            print '%-8s: %s (%s)' \
                % (feed_id, feed['feed_title'], feed['feed_link'])

    def _get_dups(self):
        feed_links = defaultdict(list)
        for feed_id, feed in self._feeds['feeds'].iteritems():
            feed_link = feed['feed_link'].rstrip('/')
            feed_links[feed_link].append((feed_id, feed))
        for feed_link, feeds in feed_links.iteritems():
            if len(feeds) > 1:
                for feed_id, feed in feeds:
                    yield feed_id, feed

    def do_dups(self, line):
        for feed_id, feed in self._get_dups():
            print '%-8s: %s (%s)' \
                % (feed_id, feed['feed_title'], feed['feed_link'])

    def do_cleanup(self, line):
        """Move all orphans to a folder named ORPHANS, and all
        duplicates to a folder name DUPLICATES."""
        self._add_folder('ORPHANS')
        self._add_folder('DUPLICATES')
        for feed_id, feed in self._get_orphans():
            self._move_feed(feed_id, '', 'ORPHANS', refetch=False)
        for feed_id, feed in self._get_dups():
            in_folder = self._find_feed_folder(feed_id)
            if in_folder != 'DUPLICATES':
                self._move_feed(feed_id, in_folder, 'DUPLICATES',
                                refetch=False)
        self._fetch_feeds()

    def do_logout(self, line):
        print self._api.logout()
        return True    # exit cmdloop

    do_exit = do_logout
    do_quit = do_logout


if __name__ == '__main__':
    cmd = NewsblurCmd()
    cmd.onecmd('login')
    NewsblurCmd().cmdloop()