gpodder/bin/gpo

#!/usr/bin/env python
# -*- coding: utf-8 -*-

#
# gPodder - A media aggregator and podcast client
# Copyright (c) 2005-2010 Thomas Perl and the gPodder Team
#
# gPodder is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# gPodder is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#


# gpo - A better command-line interface to gPodder using the gPodder API
# by Thomas Perl <thp@gpodder.org>; 2009-05-07


"""
  Usage: gpo [COMMAND] [params...]

  - Subscription management -

    subscribe URL [TITLE]      Subscribe to a new feed at URL (as TITLE)
    rename URL TITLE           Rename feed at URL to TITLE
    unsubscribe URL            Unsubscribe from feed at URL
    enable URL                 Enable feed updates for the feed at URL
    disable URL                Disable feed updates for the feed at URL

    info URL                   Show information about feed at URL
    list                       List all subscribed podcasts
    update [URL]               Check for new episodes (all or only at URL)

  - Episode management -

    download [URL]             Download new episodes (all or only from URL)
    pending [URL]              List new episodes (all or only from URL)
    episodes [URL]             List episodes (all or only from URL)

    importfiles                Check download folders for external files

  - Other commands -

    youtube [URL]              Resolve the YouTube URL to a download URL

"""

import sys
import os
import re
import inspect
import readline
import shlex
import pydoc

import termios
import fcntl
import struct

gpodder_script = sys.argv[0]
if os.path.islink(gpodder_script):
    gpodder_script = os.readlink(gpodder_script)
gpodder_dir = os.path.join(os.path.dirname(gpodder_script), '..')
prefix = os.path.abspath(os.path.normpath(gpodder_dir))

src_dir = os.path.join(prefix, 'src')
data_dir = os.path.join(prefix, 'data')

if os.path.exists(src_dir) and os.path.exists(data_dir) and \
        not prefix.startswith('/usr'):
    # Run gPodder from local source folder (not installed)
    sys.path.insert(0, src_dir)


import gpodder
_ = gpodder.gettext

# Use only the gPodder API here, so this serves both as an example
# and as a motivation to provide all functionality in the API :)
from gpodder import api

def inred(x):
    return '\033[91m' + x + '\033[0m'

def ingreen(x):
    return '\033[92m' + x + '\033[0m'

def inblue(x):
    return '\033[94m' + x + '\033[0m'

def FirstArgumentIsPodcastURL(function):
    """Decorator for functions that take a podcast URL as first arg"""
    setattr(function, '_first_arg_is_podcast', True)
    return function

def get_terminal_size():
    s = struct.pack('HHHH', 0, 0, 0, 0)
    stdout = sys.stdout.fileno()
    x = fcntl.ioctl(stdout, termios.TIOCGWINSZ, s)
    rows, cols, xp, yp = struct.unpack('HHHH', x)
    return rows, cols

class gPodderCli(object):
    COLUMNS = 80
    ALIASES = {
        '?': 'help',
    }

    def __init__(self):
        self.client = api.PodcastClient()
        self._current_action = ''

    def _start_action(self, msg, *args):
        line = msg % args
        if len(line) > self.COLUMNS-7:
            line = line[:self.COLUMNS-7-3] + '...'
        else:
            line = line + (' '*(self.COLUMNS-7-len(line)))
        self._current_action = line
        sys.stdout.write(line)
        sys.stdout.flush()

    def _update_action(self, progress):
        progress = '%3.0f%%' % (progress*100.,)
        result = '['+inblue(progress)+']'
        sys.stdout.write('\r' + self._current_action + result)
        sys.stdout.flush()

    def _finish_action(self, success=True):
        result = '['+ingreen('DONE')+']' if success else '['+inred('FAIL')+']'
        print '\r' + self._current_action + result
        self._current_action = ''

    # -------------------------------------------------------------------

    def subscribe(self, url, title=None):
        if self.client.get_podcast(url) is not None:
            self._info(_('You are already subscribed to %s.' % url))
            return True

        if self.client.create_podcast(url, title) is None:
            self._error(_('Cannot download feed for %s.') % url)
            return True

        self.client.finish()

        self._info(_('Successfully added %s.' % url))
        return True

    @FirstArgumentIsPodcastURL
    def rename(self, url, title):
        podcast = self.client.get_podcast(url)

        if podcast is None:
            self._error(_('You are not subscribed to %s.') % url)
        else:
            old_title = podcast.title
            podcast.rename(title)
            self.client.finish()
            self._info(_('Renamed %s to %s.') % (old_title, title))

        return True

    @FirstArgumentIsPodcastURL
    def unsubscribe(self, url):
        podcast = self.client.get_podcast(url)

        if podcast is None:
            self._error(_('You are not subscribed to %s.') % url)
        else:
            podcast.delete()
            self.client.finish()
            self._error(_('Unsubscribed from %s.') % url)

        return True

    def _episodesList(self, podcast):
        def status_str(episode):
            if episode.is_new:
                return ' * '
            if episode.is_downloaded:
                return ' ▉ '
            if episode.is_deleted:
                return ' ░ '
    
            return '   '

        episodes = ('%3d. %s %s' % (i+1, status_str(e), e.title) for i, e in enumerate(podcast.get_episodes()))
	return episodes

    @FirstArgumentIsPodcastURL
    def info(self, url):
        podcast = self.client.get_podcast(url)

        if podcast is None:
            self._error(_('You are not subscribed to %s.') % url)
        else:
            title, url, status = podcast.title, podcast.url, podcast.feed_update_status_msg()
            episodes = self._episodesList(podcast)
            episodes = '\n      '.join(episodes)
            self._pager("""
    Title: %(title)s
    URL: %(url)s
    Feed update is %(status)s

    Episodes:
      %(episodes)s
            """ % locals())

        return True

    @FirstArgumentIsPodcastURL
    def episodes(self, url=None):
        output = []
        for podcast in self.client.get_podcasts():
            podcast_printed = False
            if url is None or podcast.url == url:
                episodes = self._episodesList(podcast)
                episodes = '\n      '.join(episodes)
                output.append("""
    Episodes from %s:
      %s
""" % (podcast.url, episodes))

        self._pager('\n'.join(output))
        return True

    def list(self):
        for podcast in self.client.get_podcasts():
            print podcast.url

        return True

    @FirstArgumentIsPodcastURL
    def update(self, url=None):
        for podcast in self.client.get_podcasts():
            if url is None and podcast.update_enabled():
                self._start_action('Updating %s', podcast.title)
                try:
                    podcast.update()
                    self._finish_action()
                except Exception, e:
                    self._finish_action(False)
            elif podcast.url == url:
                # Don't need to check for update_enabled()
                self._start_action('Updating %s', podcast.title)
                try:
                    podcast.update()
                    self._finish_action()
                except Exception, e:
                    self._finish_action(False)

        return True

    @FirstArgumentIsPodcastURL
    def pending(self, url=None):
        count = 0
        for podcast in self.client.get_podcasts():
            podcast_printed = False
            if url is None or podcast.url == url:
                for episode in podcast.get_episodes():
                    if episode.is_new:
                        if not podcast_printed:
                            print podcast.title
                            podcast_printed = True
                        print '   ', episode.title
                        count += 1

        print count, 'episodes pending.'
        return True

    def importfiles(self):
        full_count = 0

        for podcast in self.client.get_podcasts():
            print 'Checking for files in', podcast.title
            count = podcast._podcast.import_external_files()
            if count:
                print ' => found %d valid file(s)' % count
            full_count += count

        print '%d downloaded file(s) imported.' % full_count
        self.client.finish()
        return True

    @FirstArgumentIsPodcastURL
    def download(self, url=None):
        count = 0
        for podcast in self.client.get_podcasts():
            podcast_printed = False
            if url is None or podcast.url == url:
                for episode in podcast.get_episodes():
                    if episode.is_new:
                        if not podcast_printed:
                            print inblue(podcast.title)
                            podcast_printed = True
                        self._start_action('Downloading %s', episode.title)
                        episode.download(self._update_action)
                        self._finish_action()
                        count += 1

        print count, 'episodes downloaded.'
        return True

    @FirstArgumentIsPodcastURL
    def disable(self, url):
        podcast = self.client.get_podcast(url)

        if podcast is None:
            self._error(_('You are not subscribed to %s.') % url)
        else:
            podcast.disable()
            self.client.finish()
            self._error(_('Disabling feed update from %s.') % url)

        return True

    @FirstArgumentIsPodcastURL
    def enable(self, url):
        podcast = self.client.get_podcast(url)

        if podcast is None:
            self._error(_('You are not subscribed to %s.') % url)
        else:
            podcast.enable()
            self.client.finish()
            self._error(_('Enabling feed update from %s.') % url)

        return True

    def youtube(self, url):
        yurl = self.client.youtube_url_resolver(url)
        print yurl
        return True

    def help(self):
        sys.stderr.write(stylize(__doc__))
        return True

    # -------------------------------------------------------------------

    def _pager(self, output):
        if sys.stdout.isatty():
            # Need two additional rows for command prompt
            rows_needed = len(output.splitlines()) + 2
            rows, cols = get_terminal_size()
            if rows_needed < rows:
                print output
            else:
                pydoc.pager(output)
        else:
            print output

    def _shell(self):
        print '\n'.join(x.strip() for x in ("""
        gPodder %(__version__)s (%(__date__)s) - %(__url__)s
        %(__copyright__)s
        License: %(__licence__)s

        Entering interactive shell. Type 'help' for help.
        Press Ctrl+D (EOF) or type 'quit' to quit.
        """ % gpodder.__dict__).splitlines())

        readline.parse_and_bind('tab: complete')
        readline.set_completer(self._tab_completion)
        readline.set_completer_delims(' ')

        while True:
            try:
                line = raw_input('gpo> ')
            except EOFError:
                print ''
                break
            except KeyboardInterrupt:
                print ''
                continue

            if line in ('quit', 'exit', 'bye'):
                break

            try:
                self._parse(shlex.split(line))
            except KeyboardInterrupt:
                self._error('Keyboard interrupt.')

    def _error(self, *args):
        print >>sys.stderr, inred(' '.join(args))

    def _info(self, *args):
        print >>sys.stdout, ' '.join(args)

    def _checkargs(self, func, command_line):
        args, varargs, keywords, defaults = inspect.getargspec(func)
        args.pop(0) # Remove "self" from args
        defaults = defaults or ()
        minarg, maxarg = len(args)-len(defaults), len(args)

        if len(command_line) < minarg or (len(command_line) > maxarg and \
                varargs is None):
            self._error('Wrong argument count for %s.' % func.__name__)
            return False

        return func(*command_line)

    def _tab_completion_podcast(self, text, count):
        """Tab completion for podcast URLs"""
        urls = [p.url for p in self.client.get_podcasts() if text in p.url]
        if count < len(urls):
            return urls[count]

        return NOne


    def _tab_completion(self, text, count):
        """Tab completion function for readline"""
        current_line = readline.get_line_buffer()
        if text == current_line:
            for name, func in inspect.getmembers(self):
                if inspect.ismethod(func) and \
                        name.startswith(text) and \
                        not name.startswith('_'):
                    if count == 0:
                        return name
                    else:
                        count -= 1
        else:
            args = current_line.split()
            command = args.pop(0)
            command_function = getattr(self, command, None)
            if not command_function:
                return None
            if getattr(command_function, '_first_arg_is_podcast', False):
                if not args or (len(args) == 1 and not current_line.endswith(' ')):
                    return self._tab_completion_podcast(text, count)

        return None


    def _parse(self, command_line):
        if not command_line:
            return False

        command = command_line.pop(0)
        if command.startswith('_'):
            self._error(_('This command is not available.'))
            return False

        # Resolve command aliases
        command = self.ALIASES.get(command, command)

        for name, func in inspect.getmembers(self):
            if inspect.ismethod(func) and name == command:
                return self._checkargs(func, command_line)

        self._error(_('The requested function is not available.'))
        return False


def stylize(s):
    s = re.sub(r'    .{27}', lambda m: inblue(m.group(0)), s)
    s = re.sub(r'  - .*', lambda m: ingreen(m.group(0)), s)
    return s

if __name__ == '__main__':
    cli = gPodderCli()
    args = sys.argv[1:]
    if args:
        cli._parse(args) or sys.stderr.write(stylize(__doc__))
    elif sys.stdout.isatty():
        cli._shell()
    else:
        sys.stdout.write(__doc__)