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