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 def printNicely(string):
 179     if sys.stdout.encoding:
 180         print string.encode(sys.stdout.encoding, 'replace')
 181     else:
 182         print string.encode('utf-8')
 183
 184 class StatusAction(Action):
 185     def __call__(self, twitter, options):
 186         statuses = self.getStatuses(twitter)
 187         sf = get_status_formatter(options)
 188         for status in statuses:
 189             statusStr = sf(status)
 190             if statusStr.strip():
 191                 printNicely(statusStr)
 192
 193 class AdminAction(Action):
 194     def __call__(self, twitter, options):
 195         if not options['extra_args'][0]:
 196             raise TwitterError("You need to specify a user (screen name)")
 197         af = get_admin_formatter(options)
 198         try:
 199             user = self.getUser(twitter, options['extra_args'][0])
 200         except TwitterError, e:
 201             print "There was a problem following or leaving the specified user."
 202             print "  You may be trying to follow a user you are already following;"
 203             print "  Leaving a user you are not currently following;"
 204             print "  Or the user may not exist."
 205             print "  Sorry."
 206             print
 207             print e
 208         else:
 209             printNicely(af(options['action'], user))
 210
 211 class FriendsAction(StatusAction):
 212     def getStatuses(self, twitter):
 213         return reversed(twitter.statuses.friends_timeline())
 214
 215 class PublicAction(StatusAction):
 216     def getStatuses(self, twitter):
 217         return reversed(twitter.statuses.public_timeline())
 218
 219 class RepliesAction(StatusAction):
 220     def getStatuses(self, twitter):
 221         return reversed(twitter.statuses.replies())
 222
 223 class FollowAction(AdminAction):
 224     def getUser(self, twitter, user):
 225         return twitter.friendships.create(id=user)
 226
 227 class LeaveAction(AdminAction):
 228     def getUser(self, twitter, user):
 229         return twitter.friendships.destroy(id=user)
 230
 231 class SetStatusAction(Action):
 232     def __call__(self, twitter, options):
 233         statusTxt = (u" ".join(options['extra_args'])
 234                      if options['extra_args']
 235                      else unicode(raw_input("message: ")))
 236         status = (statusTxt.encode('utf8', 'replace'))
 237         twitter.statuses.update(status=status)
 238
 239 class HelpAction(Action):
 240     def __call__(self, twitter, options):
 241         print __doc__
 242
 243 actions = {
 244     'follow': FollowAction,
 245     'friends': FriendsAction,
 246     'help': HelpAction,
 247     'leave': LeaveAction,
 248     'public': PublicAction,
 249     'replies': RepliesAction,
 250     'set': SetStatusAction,
 251 }
 252
 253 def loadConfig(filename):
 254     options = dict(OPTIONS)
 255     if os.path.exists(filename):
 256         cp = SafeConfigParser()
 257         cp.read([filename])
 258         for option in ('email', 'password', 'format'):
 259             if cp.has_option('twitter', option):
 260                 options[option] = cp.get('twitter', option)
 261     return options
 262
 263 def main(args=sys.argv[1:]):
 264     arg_options = {}
 265     try:
 266         parse_args(args, arg_options)
 267     except GetoptError, e:
 268         print >> sys.stderr, "I can't do that, %s." %(e)
 269         print >> sys.stderr
 270         sys.exit(1)
 271
 272     config_options = loadConfig(
 273         arg_options.get('config_filename') or OPTIONS.get('config_filename'))
 274
 275     # Apply the various options in order, the most important applied last.
 276     # Defaults first, then what's read from config file, then command-line
 277     # arguments.
 278     options = dict(OPTIONS)
 279     for d in config_options, arg_options:
 280         for k,v in d.items():
 281             if v: options[k] = v
 282
 283     if options['refresh'] and options['action'] not in (
 284         'friends', 'public', 'replies'):
 285         print >> sys.stderr, "You can only refresh the friends, public, or replies actions."
 286         print >> sys.stderr, "Use 'twitter -h' for help."
 287         sys.exit(1)
 288
 289     if options['email'] and not options['password']:
 290         options['password'] = getpass("Twitter password: ")
 291
 292     twitter = Twitter(options['email'], options['password'], agent=AGENT_STR)
 293     action = actions.get(options['action'], NoSuchAction)()
 294
 295     try:
 296         doAction = lambda : action(twitter, options)
 297
 298         if (options['refresh'] and isinstance(action, StatusAction)):
 299             while True:
 300                 doAction()
 301                 time.sleep(options['refresh_rate'])
 302         else:
 303             doAction()
 304
 305     except TwitterError, e:
 306         print >> sys.stderr, e.args[0]
 307         print >> sys.stderr, "Use 'twitter -h' for help."
 308         sys.exit(1)
 309     except KeyboardInterrupt:
 310         pass
 311