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