mailchimp_bible_reading_plan/mailchimpapi.py at master · ChristChurchMayfair/mailchimp_bible_reading_plan · GitHub

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
import httplib2
import json
import pprint
import urllib.parse


class MailchimpAPI:

    def __init__(self, api_token):

        _, region = api_token.split('-')

        self.base_url = 'https://{}.api.mailchimp.com/3.0'.format(region)
        self.api_token = api_token
        self.h = httplib2.Http(".cache")
        self.auth_headers = {'Authorization': 'apikey ' + self.api_token}

    def __fetch_json(self, url):
        (response, content) = self.h.request(url, "GET", headers=self.auth_headers)
        return json.loads(content)

    def __post_json(self, url, payload):
        headers = {'Content-Type': 'application/json; charset=UTF-8'}
        (response, content) = self.h.request(uri=url, method='POST', headers={**headers, **self.auth_headers},
                                             body=json.dumps(payload))

        if response['status'] != '200' and response['status'] != '204':
            print("Got Non 200 response!")
            pprint.pprint(response)
            try:
                json_content = json.loads(content)
                pprint.pprint(json_content)
            except json.decoder.JSONDecodeError:
                print(content)
            return None

        return content

    def __put_json(self, url, payload):
        headers = {'Content-Type': 'application/json; charset=UTF-8'}
        (response, content) = self.h.request(uri=url, method='PUT', headers={**headers, **self.auth_headers},
                                             body=json.dumps(payload))

        if response['status'] != '200' and response['status'] != '204':
            print("Got Non 200 response!")
            pprint.pprint(response)
            try:
                json_content = json.loads(content)
                pprint.pprint(json_content)
            except json.decoder.JSONDecodeError:
                print(content)
            return
        return

    def get_lists(self):
        url = self.base_url + "/lists"
        return self.__fetch_json(url)

    def get_list(self, list_id):
        url = self.base_url + "/lists/" + list_id
        return self.__fetch_json(url)


    def get_campaign_folders(self):
        url = self.base_url + "/campaign-folders"
        return self.__fetch_json(url)

    def create_campaign_folder(self, campaign_folder_name):

        existing_campaign_folders = self.get_campaign_folders()

        name_already_exists = any(folder['name'] == campaign_folder_name for folder in existing_campaign_folders['folders'])
        if name_already_exists:
            existing_campaign_folder = next(folder for folder in existing_campaign_folders['folders'] if folder['name'] == campaign_folder_name)
            return existing_campaign_folder['id']

        url = self.base_url + "/campaign-folders"

        payload = {"name": campaign_folder_name}

        created_campaign_folder = json.loads(self.__post_json(url, payload))
        if created_campaign_folder:
            return created_campaign_folder['id']

    def get_campaigns(self):
        url = self.base_url + "/campaigns?count=500"
        return self.__fetch_json(url)

    def get_campaign_ids_by_name(self, name):
        encoded_name = urllib.parse.quote_plus(name)
        url = self.base_url + '/search-campaigns?query=' + encoded_name
        response = self.__fetch_json(url)
        if 'results' not in response:
            return []
        real_matches = list(filter(lambda result: result['campaign']['settings']['title'] == name, response['results']))
        ids = list(map(lambda result: result['campaign']['id'], real_matches))
        return ids


    def create_campaign(self, campaign_definition):

        campaign_name = campaign_definition['settings']['title']

        existing_ids = self.get_campaign_ids_by_name(campaign_name)

        if len(existing_ids) > 1:
            print("Multiple campaigns ({}) exist with this name: {}".format(existing_ids, campaign_name))
            exit(1)

        if len(existing_ids) == 1:
            print("Campaign named {} already exists.".format(campaign_name))
            return existing_ids[0]

        print("No existing campaign named {} - creating.".format(campaign_name))

        url = self.base_url + "/campaigns"

        created_campaign = json.loads(self.__post_json(url, campaign_definition))
        if created_campaign:
            return created_campaign['id']

    def set_campaign_html_text(self, campaign_id, html_content):
        url = self.base_url + "/campaigns/" + campaign_id + "/content"

        body = {'html': html_content}

        self.__put_json(url, body)

    def send_campaign(self, campaign_id):
        url = self.base_url + '/campaigns/' + campaign_id + '/actions/send'

        response = self.__post_json(url, None)
        return response

    def schedule_campaign_for_datetime(self, campaign_id, datetime):

        url = self.base_url + '/campaigns/' + campaign_id + '/actions/schedule'

        unschedule_url = self.base_url + "/campaigns/" + campaign_id + "/actions/unschedule"

        unschedule_response = self.__post_json(unschedule_url, None)
        print(unschedule_response)

        schedule_defintion = {
            "schedule_time": datetime.isoformat(),
            "timewarp": False,
            "batch_delay": False
        }

        return self.__post_json(url, schedule_defintion)