From 06553b93c3f94071bb21241dbcb5ccf405dc08fc Mon Sep 17 00:00:00 2001 From: Norm Rasmussen Date: Fri, 26 Sep 2025 14:49:46 -0400 Subject: [PATCH] Cleaned up and minmized Chubb's CISA script to share with the client so they can run it on their own --- .../bulk_invite_and_props-minimal.py | 175 ++++++++++++++++++ Scripts/API_Tests/bulk_invite_and_props.py | 9 +- 2 files changed, 178 insertions(+), 6 deletions(-) create mode 100644 Scripts/API_Tests/bulk_invite_and_props-minimal.py diff --git a/Scripts/API_Tests/bulk_invite_and_props-minimal.py b/Scripts/API_Tests/bulk_invite_and_props-minimal.py new file mode 100644 index 00000000..4e4254d0 --- /dev/null +++ b/Scripts/API_Tests/bulk_invite_and_props-minimal.py @@ -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() diff --git a/Scripts/API_Tests/bulk_invite_and_props.py b/Scripts/API_Tests/bulk_invite_and_props.py index ddf0d4a1..80e6fea8 100644 --- a/Scripts/API_Tests/bulk_invite_and_props.py +++ b/Scripts/API_Tests/bulk_invite_and_props.py @@ -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