127 lines
4.3 KiB
Python
127 lines
4.3 KiB
Python
import requests
|
|
import csv
|
|
import pandas as pd
|
|
import time
|
|
import Apikeys
|
|
from termcolor import colored
|
|
|
|
|
|
KNOWLEDGEGROUPS = [
|
|
]
|
|
|
|
PERSON_IDS = [
|
|
]
|
|
|
|
APIKEY = Apikeys.ANTHOLOGY
|
|
BASEURL = "https://api.northpass.com/v2/"
|
|
HEADERS = {"accept": "application/json", "X-Api-Key": APIKEY}
|
|
GROUPS = []
|
|
BASEFILE = "/Users/normrasmussen/Downloads/Anthology_master_backup.csv"
|
|
|
|
|
|
def grab_person_group_ids():
|
|
file = pd.read_csv(BASEFILE)
|
|
for row in file.itertuples():
|
|
domain = f"@{row[1]}"
|
|
groups = row[2:]
|
|
tmplist = []
|
|
for group in groups:
|
|
group = str(group)
|
|
if "nan" not in group:
|
|
# Grab Group UUIDs
|
|
url = BASEURL + f"groups?filter[name][eq]={group}"
|
|
response = requests.get(url, headers=HEADERS)
|
|
response = response.json()
|
|
data = response["data"]
|
|
for name in data:
|
|
id = name["id"]
|
|
tmplist.append(id)
|
|
rowdict = {domain: tmplist}
|
|
grab_ppl_ids(rowdict)
|
|
|
|
|
|
def grab_ppl_ids(rowdict):
|
|
print colored (f"The dictionary is grabbing all the people. Here's the dict: {rowdict}", 'green')
|
|
page_count = 0
|
|
for domain, group_list in rowdict.items():
|
|
# Grab all people
|
|
person_list = []
|
|
page_count += 1
|
|
url = (
|
|
BASEURL
|
|
+ f"people/?filter[email][cont]={domain}&limit=100&page={page_count}"
|
|
)
|
|
response = requests.get(url, headers=HEADERS)
|
|
response = response.json()
|
|
nextlink = response["links"]
|
|
|
|
for data in response["data"]:
|
|
person = data["id"]
|
|
person_list.append(person)
|
|
|
|
if "next" not in nextlink:
|
|
break
|
|
|
|
if len(person_list) > 0:
|
|
print colored(f"Person list for {domain} has {len(person_list)} people.", 'blue')
|
|
bulk_remove_and_enroll(person_list, group_list)
|
|
else:
|
|
print colored(f"Person list for {domain} has {len(person_list)} people.", 'blue')
|
|
print colored(f"Skipping the bulk function.", 'yellow')
|
|
|
|
|
|
def bulk_remove_and_enroll(person_list, group_list):
|
|
print colored("Moving people and groups into bulk function.", 'green')
|
|
COUNT = 0
|
|
FINISH_SIGNAL = len(person_list)
|
|
# Get people with groups and remove them from those groups
|
|
for person in person_list:
|
|
COUNT += 1
|
|
url = BASEURL + f"people/{person}"
|
|
response = requests.get(url, headers=HEADERS)
|
|
data = response.json()
|
|
groups = data["data"]["relationships"]["groups"]
|
|
name = data["data"]["attributes"]["full_name"]
|
|
del_group_list = []
|
|
for group in groups["data"]:
|
|
del_payload_var = {"id": group["id"], "type": "membership-groups"}
|
|
del_group_list.append(del_payload_var)
|
|
del_payload_base = {"data": del_group_list}
|
|
if not del_payload_base["data"]:
|
|
pass
|
|
else:
|
|
url = BASEURL + f"people/{person}/relationships/groups"
|
|
response = requests.delete(url, headers=HEADERS, json=del_payload_base)
|
|
good_status_codes = [202, 204, 200, 203]
|
|
if response.status_code in good_status_codes:
|
|
pass
|
|
else:
|
|
print colored(f"Error: {response.status_code} with {name}", 'red')
|
|
|
|
if COUNT == FINISH_SIGNAL:
|
|
# Since we're de-enrolling one by one, let's sleep and wait.
|
|
|
|
print colored("Sleeping for 2 seconds.", 'purple')
|
|
time.sleep(2)
|
|
print colored("Sleep Complete. Hold on to your butts!", 'purple')
|
|
|
|
# Re-enroll everyone back into all the groups
|
|
payload = {"payload": {"person_ids": person_list, "group_ids": group_list}}
|
|
url = BASEURL + "bulk/people/membership/"
|
|
try:
|
|
response = requests.post(url, headers=HEADERS, json=payload)
|
|
response.raise_for_status()
|
|
except requests.exceptions.HTTPError as err:
|
|
print colored(f"Error: {response.status_code}. Exception: {err}", 'orange')
|
|
except requests.exceptions.Timeout:
|
|
print colored("Timeout Error", 'orange')
|
|
except requests.exceptions.TooManyRedirects:
|
|
print colored("Too Many Redirects Error", 'orange')
|
|
finally:
|
|
print colored(response.status_code, 'yellow')
|
|
print colored("Okay, let's see how that went.", 'orange')
|
|
|
|
|
|
if __name__ == "__main__":
|
|
grab_person_group_ids()
|