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         return(u"%s: %s" %(
 111             "Following" if action == "follow" else "Leaving", user['name']))
 112
 113 class VerboseAdminFormatter(object):
 114     def __call__(self, action, user):
 115         return(u"-- %s: %s (%s): %s" % (
 116             "Following" if action == "follow" else "Leaving",
 117             user['screen_name'],
 118             user['name'],
 119             user['url']))
 120
 121 class URLAdminFormatter(object):
 122     def __call__(self, action, user):
 123         return("Admin actions do not support the URL formatter")
 124
 125 status_formatters = {
 126     'default': StatusFormatter,
 127     'verbose': VerboseStatusFormatter,
 128     'urls': URLStatusFormatter
 129 }
 130
 131 admin_formatters = {
 132     'default': AdminFormatter,
 133     'verbose': VerboseAdminFormatter,
 134     'urls': URLAdminFormatter
 135 }
 136
 137 def get_status_formatter(options):
 138     sf = status_formatters.get(options['format'])
 139     if (not sf):
 140         raise TwitterError(
 141             "Unknown formatter '%s'" %(options['format']))
 142     return sf()
 143
 144 def get_admin_formatter(options):
 145     sf = admin_formatters.get(options['format'])
 146     if (not sf):
 147         raise TwitterError(
 148             "Unknown formatter '%s'" %(options['format']))
 149     return sf()
 150
 151 class Action(object):
 152     pass
 153
 154 class NoSuchAction(Action):
 155     def __call__(self, twitter, options):
 156         print >> sys.stderr, "No such action: ", options['action']
 157         sys.exit(1)
 158
 159 class StatusAction(Action):
 160     def __call__(self, twitter, options):
 161         statuses = self.getStatuses(twitter)
 162         sf = get_status_formatter(options)
 163         for status in statuses:
 164             statusStr = sf(status)
 165             if statusStr.strip():
 166                 print statusStr.encode(sys.stdout.encoding, 'replace')
 167
 168 class AdminAction(Action):
 169     def __call__(self, twitter, options):
 170         if (not options['extra_args'][0]):
 171             raise TwitterError("You need to specify a User (Screen Name)")
 172         af = get_admin_formatter(options)
 173         user = self.getUser(twitter, options['extra_args'][0])
 174         if(user):
 175             print af(options['action'], user).encode(sys.stdout.encoding, 'replace')
 176
 177 class FriendsAction(StatusAction):
 178     def getStatuses(self, twitter):
 179         return reversed(twitter.statuses.friends_timeline())
 180
 181 class PublicAction(StatusAction):
 182     def getStatuses(self, twitter):
 183         return reversed(twitter.statuses.public_timeline())
 184
 185 class RepliesAction(StatusAction):
 186     def getStatuses(self, twitter):
 187         return reversed(twitter.statuses.replies())
 188
 189 class FollowAction(AdminAction):
 190     def getUser(self, twitter, user):
 191         # Twitter wants /notifications/follow/user.json?id=user
 192         return twitter.notifications.follow.__getattr__(user)(id=user)
 193
 194 class LeaveAction(AdminAction):
 195     def getUser(self, twitter, user):
 196         return twitter.notifications.leave.__getattr__(user)(id=user)
 197
 198 class SetStatusAction(Action):
 199     def __call__(self, twitter, options):
 200         statusTxt = (u" ".join(options['extra_args'])
 201                      if options['extra_args']
 202                      else unicode(raw_input("message: ")))
 203         status = (statusTxt.encode('utf8', 'replace'))
 204         twitter.statuses.update(status=status)
 205
 206 actions = {
 207     'follow': FollowAction,
 208     'friends': FriendsAction,
 209     'leave': LeaveAction,
 210     'public': PublicAction,
 211     'replies': RepliesAction,
 212     'set': SetStatusAction,
 213 }
 214
 215 def loadConfig(filename):
 216     email = None
 217     password = None
 218     if os.path.exists(filename):
 219         cp = SafeConfigParser()
 220         cp.read([filename])
 221         email = cp.get('twitter', 'email', None)
 222         password = cp.get('twitter', 'password', None)
 223     return email, password
 224
 225 def main():
 226     return main_with_args(sys.argv[1:])
 227
 228 def main_with_args(args):
 229     parse_args(args, options)
 230
 231     email, password = loadConfig(options['config_filename'])
 232     if not options['email']: options['email'] = email
 233     if not options['password']: options['password'] = password
 234
 235     #Maybe check for AdminAction here, but whatever you do, don't write TODO
 236     if options['refresh'] and options['action'] == 'set':
 237         print >> sys.stderr, "You can't repeatedly set your status, silly"
 238         print >> sys.stderr, "Use 'twitter -h' for help."
 239         sys.exit(1)
 240     if options['email'] and not options['password']:
 241         options['password'] = getpass("Twitter password: ")
 242     twitter = Twitter(options['email'], options['password'])
 243     action = actions.get(options['action'], NoSuchAction)()
 244     try:
 245         doAction = lambda : action(twitter, options)
 246
 247         if (options['refresh'] and isinstance(action, StatusAction)):
 248             while True:
 249                 doAction()
 250                 time.sleep(options['refresh_rate'])
 251         else:
 252             doAction()
 253
 254     except TwitterError, e:
 255         print >> sys.stderr, e.args[0]
 256         print >> sys.stderr, "Use 'twitter -h' for help."
 257         sys.exit(1)
 258     except KeyboardInterrupt:
 259         pass
 260