Extending xApi statement generation

tools/xapi-statement-generator/faker.py

+import random
+from datetime import datetime
+
+
+class XApiFaker:
+
+    providers = [
+        {
+            "name": "H5P",
+            "base_url": "http://h5p.example.com/expapi/",
+            "verbs": [
+                {
+                    "id": "experienced",
+                    "name": "Experienced",
+                    "objects": [
+                        {
+                            "id": "QKGPPiIhI4zx9YAZZksLKigqyf7yW4WF",
+                            "name": "1.1.1 Funktionen",
+                        },
+                    ],
+                },
+                {
+                    "id": "attempted",
+                    "name": "Attempted",
+                    "objects": [
+                        {
+                            "id": "VeH7S8NeGlCRM1myYRDBjHMCknLqDLgm",
+                            "name": "2.3.1 Funktion Zirkulationsleitung",
+                        },
+                    ],
+                },
+                {
+                    "id": "interacted",
+                    "name": "Interacted",
+                    "objects": [
+                        {
+                            "id": "ofN2ODcnLRaVu30lUpzrWPqF2AcG7g46",
+                            "name": "1.2.3 Kappenventil",
+                        },
+                    ],
+                },
+                {
+                    "id": "answered",
+                    "name": "Answered",
+                    "objects": [
+                        {
+                            "id": "K34IszYvGE4R0cC72Ean6msLfLCJtQ8b",
+                            "name": "7.2.1 Ventil Basics",
+                        },
+                    ],
+                },
+                {
+                    "id": "liked",
+                    "name": "Liked",
+                    "objects": [
+                        {
+                            "id": "A34IszYvGE4R0cC72Ean6msLfLCJtQ8b",
+                            "name": "Physik Kurs A3.1",
+                        },
+                        {
+                            "id": "B34IszYvGE4R0cC72Ean6msLfLCJtQ8b",
+                            "name": "Biologie Kurs A1.1",
+                        },
+                        {
+                            "id": "C34IszYvGE4R0cC72Ean6msLfLCJtQ8b",
+                            "name": "Englisch Kurs A1",
+                        },
+                    ],
+                }
+            ],
+            "actors": ["user1@polaris.com", "user2@polaris.com"],
+        },
+        {
+            "name": "Moodle",
+            "base_url": "http://moodle.example.com/expapi/",
+            "verbs": [
+                {
+                    "id": "completed",
+                    "name": "Completed",
+                    "objects": [
+                        {
+                            "id": "programming-course-python",
+                            "name": "Python Programming Course",
+                        },
+                        {
+                            "id": "foreign-language-course",
+                            "name": "Foreign language course",
+                        },
+                    ],
+                },
+                {
+                    "id": "started",
+                    "name": "Started",
+                    "objects": [
+                        {
+                            "id": "programming-course-python",
+                            "name": "Python Programming Course",
+                        },
+                        {
+                            "id": "foreign-language-course",
+                            "name": "Foreign language course",
+                        },
+                    ],
+                },
+                {
+                    "id": "paused",
+                    "name": "Paused",
+                    "objects": [
+                        {
+                            "id": "programming-course-python",
+                            "name": "Python Programming Course",
+                        },
+                        {
+                            "id": "foreign-language-course",
+                            "name": "Foreign language course",
+                        },
+                    ],
+                },
+                {
+                    "id": "created",
+                    "name": "Created",
+                    "objects": [
+                        {
+                            "id": "poll-preferred-course-level",
+                            "name": "Poll: Preferred course level",
+                        },
+                    ],
+                },
+                {
+                    "id": "answered",
+                    "name": "Answered",
+                    "objects": [
+                        {
+                            "id": "poll-preferred-course-level",
+                            "name": "Poll: Preferred course level",
+                        },
+                    ],
+                },
+            ],
+            "actors": ["user1@polaris.com", "user2@polaris.com"],
+        },
+    ]
+
+    def __get_xapi_verb(self, base_url, verb_config):
+        return {
+            "id": base_url + "verbs/" + verb_config["id"],
+            "display": {"en-US": verb_config["name"]},
+        }
+
+    def __get_xapi_object(self, base_url, verb_config):
+        object = random.choice(verb_config["objects"])
+        return {
+            "id": base_url + "activity/" + object["id"],
+            "definition": {"name": {"de-DE": object["name"]}},
+            "objectType": "Activity",
+        }
+
+    def fake_statement(self):
+        provider_config = random.choice(self.providers)
+
+        verb_config = random.choice(provider_config["verbs"])
+        xapi_verb = self.__get_xapi_verb(provider_config["base_url"], verb_config)
+        xapi_object = self.__get_xapi_object(provider_config["base_url"], verb_config)
+
+        now = datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ")
+        actor = random.choice(provider_config["actors"])
+        return {
+            "actor": {"mbox": f"mailto:{actor}"},
+            "verb": xapi_verb,
+            "object": xapi_object,
+            "timestamp": now,
+        }

tools/xapi-statement-generator/generator.py

 import argparse
 import json
-import requests
-from datetime import datetime
-import random
-import uuid
+import logging
+import sys
 import time
+from datetime import datetime
+
+import requests
 
-VERBS = ['experienced', 'attempted', 'interacted', 'answered']
+from faker import XApiFaker
+from statistic import Statistic
+
+logging.basicConfig(
+    format="%(asctime)s [%(levelname)s] %(message)s",
+    handlers=[
+        logging.FileHandler(f"lrs_import_{time.strftime('%Y_%m_%d_%H_%M_%S')}.log"),
+        logging.StreamHandler(sys.stdout),
+    ],
+)
+logger = logging.getLogger()
+logger.setLevel(logging.INFO)
 
 
 def read_json_dump(file_name):
@@ -17,34 +29,51 @@ def read_json_dump(file_name):
 
 
 def transform_statement(statement, overwrite_ts):
-    actor = json.loads(statement.get('actor'))
-    object = statement.get('object')
+    actor = json.loads(statement.get("actor"))
+    object = statement.get("object")
     if object:
         object = json.loads(object)
-        email = actor.get('email')
-        verb_str = statement.get('verb')
-        created_at = statement.get('created_at')
-        date = datetime.strptime(
-            created_at, '%Y-%m-%d %H:%M:%S').strftime('%Y-%m-%dT%H:%M:%SZ')
-        now = datetime.now().strftime('%Y-%m-%dT%H:%M:%SZ')
+        email = actor.get("email")
+        verb_str = statement.get("verb")
+        created_at = statement.get("created_at")
+        date = datetime.strptime(created_at, "%Y-%m-%d %H:%M:%S").strftime(
+            "%Y-%m-%dT%H:%M:%SZ"
+        )
+        now = datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ")
         timestamp = now if overwrite_ts else date
-        if not email.__contains__('@'):
+        if not email.__contains__("@"):
             email = f"{email}@test.de"
         return {
             "actor": {"mbox": f"mailto:{email}"},
             "verb": json.loads(verb_str),
             "object": {
                 "id": f"http://educa-portal.de/expapi/activity/{object.get('id')}",
-                "objectType": "Activity"},
-            "timestamp": timestamp
+                "objectType": "Activity",
+            },
+            "timestamp": timestamp,
         }
     return None
 
 
-def send_x_api_statement(url, statement):
+def send_x_api_statement(statistic, url, statement):
     request = requests.post(url, json=statement)
-    print(request.content)
-
+    try:
+        response_dict = request.json()
+        if request.ok:
+            logging.info(
+                f"Accepted: {statement.get('actor').get('mbox')} {statement.get('verb').get('id')} - {response_dict.get('provider')}"
+            )
+            statistic.add(response_dict.get("provider"), True)
+        else:
+            logging.warning(
+                f"Rejected: {statement.get('actor').get('mbox')} {statement.get('verb').get('id')} - {response_dict.get('provider')} - {response_dict.get('message')}"
+            )
+            statistic.add(response_dict.get("provider"), False)
+        return [request.status_code, response_dict.get("message")]
+    except json.JSONDecodeError:
+        logging.error(
+            f"Failed: {statement.get('actor').get('mbox')} {statement.get('verb').get('id')}"
+        )
 
 def import_json(args):
     file_name = args.input[0]
@@ -54,77 +83,80 @@ def import_json(args):
 
     url = args.target[0]
 
-    print('Running...')
+    statistic = Statistic()
+
+    print("Running...")
     for statement in x_api_statements:
 
         valid_statement = transform_statement(statement, overwrite_ts)
         if valid_statement:
-            send_x_api_statement(url, statement)
-    else:
-        print("Skipped xAPI statement")
-
+            send_x_api_statement(statistic, url, statement)
+        else:
+            logging.error("Skipped xAPI statement: %s", statement.get("verb"))
 
-def get_rand_verb():
-    return {
-        "id": f"http://educa-portal.de/expapi/verbs/{random.choice(VERBS)}",
-        "display": {"en-US": random.choice(VERBS)},
-    }
-
-
-def get_rand_obj():
-    rand_id = uuid.uuid1()
-    return {
-        "id": f"http://educa-portal.de/expapi/activity/{rand_id}",
-        "objectType": "Activity"
-    }
-
-
-def get_rand_statement():
-    now = datetime.now().strftime('%Y-%m-%dT%H:%M:%SZ')
-    email = 'user1@polaris.com'
-    return {
-        "actor": {"mbox": f"mailto:{email}"},
-        "verb": get_rand_verb(),
-        "object": get_rand_obj(),
-        "timestamp": now
-    }
+    print(statistic)
 
 
 def start_random_generation(args):
-
     url = args.target[0]
-    print('Running...')
+    print("Running...")
+    statistic = Statistic()
+    faker = XApiFaker()
     while True:
-        send_x_api_statement(url, get_rand_statement())
-        time.sleep(0.2)
+        try:
+            send_x_api_statement(statistic, url, faker.fake_statement())
+            time.sleep(0.2)
+        except KeyboardInterrupt:
+            print(statistic)
+            raise SystemExit
 
 
 def main():
-    parser = argparse.ArgumentParser(
-        description="Simple xAPI Statement Generator.")
-
-    parser.add_argument("-t", "--target", type=str, nargs=1,
-                        metavar="target_url", default=None,
-                        help="Address to which xAPI statements are sent.", required=True)
-
-    parser.add_argument("-i", "--input", type=str, nargs=1,
-                        metavar="file_name", default=None,
-                        help="Reads the specified JSON file as indiviual xAPI statements.")
-
-    parser.add_argument("-o", "--overwrite_ts", default=False, action=argparse.BooleanOptionalAction,
-                        help="Overwrite timestamp with current date, which results in new LRS entries in case the xAPI statements are already stored in the LRS."
-                        )
-
-    parser.add_argument("-r", "--random", default=False, action=argparse.BooleanOptionalAction,
-                        help="Creates and sends a random xAPI statement to an LRS in an interval of 1/s."
-                        )
+    parser = argparse.ArgumentParser(description="Simple xAPI Statement Generator.")
+
+    parser.add_argument(
+        "-t",
+        "--target",
+        type=str,
+        nargs=1,
+        metavar="target_url",
+        default=None,
+        help="Address to which xAPI statements are sent.",
+        required=True,
+    )
+
+    parser.add_argument(
+        "-i",
+        "--input",
+        type=str,
+        nargs=1,
+        metavar="file_name",
+        default=None,
+        help="Reads the specified JSON file as indiviual xAPI statements.",
+    )
+
+    parser.add_argument(
+        "-o",
+        "--overwrite_ts",
+        default=False,
+        action=argparse.BooleanOptionalAction,
+        help="Overwrite timestamp with current date, which results in new LRS entries in case the xAPI statements are already stored in the LRS.",
+    )
+
+    parser.add_argument(
+        "-r",
+        "--random",
+        default=False,
+        action=argparse.BooleanOptionalAction,
+        help="Creates and sends a random xAPI statement to an LRS in an interval of 1/s.",
+    )
 
     args = parser.parse_args()
 
-    if args.target != None and args.input != None and not args.random:
+    if args.target is not None and args.input is not None and not args.random:
         import_json(args)
 
-    if args.random and args.target != None:
+    if args.random and args.target is not None:
         start_random_generation(args)
 
 

tools/xapi-statement-generator/statistic.py

+class Statistic:
+
+    providers = {}
+
+    def add(self, provider_name, accepted):
+        existing_provider = self.providers.get(
+            provider_name, {"accepted": 0, "rejected": 0}
+        )
+        provider = {
+            provider_name: {
+                "accepted": existing_provider["accepted"] + (1 if accepted else 0),
+                "rejected": existing_provider["rejected"] + (0 if accepted else 1),
+            }
+        }
+        self.providers.update(provider)
+
+    def __str__(self):
+        accepted_statements = sum(
+            [provider["accepted"] for provider in self.providers.values()]
+        )
+        rejected_statements = sum(
+            [provider["rejected"] for provider in self.providers.values()]
+        )
+
+        success_rate = int(
+            (accepted_statements / (accepted_statements + rejected_statements)) * 100
+        )
+
+        output = f"\nImporting {accepted_statements + rejected_statements} statements completed"
+        output += f"\nLRS accepted {accepted_statements} statements ({success_rate}%) and rejected {rejected_statements} statements ({100 - success_rate}%)\n"
+
+        for (provider_name, data) in self.providers.items():
+            provider_success_rate = int(
+                (data["accepted"] / (data["accepted"] + data["rejected"]) * 100)
+            )
+            output += f"\nProvider: {provider_name}"
+            output += f"\n\tAccepted statements: {data['accepted']} ({provider_success_rate}%)"
+            output += f"\n\tRejected statements: {data['rejected']} ({100 - provider_success_rate}%)"
+
+        return output