-
Notifications
You must be signed in to change notification settings - Fork 19
/
bwproject.py
336 lines (278 loc) · 13.1 KB
/
bwproject.py
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
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
"""
bwproject contains the BWUser and BWProject classes
"""
import os
import requests
import time
import logging
logger = logging.getLogger("bwapi")
handler = logging.StreamHandler()
formatter = logging.Formatter("%(asctime)s %(levelname)s: %(message)s", "%H:%M:%S")
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.setLevel(logging.DEBUG)
class BWUser:
"""
This class handles user-level tasks in the Brandwatch API, including authentication and HTTP requests. For tasks which are bound to a project
(e.g. working with queries or groups) use the subclass BWProject instead.
Attributes:
apiurl: Brandwatch API url. All API requests will be appended to this url.
oauthpath: Path to append to the API url to get an access token.
username: Brandwatch username.
password: Brandwatch password.
token: Access token.
"""
def __init__(self, token=None, token_path="tokens.txt", username=None, password=None, grant_type="api-password", client_id="brandwatch-api-client", apiurl="https://api.brandwatch.com/"):
"""
Creates a BWUser object.
Args:
username: Brandwatch username.
password: Brandwatch password - Optional if you already have an access token.
token: Access token - Optional.
token_path: File path to the file where access tokens will be read from and written to - Optional. Defaults to tokens.txt, pass None to disable.
"""
self.apiurl = apiurl
self.oauthpath = "oauth/token"
if token:
self.username, self.token = self._test_auth(username, token)
if token_path is not None:
self._write_auth(token_path)
elif username is not None and password is not None:
self.username, self.token = self._get_auth(username, password, token_path, grant_type, client_id)
if token_path is not None:
self._write_auth(token_path)
elif username is not None:
self.username, self.token = self._read_auth(username, token_path)
else:
raise KeyError("Must provide valid token, username and password, or username and path to token file")
def _test_auth(self, username, token):
user = requests.get(self.apiurl + "me", params={"access_token": token}).json()
if "username" in user:
if username is None:
return user["username"], token
elif user["username"] == username:
return username, token
else:
raise KeyError("Username " + username + " does not match provided token", user)
else:
raise KeyError("Could not validate provided token", user)
def _get_auth(self, username, password, token_path, grant_type, client_id):
token = requests.post(
self.apiurl + self.oauthpath,
params={
"username": username,
"password": password,
"grant_type": grant_type,
"client_id": client_id
}).json()
if "access_token" in token:
return username, token["access_token"]
else:
raise KeyError("Authentication failed", token)
def _read_auth(self, username, token_path):
user_tokens = self._read_auth_file(token_path)
if username in user_tokens:
return self._test_auth(username, user_tokens[username])
else:
raise KeyError("Token not found in file: " + token_path)
def _write_auth(self, token_path):
user_tokens = self._read_auth_file(token_path)
user_tokens[self.username.lower()] = self.token
with open(token_path, "w") as token_file:
token_file.write("\n".join(["\t".join(item) for item in user_tokens.items()]))
def _read_auth_file(self, token_path):
user_tokens = {}
if os.path.isfile(token_path):
with open(token_path) as token_file:
for line in token_file:
try:
user, token = line.split()
except ValueError:
pass
user_tokens[user.lower()] = token
return user_tokens
def get_projects(self):
"""
Gets a list of projects accessible to the user.
Returns:
List of dictionaries, where each dictionary is the information (name, id, clientName, timezone, ....) for one project.
"""
response = self.request(verb=requests.get, address="projects")
return response["results"] if "results" in response else response
def get_self(self):
""" Gets username and id """
return self.request(verb=requests.get, address="me")
def validate_query_search(self, **kwargs):
"""
Checks a query search to see if it contains errors. Same query debugging as used in the front end.
Keyword Args:
query: Search terms included in the query.
language: List of the languages in which you'd like to test the query - Optional.
Raises:
KeyError: If you don't pass a search or if the search has errors in it.
"""
if "query" not in kwargs:
raise KeyError("Must pass: query = 'search terms'")
if "language" not in kwargs:
kwargs["language"] = ["en"]
valid_search = self.request(verb=requests.get, address="query-validation", params=kwargs)
def validate_rule_search(self, **kwargs):
"""
Checks a rule search to see if it contains errors. Same rule debugging as used in the front end.
Keyword Args:
query: Search terms included in the rule.
language: List of the languages in which you'd like to test the query - Optional.
Raises:
KeyError: If you don't pass a search or if the search has errors in it.
"""
if "query" not in kwargs:
raise KeyError("Must pass: query = 'search terms'")
if "language" not in kwargs:
kwargs["language"] = ["en"]
valid_search = self.request(verb=requests.get, address="query-validation/searchwithin", params=kwargs)
def request(self, verb, address, params={}, data={}):
"""
Makes a request to the Brandwatch API.
Args:
verb: Type of request you want to make (e.g. 'requests.get').
address: Address to append to the Brandwatch API url.
params: Any additional parameters - Optional.
data: Any additional data - Optional.
Returns:
The response json
"""
return self.bare_request(verb=verb, address_root=self.apiurl, address_suffix=address, access_token=self.token,
params=params, data=data)
def bare_request(self, verb, address_root, address_suffix, access_token="", params={}, data={}):
"""
Makes a request to the Brandwatch API.
Args:
verb: Type of request you want to make (e.g. 'requests.get').
address_root: In most cases this will the the Brandwatch API url, but we leave the flexibility to change this for a different root address if needed.
address_suffix: Address to append to the root url.
access_token: Access token - Optional.
params: Any additional parameters - Optional.
data: Any additional data - Optional.
Returns:
The response json
"""
time.sleep(.5)
if access_token:
params["access_token"] = access_token
if data == {}:
response = verb(address_root + address_suffix, params=params)
else:
response = verb(address_root + address_suffix,
params=params,
data=data,
headers={"Content-type": "application/json"})
if "errors" in response.json() and response.json()["errors"]:
logger.error("There was an error with this request: \n{}\n{}\n{}".format(response.url, data, response.json()["errors"]))
raise RuntimeError(response.json()["errors"])
logger.debug(response.url)
return response.json()
class BWProject(BWUser):
"""
This class is required for working with project-level resources, such as queries or groups.
Attributes:
project_name: Brandwatch project name.
project_id: Brandwatch project id.
project_address: Path to append to the Brandwatch API url to make any project level calls.
"""
def __init__(self, project, token=None, token_path="tokens.txt", username=None, password=None, grant_type="api-password", client_id="brandwatch-api-client", apiurl="https://api.brandwatch.com/"):
"""
Creates a BWProject object - inheriting directly from the BWUser class.
Args:
username: Brandwatch username.
project: Brandwatch project name.
password: Brandwatch password - Optional if you already have an access token.
token: Access token - Optional.
token_path: File path to the file where access tokens will be read from and written to - Optional.
"""
super().__init__(token=token, token_path=token_path, username=username, password=password, grant_type=grant_type, client_id=client_id, apiurl=apiurl)
self.project_name = ""
self.project_id = -1
self.project_address = ""
self.get_project(project)
def get_project(self, project):
"""
Returns a dictionary of the project information (name, id, clientName, timezone, ....).
Args:
project: Brandwatch project.
"""
projects = self.get_projects()
project_found = False
try:
int(project)
numerical = True
except:
numerical = False
for p in projects:
found = False
if numerical:
if p["id"] == int(project):
found = True
else:
if p["name"] == project:
found = True
if found:
self.project_name = p["name"]
self.project_id = p["id"]
self.project_address = "projects/" + str(self.project_id) + "/"
project_found = True
break
if not project_found:
raise KeyError("Project " + project + " not found")
def get(self, endpoint, params={}):
"""
Makes a project level GET request
Args:
endpoint: Path to append to the Brandwatch project API url. Warning: project information is already included so you don't have to re-append that bit.
params: Additional parameters.
Returns:
Server's response to the HTTP request.
"""
return self.request(verb=requests.get, address=self.project_address + endpoint, params=params)
def delete(self, endpoint, params={}):
"""
Makes a project level DELETE request
Args:
endpoint: Path to append to the Brandwatch project API url. Warning: project information is already included so you don't have to re-append that bit.
params: Additional parameters.
Returns:
Server's response to the HTTP request.
"""
return self.request(verb=requests.delete, address=self.project_address + endpoint, params=params)
def post(self, endpoint, params={}, data={}):
"""
Makes a project level POST request
Args:
endpoint: Path to append to the Brandwatch project API url. Warning: project information is already included so you don't have to re-append that bit.
params: Additional parameters.
data: Additional data.
Returns:
Server's response to the HTTP request.
"""
return self.request(verb=requests.post, address=self.project_address + endpoint, params=params, data=data)
def put(self, endpoint, params={}, data={}):
"""
Makes a project level PUT request
Args:
endpoint: Path to append to the Brandwatch project API url. Warning: project information is already included so you don't have to re-append that bit.
params: Additional parameters.
data: Additional data.
Returns:
Server's response to the HTTP request.
"""
return self.request(verb=requests.put, address=self.project_address + endpoint, params=params, data=data)
def patch(self, endpoint, params={}, data={}):
"""
Makes a project level PATCH request
Args:
endpoint: Path to append to the Brandwatch project API url. Warning: project information is already included so you don't have to re-append that bit.
params: Additional parameters.
data: Additional data.
Returns:
Server's response to the HTTP request.
"""
return self.request(verb=requests.patch, address=self.project_address + endpoint, params=params, data=data)