norm/Gainsight
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Cleaned up and minmized Chubb's CISA script to share with the client so they can run it on their own

Browse Source
This commit is contained in:
Norm Rasmussen
2025-09-26 14:49:46 -04:00
parent 29d3db5c20
commit 06553b93c3
2 changed files with 178 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

175
Scripts/API_Tests/bulk_invite_and_props-minimal.py Normal file
View File

@ -0,0 +1,175 @@
import pandas as pd
import requests
import time
"""
Note that all single-line hash comments are referencing the code _below_ the comment line.
"""
APIKEY = "INSERT API KEY BETWEEN QUOTES"
HEADERS = {
"accept": "application/json",
"X-Api-Key": APIKEY,
}
BASEURL = "https://api.northpass.com/v2/"
IMPORTFILE = "INSERT THE PATH TO THE CSV ON YOUR LOCAL MACHINE"
def bulk_invite_and_group():
"""
Bulk endpoint which invites new people and adds them to a group via this structure:
{
"email": "me@mac.com"
"groups": "GroupA"
}
This function looks for the group in the CSV as well.
"""
# This first line reads the CSV using pandas and puts the dataframe into a variable
data = pd.read_csv(IMPORTFILE)
# Grab all unique values from the Group column. It should only be 3 values (the 3 groups you're adding members to)
groups = data["Group"].unique()
# Turn it into a list so we can loop through it.
groups = list(groups)
# Outputting status updates for your reference
print("Here are all groups within the CSV:")
print(groups)
print(" ")
# Now we loop through the groups
for group in groups:
payload = ""
# You can comment this out if you want. This just confirms the loops is moving through all the groups.
print(group)
# Creating a temporary variable for one group at a time
tmp_group = data[data.Group == group]
# For that group, grab all the emails using the column called "Email"
people = list(tmp_group["Email"])
# Making sure the emails don't have commas or extra spaces.
group = str(tmp_group["Group"].unique())[2:-2]
# Show you how many people are in that group! We're making progress!
print(f"Group --> {group} ... Amount of People --> {len(people)}")
# Now let's prepare the actual API call.
url = f"{BASEURL}bulk/people"
# The reason we are checking for length is because our bulk endpoints can't take more than 1500 parameters at a time.
# So if the amount of people in this group (remember, we're still in the for loop!) is greater than 1500, we'll have to break it up
if len(people) > 1500:
# Take 1500 people at a time, and add them to a mini payload (called mini load) and then we'll insert that into the bigger payload.
for chunk in range(0, len(people), 1500):
i = chunk
payload_1 = []
i_to_add = people[i : i + 1500]
for person in i_to_add:
miniload = {"email": person, "groups": group}
payload_1.append(miniload)
# This print statement should show 1500 people.
print(f"The long length {group} payload has {len(payload_1)}")
# Here's where we add the mini-payload list to the formatted payload that the API needs
payload = {"data": {"attributes": {"people": payload_1}}}
# And here we go! Making the call to our endpoint
response = requests.post(url, headers=HEADERS, json=payload)
# Printing the status code for your awareness
print(f"Completed. Status code is {response.status_code}")
# What if the amount of all the people per group is less than 1500? Well, let's run this code block/else statement!
else:
payload_1 = []
# Pretty straight forward, we don't need to do any chunking, so just add everyone to the mini-payload, as above.
for person in people:
miniload = {"email": person, "groups": group}
payload_1.append(miniload)
# Same print statement as above for continuity.
print(f"The {group} payload has {len(payload_1)}")
# Constructing the payload the API endpoint expects!
payload = {"data": {"attributes": {"people": payload_1}}}
# And here we go again! Pushing the data to your academy.
response = requests.post(url, headers=HEADERS, json=payload)
# Showing you the status code - you should expect 200.
print(f"Completed. Status code is {response.status_code}")
print(response.text)
# Next, we need to add all the properties (the Agency Name) to each person.
# The reason why we use time.sleep() is to just give the application a few minutes to fully add everyone.
# It's not really needed, but better to be safe than sorry
print("Running add props from func...")
time.sleep(3)
add_props_from_func(people, data, group)
def add_props_from_func(people, data, group):
# Creating an empty list of emails that cause errors, to be filled if an error arises.
errorlist = []
# Move through all the people
for learner_email in people:
# Grab the Agency name from the original dataframe, looking for the column called "AgencyName"
agency_name = data.loc[data["Email"] == learner_email, "AgencyName"]
# Clean up white space and commas
agname = str(agency_name.values)[2:-2]
# Print out confirmation that the email, agency, and group all align.
print(f"Learner: {learner_email} --> Agency: {agname} from Group: {group}")
# We first need the learner's UUID since that isn't passed in the CSV and is required for the properties endpoint.
ppl_search = f"{BASEURL}people?filter[email][eq]={learner_email}"
ppl_response = requests.get(ppl_search, headers=HEADERS)
# Let's give it a shot to see if the person exists (they should, we just added them!)
# If they exist, grab the ID and go straight into adding their properties via a different endpoint.
try:
ppl_data = ppl_response.json()
learner_uuid = ppl_data["data"][0]["id"]
# Now update the props
prop_url = f"{BASEURL}properties/people/bulk"
payload = {
"data": [
{
"attributes": {"properties": {"agency_name": agname}},
"id": learner_uuid,
"type": "person_properties",
}
]
}
# This is the actual call for adding the property to the person
propresponse = requests.post(prop_url, json=payload, headers=HEADERS)
# However, if we don't get a 200 status code, let's add the person to the error list for later investigation.
if propresponse.status_code != 200:
error_tupe = (learner_uuid, learner_email, agname)
errorlist.append(error_tupe)
# But, if the status code is 200, then we're good! And we output that it was successful.
else:
print(f"Looks like {learner_email} and {agname} was successful. Received status code: {propresponse.status_code}.")
except (TypeError, IndexError) as e:
error_tupe = (0, learner_email, agency_name)
errorlist.append(error_tupe)
print(f"{e} has occurred with {learner_email}")
finally:
pass
# As one last print, show the error list. It should be empty!
print(f"Error list: {errorlist}")
if __name__ == "__main__":
bulk_invite_and_group()

9
Scripts/API_Tests/bulk_invite_and_props.py
View File

@ -24,11 +24,12 @@ def bulk_invite_and_group():
This function looks for the group in the CSV as well.
"""
df = pd.DataFrame()
data = pd.read_csv(IMPORTFILE)
groups = data["Group"].unique()
groups = list(groups)
print("Here are all groups within the CSV:")
print(groups)
print(" ")
for group in groups:
payload = ""
print(group)
@ -89,12 +90,11 @@ def add_props_from_func(people, data, group):
]
}
propresponse = requests.post(prop_url, json=payload, headers=HEADERS)
print(propresponse.status_code)
if propresponse.status_code != 200:
error_tupe = (learner_uuid, learner_email, agname)
errorlist.append(error_tupe)
else:
print(f"Looks like {learner_email} and {agname} was successful.")
print(f"Looks like {learner_email} and {agname} was successful. Received status code: {propresponse.status_code}.")
except (TypeError, IndexError) as e:
error_tupe = (0, learner_email, agency_name)
errorlist.append(error_tupe)
@ -106,18 +106,15 @@ def add_props_from_func(people, data, group):
def add_props_from_csv():
errorlist = []
df = pd.DataFrame()
data = pd.read_csv(IMPORTFILE)
for dat in data.iterrows():
agency_name = dat[1][3]
# agency_name = "EMPLOYEE"
learner_email = dat[1][2]
# print(learner_email)
ppl_search = f"{BASEURL}people?filter[email][eq]={learner_email}"
ppl_response = requests.get(ppl_search, headers=HEADERS)
try:
ppl_data = ppl_response.json()
nextlink = ppl_data["links"]
learner_uuid = ppl_data["data"][0]["id"]
# Now update the props