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
  15 OPTIONS:
  16
  17  -e --email <email>         your email to login to twitter
  18  -p --password <password>   your twitter password
  19  -r --refresh               run this command forever, polling every once
  20                             in a while (default: every 5 minutes)
  21  -R --refresh-rate <rate>   set the refresh rate (in seconds)
  22  -f --format <format>       specify the output format for status updates
  23  -c --config <filename>     read username and password from given config
  24                               file (default ~/.twitter)
  25
  26 FORMATS for the --format option
  27
  28  default         one line per status
  29  verbose         multiple lines per status, more verbose status info
  30  urls            nothing but URLs
  31  ansi            ansi colour (rainbow mode)
  32
  33 CONFIG FILES
  34
  35  The config file should contain a [twitter] header, and all the desired options
  36  you wish to set, like so:
  37
  38 [twitter]
  39 email: <username>
  40 password: <password>
  41 format: <desired_default_format_for_output>
  42 """
  43
  44 import sys
  45 import time
  46 from getopt import getopt, GetoptError
  47 from getpass import getpass
  48 import re
  49 import os.path
  50 from ConfigParser import SafeConfigParser
  51
  52 from api import Twitter, TwitterError
  53 import ansi
  54
  55 # Please don't change this, it was provided by the fine folks at Twitter.
  56 # If you change it, it will not work.
  57 AGENT_STR = "twittercommandlinetoolpy"
  58
  59 OPTIONS = {
  60     'email': None,
  61     'password': None,
  62     'action': 'friends',
  63     'refresh': False,
  64     'refresh_rate': 600,
  65     'format': 'default',
  66     'config_filename': os.environ.get('HOME', '') + os.sep + '.twitter',
  67     'extra_args': []
  68 }
  69
  70 def parse_args(args, options):
  71     long_opts = ['email', 'password', 'help', 'format', 'refresh',
  72                  'refresh-rate', 'config']
  73     short_opts = "e:p:f:h?rR:c:"
  74     opts, extra_args = getopt(args, short_opts, long_opts)
  75
  76     for opt, arg in opts:
  77         if opt in ('-e', '--email'):
  78             options['email'] = arg
  79         elif opt in ('-p', '--password'):
  80             options['password'] = arg
  81         elif opt in ('-f', '--format'):
  82             options['format'] = arg
  83         elif opt in ('-r', '--refresh'):
  84             options['refresh'] = True
  85         elif opt in ('-R', '--refresh-rate'):
  86             options['refresh_rate'] = int(arg)
  87         elif opt in ('-?', '-h', '--help'):
  88             print __doc__
  89             sys.exit(0)
  90         elif opt in ('-c', '--config'):
  91             options['config_filename'] = arg
  92
  93     if extra_args:
  94         options['action'] = extra_args[0]
  95     options['extra_args'] = extra_args[1:]
  96
  97 class StatusFormatter(object):
  98     def __call__(self, status):
  99         return (u"%s %s" %(
 100             status['user']['screen_name'], status['text']))
 101
 102 class AnsiStatusFormatter(object):
 103     def __init__(self):
 104         self._colourMap = ansi.ColourMap()
 105
 106     def __call__(self, status):
 107         colour = self._colourMap.colourFor(status['user']['screen_name'])
 108         return (u"%s%s%s %s" %(
 109             ansi.cmdColour(colour), status['user']['screen_name'],
 110             ansi.cmdReset(), status['text']))
 111
 112 class VerboseStatusFormatter(object):
 113     def __call__(self, status):
 114         return (u"-- %s (%s) on %s\n%s\n" %(
 115             status['user']['screen_name'],
 116             status['user']['location'],
 117             status['created_at'],
 118             status['text']))
 119
 120 class URLStatusFormatter(object):
 121     urlmatch = re.compile(r'https?://\S+')
 122     def __call__(self, status):
 123         urls = self.urlmatch.findall(status['text'])
 124         return u'\n'.join(urls) if urls else ""
 125
 126 class AdminFormatter(object):
 127     def __call__(self, action, user):
 128         user_str = u"%s (%s)" %(user['screen_name'], user['name'])
 129         if action == "follow":
 130             return u"You are now following %s.\n" %(user_str)
 131         else:
 132             return u"You are no longer following %s.\n" %(user_str)
 133
 134 class VerboseAdminFormatter(object):
 135     def __call__(self, action, user):
 136         return(u"-- %s: %s (%s): %s" % (
 137             "Following" if action == "follow" else "Leaving",
 138             user['screen_name'],
 139             user['name'],
 140             user['url']))
 141
 142 status_formatters = {
 143     'default': StatusFormatter,
 144     'verbose': VerboseStatusFormatter,
 145     'urls': URLStatusFormatter,
 146     'ansi': AnsiStatusFormatter
 147 }
 148
 149 admin_formatters = {
 150     'default': AdminFormatter,
 151     'verbose': VerboseAdminFormatter,
 152     'urls': AdminFormatter,
 153     'ansi': AdminFormatter
 154 }
 155
 156 def get_status_formatter(options):
 157     sf = status_formatters.get(options['format'])
 158     if (not sf):
 159         raise TwitterError(
 160             "Unknown formatter '%s'" %(options['format']))
 161     return sf()
 162
 163 def get_admin_formatter(options):
 164     sf = admin_formatters.get(options['format'])
 165     if (not sf):
 166         raise TwitterError(
 167             "Unknown formatter '%s'" %(options['format']))
 168     return sf()
 169
 170 class Action(object):
 171     pass
 172
 173 class NoSuchAction(Action):
 174     def __call__(self, twitter, options):
 175         print >> sys.stderr, "No such action: ", options['action']
 176         sys.exit(1)
 177
 178 class StatusAction(Action):
 179     def __call__(self, twitter, options):
 180         statuses = self.getStatuses(twitter)
 181         sf = get_status_formatter(options)
 182         for status in statuses:
 183             statusStr = sf(status)
 184             if statusStr.strip():
 185                 if sys.stdout.encoding:
 186                     print statusStr.encode(sys.stdout.encoding, 'replace')
 187                 else:
 188                     print statusStr.encode('utf-8')
 189
 190 class AdminAction(Action):
 191     def __call__(self, twitter, options):
 192         if not options['extra_args'][0]:
 193             raise TwitterError("You need to specify a user (screen name)")
 194         af = get_admin_formatter(options)
 195         try:
 196             user = self.getUser(twitter, options['extra_args'][0])
 197         except TwitterError, e:
 198             print "There was a problem following or leaving the specified user."
 199             print "  You may be trying to follow a user you are already following;"
 200             print "  Leaving a user you are not currently following;"
 201             print "  Or the user may not exist."
 202             print "  Sorry."
 203             print
 204             print e
 205         else:
 206             print af(options['action'], user).encode(sys.stdout.encoding, 'replace')
 207
 208 class FriendsAction(StatusAction):
 209     def getStatuses(self, twitter):
 210         return reversed(twitter.statuses.friends_timeline())
 211
 212 class PublicAction(StatusAction):
 213     def getStatuses(self, twitter):
 214         return reversed(twitter.statuses.public_timeline())
 215
 216 class RepliesAction(StatusAction):
 217     def getStatuses(self, twitter):
 218         return reversed(twitter.statuses.replies())
 219
 220 class FollowAction(AdminAction):
 221     def getUser(self, twitter, user):
 222         return twitter.friendships.create(id=user)
 223
 224 class LeaveAction(AdminAction):
 225     def getUser(self, twitter, user):
 226         return twitter.friendships.destroy(id=user)
 227
 228 class SetStatusAction(Action):
 229     def __call__(self, twitter, options):
 230         statusTxt = (u" ".join(options['extra_args'])
 231                      if options['extra_args']
 232                      else unicode(raw_input("message: ")))
 233         status = (statusTxt.encode('utf8', 'replace'))
 234         twitter.statuses.update(status=status)
 235
 236 class HelpAction(Action):
 237     def __call__(self, twitter, options):
 238         print __doc__
 239
 240 actions = {
 241     'follow': FollowAction,
 242     'friends': FriendsAction,
 243     'help': HelpAction,
 244     'leave': LeaveAction,
 245     'public': PublicAction,
 246     'replies': RepliesAction,
 247     'set': SetStatusAction,
 248 }
 249
 250 def loadConfig(filename):
 251     options = dict(OPTIONS)
 252     if os.path.exists(filename):
 253         cp = SafeConfigParser()
 254         cp.read([filename])
 255         for option in ('email', 'password', 'format'):
 256             if cp.has_option('twitter', option):
 257                 options[option] = cp.get('twitter', option)
 258     return options
 259
 260 def main(args=sys.argv[1:]):
 261     arg_options = {}
 262     try:
 263         parse_args(args, arg_options)
 264     except GetoptError, e:
 265         print >> sys.stderr, "I can't do that, %s." %(e)
 266         print >> sys.stderr
 267         sys.exit(1)
 268
 269     config_options = loadConfig(
 270         arg_options.get('config_filename') or OPTIONS.get('config_filename'))
 271
 272     # Apply the various options in order, the most important applied last.
 273     # Defaults first, then what's read from config file, then command-line
 274     # arguments.
 275     options = dict(OPTIONS)
 276     for d in config_options, arg_options:
 277         for k,v in d.items():
 278             if v: options[k] = v
 279
 280     if options['refresh'] and options['action'] not in (
 281         'friends', 'public', 'replies'):
 282         print >> sys.stderr, "You can only refresh the friends, public, or replies actions."
 283         print >> sys.stderr, "Use 'twitter -h' for help."
 284         sys.exit(1)
 285
 286     if options['email'] and not options['password']:
 287         options['password'] = getpass("Twitter password: ")
 288
 289     twitter = Twitter(options['email'], options['password'], agent=AGENT_STR)
 290     action = actions.get(options['action'], NoSuchAction)()
 291
 292     try:
 293         doAction = lambda : action(twitter, options)
 294
 295         if (options['refresh'] and isinstance(action, StatusAction)):
 296             while True:
 297                 doAction()
 298                 time.sleep(options['refresh_rate'])
 299         else:
 300             doAction()
 301
 302     except TwitterError, e:
 303         print >> sys.stderr, e.args[0]
 304         print >> sys.stderr, "Use 'twitter -h' for help."
 305         sys.exit(1)
 306     except KeyboardInterrupt:
 307         pass
 308