]>
jfr.im git - z_archive/twitter.git/blob - twitter/cmdline.py
59113de35c26ae928da52803b2a2a171fb547895
6 twitter [action] [options]
10 authorize authorize the command-line tool to interact with Twitter
11 follow add the specified user to your follow list
12 friends get latest tweets from your friends (default action)
13 help print this help text that you are currently reading
14 leave remove the specified user from your following list
15 public get latest public tweets
16 list get list of user lists
17 replies get latest replies
18 search search twitter (Beware: octothorpe, escape it)
19 set set your twitter status
20 shell login the twitter shell
25 -r --refresh run this command forever, polling every once
26 in a while (default: every 5 minutes)
27 -R --refresh-rate <rate> set the refresh rate (in seconds)
28 -f --format <format> specify the output format for status updates
29 -c --config <filename> read username and password from given config
30 file (default ~/.twitter)
31 -l --length <count> specify number of status updates shown
32 (default: 20, max: 200)
33 -t --timestamp show time before status lines
34 -d --datestamp shoe date before status lines
35 --no-ssl use HTTP instead of more secure HTTPS
36 --oauth <filename> filename to read/store oauth credentials to
38 FORMATS for the --format option
40 default one line per status
41 verbose multiple lines per status, more verbose status info
43 ansi ansi colour (rainbow mode)
48 The config file should be placed in your home directory and be named .twitter.
49 It must contain a [twitter] header, and all the desired options you wish to
53 format: <desired_default_format_for_output>
54 prompt: <twitter_shell_prompt e.g. '[cyan]twitter[R]> '>
56 OAuth authentication tokens are stored in the file .twitter_oauth in your
60 CONSUMER_KEY
='uS6hO2sV6tDKIOeVjhnFnQ'
61 CONSUMER_SECRET
='MEYTOS97VvlHX7K1rwHPEqVpTSqZ71HtvoK4sVuYk'
65 from getopt
import gnu_getopt
as getopt
, GetoptError
66 from getpass
import getpass
69 from ConfigParser
import SafeConfigParser
71 from urllib
import quote
74 from api
import Twitter
, TwitterError
75 from oauth
import OAuth
, write_token_file
, read_token_file
76 from oauth_dance
import oauth_dance
84 'prompt': '[cyan]twitter[R]> ',
85 'config_filename': os
.environ
.get('HOME', '') + os
.sep
+ '.twitter',
86 'oauth_filename': os
.environ
.get('HOME', '') + os
.sep
+ '.twitter_oauth',
94 def parse_args(args
, options
):
95 long_opts
= ['help', 'format=', 'refresh', 'oauth=',
96 'refresh-rate=', 'config=', 'length=', 'timestamp',
97 'datestamp', 'no-ssl']
98 short_opts
= "e:p:f:h?rR:c:l:td"
99 opts
, extra_args
= getopt(args
, short_opts
, long_opts
)
101 for opt
, arg
in opts
:
102 if opt
in ('-f', '--format'):
103 options
['format'] = arg
104 elif opt
in ('-r', '--refresh'):
105 options
['refresh'] = True
106 elif opt
in ('-R', '--refresh-rate'):
107 options
['refresh_rate'] = int(arg
)
108 elif opt
in ('-l', '--length'):
109 options
["length"] = int(arg
)
110 elif opt
in ('-t', '--timestamp'):
111 options
["timestamp"] = True
112 elif opt
in ('-d', '--datestamp'):
113 options
["datestamp"] = True
114 elif opt
in ('-?', '-h', '--help'):
115 options
['action'] = 'help'
116 elif opt
in ('-c', '--config'):
117 options
['config_filename'] = arg
118 elif opt
== '--no-ssl':
119 options
['secure'] = False
120 elif opt
== '--oauth':
121 options
['oauth_filename'] = arg
123 if extra_args
and not ('action' in options
and options
['action'] == 'help'):
124 options
['action'] = extra_args
[0]
125 options
['extra_args'] = extra_args
[1:]
127 def get_time_string(status
, options
, format
="%a %b %d %H:%M:%S +0000 %Y"):
128 timestamp
= options
["timestamp"]
129 datestamp
= options
["datestamp"]
130 t
= time
.strptime(status
['created_at'], format
)
131 i_hate_timezones
= time
.timezone
133 i_hate_timezones
= time
.altzone
134 dt
= datetime
.datetime(*t
[:-3]) - datetime
.timedelta(
135 seconds
=i_hate_timezones
)
137 if timestamp
and datestamp
:
138 return time
.strftime("%Y-%m-%d %H:%M:%S ", t
)
140 return time
.strftime("%H:%M:%S ", t
)
142 return time
.strftime("%Y-%m-%d ", t
)
145 class StatusFormatter(object):
146 def __call__(self
, status
, options
):
147 return (u
"%s%s %s" %(
148 get_time_string(status
, options
),
149 status
['user']['screen_name'], status
['text']))
151 class AnsiStatusFormatter(object):
153 self
._colourMap
= ansi
.ColourMap()
155 def __call__(self
, status
, options
):
156 colour
= self
._colourMap
.colourFor(status
['user']['screen_name'])
157 return (u
"%s%s%s%s %s" %(
158 get_time_string(status
, options
),
159 ansi
.cmdColour(colour
), status
['user']['screen_name'],
160 ansi
.cmdReset(), status
['text']))
162 class VerboseStatusFormatter(object):
163 def __call__(self
, status
, options
):
164 return (u
"-- %s (%s) on %s\n%s\n" %(
165 status
['user']['screen_name'],
166 status
['user']['location'],
167 status
['created_at'],
170 class URLStatusFormatter(object):
171 urlmatch
= re
.compile(r
'https?://\S+')
172 def __call__(self
, status
, options
):
173 urls
= self
.urlmatch
.findall(status
['text'])
174 return u
'\n'.join(urls
) if urls
else ""
176 class ListsFormatter(object):
177 def __call__(self
, list):
178 if list['description']:
179 list_str
= u
"%-30s (%s)" % (list['name'], list['description'])
181 list_str
= u
"%-30s" % (list['name'])
182 return u
"%s\n" % list_str
184 class ListsVerboseFormatter(object):
185 def __call__(self
, list):
186 list_str
= u
"%-30s\n description: %s\n members: %s\n mode:%s\n" % (list['name'], list['description'], list['member_count'], list['mode'])
189 class AnsiListsFormatter(object):
191 self
._colourMap
= ansi
.ColourMap()
193 def __call__(self
, list):
194 colour
= self
._colourMap
.colourFor(list['name'])
195 return (u
"%s%-15s%s %s" %(
196 ansi
.cmdColour(colour
), list['name'],
197 ansi
.cmdReset(), list['description']))
200 class AdminFormatter(object):
201 def __call__(self
, action
, user
):
202 user_str
= u
"%s (%s)" %(user
['screen_name'], user
['name'])
203 if action
== "follow":
204 return u
"You are now following %s.\n" %(user_str)
206 return u
"You are no longer following %s.\n" %(user_str)
208 class VerboseAdminFormatter(object):
209 def __call__(self
, action
, user
):
210 return(u
"-- %s: %s (%s): %s" % (
211 "Following" if action
== "follow" else "Leaving",
216 class SearchFormatter(object):
217 def __call__(self
, result
, options
):
219 get_time_string(result
, options
, "%a, %d %b %Y %H:%M:%S +0000"),
220 result
['from_user'], result
['text']))
222 class VerboseSearchFormatter(SearchFormatter
):
223 pass #Default to the regular one
225 class URLSearchFormatter(object):
226 urlmatch
= re
.compile(r
'https?://\S+')
227 def __call__(self
, result
, options
):
228 urls
= self
.urlmatch
.findall(result
['text'])
229 return u
'\n'.join(urls
) if urls
else ""
231 class AnsiSearchFormatter(object):
233 self
._colourMap
= ansi
.ColourMap()
235 def __call__(self
, result
, options
):
236 colour
= self
._colourMap
.colourFor(result
['from_user'])
237 return (u
"%s%s%s%s %s" %(
238 get_time_string(result
, options
, "%a, %d %b %Y %H:%M:%S +0000"),
239 ansi
.cmdColour(colour
), result
['from_user'],
240 ansi
.cmdReset(), result
['text']))
242 _term_encoding
= None
243 def get_term_encoding():
244 global _term_encoding
245 if not _term_encoding
:
246 lang
= os
.getenv('LANG', 'unknown.UTF-8').split('.')
248 _term_encoding
= lang
[1]
250 _term_encoding
= 'UTF-8'
251 return _term_encoding
254 status_formatters
= {
255 'default': StatusFormatter
,
256 'verbose': VerboseStatusFormatter
,
257 'urls': URLStatusFormatter
,
258 'ansi': AnsiStatusFormatter
260 formatters
['status'] = status_formatters
263 'default': AdminFormatter
,
264 'verbose': VerboseAdminFormatter
,
265 'urls': AdminFormatter
,
266 'ansi': AdminFormatter
268 formatters
['admin'] = admin_formatters
270 search_formatters
= {
271 'default': SearchFormatter
,
272 'verbose': VerboseSearchFormatter
,
273 'urls': URLSearchFormatter
,
274 'ansi': AnsiSearchFormatter
276 formatters
['search'] = search_formatters
279 'default': ListsFormatter
,
280 'verbose': ListsVerboseFormatter
,
282 'ansi': AnsiListsFormatter
284 formatters
['lists'] = lists_formatters
286 def get_formatter(action_type
, options
):
287 formatters_dict
= formatters
.get(action_type
)
288 if (not formatters_dict
):
290 "There was an error finding a class of formatters for your type (%s)"
292 f
= formatters_dict
.get(options
['format'])
295 "Unknown formatter '%s' for status actions" %(options
['format']))
298 class Action(object):
300 def ask(self
, subject
='perform this action', careful
=False):
302 Requests fromt he user using `raw_input` if `subject` should be
303 performed. When `careful`, the default answer is NO, otherwise YES.
304 Returns the user answer in the form `True` or `False`.
310 prompt
= 'You really want to %s %s? ' %(subject
, sample
)
312 answer
= raw_input(prompt
).lower()
314 return answer
in ('yes', 'y')
316 return answer
not in ('no', 'n')
318 print >>sys
.stderr
# Put Newline since Enter was never pressed
320 # Figure out why on OS X the raw_input keeps raising
321 # EOFError and is never able to reset and get more input
322 # Hint: Look at how IPython implements their console
328 def __call__(self
, twitter
, options
):
329 action
= actions
.get(options
['action'], NoSuchAction
)()
331 doAction
= lambda : action(twitter
, options
)
332 if (options
['refresh'] and isinstance(action
, StatusAction
)):
335 time
.sleep(options
['refresh_rate'])
338 except KeyboardInterrupt:
339 print >>sys
.stderr
, '\n[Keyboard Interrupt]'
342 class NoSuchActionError(Exception):
345 class NoSuchAction(Action
):
346 def __call__(self
, twitter
, options
):
347 raise NoSuchActionError("No such action: %s" %(options
['action']))
349 def printNicely(string
):
350 if sys
.stdout
.encoding
:
351 print string
.encode(sys
.stdout
.encoding
, 'replace')
353 print string
.encode('utf-8')
355 class StatusAction(Action
):
356 def __call__(self
, twitter
, options
):
357 statuses
= self
.getStatuses(twitter
, options
)
358 sf
= get_formatter('status', options
)
359 for status
in statuses
:
360 statusStr
= sf(status
, options
)
361 if statusStr
.strip():
362 printNicely(statusStr
)
364 class SearchAction(Action
):
365 def __call__(self
, twitter
, options
):
366 # We need to be pointing at search.twitter.com to work, and it is less
367 # tangly to do it here than in the main()
368 twitter
.domain
="search.twitter.com"
370 # We need to bypass the TwitterCall parameter encoding, so we
371 # don't encode the plus sign, so we have to encode it ourselves
372 query_string
= "+".join(
373 [quote(term
.decode(get_term_encoding()))
374 for term
in options
['extra_args']])
376 results
= twitter
.search(q
=query_string
)['results']
377 f
= get_formatter('search', options
)
378 for result
in results
:
379 resultStr
= f(result
, options
)
380 if resultStr
.strip():
381 printNicely(resultStr
)
383 class AdminAction(Action
):
384 def __call__(self
, twitter
, options
):
385 if not (options
['extra_args'] and options
['extra_args'][0]):
386 raise TwitterError("You need to specify a user (screen name)")
387 af
= get_formatter('admin', options
)
389 user
= self
.getUser(twitter
, options
['extra_args'][0])
390 except TwitterError
, e
:
391 print "There was a problem following or leaving the specified user."
392 print "You may be trying to follow a user you are already following;"
393 print "Leaving a user you are not currently following;"
394 print "Or the user may not exist."
399 printNicely(af(options
['action'], user
))
401 class ListsAction(StatusAction
):
402 def getStatuses(self
, twitter
, options
):
403 screen_name
= twitter
.account
.verify_credentials()['screen_name']
404 if not (options
['extra_args'] and options
['extra_args'][0]):
405 for list in twitter
.user
.lists(user
=screen_name
)['lists']:
406 lf
= get_formatter('lists', options
)
407 printNicely(lf(list))
409 return reversed(twitter
.user
.lists
.list.statuses(user
=screen_name
, list=options
['extra_args'][0]))
411 class FriendsAction(StatusAction
):
412 def getStatuses(self
, twitter
, options
):
413 return reversed(twitter
.statuses
.friends_timeline(count
=options
["length"]))
415 class PublicAction(StatusAction
):
416 def getStatuses(self
, twitter
, options
):
417 return reversed(twitter
.statuses
.public_timeline(count
=options
["length"]))
419 class RepliesAction(StatusAction
):
420 def getStatuses(self
, twitter
, options
):
421 return reversed(twitter
.statuses
.replies(count
=options
["length"]))
423 class FollowAction(AdminAction
):
424 def getUser(self
, twitter
, user
):
425 return twitter
.friendships
.create(id=user
)
427 class LeaveAction(AdminAction
):
428 def getUser(self
, twitter
, user
):
429 return twitter
.friendships
.destroy(id=user
)
431 class SetStatusAction(Action
):
432 def __call__(self
, twitter
, options
):
433 statusTxt
= (" ".join(options
['extra_args']).decode(get_term_encoding())
434 if options
['extra_args']
435 else unicode(raw_input("message: ")))
436 status
= (statusTxt
.encode('utf8', 'replace'))
437 twitter
.statuses
.update(status
=status
)
439 class TwitterShell(Action
):
441 def render_prompt(self
, prompt
):
442 '''Parses the `prompt` string and returns the rendered version'''
443 prompt
= prompt
.strip("'").replace("\\'","'")
444 for colour
in ansi
.COLOURS_NAMED
:
445 if '[%s]' %(colour) in prompt
:
446 prompt
= prompt
.replace(
447 '[%s]' %(colour), ansi
.cmdColourNamed(colour
))
448 prompt
= prompt
.replace('[R]', ansi
.cmdReset())
451 def __call__(self
, twitter
, options
):
452 prompt
= self
.render_prompt(options
.get('prompt', 'twitter> '))
454 options
['action'] = ""
456 args
= raw_input(prompt
).split()
457 parse_args(args
, options
)
458 if not options
['action']:
460 elif options
['action'] == 'exit':
462 elif options
['action'] == 'shell':
463 print >>sys
.stderr
, 'Sorry Xzibit does not work here!'
465 elif options
['action'] == 'help':
466 print >>sys
.stderr
, '''\ntwitter> `action`\n
467 The Shell Accepts all the command line actions along with:
469 exit Leave the twitter shell (^D may also be used)
471 Full CMD Line help is appended below for your convinience.'''
472 Action()(twitter
, options
)
473 options
['action'] = ''
474 except NoSuchActionError
, e
:
475 print >>sys
.stderr
, e
476 except KeyboardInterrupt:
477 print >>sys
.stderr
, '\n[Keyboard Interrupt]'
480 leaving
= self
.ask(subject
='Leave')
482 print >>sys
.stderr
, 'Excellent!'
486 class HelpAction(Action
):
487 def __call__(self
, twitter
, options
):
490 class DoNothingAction(Action
):
491 def __call__(self
, twitter
, options
):
495 'authorize' : DoNothingAction
,
496 'follow' : FollowAction
,
497 'friends' : FriendsAction
,
498 'list' : ListsAction
,
500 'leave' : LeaveAction
,
501 'public' : PublicAction
,
502 'replies' : RepliesAction
,
503 'search' : SearchAction
,
504 'set' : SetStatusAction
,
505 'shell' : TwitterShell
,
508 def loadConfig(filename
):
509 options
= dict(OPTIONS
)
510 if os
.path
.exists(filename
):
511 cp
= SafeConfigParser()
513 for option
in ('format', 'prompt'):
514 if cp
.has_option('twitter', option
):
515 options
[option
] = cp
.get('twitter', option
)
518 def main(args
=sys
.argv
[1:]):
521 parse_args(args
, arg_options
)
522 except GetoptError
, e
:
523 print >> sys
.stderr
, "I can't do that, %s." %(e)
527 config_path
= os
.path
.expanduser(
528 arg_options
.get('config_filename') or OPTIONS
.get('config_filename'))
529 config_options
= loadConfig(config_path
)
531 # Apply the various options in order, the most important applied last.
532 # Defaults first, then what's read from config file, then command-line
534 options
= dict(OPTIONS
)
535 for d
in config_options
, arg_options
:
536 for k
,v
in d
.items():
539 if options
['refresh'] and options
['action'] not in (
540 'friends', 'public', 'replies'):
541 print >> sys
.stderr
, "You can only refresh the friends, public, or replies actions."
542 print >> sys
.stderr
, "Use 'twitter -h' for help."
545 oauth_filename
= os
.path
.expanduser(options
['oauth_filename'])
547 if (options
['action'] == 'authorize'
548 or not os
.path
.exists(oauth_filename
)):
550 "the Command-Line Tool", CONSUMER_KEY
, CONSUMER_SECRET
,
551 options
['oauth_filename'])
553 oauth_token
, oauth_token_secret
= read_token_file(oauth_filename
)
557 oauth_token
, oauth_token_secret
, CONSUMER_KEY
, CONSUMER_SECRET
),
558 secure
=options
['secure'],
560 domain
='api.twitter.com')
563 Action()(twitter
, options
)
564 except NoSuchActionError
, e
:
565 print >>sys
.stderr
, e
567 except TwitterError
, e
:
568 print >> sys
.stderr
, str(e
)
569 print >> sys
.stderr
, "Use 'twitter -h' for help."