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