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

Some completed tasks. Script for large Walmart enrollment (which brought down the app... oops).

Browse Source
This commit is contained in:
Norm Rasmussen
2025-04-29 15:47:51 -04:00
parent 603e7baf6a
commit 00d628a532
4 changed files with 78 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
Scripts/API_Tests/bulk_add_ppl_from_csv.py
View File

@ -1,18 +1,23 @@
import requests
import json
import pandas as pd
import pprint
import csv
import Apikeys
# ************************
# IF YOU RUN THIS SCRIPT AGAIN WITH HUNDREDS OF THOUSANDS OF ENROLLMENTS
# SPREAD IT OUT OVER MANY BATCHES, IDEALLY ALL DAY
# ************************
pp = pprint.PrettyPrinter(indent=4)
APIKEY = Apikeys.CHUBB
IMPORTFILE = "~/Downloads/walmart-supplier-mass-grouping.csv"
APIKEY = Apikeys.SUPPLIERPROD
HEADERS = {
"accept": "application/json",
"X-Api-Key": APIKEY,
}
BASEURL = "https://api.northpass.com/v2/"
BASEURL = "https://api2.northpass.com/v2/"
def bulk_invite_ppl():
def bulk_invite_ppl_under_1500():
"""
Bulk endpoint which invites new people and adds them to a group via this structure:
{
@ -35,6 +40,43 @@ def bulk_invite_ppl():
response = requests.post(url, headers=HEADERS, json=payload)
print(f"Completed. Status code is {response.status_code}")
def bulk_invite_ppl_over_1500():
data = pd.read_csv(IMPORTFILE, encoding = "utf-8")
emails = list(data["EmailID"].unique())
url = f"{BASEURL}bulk/people"
count = 0
if len(emails) > 499:
for chunk in range(0, len(emails), 499):
i = chunk
payload_1 = []
i_to_add = emails[i: i + 499]
for person in i_to_add:
miniload = {"email": person, "groups": "Active Suppliers" }
miniload1 = {"email": person, "groups": "All Users" }
miniload2 = {"email": person, "groups": "Grow With Walmart" }
payload_1.append(miniload)
payload_1.append(miniload2)
payload_1.append(miniload1)
print(f"The long length payload has {len(payload_1)} people emails")
payload = {"data": {"attributes": {"people": payload_1}}}
response = requests.post(url, headers=HEADERS, json=payload)
count += 1
print(f"Iteration number {count} out of {int(len(emails))/1500}")
print(f"Completed. Status code is {response.status_code}")
else:
payload_1 = []
for person in emails:
miniload = {"email": person, "groups": "Active Suppliers" }
miniload1 = {"email": person, "groups": "All Users" }
miniload2 = {"email": person, "groups": "Grow With Walmart" }
payload_1.append(miniload)
payload_1.append(miniload2)
payload_1.append(miniload1)
print(f"The payload has {len(payload_1)}")
payload = {"data": {"attributes": {"people": payload_1}}}
response = requests.post(url, headers=HEADERS, json=payload)
print(f"Completed. Status code is {response.status_code}")
print(response.text)
if __name__ == "__main__":
bulk_invite_ppl()
bulk_invite_ppl_over_1500()

27
Scripts/API_Tests/get_all_people_toggle_prop.py
View File

@ -13,27 +13,33 @@ HEADERS = {
}
def get_people():
count = 0
count = 1
while True:
ppl_uuids = []
url = f"{BASEURL}/people?page={count}&limit=100"
print(url)
count += 1
response = requests.get(url, headers=HEADERS).json()
nextlink = response["links"]
for data in response['data']:
person_uuid = data['id']
if person_uuid == "c09c6319-7569-46ac-94d5-3885d9de4c2f" or person_uuid == "477de2a1-15cc-4d67-9e95-7e3910dca5d3":
print("Found Claudia or Fabian")
print(person_uuid)
print(data['attributes']['full_name'])
ppl_uuids.append(person_uuid)
if len(ppl_uuids) == 10:
print(len(ppl_uuids))
if len(ppl_uuids) == 100:
apply_prop(ppl_uuids)
# break
if "next" not in nextlink:
break
def apply_prop(ppl_uuids):
inserted_load = []
for person in ppl_uuids:
print(person)
miniload = {
"attributes": { "properties": { "new_content": "true" } },
"attributes": { "properties": { "new_content": "True" } },
"id": person,
"type": "person_properties"
}
@ -41,7 +47,16 @@ def apply_prop(ppl_uuids):
prop_url = f"{BASEURL}/properties/people/bulk"
payload = { "data": inserted_load }
print(payload)
# apply_payload = requests.posts(prop_url, json=payload, headers=HEADERS)
apply_payload = requests.post(prop_url, json=payload, headers=HEADERS)
stat = apply_payload.status_code
print(stat)
if "4" in str(stat):
print("....!")
print("Error! Error!")
print(stat.text)
print(payload)
print("....!")
if __name__ == "__main__":
get_people()