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