103 lines
4.0 KiB
Python
103 lines
4.0 KiB
Python
import csv
|
|
import json
|
|
import urllib3
|
|
import sys
|
|
from io import StringIO
|
|
from xml.etree import ElementTree as ET
|
|
import requests
|
|
|
|
from passwords import *
|
|
|
|
# it's fine, ssl certs are for losers anyway
|
|
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
|
|
|
config = json.load(open("config.json"))
|
|
|
|
ET.register_namespace("hid", "http://www.hidglobal.com/VertX")
|
|
ET.register_namespace("hid", "http://www.hidcorp.com/VertX")
|
|
|
|
fieldnames = "CardNumber,CardFormat,PinRequired,PinCode,ExtendedAccess,ExpiryDate,Forename,Initial,Surname,Email,Phone,Custom1,Custom2,Schedule1,Schedule2,Schedule3,Schedule4,Schedule5,Schedule6,Schedule7,Schedule8".split(",")
|
|
|
|
def getMembershipworksData(columns):
|
|
""" Pull the members csv from the membershipworks api
|
|
columns: which columns to get"""
|
|
BASE_URL = "https://api.membershipworks.com/v1/"
|
|
|
|
# login
|
|
r = requests.post(BASE_URL + 'usr',
|
|
data={"_st": "all",
|
|
"eml": MEMBERSHIPWORKS_USERNAME,
|
|
"org": "10000",
|
|
"pwd": MEMBERSHIPWORKS_PASSWORD})
|
|
if r.status_code != 200 or 'SF' not in r.json():
|
|
print("MembershipWorks Login Error: ", r.status_code, r.reason)
|
|
print(r.text)
|
|
sys.exit(1)
|
|
SFtoken = r.json()['SF']
|
|
|
|
# get list of member/staff IDs
|
|
r = requests.get(BASE_URL + "ylp",
|
|
params={"SF": SFtoken,
|
|
"lbl": "5ae37979f033bfe8534f8799,5771675edcdf126302a2f6b9", # members and staff
|
|
"org": "15475", # unknown
|
|
"var": "_id,nam,ctc"})
|
|
if r.status_code != 200 or 'usr' not in r.json():
|
|
print("MembershipWorks User Listing Error: ", r.status_code, r.reason)
|
|
print(r.text)
|
|
sys.exit(1)
|
|
ids = [user['_id'] for user in r.json()['usr']]
|
|
|
|
# get members CSV
|
|
# TODO: maybe can just use previous get instead? would return JSON
|
|
r = requests.post(BASE_URL + "csv",
|
|
params={"SF": SFtoken},
|
|
data={"_rt": "946702800", # unknown
|
|
"mux": "", # unknown
|
|
"tid": ",".join(ids), # ids of members to get
|
|
"var": columns})
|
|
if r.status_code != 200:
|
|
print("MembershipWorks CSV Generation Error: ", r.status_code, r.reason)
|
|
print(r.text)
|
|
sys.exit(1)
|
|
|
|
return list(csv.DictReader(StringIO(r.text)))
|
|
|
|
def doImportRequest(ip, params=None, files=None):
|
|
"""Send a request to the door control import script"""
|
|
r = requests.post(
|
|
'https://' + ip + '/cgi-bin/import.cgi',
|
|
params=params,
|
|
files=files,
|
|
auth=requests.auth.HTTPDigestAuth(DOOR_USERNAME, DOOR_PASSWORD),
|
|
timeout=60,
|
|
verify=False) # ignore insecure SSL
|
|
xml = ET.XML(r.text)
|
|
if r.status_code != 200 \
|
|
or len(xml.findall("{http://www.hidglobal.com/VertX}Error")) > 0:
|
|
print("Door Updating Error: ", r.status_code, r.reason)
|
|
print(r.text)
|
|
sys.exit(1)
|
|
|
|
def doCSVImport(doorIP, csv):
|
|
"""Do the CSV import procedure on a door control"""
|
|
doImportRequest(doorIP, {"task": "importInit"})
|
|
doImportRequest(doorIP,
|
|
{"task": "importCardsPeople", "name": "cardspeopleschedule.csv"},
|
|
{"importCardsPeopleButton": ("cardspeopleschedule.csv", csv, 'text/csv')})
|
|
doImportRequest(doorIP, {"task": "importDone"})
|
|
|
|
def doXMLRequest(doorIP, xml, prefix=b'<?xml version="1.0" encoding="UTF-8"?>'):
|
|
r = requests.get(
|
|
'https://' + doorIP + '/cgi-bin/vertx_xml.cgi',
|
|
params={'XML': prefix + xml},
|
|
auth=requests.auth.HTTPDigestAuth(DOOR_USERNAME, DOOR_PASSWORD),
|
|
verify=False)
|
|
xml = ET.XML(r.text)
|
|
# probably meed to be more sane about this
|
|
if r.status_code != 200 \
|
|
or len(xml.findall("{http://www.hidglobal.com/VertX}Error")) > 0:
|
|
print("Door Updating Error: ", r.status_code, r.reason)
|
|
print(r.text)
|
|
sys.exit(1)
|
|
return xml
|