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