Lots of progress on migration tool for Luma's monthly ask. It's 90% working, but the enrollments endpoint is giving me issues.

Scripts/Migration_tool/Luma_mass_completions/data.csv

@@ -0,0 +1,16 @@
+Completion Date,Email Address,Course Name
+2024-04-19,norm@rsmsn.co,This is War
+2024-04-20,norm+climbing@northpass.com,This is War
+2024-04-21,gasssygas@gas.com,This is War
+2024-04-22,boy@morrison.org,This is War
+2024-04-23,norm+mtnclimb@northpass.com,This is War
+2024-04-24,n,This is War
+2024-04-25,jim@morrison.org,This is War
+2024-04-25,nrasmussen+admincomms@northpass.com,This is War
+2024-04-26,nrasmussen+test2@gainsight.com,This is War
+2024-04-27,paulmiller@paulmiller.com,This is War
+2024-04-28,potatobox@gas.com,This is War
+2024-05-01,kfolsomtest@northpass.com,This is War
+2024-05-02,biggum@gas.com,This is War
+2024-05-03,aliens@destination.com,This is War
+2024-05-04,y,This is War

Scripts/Migration_tool/Luma_mass_completions/manage_csv.py

@@ -0,0 +1,6 @@
+import pandas as pd
+
+def import_as_dataframe():
+    importfile = "./data.csv"
+    df = pd.read_csv(importfile)
+    return df

Scripts/Migration_tool/Luma_mass_completions/mark_course_as_complete.py

@@ -0,0 +1,225 @@
+"""
+Order of operations:
+1. Get input of people and course
+2. Create Project
+3. Create Item with Course_attempt type
+4. Create Course Attempt Resource with the same type.
+5. Start Migration.
+"""
+
+from sys import argv
+from utils import calls
+import manage_csv
+from datetime import datetime, timedelta
+import uuid
+
+baseurl = calls.BASEURL
+
+probject = {}
+items = {}
+courses = {}
+people = {}
+miniattempt_loads = []
+minienroll_loads = []
+
+
+item_types = [
+    "courses",
+    "sections",
+    "activities",
+    "people",
+    "enrollment_resources",
+    "course_attempts",
+    "quiz_attempts",
+    "certificates",
+    "learning_path_attempts",
+]
+project_name = argv[1]
+
+
+def get_data():
+    df = manage_csv.import_as_dataframe()
+    df.columns = df.columns.str.lower()
+    row_iterator = df.iterrows()
+    _, last = next(row_iterator)
+    for i, row in row_iterator:
+        email = row["email address"]
+        course = row["course name"]
+        date = row["completion date"]
+        people_tuple = get_people(email)
+        if people_tuple is None:
+            pass
+        else:
+            course_uuid = get_courses(course)
+            attempt_payload = create_attempt_payload(course_uuid, people_tuple[0], date)
+            enrollment_payload = create_enrollment_payload(
+                course_uuid, people_tuple[0], date
+            )
+    create_attempt_resource(attempt_payload)
+    create_enrollment_resource(enrollment_payload)
+
+
+def get_people(email):
+    url = f"{baseurl}/people?filter[email][eq]={email}"
+    returned = calls.get(url)
+    if returned["data"] == []:
+        print("I couldn't find this user.")
+    else:
+        for api_items in returned["data"]:
+            if api_items["attributes"]["registration_status"] == "activated":
+                single_uuid = api_items["id"]
+                single_email = api_items["attributes"]["email"]
+                return (single_uuid, single_email)
+            else:
+                print(f"Person not found or activated. {email}")
+
+
+def get_courses(course):
+    url = f"{baseurl}/courses?filter[name][eq]={course}"
+    returned = calls.get(url)
+    for items in returned["data"]:
+        course_uuid = items["id"]
+        course_name = items["attributes"]["name"]
+        print(f"Cool. Course {course_uuid} exists.")
+        return course_uuid
+
+
+def create_attempt_payload(course_uuid, learner_uuid, date_str):
+    print(f"Creating Course Attempt Payload")
+    now = datetime.now()
+    date_format = "%Y-%m-%d %H:%M:%S"
+    if type(date_str) == str:
+        if len(date_str) >= 8 and len(date_str) <= 10:
+            date_str = f"{date_str} 00:00:00"
+            formatted_date = datetime.strptime(date_str, date_format)
+            formatted_date = str(formatted_date)
+        else:
+            formatted_date = datetime.strptime(date_str, date_format)
+            formatted_date = str(formatted_date)
+    mini_attempt_payload = {
+        "type": "course_attempt_resources",
+        "attributes": {
+            "uuid": str(uuid.uuid4()),
+            "display_name": f"{learner_uuid}s Attempt for course {course_uuid}",
+            "data": {
+                "learner_id": f"{learner_uuid}",
+                "course_id": f"{course_uuid}",
+                "progress": 100,
+                "enrolled_at": formatted_date,
+                "started_at": formatted_date,
+                "completed_at": formatted_date,
+                "completed_activities": [
+                    {"id": str(uuid.uuid4()), "completed_at": formatted_date}
+                ],
+            },
+        },
+    }
+    miniattempt_loads.append(mini_attempt_payload)
+    attempt_payload = { "data": miniattempt_loads }
+    return attempt_payload
+
+
+def create_enrollment_payload(course_uuid, learner_uuid, date_str):
+    print(f"Creating Course Enrollment Payload")
+    now = datetime.now()
+    date_format = "%Y-%m-%d %H:%M:%S"
+    if type(date_str) == str:
+        if len(date_str) >= 8 and len(date_str) <= 10:
+            date_str = f"{date_str} 00:00:00"
+            formatted_date = datetime.strptime(date_str, date_format)
+            formatted_date = str(formatted_date)
+        else:
+            formatted_date = datetime.strptime(date_str, date_format)
+            formatted_date = str(formatted_date)
+    mini_enrollment_payload = {
+        "type": "enrollment_resource",
+        "attributes": {
+            "uuid": str(uuid.uuid4()),
+            "display_name": f"{learner_uuid}s enrollment for course {course_uuid}/",
+            "data": {
+                "enrolled_at": formatted_date,
+                "course_id": course_uuid,
+                "person_id": learner_uuid
+            }
+        }
+    }
+    minienroll_loads.append(mini_enrollment_payload)
+    enrollment_payload = {"data": minienroll_loads }
+    with open("payload.json", "w") as f:
+        f.write(str(enrollment_payload))
+    print(enrollment_payload)
+    return enrollment_payload
+
+
+def create_enrollment_resource(enrollment_payload):
+    item = items["enrollments"]
+    enrollment_url = f"{baseurl}/migration/projects/{list(probject.values())[0]}/items/{item}/enrollment_resources"
+    print(enrollment_url)
+    print(f"Enrollment Payload is: {enrollment_payload}")
+    enrollment_call = calls.post(enrollment_url, enrollment_payload)
+    print(
+        f"In trying the enrollment resource, the following code was returned: {enrollment_call.status_code}"
+    )
+
+
+def create_attempt_resource(attempt_payload):
+    item = items["course_attempts"]
+    attempt_url = f"{baseurl}/migration/projects/{list(probject.values())[0]}/items/{item}/course_attempt_resources"
+    print(attempt_url)
+    print(f"Attempt Payload is: {attempt_payload}")
+    attempt_call = calls.post(attempt_url, attempt_payload)
+    print(
+        f"In trying the attempt resource, the following code was returned: {attempt_call.status_code}"
+    )
+
+def create_project(project_name):
+    proj_payload = {
+        "data": {
+            "type": "migration_projects",
+            "attributes": {"name": project_name},
+        }
+    }
+    proj_full_url = f"{baseurl}/migration/projects"
+    tmp_p = calls.post(proj_full_url, proj_payload)
+    probject[project_name] = tmp_p["data"]["id"]
+    print(f"Created Project: {probject}")
+
+
+def create_item(i_type):
+    print("Creating Item")
+    # Item Type Options: 'courses', 'sections', 'activities', 'people', 'enrollments', 'course_attempts', 'quiz_attempts', 'certificates', 'learning_path_attempts'
+    item_full_url = f"{baseurl}/migration/projects/{list(probject.values())[0]}/items"
+    item_type = i_type
+    print(item_type)
+    item_payload = {
+        "data": {
+            "attributes": {"type": item_type},
+        }
+    }
+    # "type": "migration_items",
+    item_return = calls.post(item_full_url, item_payload)
+    items[item_type] = item_return["data"]["id"]
+    item_id = item_return["data"]["id"]
+    print(f"Created Item ID: { items[item_type] }")
+    return item_id
+
+
+def start_migration():
+    print("Starting Migration Function")
+    start_url = (
+        f"{baseurl}/migration/projects/{list(probject.values())[0]}/start_migration"
+    )
+    empty = ""
+    mig = calls.post(start_url, empty)
+    print(mig.text)
+    print(mig)
+
+
+if __name__ == "__main__":
+    create_project(project_name)
+    create_item("course_attempts")
+    create_item("enrollments")
+    print(items)
+    print(probject)
+    get_data()
+    start_migration()

Scripts/Migration_tool/Luma_mass_completions/utils/apikeys.py

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

Scripts/Migration_tool/Luma_mass_completions/utils/calls.py

@@ -0,0 +1,63 @@
+import requests
+from utils import apikeys
+import pprint
+from json import JSONDecodeError
+
+
+PP = pprint.PrettyPrinter(indent=4)
+APIKEY = apikeys.SANDBOX
+HEADERS = {"content-type": "application/json", "X-Api-Key": APIKEY}
+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}")
+    except HTTPError as h:
+        print(
+            f"Error occurred. Here's the info: {h} and status code: {get_response.status_code}"
+        )
+    finally:
+        json_get = get_response.json()
+        # PP.pprint(json_get)
+        return json_get
+
+
+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}")
+    except Exception as h:
+        print(
+            f"Error occurred. Here's the info: {h} and text: {post_response.text}"
+        )
+    finally:
+        try:
+            json_post = post_response.json()
+            # print(f"JSONResponse: {json_post}")
+            return json_post
+        except JSONDecodeError as e:
+            print(f"Error occurred. Here's the info: {e} and text: {post_response.text}")
+            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:
+        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: {delete_response.status_code}"
+        )
+    finally:
+        return delete_response