forked from CMS/memberPlumbing
83 lines
3.1 KiB
Python
83 lines
3.1 KiB
Python
import csv
|
|
from io import StringIO
|
|
import requests
|
|
|
|
BASE_URL = "https://api.membershipworks.com/v1/"
|
|
|
|
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.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.auth_token = r.json()['SF']
|
|
self.org_num = r.json()['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 get_member_ids(self, folders):
|
|
# TODO: this is hardcoded for CMS
|
|
folder_map = {
|
|
'members': '5ae37979f033bfe8534f8799',
|
|
'staff': '5771675edcdf126302a2f6b9',
|
|
'misc': '5b69ee9bf033bf8e7346c434'
|
|
}
|
|
|
|
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/staff IDs
|
|
return [user['uid'] for user in r.json()['usr']]
|
|
|
|
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)))
|