import csv from io import StringIO import requests BASE_URL = "https://api.membershipworks.com/v1/" # extracted from `SF._is.crm` in https://cdn.membershipworks.com/all.js CRM = {0: "Note", 4: "Profile Updated", 8: "Scheduled/Reminder Email", 9: "Renewal Notice", 10: "Join Date", 11: "Next Renewal Date", 12: "Membership Payment", 13: "Donation", 14: "Event Activity", 15: "Conversation", 16: "Contact Change", 17: "Label Change", 18: "Other Payment", 19: "Cart Payment", 20: "Payment Failed", 21: "Billing Updated", 22: "Form Checkout", 23: "Event Payment", 24: "Invoice", 25: "Invoice Payment", 26: "Renewal", 27: "Payment"} # Types of fields, extracted from a html snippet in all.js + some guessing typ = { 1: "Text input", 2: "Password", # inferred from data 3: "Simple text area", 4: "Rich text area", 7: "Address", 8: "Check box", 9: "Select", 11: "Display value stored in field (ie. read only)", 12: "Required waiver/terms"} # more constants, this time extracted from the members csv export in all.js staticFlags = { "pos": {"lbl": "Position/relation (contacts)"}, "nte": {"lbl": "Admin note (contacts)"}, "pwd": {"lbl": "Password"}, "lgo": {"lbl": "Business card image URLs"}, "pfx": {"lbl": "Profile gallery image URLs"}, "lvl": {"lbl": "Membership levels"}, "aon": {"lbl": "Membership add-ons"}, "lbl": {"lbl": "Labels"}, "joi": {"lbl": "Join date"}, "end": {"lbl": "Renewal date"}, "spy": {"lbl": "Billing method"}, "rid": {"lbl": "Auto recurring billing ID"}, "ipa": {"lbl": "IP address"}, "_id": {"lbl": "Account ID"} } class MembershipWorksRemoteError(Exception): def __init__(self, reason, r): super().__init__( f"Error when attempting {reason}: {r.status_code} {r.reason}\n{r.text}") class MembershipWorks: def __init__(self): self.org_info = None self.auth_token = None self.org_num = None def login(self, username, password): """Authenticate against the membershipworks api""" r = requests.post(BASE_URL + 'usr', data={"_st": "all", "eml": username, "org": "10000", "pwd": password}) if r.status_code != 200 or 'SF' not in r.json(): raise MembershipWorksRemoteError('login', r) self.org_info = r.json() self.auth_token = self.org_info['SF'] self.org_num = self.org_info['org'] def _inject_auth(self, kwargs): # TODO: should probably be a decorator or something if self.auth_token is None: raise RuntimeError('Not Logged in to MembershipWorks') # add auth token to params if 'params' not in kwargs: kwargs['params'] = {} kwargs['params']["SF"] = self.auth_token def _get(self, *args, **kwargs): self._inject_auth(kwargs) # TODO: should probably do some error handling in here return requests.get(*args, **kwargs) def _post(self, *args, **kwargs): self._inject_auth(kwargs) # TODO: should probably do some error handling in here return requests.post(*args, **kwargs) def _all_fields(self): """Parse out a list of fields from the org data. Is this terrible? Yes. Also, not dissimilar to how MW does it in all.js. """ fields = staticFlags.copy() # TODO: this will take the later option, if the same field # used in mulitple places. I don't know which lbl is used for # csv export # anm: member signup, acc: member manage, adm: admin manage for screen_type in ['anm', 'acc', 'adm']: for box in self.org_info['tpl'][screen_type]: for element in box['box']: if (type(element['dat']) != str): for field in element['dat']: if '_id' in field: if field['_id'] not in fields: fields[field['_id']] = field return fields def _parse_flags(self): """Parse the flags out of the org data. This is terrible, and there might be a better way to do this. """ ret = {"folders": {}, "levels": {}, "addons": {}, "labels": {}} for dek in self.org_info['dek']: # TODO: there must be a better way. this is stupid if dek['dek'] == 1: ret["folders"][dek['lbl']] = dek['_id'] elif 'cur' in dek: ret["levels"][dek['lbl']] = dek['_id'] elif 'mux' in dek: ret["addons"][dek['lbl']] = dek['_id'] else: ret["labels"][dek['lbl']] = dek['_id'] return ret def get_member_ids(self, folders): folder_map = self._parse_flags()["folders"] r = self._get(BASE_URL + "ylp", params={ "lbl": ",".join([folder_map[f] for f in folders]), "org": self.org_num, "var": "_id,nam,ctc"}) if r.status_code != 200 or 'usr' not in r.json(): raise MembershipWorksRemoteError('user listing', r) # get list of member ID matching the search return [user['uid'] for user in r.json()['usr']] # TODO: has issues with aliasing header names: # ex: "Personal Studio Space" Label vs Membership Addon/Field def get_members(self, folders, columns): """ Pull the members csv from the membershipworks api folders: a list of the names of the folders to get (see folder_map in this function for mapping to ids) columns: which columns to get""" ids = self.get_member_ids(folders) # get members CSV # TODO: maybe can just use previous get instead? would return JSON r = self._post(BASE_URL + "csv", data={"_rt": "946702800", # unknown "mux": "", # unknown "tid": ",".join(ids), # ids of members to get "var": columns}) if r.status_code != 200: raise MembershipWorksRemoteError('csv generation', r) return list(csv.DictReader(StringIO(r.text))) def get_transactions(self, start_date, end_date, json=False): """Get the transactions between start_date and end_date Dates can be datetime.date or datetime.datetime json gets a different version of the transactions list, which contains a different set information """ r = self._get(BASE_URL + "csv", params={'crm': '12,13,14,18,19', # transaction types, see CRM **({'txl': ''} if json else {}), 'sdp': start_date.strftime('%s'), 'edp': end_date.strftime('%s')}) if r.status_code != 200: raise MembershipWorksRemoteError('csv generation', r) if json: return r.json() else: return list(csv.DictReader(StringIO(r.text))) def get_all_members(self): """Get all the data for all the members""" folders = self._parse_flags()["folders"].keys() fields = self._all_fields() members = self.get_members(folders, ",".join(fields.keys())) return members