You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
50 lines
1.5 KiB
50 lines
1.5 KiB
4 years ago
|
import requests
|
||
|
|
|
||
|
|
|
||
|
|
class Twitivity:
|
||
|
|
def __init__(self, api, environment):
|
||
|
|
self.api = api
|
||
|
|
self.environment = environment
|
||
|
|
|
||
|
|
def _call(self, method, endpoint, data=None):
|
||
|
|
url = 'https://{0}{1}/account_activity/all/{2}'.format(self.api.host, self.api.api_root, self.environment)
|
||
|
|
r = requests.request(method, '{0}/{1}.json'.format(url, endpoint), auth=self.api.auth.apply_auth(), data=data)
|
||
|
|
if not r.ok:
|
||
|
|
raise TwitivityError(r.json())
|
||
|
|
if r.status_code == 204:
|
||
|
|
return None
|
||
|
|
return r.json()
|
||
|
|
|
||
|
|
def check_webhook(self):
|
||
|
|
data = self._call('GET', 'webhooks')
|
||
|
|
if not data:
|
||
|
|
return None
|
||
|
|
return data.pop().get('id')
|
||
|
|
|
||
|
|
def register_webhook(self, url):
|
||
|
|
data = self._call('POST', 'webhooks', data=dict(url=url))
|
||
|
|
if not data:
|
||
|
|
return None
|
||
|
|
return data.get('id')
|
||
|
|
|
||
|
|
def remove_webhook(self, id):
|
||
|
|
data = self._call('DELETE', 'webhooks/{0}'.format(id))
|
||
|
|
|
||
|
|
def poke_webhook(self, id):
|
||
|
|
data = self._call('PUT', 'webhooks/{0}'.format(id))
|
||
|
|
|
||
|
|
def check_subscription(self):
|
||
|
|
self._call('GET', 'subscriptions')
|
||
|
|
|
||
|
|
def subscribe(self):
|
||
|
|
self._call('POST', 'subscriptions')
|
||
|
|
|
||
|
|
def unsubscribe(self):
|
||
|
|
self._call('DELETE', 'subscriptions')
|
||
|
|
|
||
|
|
|
||
|
|
class TwitivityError(Exception):
|
||
|
|
def __init__(self, data):
|
||
|
|
message = data.get('errors', [dict(message='Unknown error.')]).pop(0).get('message')
|
||
|
|
super().__init__('Twitter Account Activity API error: {0}'.format(message))
|