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

TONS of migration progress! I'm so close

Browse Source 
...
This commit is contained in:
Norm Rasmussen
2024-09-04 17:18:51 -04:00
parent 8385bd1240
commit c45ca38934
11 changed files with 125 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
Scripts/API_Tests/__pycache__/Apikeys.cpython-310.pyc
View File

Binary file not shown.

4
Scripts/API_Tests/add-pgaids.py
View File

@ -4,7 +4,7 @@ from pathlib import Path
import Apikeys
import os
basefile = "/Users/normrasmussen/Downloads/mizuno-july-completions.csv"
basefile = "/Users/normrasmussen/Downloads/mizuno-aug2024-completions.csv"
api_key = Apikeys.MIZUNO
uuid_url = "https://api.northpass.com/v2/people?filter[email][eq]="
prop_url = "https://api.northpass.com/v2/properties/people/"
@ -51,7 +51,7 @@ def load_file(basefile):
completions = completions[completions["userid"] != "0"]
# completions = completions.iloc[:, 0:]
completions.to_csv(
"/Users/normrasmussen/Downloads/Mizuno-July2024-with-PGAID.csv",
"/Users/normrasmussen/Downloads/Mizuno-August2024-with-PGAID.csv",
index=False,
)

2
Scripts/API_Tests/bulk_invite_and_props.py
View File

@ -13,7 +13,7 @@ HEADERS = {
"X-Api-Key": APIKEY,
}
BASEURL = "https://api.northpass.com/v2/"
IMPORTFILE = "/Users/normrasmussen/Downloads/bloomerang-mass-import.csv"
IMPORTFILE = "/Users/normrasmussen/Downloads/bloomberg-vms-invite.csv"
def bulk_invite_and_group():

BIN
Scripts/Migration_tool/Mark_Course_As_Complete/__pycache__/Calls.cpython-310.pyc
View File

Binary file not shown.

171
Scripts/Migration_tool/Mark_Course_As_Complete/mark_course_as_complete.py
View File

@ -1,7 +1,7 @@
import Calls
from utils import calls
from datetime import datetime, timedelta
baseurl = Calls.BASEURL
baseurl = calls.BASEURL
"""
Order of operations:
@ -59,12 +59,22 @@ Order of operations:
# "joe schmo": "5555555"
# }
# }
probject = {
"items": {},
"courses": {},
"people" : {}
}
ITEM_TYPES = ['courses', 'sections', 'activities', 'people', 'enrollments', 'course_attempts', 'quiz_attempts', 'certificates', 'learning_path_attempts']
probject = {}
items = {}
courses = {}
people = {}
ITEM_TYPES = [
"courses",
"sections",
"activities",
"people",
"enrollments",
"course_attempts",
"quiz_attempts",
"certificates",
"learning_path_attempts",
]
PPL_URL = "people"
COURSE_URL = "courses"
@ -78,7 +88,7 @@ def get_people():
learner = get_individual_person(email)
ppl_uuid = learner[0]
ppl_email = learner[1]
probject["people"][ppl_email] = ppl_uuid
people[ppl_email] = ppl_uuid
get_courses(ppl_uuid)
elif type(email) is list:
person_uuids = get_group_person(email)
@ -98,71 +108,95 @@ def get_courses(learner_uuid):
def get_individual_course(name, learner_uuid):
url = f"{baseurl}/courses?filter[name][eq]={name}"
returned = Calls.get(url)
returned = calls.get(url)
for items in returned["data"]:
course_uuid = items["id"]
course_name = items["attributes"]["name"]
probject["courses"][course_name] = course_uuid
courses[course_name] = course_uuid
print(f"Cool. Course {course_uuid} exists. Checking enrollments.")
# enrollment = get_enrollment_status(course_uuid, learner_uuid)
enrollment = get_enrollment_status(course_uuid, learner_uuid)
# Use this to get a specific value from the courses sub-dict
print(list(probject["people"].values())[0])
def get_enrollment_status(uuid, learner_uuid):
enrollment_url = (
f"{baseurl}/courses/{uuid}/enrollments?filter[person_id][eq]={learner_uuid}"
)
enrolled = Calls.get(enrollment_url)
print(enrolled["data"])
enrolled = calls.get(enrollment_url)
if enrolled["data"] == []:
now = datetime.now()
formatted_now = now.strftime("%Y-%m-%d %H:%M:%S")
print("Oof, no enrollments. Not to worry. We'll get one created for you.")
mig_enroll_payload = {
"data": {
"attributes": {
{"enrolled_at": formatted_now},
{"course_id": uuid},
{"person_id": learner_uuid},
"data": [
{
"type": f"{ list(items.keys())[0] }",
"attributes": {
"uuid": f"{ list(items.values())[0] }",
"display_name": "Enrollment to Course 1",
"data": {
"enrolled_at": formatted_now,
"course_id": f"{ list(courses.values())[0] }",
"person_id": f"{ list(people.values())[0] }"
}
}
}
}
]
}
mig_enroll_url = Calls.post(
f"{baseurl}migration/projects/{probject}/items/{item_id}/enrollment_resources",
mig_enroll_payload,
)
mig_enroll_url = f"{baseurl}/migration/projects/{list(probject.values())[0]}/items/{list(items.values())[0]}/enrollment_resources"
mig_enroll = calls.post(mig_enroll_url, mig_enroll_payload)
print("Cool, enrollment resource has been created. Let's check that it exists.")
check_resources()
resource_check = check_resources()
if resource_check['data'] != []:
print("We're in! Enrollment confirmed. Let's create an attempt")
create_attempt()
else:
print("Nice! We have an enrollment. So now we just need to update progress.")
# create_attempt(enrolled["data"])
create_attempt(mig_enroll_url["data"])
def create_attempt(data):
print(f"Is there a project ID? {PROJ_ID}")
attempt_url = f"{baseurl}/migration/projects/{PROJ_ID}/items/{ITEM_ID}/course_attempt_resources"
now = datetime.now()
formatted_now = now.strftime("%Y-%m-%d %H:%M:%S")
new_attempt_payload = {
"data": {
"attributes": {
"uuid": {"which uuid?"},
"display_name": {f"{PPL_EMAIL}'s Attempt for course {COURSE_NAME}"},
"learner_id": {PPL_UUID},
"course_id": {COURSE_UUID},
"progress": {"100"},
"started_at": {datetime.now() - timedelta(hours=2)},
"completed_at": {formatted_now},
"completed_activities": [
{"uuid": {"1111"}, "completed_at": {formatted_now}}
],
def check_item(itype):
if itype not in items:
create_item(i_type= "attempts")
print(f"Is there a project ID? {list(probject.values())[0]}")
attempt_url = f"{baseurl}/migration/projects/{list(probject.values())[0]}/items/{list(items.values())[0]}/course_attempt_resources"
print(attempt_url)
now = datetime.now()
formatted_now = now.strftime("%Y-%m-%d %H:%M:%S")
new_attempt_payload = { "data": [
{
"type": "attempts",
"attributes": {
"uuid": "1234567890",
"display_name": f"{list(people.keys())[0]}'s Attempt for course {list(courses.keys())[0]}",
"data": {
"learner_id": f"{list(people.keys())[0]}",
"course_id": f"{list(courses.values())[0]}",
"progress": "100",
"started_at": f"{datetime.now() - timedelta(hours=2)}",
"enrolled_at": f"{datetime.now() - timedelta(hours=3)}",
"completed_at": formatted_now,
"completed_activities": [
{
"uuid": "1111",
"completed_at": formatted_now
}
]
}
}
}
}
]
}
check_resources()
mig_attempt_url = f"{baseurl}migration/projects/{list(probject.values())[0]}/items/{list(probject.values())[0]}/enrollment_resources"
mig_attempt_call = calls.post(attempt_url, new_attempt_payload)
print("**********")
print(f"MigAttemptURL")
print(mig_attempt_call)
print("**********")
check_resources()
"""
@ -177,49 +211,46 @@ def create_attempt(data):
def check_resources():
print("Checking resources")
get_resources_url = (
f"{baseurl}/migration/projects/{PROJ_ID}/items/{ITEM_ID}/resources"
f"{baseurl}/migration/projects/{list(probject.values())[0]}/items/{list(items.values())[0]}/resources"
)
get_resources = Calls.get(get_resources_url)
get_resources = calls.get(get_resources_url)
if get_resources["data"] == []:
print("Something went wrong. No resources were created.")
else:
print("We're in! Here's the data:")
print(get_resources)
return get_resources
def create_project():
project_name = "Testing Project 2"
proj_payload = {
"data":
{
"data": {
"type": "migration_projects",
"attributes": {"name": project_name},
}
}
proj_full_url = f"{baseurl}/migration/{PROJ_URL}"
tmp_p = Calls.post(proj_full_url, proj_payload)
tmp_p = calls.post(proj_full_url, proj_payload)
probject[project_name] = tmp_p["data"]["id"]
print(f"Created Project: {probject}")
i_type = "enrollments"
create_item(i_type, project_name)
create_item(i_type)
def create_item(i_type, project_name):
print(project_name)
def create_item(i_type):
# Item Type Options: 'courses', 'sections', 'activities', 'people', 'enrollments', 'course_attempts', 'quiz_attempts', 'certificates', 'learning_path_attempts'
item_full_url = f"{baseurl}/migration/{PROJ_URL}/{probject[project_name]}/{ITEM_URL}"
item_full_url = (
f"{baseurl}/migration/{PROJ_URL}/{list(probject.values())[0]}/{ITEM_URL}"
)
item_type = i_type
item_payload = {
"data":
{
"data": {
"type": "migration_items",
"attributes": {
"type": item_type
},
}
"attributes": {"type": item_type},
}
}
item_return = Calls.post(item_full_url, item_payload)
probject['items'][item_type] = item_return["data"]["id"]
print(f"Created Item ID: { probject['items'][item_type] }")
item_return = calls.post(item_full_url, item_payload)
items[item_type] = item_return["data"]["id"]
print(f"Created Item ID: { items[item_type] }")
print(probject)
@ -227,7 +258,7 @@ def get_group_course(list):
multiple_uuids = []
for person in email:
url = f"{baseurl}/people?filter[email][eq]={email}"
returned = Calls.get(url)
returned = calls.get(url)
for items in returned["data"]:
multiple_uuids.append(learner_uuid)
@ -235,7 +266,7 @@ def get_group_course(list):
def get_individual_person(email):
url = f"{baseurl}/people?filter[email][eq]={email}"
returned = Calls.get(url)
returned = calls.get(url)
for items in returned["data"]:
if items["attributes"]["registration_status"] == "activated":
@ -253,7 +284,7 @@ def get_group_person(list):
multiple_uuids = []
for person in email:
url = f"{baseurl}/people?filter[email][eq]={email}"
returned = Calls.get(url)
returned = calls.get(url)
for items in returned["data"]:
if items["attributes"]["registration_status"] == "activated":

0
Scripts/Migration_tool/Mark_Course_As_Complete/utils/__init__.py Normal file
View File

BIN
Scripts/Migration_tool/Mark_Course_As_Complete/utils/__pycache__/__init__.cpython-310.pyc Normal file
View File

Binary file not shown.

BIN
Scripts/Migration_tool/Mark_Course_As_Complete/utils/__pycache__/apikeys.cpython-310.pyc Normal file
View File

Binary file not shown.

BIN
Scripts/Migration_tool/Mark_Course_As_Complete/utils/__pycache__/calls.cpython-310.pyc Normal file
View File

Binary file not shown.

1
Scripts/Migration_tool/Mark_Course_As_Complete/utils/apikeys.py Normal file
View File

@ -0,0 +1 @@
SANDBOX = "SlpQlju219WnWogn94dQUT6Yt"

37
Scripts/Migration_tool/Mark_Course_As_Complete/Calls.py → Scripts/Migration_tool/Mark_Course_As_Complete/utils/calls.py
View File

@ -1,11 +1,11 @@
import requests
import Apikeys
from utils import apikeys
import pprint
import json
from json import JSONDecodeError
PP = pprint.PrettyPrinter(indent=4)
APIKEY = Apikeys.SANDBOX
APIKEY = apikeys.SANDBOX
HEADERS = {"content-type": "application/json", "X-Api-Key": APIKEY}
BASEURL = "https://api.northpass.com/v2"
@ -13,7 +13,7 @@ BASEURL = "https://api.northpass.com/v2"
def get(url):
try:
get_response = requests.get(url, headers=HEADERS)
# print(f"Executed Get Request. Status code is {get_response.status_code}")
print(f"Executed Get Request. Status code is {get_response.status_code}")
except HTTPError as h:
print(
f"Error occurred. Here's the info: {h} and status code: {get_response.status_code}"
@ -28,33 +28,36 @@ def post(url, payload):
try:
post_response = requests.post(url, headers=HEADERS, json=payload)
print(f"Executed Post Request. Status code is {post_response.status_code}")
# if post_response.status_code == 404:
# print(f"Received 404 Response. Here's the returned text: {post_response.text}")
except HTTPError as h:
except TypeError as h:
print(
f"Error occurred. Here's the info: {h} and status code: {post_response.status_code}"
)
finally:
try:
json_post = post_response.json()
# PP.pprint(json_post)
# print(f"JSONResponse: {json_post}")
return json_post
except JSONDecodeError as j:
print(
f"Error occurred. Here's the info: {h} and status code: {post_response.status_code}"
)
except JSONDecodeError as e:
print(f"Error occurred. Here's the info: {e}.")
# print(f"PostResponse: {post_response}")
return post_response
finally:
# PP.pprint(json_get)
pass
#
# def post(url, payload):
# post_response = requests.post(url, headers=HEADERS, json=payload)
# print(f"Executed Post Request. Status code is {post_response.status_code}")
# print(post_response.text)
# return post_response
def delete(url):
try:
get_response = requests.delete(url, headers=HEADERS)
# print(f"Executed Get Request. Status code is {get_response.status_code}")
delete_response = requests.delete(url, headers=HEADERS)
print(f"Executed Delete Request. Status code is {delete_response.status_code}")
except HTTPError as h:
print(
f"Error occurred. Here's the info: {h} and status code: {get_response.status_code}"
f"Error occurred. Here's the info: {h} and status code: {delete_response.status_code}"
)
finally:
return get_response
return delete_response