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