twitter/cmdline.py

   1 """
   2 USAGE:
   3
   4  twitter [action] [options]
   5
   6 ACTIONS:
   7  follow         add the specified user to your follow list
   8  friends        get latest tweets from your friends (default action)
   9  help           print this help text that you are currently reading
  10  leave          remove the specified user from your following list
  11  public         get latest public tweets
  12  replies        get latest replies
  13  set            set your twitter status
  14  shell          login the twitter shell
  15
  16 OPTIONS:
  17
  18  -e --email <email>         your email to login to twitter
  19  -p --password <password>   your twitter password
  20  -r --refresh               run this command forever, polling every once
  21                             in a while (default: every 5 minutes)
  22  -R --refresh-rate <rate>   set the refresh rate (in seconds)
  23  -f --format <format>       specify the output format for status updates
  24  -c --config <filename>     read username and password from given config
  25                               file (default ~/.twitter)
  26
  27 FORMATS for the --format option
  28
  29  default         one line per status
  30  verbose         multiple lines per status, more verbose status info
  31  urls            nothing but URLs
  32  ansi            ansi colour (rainbow mode)
  33
  34 CONFIG FILES
  35
  36  The config file should contain a [twitter] header, and all the desired options
  37  you wish to set, like so:
  38
  39 [twitter]
  40 email: <username>
  41 password: <password>
  42 format: <desired_default_format_for_output>
  43 prompt: <twitter_shell_prompt e.g. '[cyan]twitter[R]> '>
  44 """
  45
  46 import sys
  47 import time
  48 from getopt import gnu_getopt as getopt, GetoptError
  49 from getpass import getpass
  50 import re
  51 import os.path
  52 from ConfigParser import SafeConfigParser
  53
  54 from api import Twitter, TwitterError
  55 import ansi
  56
  57 # Please don't change this, it was provided by the fine folks at Twitter.
  58 # If you change it, it will not work.
  59 AGENT_STR = "twittercommandlinetoolpy"
  60
  61 OPTIONS = {
  62     'email': None,
  63     'password': None,
  64     'action': 'friends',
  65     'refresh': False,
  66     'refresh_rate': 600,
  67     'format': 'default',
  68     'prompt': '[cyan]twitter[R]> ',
  69     'config_filename': os.environ.get('HOME', '') + os.sep + '.twitter',
  70     'extra_args': []
  71 }
  72
  73 def parse_args(args, options):
  74     long_opts = ['email', 'password', 'help', 'format', 'refresh',
  75                  'refresh-rate', 'config']
  76     short_opts = "e:p:f:h?rR:c:"
  77     opts, extra_args = getopt(args, short_opts, long_opts)
  78
  79     for opt, arg in opts:
  80         if opt in ('-e', '--email'):
  81             options['email'] = arg
  82         elif opt in ('-p', '--password'):
  83             options['password'] = arg
  84         elif opt in ('-f', '--format'):
  85             options['format'] = arg
  86         elif opt in ('-r', '--refresh'):
  87             options['refresh'] = True
  88         elif opt in ('-R', '--refresh-rate'):
  89             options['refresh_rate'] = int(arg)
  90         elif opt in ('-?', '-h', '--help'):
  91             options['action'] = 'help'
  92         elif opt in ('-c', '--config'):
  93             options['config_filename'] = arg
  94
  95     if extra_args and not ('action' in options and options['action'] == 'help'):
  96         options['action'] = extra_args[0]
  97     options['extra_args'] = extra_args[1:]
  98
  99 class StatusFormatter(object):
 100     def __call__(self, status):
 101         return (u"%s %s" %(
 102             status['user']['screen_name'], status['text']))
 103
 104 class AnsiStatusFormatter(object):
 105     def __init__(self):
 106         self._colourMap = ansi.ColourMap()
 107
 108     def __call__(self, status):
 109         colour = self._colourMap.colourFor(status['user']['screen_name'])
 110         return (u"%s%s%s %s" %(
 111             ansi.cmdColour(colour), status['user']['screen_name'],
 112             ansi.cmdReset(), status['text']))
 113
 114 class VerboseStatusFormatter(object):
 115     def __call__(self, status):
 116         return (u"-- %s (%s) on %s\n%s\n" %(
 117             status['user']['screen_name'],
 118             status['user']['location'],
 119             status['created_at'],
 120             status['text']))
 121
 122 class URLStatusFormatter(object):
 123     urlmatch = re.compile(r'https?://\S+')
 124     def __call__(self, status):
 125         urls = self.urlmatch.findall(status['text'])
 126         return u'\n'.join(urls) if urls else ""
 127
 128 class AdminFormatter(object):
 129     def __call__(self, action, user):
 130         user_str = u"%s (%s)" %(user['screen_name'], user['name'])
 131         if action == "follow":
 132             return u"You are now following %s.\n" %(user_str)
 133         else:
 134             return u"You are no longer following %s.\n" %(user_str)
 135
 136 class VerboseAdminFormatter(object):
 137     def __call__(self, action, user):
 138         return(u"-- %s: %s (%s): %s" % (
 139             "Following" if action == "follow" else "Leaving",
 140             user['screen_name'],
 141             user['name'],
 142             user['url']))
 143
 144 status_formatters = {
 145     'default': StatusFormatter,
 146     'verbose': VerboseStatusFormatter,
 147     'urls': URLStatusFormatter,
 148     'ansi': AnsiStatusFormatter
 149 }
 150
 151 admin_formatters = {
 152     'default': AdminFormatter,
 153     'verbose': VerboseAdminFormatter,
 154     'urls': AdminFormatter,
 155     'ansi': AdminFormatter
 156 }
 157
 158 def get_status_formatter(options):
 159     sf = status_formatters.get(options['format'])
 160     if (not sf):
 161         raise TwitterError(
 162             "Unknown formatter '%s'" %(options['format']))
 163     return sf()
 164
 165 def get_admin_formatter(options):
 166     sf = admin_formatters.get(options['format'])
 167     if (not sf):
 168         raise TwitterError(
 169             "Unknown formatter '%s'" %(options['format']))
 170     return sf()
 171
 172 class Action(object):
 173     @staticmethod
 174     def ask(subject='perform this action', careful=False):
 175         '''
 176         Requests fromt he user using `raw_input` if `subject` should be
 177         performed. When `careful`, the default answer is NO, otherwise YES.
 178         Returns the user answer in the form `True` or `False`.
 179         '''
 180         sample = '(y/N)' if careful else '(Y/n)'
 181         prompt = 'You really want to %s %s? ' %(subject, sample)
 182         try:
 183             answer = raw_input(prompt).lower()
 184             if careful:
 185                 if answer not in ('yes', 'y'):
 186                     return False
 187                 else:
 188                     return True
 189             else:
 190                 if answer in ('no', 'n'):
 191                     return False
 192                 else:
 193                     return True
 194         except EOFError:
 195             print >>sys.stderr # Put Newline since Enter was never pressed
 196             # TODO:
 197                 #   Figure out why on OS X the raw_input keeps raising
 198                 #   EOFError and is never able to reset and get more input
 199                 #   Hint: Look at how IPython implements their console
 200             default = False if careful else True
 201             return default
 202     def __call__(self, twitter, options):
 203         action = actions.get(options['action'], NoSuchAction)()
 204         try:
 205             doAction = lambda : action(twitter, options)
 206             if (options['refresh'] and isinstance(action, StatusAction)):
 207                 while True:
 208                     doAction()
 209                     time.sleep(options['refresh_rate'])
 210             else:
 211                 doAction()
 212         except KeyboardInterrupt:
 213             print >>sys.stderr, '\n[Keyboard Interrupt]'
 214             pass
 215
 216 class NoSuchActionError(Exception):
 217     pass
 218
 219 class NoSuchAction(Action):
 220     def __call__(self, twitter, options):
 221         raise NoSuchActionError("No such action: %s" %(options['action']))
 222
 223 class StatusAction(Action):
 224     def __call__(self, twitter, options):
 225         statuses = self.getStatuses(twitter)
 226         sf = get_status_formatter(options)
 227         for status in statuses:
 228             statusStr = sf(status)
 229             if statusStr.strip():
 230                 print statusStr.encode(sys.stdout.encoding, 'replace')
 231
 232 class AdminAction(Action):
 233     def __call__(self, twitter, options):
 234         if not options['extra_args'][0]:
 235             raise TwitterError("You need to specify a user (screen name)")
 236         af = get_admin_formatter(options)
 237         try:
 238             user = self.getUser(twitter, options['extra_args'][0])
 239         except TwitterError, e:
 240             print "There was a problem following or leaving the specified user."
 241             print "  You may be trying to follow a user you are already following;"
 242             print "  Leaving a user you are not currently following;"
 243             print "  Or the user may not exist."
 244             print "  Sorry."
 245             print
 246             print e
 247         else:
 248             print af(options['action'], user).encode(sys.stdout.encoding, 'replace')
 249
 250 class FriendsAction(StatusAction):
 251     def getStatuses(self, twitter):
 252         return reversed(twitter.statuses.friends_timeline())
 253
 254 class PublicAction(StatusAction):
 255     def getStatuses(self, twitter):
 256         return reversed(twitter.statuses.public_timeline())
 257
 258 class RepliesAction(StatusAction):
 259     def getStatuses(self, twitter):
 260         return reversed(twitter.statuses.replies())
 261
 262 class FollowAction(AdminAction):
 263     def getUser(self, twitter, user):
 264         return twitter.notifications.follow(id=user)
 265
 266 class LeaveAction(AdminAction):
 267     def getUser(self, twitter, user):
 268         return twitter.notifications.leave(id=user)
 269
 270 class SetStatusAction(Action):
 271     def __call__(self, twitter, options):
 272         statusTxt = (u" ".join(options['extra_args'])
 273                      if options['extra_args']
 274                      else unicode(raw_input("message: ")))
 275         status = (statusTxt.encode('utf8', 'replace'))
 276         twitter.statuses.update(status=status)
 277
 278 class TwitterShell(Action):
 279     @staticmethod
 280     def render_prompt(prompt):
 281         '''Parses the `prompt` string and returns the rendered version'''
 282         prompt = prompt.strip("'").replace("\\'","'")
 283         for colour in ansi.COLOURS_NAMED:
 284             if '[%s]' %(colour) in prompt:
 285                 prompt = prompt.replace(
 286                             '[%s]' %(colour), ansi.cmdColourNamed(colour))
 287         prompt = prompt.replace('[R]', ansi.cmdReset())
 288         return prompt
 289     def __call__(self, twitter, options):
 290         prompt = self.render_prompt(options.get('prompt', 'twitter> '))
 291         while True:
 292             try:
 293                 args = raw_input(prompt).split()
 294                 parse_args(args, options)
 295                 if not options['action']:
 296                     continue
 297                 elif options['action'] == 'exit':
 298                     raise SystemExit(0)
 299                 elif options['action'] == 'shell':
 300                     print >>sys.stderr, 'Sorry Xzibit does not work here!'
 301                     continue
 302                 elif options['action'] == 'help':
 303                     print >>sys.stderr, '''\ntwitter> `action`\n
 304         The Shell Accepts all the command line actions along with:
 305
 306             exit    Leave the twitter shell (^D may also be used)
 307
 308         Full CMD Line help is appended below for your convinience.'''
 309                 Action()(twitter, options)
 310                 options['action'] = ''
 311             except NoSuchActionError, e:
 312                 print >>sys.stderr, e
 313             except KeyboardInterrupt:
 314                 print >>sys.stderr, '\n[Keyboard Interrupt]'
 315             except EOFError:
 316                 print >>sys.stderr
 317                 leaving = self.ask(subject='Leave')
 318                 if not leaving:
 319                     print >>sys.stderr, 'Excellent!'
 320                 else:
 321                     raise SystemExit(0)
 322
 323 class HelpAction(Action):
 324     def __call__(self, twitter, options):
 325         print __doc__
 326
 327 actions = {
 328     'follow'    : FollowAction,
 329     'friends'   : FriendsAction,
 330     'help'      : HelpAction,
 331     'leave'     : LeaveAction,
 332     'public'    : PublicAction,
 333     'replies'   : RepliesAction,
 334     'set'       : SetStatusAction,
 335     'shell'     : TwitterShell,
 336 }
 337
 338 def loadConfig(filename):
 339     options = dict(OPTIONS)
 340     if os.path.exists(filename):
 341         cp = SafeConfigParser()
 342         cp.read([filename])
 343         for option in ('email', 'password', 'format', 'prompt'):
 344             if cp.has_option('twitter', option):
 345                 options[option] = cp.get('twitter', option)
 346     return options
 347
 348 def main(args=sys.argv[1:]):
 349     arg_options = {}
 350     try:
 351         parse_args(args, arg_options)
 352     except GetoptError, e:
 353         print >> sys.stderr, "I can't do that, %s." %(e)
 354         print >> sys.stderr
 355         raise SystemExit(1)
 356
 357     config_options = loadConfig(
 358         arg_options.get('config_filename') or OPTIONS.get('config_filename'))
 359
 360     # Apply the various options in order, the most important applied last.
 361     # Defaults first, then what's read from config file, then command-line
 362     # arguments.
 363     options = dict(OPTIONS)
 364     for d in config_options, arg_options:
 365         for k,v in d.items():
 366             if v: options[k] = v
 367
 368     if options['refresh'] and options['action'] not in (
 369         'friends', 'public', 'replies'):
 370         print >> sys.stderr, "You can only refresh the friends, public, or replies actions."
 371         print >> sys.stderr, "Use 'twitter -h' for help."
 372         raise SystemExit(1)
 373
 374     if options['email'] and not options['password']:
 375         options['password'] = getpass("Twitter password: ")
 376
 377     twitter = Twitter(options['email'], options['password'], agent=AGENT_STR)
 378     try:
 379         Action()(twitter, options)
 380     except NoSuchActionError, e:
 381         print >>sys.stderr, e
 382         raise SystemExit(1)
 383     except TwitterError, e:
 384         print >> sys.stderr, e.args[0]
 385         print >> sys.stderr, "Use 'twitter -h' for help."
 386         raise SystemExit(1)
 387