Add unsubscribe script

twitivity.py

@@ -0,0 +1,49 @@
+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))

unsubscribe.py

@@ -0,0 +1,41 @@
+import configparser
+import sys
+
+import tweepy
+
+from twitivity import Twitivity, TwitivityError
+
+
+config = configparser.ConfigParser()
+config.read('settings.cfg')
+
+
+consumer_key = config.get('Twitter', 'consumer_key')
+consumer_secret = config.get('Twitter', 'consumer_secret')
+environment_name = config.get('Twitter', 'environment_name')
+
+
+def unsubscribe(access_token, access_token_secret):
+    handler = tweepy.OAuthHandler(consumer_key, consumer_secret)
+    handler.set_access_token(access_token, access_token_secret)
+    twitivity = Twitivity(tweepy.API(handler), environment_name)
+    try:
+        twitivity.check_subscription()
+    except TwitivityError:
+        return
+    twitivity.unsubscribe()
+
+
+def main():
+    try:
+        unsubscribe(*sys.argv[1:3])
+    except TypeError:
+        print('Not enough arguments!')
+    except TwitivityError as e:
+        print(str(e))
+    else:
+        print('Success.')
+    
+
+if __name__ == '__main__':
+    main()