forked from geduldig/TwitterAPI
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathcli.py
More file actions
139 lines (116 loc) · 3.96 KB
/
cli.py
File metadata and controls
139 lines (116 loc) · 3.96 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
"""
A Command-Line Interface to Twitter's REST API and Streaming API.
-----------------------------------------------------------------
Run this command line script with any Twitter endpoint. The json-formatted
response is printed to the console. The script works with both Streaming API and
REST API endpoints.
IMPORTANT: Before using this script, you must enter your Twitter application's OAuth
credentials in TwitterAPI/credentials.txt. Log into http://dev.twitter.com to create
your application.
Examples:
::
python -u -m TwitterAPI.cli -endpoint search/tweets -parameters q=zzz
python -u -m TwitterAPI.cli -endpoint statuses/filter -parameters track=zzz
These examples print the raw json response. You can also print one or more fields
from the response, for instance the tweet 'text' field, like this:
::
python -u -m TwitterAPI.cli -endpoint statuses/filter -parameters track=zzz -fields text
Documentation for all Twitter endpoints is located at:
https://dev.twitter.com/docs/api/1.1
"""
__author__ = "Jonas Geduldig"
__date__ = "June 7, 2013"
__license__ = "MIT"
import argparse
import codecs
import json
import sys
from .TwitterOAuth import TwitterOAuth
from .TwitterAPI import TwitterAPI
def _search(name, obj):
"""Breadth-first search for name in the JSON response and return value."""
q = []
q.append(obj)
while q:
obj = q.pop(0)
if hasattr(obj, '__iter__'):
isdict = isinstance(obj, dict)
if isdict and name in obj:
return obj[name]
for k in obj:
q.append(obj[k] if isdict else k)
else:
return None
def _to_dict(param_list):
"""Convert a list of key=value to dict[key]=value"""
if param_list:
return {
name: value for (
name,
value) in [
param.split('=') for param in param_list]}
else:
return None
if __name__ == '__main__':
# print UTF-8 to the console
try:
# python 3
sys.stdout = codecs.getwriter('utf8')(sys.stdout.buffer)
except:
# python 2
sys.stdout = codecs.getwriter('utf8')(sys.stdout)
parser = argparse.ArgumentParser(
description='Request any Twitter Streaming or REST API endpoint')
parser.add_argument(
'-oauth',
metavar='FILENAME',
type=str,
help='file containing OAuth credentials')
parser.add_argument(
'-endpoint',
metavar='ENDPOINT',
type=str,
help='Twitter endpoint',
required=True)
parser.add_argument(
'-parameters',
metavar='NAME_VALUE',
type=str,
help='parameter NAME=VALUE',
nargs='+')
parser.add_argument(
'-fields',
metavar='NAME',
type=str,
help='print a top-level field in the json response',
nargs='+')
parser.add_argument(
'-indent',
metavar='SPACES',
type=int,
help='number of spaces to indent json output',
default=None)
args = parser.parse_args()
try:
params = _to_dict(args.parameters)
oauth = TwitterOAuth.read_file(args.oauth)
api = TwitterAPI(
oauth.consumer_key,
oauth.consumer_secret,
oauth.access_token_key,
oauth.access_token_secret)
response = api.request(args.endpoint, params)
for item in response.get_iterator():
if 'message' in item:
print('ERROR %s: %s' % (item['code'], item['message']))
elif not args.fields:
print(json.dumps(item, ensure_ascii='False', indent=args.indent))
else:
for name in args.fields:
value = _search(name, item)
if value:
print('%s: %s' % (name, value))
except KeyboardInterrupt:
print('\nTerminated by user')
except Exception as e:
print('*** STOPPED %s' % str(e))