]>
jfr.im git - z_archive/twitter.git/blob - twitter/cmdline.py
5 twitter [action] [options]
9 authorize authorize the command-line tool to interact with Twitter
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
16 mylist get list of your lists; give a list name to get tweets
18 pyprompt start a Python prompt for interacting with the twitter
20 replies get latest replies to you
21 search search twitter (Beware: octothorpe, escape it)
22 set set your twitter status
23 shell login to the twitter shell
24 rate get your current rate limit status (remaining API reqs)
29 -r --refresh run this command forever, polling every once
30 in a while (default: every 5 minutes)
31 -R --refresh-rate <rate> set the refresh rate (in seconds)
32 -f --format <format> specify the output format for status updates
33 -c --config <filename> read username and password from given config
34 file (default ~/.twitter)
35 -l --length <count> specify number of status updates shown
36 (default: 20, max: 200)
37 -t --timestamp show time before status lines
38 -d --datestamp show date before status lines
39 --no-ssl use less-secure HTTP instead of HTTPS
40 --oauth <filename> filename to read/store oauth credentials to
42 FORMATS for the --format option
44 default one line per status
45 verbose multiple lines per status, more verbose status info
46 json raw json data from the api on each line
48 ansi ansi colour (rainbow mode)
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
58 format: <desired_default_format_for_output>
59 prompt: <twitter_shell_prompt e.g. '[cyan]twitter[R]> '>
61 OAuth authentication tokens are stored in the file .twitter_oauth in your
65 from __future__
import print_function
68 input = __builtins__
.raw_input
69 except (AttributeError, KeyError):
73 CONSUMER_KEY
= 'uS6hO2sV6tDKIOeVjhnFnQ'
74 CONSUMER_SECRET
= 'MEYTOS97VvlHX7K1rwHPEqVpTSqZ71HtvoK4sVuYk'
76 from getopt
import gnu_getopt
as getopt
, GetoptError
77 from getpass
import getpass
87 from ConfigParser
import SafeConfigParser
89 from configparser
import ConfigParser
as SafeConfigParser
92 from urllib
.parse
import quote
94 from urllib2
import quote
98 import html
.parser
as HTMLParser
102 from .api
import Twitter
, TwitterError
103 from .oauth
import OAuth
, write_token_file
, read_token_file
104 from .oauth_dance
import oauth_dance
106 from .util
import smrt_input
, printNicely
, align_text
113 'prompt': '[cyan]twitter[R]> ',
114 'config_filename': os
.environ
.get('HOME', os
.environ
.get('USERPROFILE', '')) + os
.sep
+ '.twitter',
115 'oauth_filename': os
.environ
.get('HOME', os
.environ
.get('USERPROFILE', '')) + os
.sep
+ '.twitter_oauth',
121 'invert_split': False,
125 gHtmlParser
= HTMLParser
.HTMLParser()
126 hashtagRe
= re
.compile(r
'(?P<hashtag>#\S+)')
127 profileRe
= re
.compile(r
'(?P<profile>\@\S+)')
128 ansiFormatter
= ansi
.AnsiCmd(False)
130 def parse_args(args
, options
):
131 long_opts
= ['help', 'format=', 'refresh', 'oauth=',
132 'refresh-rate=', 'config=', 'length=', 'timestamp',
133 'datestamp', 'no-ssl', 'force-ansi']
134 short_opts
= "e:p:f:h?rR:c:l:td"
135 opts
, extra_args
= getopt(args
, short_opts
, long_opts
)
136 if extra_args
and hasattr(extra_args
[0], 'decode'):
137 extra_args
= [arg
.decode(locale
.getpreferredencoding())
138 for arg
in extra_args
]
140 for opt
, arg
in opts
:
141 if opt
in ('-f', '--format'):
142 options
['format'] = arg
143 elif opt
in ('-r', '--refresh'):
144 options
['refresh'] = True
145 elif opt
in ('-R', '--refresh-rate'):
146 options
['refresh_rate'] = int(arg
)
147 elif opt
in ('-l', '--length'):
148 options
["length"] = int(arg
)
149 elif opt
in ('-t', '--timestamp'):
150 options
["timestamp"] = True
151 elif opt
in ('-d', '--datestamp'):
152 options
["datestamp"] = True
153 elif opt
in ('-?', '-h', '--help'):
154 options
['action'] = 'help'
155 elif opt
in ('-c', '--config'):
156 options
['config_filename'] = arg
157 elif opt
== '--no-ssl':
158 options
['secure'] = False
159 elif opt
== '--oauth':
160 options
['oauth_filename'] = arg
161 elif opt
== '--force-ansi':
162 options
['force-ansi'] = True
164 if extra_args
and not ('action' in options
and options
['action'] == 'help'):
165 options
['action'] = extra_args
[0]
166 options
['extra_args'] = extra_args
[1:]
168 def get_time_string(status
, options
, format
="%a %b %d %H:%M:%S +0000 %Y"):
169 timestamp
= options
["timestamp"]
170 datestamp
= options
["datestamp"]
171 t
= time
.strptime(status
['created_at'], format
)
172 i_hate_timezones
= time
.timezone
174 i_hate_timezones
= time
.altzone
175 dt
= datetime
.datetime(*t
[:-3]) - datetime
.timedelta(
176 seconds
=i_hate_timezones
)
178 if timestamp
and datestamp
:
179 return time
.strftime("%Y-%m-%d %H:%M:%S ", t
)
181 return time
.strftime("%H:%M:%S ", t
)
183 return time
.strftime("%Y-%m-%d ", t
)
188 'clear': ansiFormatter
.cmdReset(),
189 'hashtag': ansiFormatter
.cmdBold(),
190 'profile': ansiFormatter
.cmdUnderline(),
197 s
= '%s%s%s' % (ansiTypes
[mkey
], m
.group(mkey
), ansiTypes
['clear'])
202 def replaceInStatus(status
):
203 txt
= gHtmlParser
.unescape(status
)
204 txt
= re
.sub(hashtagRe
, reRepl
, txt
)
205 txt
= re
.sub(profileRe
, reRepl
, txt
)
208 class StatusFormatter(object):
209 def __call__(self
, status
, options
):
210 return ("%s%s %s" % (
211 get_time_string(status
, options
),
212 status
['user']['screen_name'], gHtmlParser
.unescape(status
['text'])))
214 class AnsiStatusFormatter(object):
216 self
._colourMap
= ansi
.ColourMap()
218 def __call__(self
, status
, options
):
219 colour
= self
._colourMap
.colourFor(status
['user']['screen_name'])
220 ret
= "%s%s% 16s%s " %(
221 get_time_string(status
, options
),
222 ansi
.cmdColour(colour
), status
['user']['screen_name'],
224 ret
+= "%s" % align_text(status
['text'])
227 class VerboseStatusFormatter(object):
228 def __call__(self
, status
, options
):
229 return ("-- %s (%s) on %s\n%s\n" % (
230 status
['user']['screen_name'],
231 status
['user']['location'],
232 status
['created_at'],
233 gHtmlParser
.unescape(status
['text'])))
235 class JSONStatusFormatter(object):
236 def __call__(self
, status
, options
):
237 status
['text'] = gHtmlParser
.unescape(status
['text'])
238 return json
.dumps(status
)
240 class URLStatusFormatter(object):
241 urlmatch
= re
.compile(r
'https?://\S+')
242 def __call__(self
, status
, options
):
243 urls
= self
.urlmatch
.findall(status
['text'])
244 return '\n'.join(urls
) if urls
else ""
247 class ListsFormatter(object):
248 def __call__(self
, list):
249 if list['description']:
250 list_str
= "%-30s (%s)" % (list['name'], list['description'])
252 list_str
= "%-30s" % (list['name'])
253 return "%s\n" % list_str
255 class ListsVerboseFormatter(object):
256 def __call__(self
, list):
257 list_str
= "%-30s\n description: %s\n members: %s\n mode:%s\n" % (list['name'], list['description'], list['member_count'], list['mode'])
260 class AnsiListsFormatter(object):
262 self
._colourMap
= ansi
.ColourMap()
264 def __call__(self
, list):
265 colour
= self
._colourMap
.colourFor(list['name'])
266 return ("%s%-15s%s %s" % (
267 ansiFormatter
.cmdColour(colour
), list['name'],
268 ansiFormatter
.cmdReset(), list['description']))
271 class AdminFormatter(object):
272 def __call__(self
, action
, user
):
273 user_str
= "%s (%s)" % (user
['screen_name'], user
['name'])
274 if action
== "follow":
275 return "You are now following %s.\n" % (user_str
)
277 return "You are no longer following %s.\n" % (user_str
)
279 class VerboseAdminFormatter(object):
280 def __call__(self
, action
, user
):
281 return("-- %s: %s (%s): %s" % (
282 "Following" if action
== "follow" else "Leaving",
287 class SearchFormatter(object):
288 def __call__(self
, result
, options
):
290 get_time_string(result
, options
, "%a, %d %b %Y %H:%M:%S +0000"),
291 result
['from_user'], result
['text']))
293 class VerboseSearchFormatter(SearchFormatter
):
294 pass # Default to the regular one
296 class URLSearchFormatter(object):
297 urlmatch
= re
.compile(r
'https?://\S+')
298 def __call__(self
, result
, options
):
299 urls
= self
.urlmatch
.findall(result
['text'])
300 return '\n'.join(urls
) if urls
else ""
302 class AnsiSearchFormatter(object):
304 self
._colourMap
= ansi
.ColourMap()
306 def __call__(self
, result
, options
):
307 colour
= self
._colourMap
.colourFor(result
['from_user'])
308 return ("%s%s%s%s %s" % (
309 get_time_string(result
, options
, "%a, %d %b %Y %H:%M:%S +0000"),
310 ansiFormatter
.cmdColour(colour
), result
['from_user'],
311 ansiFormatter
.cmdReset(), result
['text']))
313 _term_encoding
= None
314 def get_term_encoding():
315 global _term_encoding
316 if not _term_encoding
:
317 lang
= os
.getenv('LANG', 'unknown.UTF-8').split('.')
319 _term_encoding
= lang
[1]
321 _term_encoding
= 'UTF-8'
322 return _term_encoding
325 status_formatters
= {
326 'default': StatusFormatter
,
327 'verbose': VerboseStatusFormatter
,
328 'json': JSONStatusFormatter
,
329 'urls': URLStatusFormatter
,
330 'ansi': AnsiStatusFormatter
332 formatters
['status'] = status_formatters
335 'default': AdminFormatter
,
336 'verbose': VerboseAdminFormatter
,
337 'urls': AdminFormatter
,
338 'ansi': AdminFormatter
340 formatters
['admin'] = admin_formatters
342 search_formatters
= {
343 'default': SearchFormatter
,
344 'verbose': VerboseSearchFormatter
,
345 'urls': URLSearchFormatter
,
346 'ansi': AnsiSearchFormatter
348 formatters
['search'] = search_formatters
351 'default': ListsFormatter
,
352 'verbose': ListsVerboseFormatter
,
354 'ansi': AnsiListsFormatter
356 formatters
['lists'] = lists_formatters
358 def get_formatter(action_type
, options
):
359 formatters_dict
= formatters
.get(action_type
)
360 if (not formatters_dict
):
362 "There was an error finding a class of formatters for your type (%s)"
364 f
= formatters_dict
.get(options
['format'])
367 "Unknown formatter '%s' for status actions" % (options
['format']))
370 class Action(object):
372 def ask(self
, subject
='perform this action', careful
=False):
374 Requests from the user using `raw_input` if `subject` should be
375 performed. When `careful`, the default answer is NO, otherwise YES.
376 Returns the user answer in the form `True` or `False`.
382 prompt
= 'You really want to %s %s? ' % (subject
, sample
)
384 answer
= input(prompt
).lower()
386 return answer
in ('yes', 'y')
388 return answer
not in ('no', 'n')
390 print(file=sys
.stderr
) # Put Newline since Enter was never pressed
392 # Figure out why on OS X the raw_input keeps raising
393 # EOFError and is never able to reset and get more input
394 # Hint: Look at how IPython implements their console
400 def __call__(self
, twitter
, options
):
401 action
= actions
.get(options
['action'], NoSuchAction
)()
403 doAction
= lambda : action(twitter
, options
)
404 if (options
['refresh'] and isinstance(action
, StatusAction
)):
408 time
.sleep(options
['refresh_rate'])
411 except KeyboardInterrupt:
412 print('\n[Keyboard Interrupt]', file=sys
.stderr
)
415 class NoSuchActionError(Exception):
418 class NoSuchAction(Action
):
419 def __call__(self
, twitter
, options
):
420 raise NoSuchActionError("No such action: %s" % (options
['action']))
422 class StatusAction(Action
):
423 def __call__(self
, twitter
, options
):
424 statuses
= self
.getStatuses(twitter
, options
)
425 sf
= get_formatter('status', options
)
426 for status
in statuses
:
427 statusStr
= sf(status
, options
)
428 if statusStr
.strip():
429 printNicely(statusStr
)
431 class SearchAction(Action
):
432 def __call__(self
, twitter
, options
):
433 # We need to be pointing at search.twitter.com to work, and it is less
434 # tangly to do it here than in the main()
435 twitter
.domain
= "search.twitter.com"
436 twitter
.uriparts
= ()
437 # We need to bypass the TwitterCall parameter encoding, so we
438 # don't encode the plus sign, so we have to encode it ourselves
439 query_string
= "+".join(
441 for term
in options
['extra_args']])
443 results
= twitter
.search(q
=query_string
)['results']
444 f
= get_formatter('search', options
)
445 for result
in results
:
446 resultStr
= f(result
, options
)
447 if resultStr
.strip():
448 printNicely(resultStr
)
450 class AdminAction(Action
):
451 def __call__(self
, twitter
, options
):
452 if not (options
['extra_args'] and options
['extra_args'][0]):
453 raise TwitterError("You need to specify a user (screen name)")
454 af
= get_formatter('admin', options
)
456 user
= self
.getUser(twitter
, options
['extra_args'][0])
457 except TwitterError
as e
:
458 print("There was a problem following or leaving the specified user.")
459 print("You may be trying to follow a user you are already following;")
460 print("Leaving a user you are not currently following;")
461 print("Or the user may not exist.")
466 printNicely(af(options
['action'], user
))
468 class ListsAction(StatusAction
):
469 def getStatuses(self
, twitter
, options
):
470 if not options
['extra_args']:
471 raise TwitterError("Please provide a user to query for lists")
473 screen_name
= options
['extra_args'][0]
475 if not options
['extra_args'][1:]:
476 lists
= twitter
.lists
.list(screen_name
=screen_name
)
478 printNicely("This user has no lists.")
480 lf
= get_formatter('lists', options
)
481 printNicely(lf(list))
484 return reversed(twitter
.user
.lists
.list.statuses(
485 user
=screen_name
, list=options
['extra_args'][1]))
488 class MyListsAction(ListsAction
):
489 def getStatuses(self
, twitter
, options
):
490 screen_name
= twitter
.account
.verify_credentials()['screen_name']
491 options
['extra_args'].insert(0, screen_name
)
492 return ListsAction
.getStatuses(self
, twitter
, options
)
495 class FriendsAction(StatusAction
):
496 def getStatuses(self
, twitter
, options
):
497 return reversed(twitter
.statuses
.home_timeline(count
=options
["length"]))
499 class RepliesAction(StatusAction
):
500 def getStatuses(self
, twitter
, options
):
501 return reversed(twitter
.statuses
.mentions_timeline(count
=options
["length"]))
503 class FollowAction(AdminAction
):
504 def getUser(self
, twitter
, user
):
505 return twitter
.friendships
.create(screen_name
=user
)
507 class LeaveAction(AdminAction
):
508 def getUser(self
, twitter
, user
):
509 return twitter
.friendships
.destroy(screen_name
=user
)
511 class SetStatusAction(Action
):
512 def __call__(self
, twitter
, options
):
513 statusTxt
= (" ".join(options
['extra_args'])
514 if options
['extra_args']
515 else str(input("message: ")))
517 ptr
= re
.compile("@[\w_]+")
519 s
= ptr
.match(statusTxt
)
520 if s
and s
.start() == 0:
521 replies
.append(statusTxt
[s
.start():s
.end()])
522 statusTxt
= statusTxt
[s
.end() + 1:]
525 replies
= " ".join(replies
)
526 if len(replies
) >= 140:
533 limit
= 140 - len(replies
)
534 if len(statusTxt
) > limit
:
535 end
= string
.rfind(statusTxt
, ' ', 0, limit
)
538 splitted
.append(" ".join((replies
, statusTxt
[:end
])))
539 statusTxt
= statusTxt
[end
:]
541 if options
['invert_split']:
543 for status
in splitted
:
544 twitter
.statuses
.update(status
=status
)
546 class TwitterShell(Action
):
548 def render_prompt(self
, prompt
):
549 '''Parses the `prompt` string and returns the rendered version'''
550 prompt
= prompt
.strip("'").replace("\\'", "'")
551 for colour
in ansi
.COLOURS_NAMED
:
552 if '[%s]' % (colour
) in prompt
:
553 prompt
= prompt
.replace(
554 '[%s]' % (colour
), ansiFormatter
.cmdColourNamed(colour
))
555 prompt
= prompt
.replace('[R]', ansiFormatter
.cmdReset())
558 def __call__(self
, twitter
, options
):
559 prompt
= self
.render_prompt(options
.get('prompt', 'twitter> '))
561 options
['action'] = ""
563 args
= input(prompt
).split()
564 parse_args(args
, options
)
565 if not options
['action']:
567 elif options
['action'] == 'exit':
569 elif options
['action'] == 'shell':
570 print('Sorry Xzibit does not work here!', file=sys
.stderr
)
572 elif options
['action'] == 'help':
573 print('''\ntwitter> `action`\n
574 The Shell Accepts all the command line actions along with:
576 exit Leave the twitter shell (^D may also be used)
578 Full CMD Line help is appended below for your convinience.''', file=sys
.stderr
)
579 Action()(twitter
, options
)
580 options
['action'] = ''
581 except NoSuchActionError
as e
:
582 print(e
, file=sys
.stderr
)
583 except KeyboardInterrupt:
584 print('\n[Keyboard Interrupt]', file=sys
.stderr
)
586 print(file=sys
.stderr
)
587 leaving
= self
.ask(subject
='Leave')
589 print('Excellent!', file=sys
.stderr
)
593 class PythonPromptAction(Action
):
594 def __call__(self
, twitter
, options
):
597 smrt_input(globals(), locals())
601 class HelpAction(Action
):
602 def __call__(self
, twitter
, options
):
605 class DoNothingAction(Action
):
606 def __call__(self
, twitter
, options
):
609 class RateLimitStatus(Action
):
610 def __call__(self
, twitter
, options
):
611 rate
= twitter
.application
.rate_limit_status()
612 print("Remaining API requests: %s / %s (hourly limit)" % (rate
['remaining_hits'], rate
['hourly_limit']))
613 print("Next reset in %ss (%s)" % (int(rate
['reset_time_in_seconds'] - time
.time()),
614 time
.asctime(time
.localtime(rate
['reset_time_in_seconds']))))
617 'authorize' : DoNothingAction
,
618 'follow' : FollowAction
,
619 'friends' : FriendsAction
,
620 'list' : ListsAction
,
621 'mylist' : MyListsAction
,
623 'leave' : LeaveAction
,
624 'pyprompt' : PythonPromptAction
,
625 'replies' : RepliesAction
,
626 'search' : SearchAction
,
627 'set' : SetStatusAction
,
628 'shell' : TwitterShell
,
629 'rate' : RateLimitStatus
,
632 def loadConfig(filename
):
633 options
= dict(OPTIONS
)
634 if os
.path
.exists(filename
):
635 cp
= SafeConfigParser()
637 for option
in ('format', 'prompt'):
638 if cp
.has_option('twitter', option
):
639 options
[option
] = cp
.get('twitter', option
)
641 for option
in ('invert_split',):
642 if cp
.has_option('twitter', option
):
643 options
[option
] = cp
.getboolean('twitter', option
)
646 def main(args
=sys
.argv
[1:]):
649 parse_args(args
, arg_options
)
650 except GetoptError
as e
:
651 print("I can't do that, %s." % (e
), file=sys
.stderr
)
652 print(file=sys
.stderr
)
655 config_path
= os
.path
.expanduser(
656 arg_options
.get('config_filename') or OPTIONS
.get('config_filename'))
657 config_options
= loadConfig(config_path
)
659 # Apply the various options in order, the most important applied last.
660 # Defaults first, then what's read from config file, then command-line
662 options
= dict(OPTIONS
)
663 for d
in config_options
, arg_options
:
664 for k
, v
in list(d
.items()):
667 if options
['refresh'] and options
['action'] not in (
668 'friends', 'replies'):
669 print("You can only refresh the friends or replies actions.", file=sys
.stderr
)
670 print("Use 'twitter -h' for help.", file=sys
.stderr
)
673 oauth_filename
= os
.path
.expanduser(options
['oauth_filename'])
675 if (options
['action'] == 'authorize'
676 or not os
.path
.exists(oauth_filename
)):
678 "the Command-Line Tool", CONSUMER_KEY
, CONSUMER_SECRET
,
679 options
['oauth_filename'])
682 ansiFormatter
= ansi
.AnsiCmd(options
["force-ansi"])
684 oauth_token
, oauth_token_secret
= read_token_file(oauth_filename
)
688 oauth_token
, oauth_token_secret
, CONSUMER_KEY
, CONSUMER_SECRET
),
689 secure
=options
['secure'],
691 domain
='api.twitter.com')
694 Action()(twitter
, options
)
695 except NoSuchActionError
as e
:
696 print(e
, file=sys
.stderr
)
698 except TwitterError
as e
:
699 print(str(e
), file=sys
.stderr
)
700 print("Use 'twitter -h' for help.", file=sys
.stderr
)