twitter/cmdline.py

   1 # encoding: utf-8
   2 """
   3 USAGE:
   4
   5  twitter [action] [options]
   6
   7
   8 ACTIONS:
   9  authorize      authorize the command-line tool to interact with Twitter
  10  follow         follow a user
  11  friends        get latest tweets from your friends (default action)
  12  help           print this help text that you are currently reading
  13  leave          stop following a user
  14  list           get list of a user's lists; give a list name to get
  15                     tweets from that list
  16  mylist         get list of your lists; give a list name to get tweets
  17                     from that list
  18  public         get latest public tweets
  19  pyprompt       start a Python prompt for interacting with the twitter
  20                     object directly
  21  replies        get latest replies to you
  22  search         search twitter (Beware: octothorpe, escape it)
  23  set            set your twitter status
  24  shell          login to the twitter shell
  25  rate           get your current rate limit status (remaining API reqs)
  26
  27
  28 OPTIONS:
  29
  30  -r --refresh               run this command forever, polling every once
  31                             in a while (default: every 5 minutes)
  32  -R --refresh-rate <rate>   set the refresh rate (in seconds)
  33  -f --format <format>       specify the output format for status updates
  34  -c --config <filename>     read username and password from given config
  35                             file (default ~/.twitter)
  36  -l --length <count>        specify number of status updates shown
  37                             (default: 20, max: 200)
  38  -t --timestamp             show time before status lines
  39  -d --datestamp             show date before status lines
  40     --no-ssl                use less-secure HTTP instead of HTTPS
  41     --oauth <filename>      filename to read/store oauth credentials to
  42
  43 FORMATS for the --format option
  44
  45  default         one line per status
  46  verbose         multiple lines per status, more verbose status info
  47  urls            nothing but URLs
  48  ansi            ansi colour (rainbow mode)
  49
  50
  51 CONFIG FILES
  52
  53  The config file should be placed in your home directory and be named .twitter.
  54  It must contain a [twitter] header, and all the desired options you wish to
  55  set, like so:
  56
  57 [twitter]
  58 format: <desired_default_format_for_output>
  59 prompt: <twitter_shell_prompt e.g. '[cyan]twitter[R]> '>
  60
  61  OAuth authentication tokens are stored in the file .twitter_oauth in your
  62  home directory.
  63 """
  64
  65 from __future__ import print_function
  66
  67 CONSUMER_KEY='uS6hO2sV6tDKIOeVjhnFnQ'
  68 CONSUMER_SECRET='MEYTOS97VvlHX7K1rwHPEqVpTSqZ71HtvoK4sVuYk'
  69
  70 import sys
  71 import time
  72 from getopt import gnu_getopt as getopt, GetoptError
  73 from getpass import getpass
  74 import re
  75 import os.path
  76 import locale
  77 import string
  78
  79 try:
  80     from ConfigParser import SafeConfigParser
  81 except ImportError:
  82     from configparser import ConfigParser as SafeConfigParser
  83 import datetime
  84 try:
  85     from urllib.parse import quote
  86 except ImportError:
  87     from urllib2 import quote
  88 import webbrowser
  89
  90 from .api import Twitter, TwitterError
  91 from .oauth import OAuth, write_token_file, read_token_file
  92 from .oauth_dance import oauth_dance
  93 from . import ansi
  94 from .util import smrt_input, printNicely
  95
  96 OPTIONS = {
  97     'action': 'friends',
  98     'refresh': False,
  99     'refresh_rate': 600,
 100     'format': 'default',
 101     'prompt': '[cyan]twitter[R]> ',
 102     'config_filename': os.environ.get('HOME', '') + os.sep + '.twitter',
 103     'oauth_filename': os.environ.get('HOME', '') + os.sep + '.twitter_oauth',
 104     'length': 20,
 105     'timestamp': False,
 106     'datestamp': False,
 107     'extra_args': [],
 108     'secure': True,
 109 }
 110
 111 def parse_args(args, options):
 112     long_opts = ['help', 'format=', 'refresh', 'oauth=',
 113                  'refresh-rate=', 'config=', 'length=', 'timestamp',
 114                  'datestamp', 'no-ssl']
 115     short_opts = "e:p:f:h?rR:c:l:td"
 116     opts, extra_args = getopt(args, short_opts, long_opts)
 117     extra_args = [arg.decode(locale.getpreferredencoding())
 118                   for arg in extra_args]
 119
 120     for opt, arg in opts:
 121         if opt in ('-f', '--format'):
 122             options['format'] = arg
 123         elif opt in ('-r', '--refresh'):
 124             options['refresh'] = True
 125         elif opt in ('-R', '--refresh-rate'):
 126             options['refresh_rate'] = int(arg)
 127         elif opt in ('-l', '--length'):
 128             options["length"] = int(arg)
 129         elif opt in ('-t', '--timestamp'):
 130             options["timestamp"] = True
 131         elif opt in ('-d', '--datestamp'):
 132             options["datestamp"] = True
 133         elif opt in ('-?', '-h', '--help'):
 134             options['action'] = 'help'
 135         elif opt in ('-c', '--config'):
 136             options['config_filename'] = arg
 137         elif opt == '--no-ssl':
 138             options['secure'] = False
 139         elif opt == '--oauth':
 140             options['oauth_filename'] = arg
 141
 142     if extra_args and not ('action' in options and options['action'] == 'help'):
 143         options['action'] = extra_args[0]
 144     options['extra_args'] = extra_args[1:]
 145
 146 def get_time_string(status, options, format="%a %b %d %H:%M:%S +0000 %Y"):
 147     timestamp = options["timestamp"]
 148     datestamp = options["datestamp"]
 149     t = time.strptime(status['created_at'], format)
 150     i_hate_timezones = time.timezone
 151     if (time.daylight):
 152         i_hate_timezones = time.altzone
 153     dt = datetime.datetime(*t[:-3]) - datetime.timedelta(
 154         seconds=i_hate_timezones)
 155     t = dt.timetuple()
 156     if timestamp and datestamp:
 157         return time.strftime("%Y-%m-%d %H:%M:%S ", t)
 158     elif timestamp:
 159         return time.strftime("%H:%M:%S ", t)
 160     elif datestamp:
 161         return time.strftime("%Y-%m-%d ", t)
 162     return ""
 163
 164 class StatusFormatter(object):
 165     def __call__(self, status, options):
 166         return ("%s%s %s" %(
 167             get_time_string(status, options),
 168             status['user']['screen_name'], status['text']))
 169
 170 class AnsiStatusFormatter(object):
 171     def __init__(self):
 172         self._colourMap = ansi.ColourMap()
 173
 174     def __call__(self, status, options):
 175         colour = self._colourMap.colourFor(status['user']['screen_name'])
 176         return ("%s%s%s%s %s" %(
 177             get_time_string(status, options),
 178             ansi.cmdColour(colour), status['user']['screen_name'],
 179             ansi.cmdReset(), status['text']))
 180
 181 class VerboseStatusFormatter(object):
 182     def __call__(self, status, options):
 183         return ("-- %s (%s) on %s\n%s\n" %(
 184             status['user']['screen_name'],
 185             status['user']['location'],
 186             status['created_at'],
 187             status['text']))
 188
 189 class URLStatusFormatter(object):
 190     urlmatch = re.compile(r'https?://\S+')
 191     def __call__(self, status, options):
 192         urls = self.urlmatch.findall(status['text'])
 193         return '\n'.join(urls) if urls else ""
 194
 195
 196 class ListsFormatter(object):
 197     def __call__(self, list):
 198         if list['description']:
 199             list_str = "%-30s (%s)" % (list['name'], list['description'])
 200         else:
 201             list_str = "%-30s" % (list['name'])
 202         return "%s\n" % list_str
 203
 204 class ListsVerboseFormatter(object):
 205     def __call__(self, list):
 206         list_str = "%-30s\n description: %s\n members: %s\n mode:%s\n" % (list['name'], list['description'], list['member_count'], list['mode'])
 207         return list_str
 208
 209 class AnsiListsFormatter(object):
 210     def __init__(self):
 211         self._colourMap = ansi.ColourMap()
 212
 213     def __call__(self, list):
 214         colour = self._colourMap.colourFor(list['name'])
 215         return ("%s%-15s%s %s" %(
 216             ansi.cmdColour(colour), list['name'],
 217             ansi.cmdReset(), list['description']))
 218
 219
 220 class AdminFormatter(object):
 221     def __call__(self, action, user):
 222         user_str = "%s (%s)" %(user['screen_name'], user['name'])
 223         if action == "follow":
 224             return "You are now following %s.\n" %(user_str)
 225         else:
 226             return "You are no longer following %s.\n" %(user_str)
 227
 228 class VerboseAdminFormatter(object):
 229     def __call__(self, action, user):
 230         return("-- %s: %s (%s): %s" % (
 231             "Following" if action == "follow" else "Leaving",
 232             user['screen_name'],
 233             user['name'],
 234             user['url']))
 235
 236 class SearchFormatter(object):
 237     def __call__(self, result, options):
 238         return("%s%s %s" %(
 239             get_time_string(result, options, "%a, %d %b %Y %H:%M:%S +0000"),
 240             result['from_user'], result['text']))
 241
 242 class VerboseSearchFormatter(SearchFormatter):
 243     pass #Default to the regular one
 244
 245 class URLSearchFormatter(object):
 246     urlmatch = re.compile(r'https?://\S+')
 247     def __call__(self, result, options):
 248         urls = self.urlmatch.findall(result['text'])
 249         return '\n'.join(urls) if urls else ""
 250
 251 class AnsiSearchFormatter(object):
 252     def __init__(self):
 253         self._colourMap = ansi.ColourMap()
 254
 255     def __call__(self, result, options):
 256         colour = self._colourMap.colourFor(result['from_user'])
 257         return ("%s%s%s%s %s" %(
 258             get_time_string(result, options, "%a, %d %b %Y %H:%M:%S +0000"),
 259             ansi.cmdColour(colour), result['from_user'],
 260             ansi.cmdReset(), result['text']))
 261
 262 _term_encoding = None
 263 def get_term_encoding():
 264     global _term_encoding
 265     if not _term_encoding:
 266         lang = os.getenv('LANG', 'unknown.UTF-8').split('.')
 267         if lang[1:]:
 268             _term_encoding = lang[1]
 269         else:
 270             _term_encoding = 'UTF-8'
 271     return _term_encoding
 272
 273 formatters = {}
 274 status_formatters = {
 275     'default': StatusFormatter,
 276     'verbose': VerboseStatusFormatter,
 277     'urls': URLStatusFormatter,
 278     'ansi': AnsiStatusFormatter
 279 }
 280 formatters['status'] = status_formatters
 281
 282 admin_formatters = {
 283     'default': AdminFormatter,
 284     'verbose': VerboseAdminFormatter,
 285     'urls': AdminFormatter,
 286     'ansi': AdminFormatter
 287 }
 288 formatters['admin'] = admin_formatters
 289
 290 search_formatters = {
 291     'default': SearchFormatter,
 292     'verbose': VerboseSearchFormatter,
 293     'urls': URLSearchFormatter,
 294     'ansi': AnsiSearchFormatter
 295 }
 296 formatters['search'] = search_formatters
 297
 298 lists_formatters = {
 299     'default': ListsFormatter,
 300     'verbose': ListsVerboseFormatter,
 301     'urls': None,
 302     'ansi': AnsiListsFormatter
 303 }
 304 formatters['lists'] = lists_formatters
 305
 306 def get_formatter(action_type, options):
 307     formatters_dict = formatters.get(action_type)
 308     if (not formatters_dict):
 309         raise TwitterError(
 310             "There was an error finding a class of formatters for your type (%s)"
 311             %(action_type))
 312     f = formatters_dict.get(options['format'])
 313     if (not f):
 314         raise TwitterError(
 315             "Unknown formatter '%s' for status actions" %(options['format']))
 316     return f()
 317
 318 class Action(object):
 319
 320     def ask(self, subject='perform this action', careful=False):
 321         '''
 322         Requests fromt he user using `raw_input` if `subject` should be
 323         performed. When `careful`, the default answer is NO, otherwise YES.
 324         Returns the user answer in the form `True` or `False`.
 325         '''
 326         sample = '(y/N)'
 327         if not careful:
 328             sample = '(Y/n)'
 329
 330         prompt = 'You really want to %s %s? ' %(subject, sample)
 331         try:
 332             answer = input(prompt).lower()
 333             if careful:
 334                 return answer in ('yes', 'y')
 335             else:
 336                 return answer not in ('no', 'n')
 337         except EOFError:
 338             print(file=sys.stderr) # Put Newline since Enter was never pressed
 339             # TODO:
 340                 #   Figure out why on OS X the raw_input keeps raising
 341                 #   EOFError and is never able to reset and get more input
 342                 #   Hint: Look at how IPython implements their console
 343             default = True
 344             if careful:
 345                 default = False
 346             return default
 347
 348     def __call__(self, twitter, options):
 349         action = actions.get(options['action'], NoSuchAction)()
 350         try:
 351             doAction = lambda : action(twitter, options)
 352             if (options['refresh'] and isinstance(action, StatusAction)):
 353                 while True:
 354                     doAction()
 355                     time.sleep(options['refresh_rate'])
 356             else:
 357                 doAction()
 358         except KeyboardInterrupt:
 359             print('\n[Keyboard Interrupt]', file=sys.stderr)
 360             pass
 361
 362 class NoSuchActionError(Exception):
 363     pass
 364
 365 class NoSuchAction(Action):
 366     def __call__(self, twitter, options):
 367         raise NoSuchActionError("No such action: %s" %(options['action']))
 368
 369 class StatusAction(Action):
 370     def __call__(self, twitter, options):
 371         statuses = self.getStatuses(twitter, options)
 372         sf = get_formatter('status', options)
 373         for status in statuses:
 374             statusStr = sf(status, options)
 375             if statusStr.strip():
 376                 printNicely(statusStr)
 377
 378 class SearchAction(Action):
 379     def __call__(self, twitter, options):
 380         # We need to be pointing at search.twitter.com to work, and it is less
 381         # tangly to do it here than in the main()
 382         twitter.domain="search.twitter.com"
 383         twitter.uriparts=()
 384         # We need to bypass the TwitterCall parameter encoding, so we
 385         # don't encode the plus sign, so we have to encode it ourselves
 386         query_string = "+".join(
 387             [quote(term)
 388              for term in options['extra_args']])
 389
 390         results = twitter.search(q=query_string)['results']
 391         f = get_formatter('search', options)
 392         for result in results:
 393             resultStr = f(result, options)
 394             if resultStr.strip():
 395                 printNicely(resultStr)
 396
 397 class AdminAction(Action):
 398     def __call__(self, twitter, options):
 399         if not (options['extra_args'] and options['extra_args'][0]):
 400             raise TwitterError("You need to specify a user (screen name)")
 401         af = get_formatter('admin', options)
 402         try:
 403             user = self.getUser(twitter, options['extra_args'][0])
 404         except TwitterError as e:
 405             print("There was a problem following or leaving the specified user.")
 406             print("You may be trying to follow a user you are already following;")
 407             print("Leaving a user you are not currently following;")
 408             print("Or the user may not exist.")
 409             print("Sorry.")
 410             print()
 411             print(e)
 412         else:
 413             printNicely(af(options['action'], user))
 414
 415 class ListsAction(StatusAction):
 416     def getStatuses(self, twitter, options):
 417         if not options['extra_args']:
 418             raise TwitterError("Please provide a user to query for lists")
 419
 420         screen_name = options['extra_args'][0]
 421
 422         if not options['extra_args'][1:]:
 423             lists = twitter.user.lists(user=screen_name)['lists']
 424             if not lists:
 425                 printNicely("This user has no lists.")
 426             for list in lists:
 427                 lf = get_formatter('lists', options)
 428                 printNicely(lf(list))
 429             return []
 430         else:
 431             return reversed(twitter.user.lists.list.statuses(
 432                     user=screen_name, list=options['extra_args'][1]))
 433
 434
 435 class MyListsAction(ListsAction):
 436     def getStatuses(self, twitter, options):
 437         screen_name = twitter.account.verify_credentials()['screen_name']
 438         options['extra_args'].insert(0, screen_name)
 439         return ListsAction.getStatuses(self, twitter, options)
 440
 441
 442 class FriendsAction(StatusAction):
 443     def getStatuses(self, twitter, options):
 444         return reversed(twitter.statuses.friends_timeline(count=options["length"]))
 445
 446 class PublicAction(StatusAction):
 447     def getStatuses(self, twitter, options):
 448         return reversed(twitter.statuses.public_timeline(count=options["length"]))
 449
 450 class RepliesAction(StatusAction):
 451     def getStatuses(self, twitter, options):
 452         return reversed(twitter.statuses.replies(count=options["length"]))
 453
 454 class FollowAction(AdminAction):
 455     def getUser(self, twitter, user):
 456         return twitter.friendships.create(id=user)
 457
 458 class LeaveAction(AdminAction):
 459     def getUser(self, twitter, user):
 460         return twitter.friendships.destroy(id=user)
 461
 462 class SetStatusAction(Action):
 463     def __call__(self, twitter, options):
 464         statusTxt = (" ".join(options['extra_args'])
 465                      if options['extra_args']
 466                      else str(input("message: ")))
 467         splitted = []
 468         while statusTxt:
 469             if len(statusTxt) > 140:
 470                 end = string.rfind(statusTxt, ' ', 0, 140)
 471             else:
 472                 end = 140
 473             splitted.append(statusTxt[:end])
 474             statusTxt = statusTxt[end:]
 475
 476         for status in splitted:
 477             twitter.statuses.update(status=status)
 478
 479 class TwitterShell(Action):
 480
 481     def render_prompt(self, prompt):
 482         '''Parses the `prompt` string and returns the rendered version'''
 483         prompt = prompt.strip("'").replace("\\'","'")
 484         for colour in ansi.COLOURS_NAMED:
 485             if '[%s]' %(colour) in prompt:
 486                 prompt = prompt.replace(
 487                     '[%s]' %(colour), ansi.cmdColourNamed(colour))
 488         prompt = prompt.replace('[R]', ansi.cmdReset())
 489         return prompt
 490
 491     def __call__(self, twitter, options):
 492         prompt = self.render_prompt(options.get('prompt', 'twitter> '))
 493         while True:
 494             options['action'] = ""
 495             try:
 496                 args = input(prompt).split()
 497                 parse_args(args, options)
 498                 if not options['action']:
 499                     continue
 500                 elif options['action'] == 'exit':
 501                     raise SystemExit(0)
 502                 elif options['action'] == 'shell':
 503                     print('Sorry Xzibit does not work here!', file=sys.stderr)
 504                     continue
 505                 elif options['action'] == 'help':
 506                     print('''\ntwitter> `action`\n
 507                           The Shell Accepts all the command line actions along with:
 508
 509                           exit    Leave the twitter shell (^D may also be used)
 510
 511                           Full CMD Line help is appended below for your convinience.''', file=sys.stderr)
 512                 Action()(twitter, options)
 513                 options['action'] = ''
 514             except NoSuchActionError as e:
 515                 print(e, file=sys.stderr)
 516             except KeyboardInterrupt:
 517                 print('\n[Keyboard Interrupt]', file=sys.stderr)
 518             except EOFError:
 519                 print(file=sys.stderr)
 520                 leaving = self.ask(subject='Leave')
 521                 if not leaving:
 522                     print('Excellent!', file=sys.stderr)
 523                 else:
 524                     raise SystemExit(0)
 525
 526 class PythonPromptAction(Action):
 527     def __call__(self, twitter, options):
 528         try:
 529             while True:
 530                 smrt_input(globals(), locals())
 531         except EOFError:
 532             pass
 533
 534 class HelpAction(Action):
 535     def __call__(self, twitter, options):
 536         print(__doc__)
 537
 538 class DoNothingAction(Action):
 539     def __call__(self, twitter, options):
 540         pass
 541
 542 class RateLimitStatus(Action):
 543     def __call__(self, twitter, options):
 544         rate = twitter.account.rate_limit_status()
 545         print("Remaining API requests: %s / %s (hourly limit)" % (rate['remaining_hits'], rate['hourly_limit']))
 546         print("Next reset in %ss (%s)" % (int(rate['reset_time_in_seconds']-time.time()),
 547                                           time.asctime(time.localtime(rate['reset_time_in_seconds']))))
 548
 549 actions = {
 550     'authorize' : DoNothingAction,
 551     'follow'    : FollowAction,
 552     'friends'   : FriendsAction,
 553     'list'      : ListsAction,
 554     'mylist'    : MyListsAction,
 555     'help'      : HelpAction,
 556     'leave'     : LeaveAction,
 557     'public'    : PublicAction,
 558     'pyprompt'  : PythonPromptAction,
 559     'replies'   : RepliesAction,
 560     'search'    : SearchAction,
 561     'set'       : SetStatusAction,
 562     'shell'     : TwitterShell,
 563     'rate'      : RateLimitStatus,
 564 }
 565
 566 def loadConfig(filename):
 567     options = dict(OPTIONS)
 568     if os.path.exists(filename):
 569         cp = SafeConfigParser()
 570         cp.read([filename])
 571         for option in ('format', 'prompt'):
 572             if cp.has_option('twitter', option):
 573                 options[option] = cp.get('twitter', option)
 574     return options
 575
 576 def main(args=sys.argv[1:]):
 577     arg_options = {}
 578     try:
 579         parse_args(args, arg_options)
 580     except GetoptError as e:
 581         print("I can't do that, %s." %(e), file=sys.stderr)
 582         print(file=sys.stderr)
 583         raise SystemExit(1)
 584
 585     config_path = os.path.expanduser(
 586         arg_options.get('config_filename') or OPTIONS.get('config_filename'))
 587     config_options = loadConfig(config_path)
 588
 589     # Apply the various options in order, the most important applied last.
 590     # Defaults first, then what's read from config file, then command-line
 591     # arguments.
 592     options = dict(OPTIONS)
 593     for d in config_options, arg_options:
 594         for k,v in list(d.items()):
 595             if v: options[k] = v
 596
 597     if options['refresh'] and options['action'] not in (
 598         'friends', 'public', 'replies'):
 599         print("You can only refresh the friends, public, or replies actions.", file=sys.stderr)
 600         print("Use 'twitter -h' for help.", file=sys.stderr)
 601         return 1
 602
 603     oauth_filename = os.path.expanduser(options['oauth_filename'])
 604
 605     if (options['action'] == 'authorize'
 606         or not os.path.exists(oauth_filename)):
 607         oauth_dance(
 608             "the Command-Line Tool", CONSUMER_KEY, CONSUMER_SECRET,
 609             options['oauth_filename'])
 610
 611     oauth_token, oauth_token_secret = read_token_file(oauth_filename)
 612
 613     twitter = Twitter(
 614         auth=OAuth(
 615             oauth_token, oauth_token_secret, CONSUMER_KEY, CONSUMER_SECRET),
 616         secure=options['secure'],
 617         api_version='1',
 618         domain='api.twitter.com')
 619
 620     try:
 621         Action()(twitter, options)
 622     except NoSuchActionError as e:
 623         print(e, file=sys.stderr)
 624         raise SystemExit(1)
 625     except TwitterError as e:
 626         print(str(e), file=sys.stderr)
 627         print("Use 'twitter -h' for help.", file=sys.stderr)
 628         raise SystemExit(1)