utils.py

import json
import os
import requests
import uuid
from datetime import datetime, timezone

ACTIVE_MODE = True
LEARN_ENDPOINT = "https://api-sg-demo.adaptemy.io/learn"
CONTENT_OBJECTS_ENDPOINT = "https://api-sg-demo.adaptemy.io/content-objects"
# LEARN_ENDPOINT = "https://api-dev-demo-app.adaptemy.io/learn"
# CONTENT_OBJECTS_ENDPOINT = "https://api-dev-demo-app.adaptemy.io/content-objects"
DEFAULT_BEARER_TOKEN = "****"  # Replace with a valid token for testing, or set via environment variable.


def get_bearer_token():
    return (
        os.getenv("BEARER_TOKEN")
        or os.getenv("DEV_BEARER_TOKEN")
        or DEFAULT_BEARER_TOKEN
    )


def get_headers():
    return {
        "Authorization": f"Bearer {get_bearer_token()}",
        "Content-Type": "application/json",
        "Accept": "application/json",
    }

def new_guid():
    return str(uuid.uuid4())

def xAPI_deploy_request(endpoint, payload, headers=None):
    request_headers = headers or get_headers()

    response = requests.post(
        endpoint,
        headers=request_headers,
        json=payload
    )

    print("Status code (content object post):", response.status_code)
    try:
        print(json.dumps(response.json(), indent=2))
    except Exception:
        print(response.text)
    return response

def xAPI_user_request(endpoint, payload, headers=None):
    request_headers = headers or get_headers()

    response = requests.post(
        endpoint,
        headers=request_headers,
        json=payload
    )

    print("Status code (experience post):", response.status_code)
    if response.status_code == 200:
        try:
            print("Sucessfull submmited the learner experience:", json.dumps(payload, indent=2))
            return( f"✅ Submitted **{len(payload)}** learner experience record(s).")
        except Exception as e:
            print("error due to:", e)
            return(f"error due to: {e}")
    else:
        print("Failed to submit learner experience.")
        print("Response:", response.text)
        return(f"Failed to submit learner experience : {response.text}")

def xAPI_validate(params, endpoint = CONTENT_OBJECTS_ENDPOINT, headers=None):
    request_headers = headers or get_headers()

    response = requests.get(
        endpoint,
        headers=request_headers,
        params=params
    )

    print("Status code (content object post):", response.status_code)
    try:
        print(json.dumps(response.json(), indent=2))
    except Exception:
        print(response.text)

def xr_experience_event_formatting(
    learner_name: str,
    duration: str,
    xr_content_object_id: str,
    session_id: str,
    platform: str = "heat-xrel",
    qoe_rating: float | None = None,
    emotion_score: float | None = None,
):
    """
    Formulates a learner's XR experience event to Adaptemy xAPI as per the defined schema.

    Args:
        learner_name: The name of the learner.
        duration: The temporal duration of the experience.
        xr_content_object_id: The ID of the XR multisensory content object.
        session_id: The ID of the learning session.
    """

    result_payload = {
        "completion": True,
        "duration": duration
    }

    if qoe_rating is not None or emotion_score is not None:
        result_extensions = {}
        if qoe_rating is not None:
            result_extensions["https://api.adaptemy.io/xapi/result/qoe-rating"] = qoe_rating
        if emotion_score is not None:
            result_extensions["https://api.adaptemy.io/xapi/result/emotion-score"] = emotion_score
        result_payload["extensions"] = result_extensions

    xr_experience_events = {
        "actor": {
            "account": {
                "homepage": "https://api-sg-demo.adaptemy.io/organization/23",
                "name": learner_name
            }
        },
        "object": {
            "id": xr_content_object_id,
        },
        "verb": {
            "id": "http://activitystrea.ms/schema/1.0/experienced"
        },
        "result": result_payload,
        "context": {
            "platform": platform,
            "registration": session_id,
            "active": ACTIVE_MODE
        }
    }

    return xr_experience_events


def xr_question_answer_event_formatting(
    learner_name: str,
    xr_ncc_object_id: str,
    session_id: str,
    result: list[dict[str, str]],
    platform: str = "heat_xrtheater",
):
    """
    Formulates a learner's NCC question-answer event for Adaptemy xAPI.

    Args:
        learner_name: The name of the learner.
        xr_ncc_object_id: The ID of the NCC question content object.
        session_id: The ID of the learning session.
        result: List of answer objects, each with 'id' and 'description' keys,
                e.g. [{"id": "1", "description": "Strongly agree"}, ...].
        platform: The platform identifier.
    """
    xr_question_answer_event = {
        "actor": {
            "account": {
                "homepage": "https://api-sg-demo.adaptemy.io/organization/23",
                "name": learner_name,
            }
        },
        "object": {
            "id": xr_ncc_object_id,
        },
        "verb": {
            "id": "http://adlnet.gov/expapi/verbs/answered",
        },
        "result": [{"id": r["id"], "description": r["description"]} for r in result],
        "context": {
            "platform": platform,
            "registration": session_id,
            "active": ACTIVE_MODE,
        },
    }

    return xr_question_answer_event


def post_xr_content_object(
    description_text: str,
    name_text: str,
    media_type: str,
    scene: str,
    media_device: str,
    concept_guids: list[str] = None
):
    """
    Formulates and posts an XR multisensory content object to Adaptemy xAPI.

    Args:
        description_text: Description of the XR experience.
        name_text: Short name of the XR experience.
        media_type: Type of media (e.g., "sensory-xr", "hologram-xr").
        scene: Scene description for media details.
        media_device: Device description for media details.
        concept_guids: Optional list of concept GUIDs related to the content.

    Returns:
        The ID of the posted XR multisensory content object.
    """
    xr_content_guid = new_guid()
    xr_content_object_id = f"https://api.adaptemy.io/content/{xr_content_guid}"

    extensions_data = {
        "https://api.adaptemy.io/contentmetadata/active": ACTIVE_MODE,
        "https://api.adaptemy.io/contentmetadata/learning-phase": "inquiry-based-learning",
        "https://api.adaptemy.io/contentmetadata/scene": scene,
        "https://api.adaptemy.io/contentmetadata/media-details": {
            "type": media_type,
            "device": media_device
        }
    }
    if concept_guids:
        extensions_data["https://api.adaptemy.io/concept"] = concept_guids

    xr_content_objects = [{
        "id": xr_content_object_id,
        "organizationId": 18,
        "definition": {
            "description": {
                "en-US": description_text
            },
            "extensions": extensions_data,
            "name": {
                "en-US": name_text
            },
            "type": "http://adlnet.gov/expapi/activities/media"
        },
        "isEnabled": True
    }]

    xAPI_deploy_request(CONTENT_OBJECTS_ENDPOINT, xr_content_objects)

    return xr_content_object_id


def build_question_ncc_object(
    question: str,
    description: str,
    answer: str,
    ncc_tag: str,
    assessment_type: str,
    ncc_senario: str = "all-users",
):
    question_guid = new_guid()
    xr_ncc_object_id = f"https://api.adaptemy.io/content/{question_guid}"

    answer_options = [opt.strip() for opt in (answer or "").split(",") if opt.strip()]
    choices = (
        [{"id": str(index), "description": option} for index, option in enumerate(answer_options, start=1)]
        if answer_options
        else [{"id": "1", "description": "open-ended"}]
    )

    ncc_content_object = {
        "id": xr_ncc_object_id,
        "organizationId": 23,
        "description": question,
        "definition": {
            "extensions": {
                "https://api.adaptemy.io/contentmetadata/active": ACTIVE_MODE,
                "https://api.adaptemy.io/contentmetadata/learning-phase": "assessment",
                "https://api.adaptemy.io/contentmetadata/assessment-type": assessment_type,
                "https://api.adaptemy.io/contentmetadata/ncc-tag": ncc_tag,
                "https://api.adaptemy.io/contentmetadata/ncc-senario": ncc_senario,
                "https://api.adaptemy.io/contentmetadata/question-details": {
                    "question": {
                        "guid": question_guid,
                        "description": description,
                    },
                    "interactions": [
                        {
                            "guid": question_guid,
                            "type": "mcq-single-answer",
                            "choices": choices,
                        }
                    ],
                },
            },
            "type": "http://adlnet.gov/expapi/activities/question",
        },
        "createdOn": datetime.now(timezone.utc).isoformat(),
        "isEnabled": True,
    }

    return xr_ncc_object_id, ncc_content_object


def post_question_ncc_object(
    question: str,
    description: str,
    answer: str,
    ncc_tag: str,
    assessment_type: str,
    ncc_senario: str = "all-users",
):
    """
    Maps NCC fields into the assessment question content-object format,
    posts it to the content-object endpoint, and returns the created object ID.

    Args:
        question: Question text displayed to the learner.
        description: Additional context or instructions for the question.
        answer: Comma-separated answer options.
        ncc_tag: NCC tag value.
        assessment_type: Assessment type value.
        ncc_senario: NCC scenario selector.
    """
    xr_ncc_object_id, ncc_content_object = build_question_ncc_object(
        question=question,
        description=description,
        answer=answer,
        ncc_tag=ncc_tag,
        assessment_type=assessment_type,
        ncc_senario=ncc_senario,
    )

    response = xAPI_deploy_request(CONTENT_OBJECTS_ENDPOINT, [ncc_content_object])

    if response.status_code != 200:
        try:
            error_detail = response.json()
        except Exception:
            error_detail = response.text
        raise RuntimeError(
            f"Failed to post NCC question object (HTTP {response.status_code}): {error_detail}"
        )

    return xr_ncc_object_id