Cin7's templates for both internal and external academies. Notes for Cin7 and G2. Wrote a backup script for Anthology in case the CSV workflow fails.
This commit is contained in:
Norm Rasmussen
@ -1,7 +1,9 @@
|
||||
import requests
|
||||
import asyncio
|
||||
import csv
|
||||
import pandas as pd
|
||||
import time
|
||||
import Apikeys
|
||||
from termcolor import colored
|
||||
|
||||
|
||||
KNOWLEDGEGROUPS = [
|
||||
@ -154,52 +156,111 @@ APIKEY = Apikeys.ANTHOLOGY
|
||||
BASEURL = "https://api.northpass.com/v2/"
|
||||
HEADERS = {"accept": "application/json", "X-Api-Key": APIKEY}
|
||||
GROUPS = []
|
||||
FINISH_SIGNAL = len(PERSON_IDS)
|
||||
COUNT = 0
|
||||
BASEFILE = "/Users/normrasmussen/Downloads/Anthology_master_backup.csv"
|
||||
|
||||
# Get people with groups and remove them from those groups
|
||||
for person in PERSON_IDS:
|
||||
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
|
||||
|
||||
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:
|
||||
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:
|
||||
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:
|
||||
print(f"Error: {response.status_code} with {name}")
|
||||
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.
|
||||
if COUNT == FINISH_SIGNAL:
|
||||
# Since we're de-enrolling one by one, let's sleep and wait.
|
||||
|
||||
print("Sleeping for 30 seconds.")
|
||||
time.sleep(30)
|
||||
print("Sleep Complete. Hold on to your butts!")
|
||||
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_IDS, "group_ids": KNOWLEDGEGROUPS}}
|
||||
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(f"Error: {response.status_code}. Exception: {err}")
|
||||
except requests.exceptions.Timeout:
|
||||
print("Timeout Error")
|
||||
except requests.exceptions.TooManyRedirects:
|
||||
print("Too Many Redirects Error")
|
||||
finally:
|
||||
print(response.status_code)
|
||||
print("Okay, let's see how that went.")
|
||||
# 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()
|
||||
|
||||
Reference in New Issue
Block a user