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