exerciseAPI.py 9.05 KB
Newer Older
1 2
import uuid
from datetime import datetime
3
import connexion
4
import rapidjson as json
5
from typing import List, Dict, Union
6
import requests
7 8
from connexion.lifecycle import ConnexionResponse
from flask import Response
9
from mcserver.app import db
10 11
from mcserver.app.models import ExerciseType, Solution, ExerciseData, AnnisResponse, Phenomenon, TextComplexity, \
    TextComplexityMeasure, ResourceType, ExerciseMC
12 13
from mcserver.app.services import AnnotationService, CorpusService, NetworkService, TextComplexityService
from mcserver.config import Config
14
from mcserver.models_auto import Exercise, TExercise, UpdateInfo
15 16 17 18 19 20 21 22 23 24 25 26 27


def adjust_solutions(exercise_data: ExerciseData, exercise_type: str, solutions: List[Solution]) -> List[Solution]:
    """Adds the content to each SolutionElement."""
    if exercise_type == ExerciseType.matching.value:
        node_id_dict: Dict[str, int] = dict(
            (exercise_data.graph.nodes[i].id, i) for i in range(len(exercise_data.graph.nodes)))
        for solution in solutions:
            solution.target.content = exercise_data.graph.nodes[node_id_dict[solution.target.salt_id]].annis_tok
            solution.value.content = exercise_data.graph.nodes[node_id_dict[solution.value.salt_id]].annis_tok
    return solutions


28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44
def get(eid: str) -> Union[Response, ConnexionResponse]:
    exercise: TExercise = db.session.query(Exercise).filter_by(eid=eid).first()
    if exercise is None:
        return connexion.problem(404, Config.ERROR_TITLE_NOT_FOUND, Config.ERROR_MESSAGE_EXERCISE_NOT_FOUND)
    ar: AnnisResponse = CorpusService.get_corpus(cts_urn=exercise.urn, is_csm=False)
    if not ar.nodes:
        return connexion.problem(404, Config.ERROR_TITLE_NOT_FOUND, Config.ERROR_MESSAGE_CORPUS_NOT_FOUND)
    exercise.last_access_time = datetime.utcnow().timestamp()
    db.session.commit()
    exercise_type: ExerciseType = ExerciseType(exercise.exercise_type)
    ar.solutions = json.loads(exercise.solutions)
    ar.uri = NetworkService.get_exercise_uri(exercise)
    ar.exercise_id = exercise.eid
    ar.exercise_type = exercise_type.value
    return NetworkService.make_json_response(ar.__dict__)


45 46 47 48 49 50 51 52 53 54 55
def get_graph_data(title: str, conll_string_or_urn: str, aqls: List[str], exercise_type: ExerciseType,
                   search_phenomena: List[Phenomenon]):
    """Sends annotated text data or a URN to the Corpus Storage Manager in order to get a graph."""
    url: str = f"{Config.INTERNET_PROTOCOL}{Config.HOST_IP_CSM}:{Config.CORPUS_STORAGE_MANAGER_PORT}"
    data: str = json.dumps(
        dict(title=title, annotations=conll_string_or_urn, aqls=aqls, exercise_type=exercise_type.name,
             search_phenomena=[x.name for x in search_phenomena]))
    response: requests.Response = requests.post(url, data=data)
    try:
        return json.loads(response.text)
    except ValueError:
56
        raise
57 58


59 60 61 62
def make_new_exercise(conll: str, correct_feedback: str, exercise_type: str, general_feedback: str,
                      graph_data_raw: dict, incorrect_feedback: str, instructions: str, partially_correct_feedback: str,
                      search_values: str, solutions: List[Solution], type_translation: str, urn: str,
                      work_author: str, work_title: str) -> AnnisResponse:
63 64 65 66 67 68 69
    """ Creates a new exercise and makes it JSON serializable. """
    # generate a GUID so we can offer the exercise XML as a file download
    xml_guid = str(uuid.uuid4())
    # assemble the mapped exercise data
    ed: ExerciseData = AnnotationService.map_graph_data_to_exercise(graph_data_raw=graph_data_raw, solutions=solutions,
                                                                    xml_guid=xml_guid)
    # for markWords exercises, add the maximum number of correct solutions to the description
70
    instructions += (f"({len(solutions)})" if exercise_type == ExerciseType.markWords.value else "")
71 72 73
    # map the exercise data to our database data model
    new_exercise: Exercise = map_exercise_data_to_database(solutions=solutions, exercise_data=ed,
                                                           exercise_type=exercise_type, instructions=instructions,
74 75 76 77 78 79 80
                                                           xml_guid=xml_guid, correct_feedback=correct_feedback,
                                                           partially_correct_feedback=partially_correct_feedback,
                                                           incorrect_feedback=incorrect_feedback,
                                                           general_feedback=general_feedback,
                                                           exercise_type_translation=type_translation, conll=conll,
                                                           work_author=work_author, work_title=work_title,
                                                           search_values=search_values, urn=urn)
81
    # create a response
82 83 84
    return AnnisResponse(
        solutions=json.loads(new_exercise.solutions), uri=f"{Config.SERVER_URI_FILE}/{new_exercise.eid}",
        exercise_id=xml_guid)
85 86 87 88 89 90 91 92 93 94


def map_exercise_data_to_database(exercise_data: ExerciseData, exercise_type: str, instructions: str, xml_guid: str,
                                  correct_feedback: str, partially_correct_feedback: str, incorrect_feedback: str,
                                  general_feedback: str, exercise_type_translation: str, search_values: str,
                                  solutions: List[Solution], conll: str, work_author: str, work_title: str, urn: str):
    """Maps the exercise data so we can save it to the database."""
    # sort the nodes according to the ordering links
    AnnotationService.sort_nodes(graph_data=exercise_data.graph)
    # add content to solutions
95 96
    solutions: List[Solution] = adjust_solutions(exercise_data=exercise_data, solutions=solutions,
                                                 exercise_type=exercise_type)
97 98 99
    quiz_solutions: str = json.dumps([x.serialize() for x in solutions])
    tc: TextComplexity = TextComplexityService.text_complexity(TextComplexityMeasure.all.name, urn, False,
                                                               exercise_data.graph)
100 101 102 103 104 105 106
    new_exercise: Exercise = ExerciseMC.from_dict(
        conll=conll, correct_feedback=correct_feedback, eid=xml_guid, exercise_type=exercise_type,
        exercise_type_translation=exercise_type_translation, general_feedback=general_feedback,
        incorrect_feedback=incorrect_feedback, instructions=instructions,
        last_access_time=datetime.utcnow().timestamp(), partially_correct_feedback=partially_correct_feedback,
        search_values=search_values, solutions=quiz_solutions, text_complexity=tc.all, work_author=work_author,
        work_title=work_title, urn=urn)
107 108
    # add the mapped exercise to the database
    db.session.add(new_exercise)
109 110 111
    ui_exercises: UpdateInfo = db.session.query(UpdateInfo).filter_by(
        resource_type=ResourceType.exercise_list.name).first()
    ui_exercises.last_modified_time = datetime.utcnow().timestamp()
112 113
    db.session.commit()
    return new_exercise
114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144


def post(exercise_data: dict) -> Union[Response, ConnexionResponse]:
    exercise_type: ExerciseType = ExerciseType(exercise_data["type"])
    search_values_list: List[str] = json.loads(exercise_data["search_values"])
    aqls: List[str] = AnnotationService.map_search_values_to_aql(search_values_list=search_values_list,
                                                                 exercise_type=exercise_type)
    search_phenomena: List[Phenomenon] = [Phenomenon[x.split("=")[0]] for x in search_values_list]
    urn: str = exercise_data.get("urn", "")
    # if there is custom text instead of a URN, immediately annotate it
    conll_string_or_urn: str = urn if CorpusService.is_urn(urn) else AnnotationService.get_udpipe(
        CorpusService.get_raw_text(urn, False))
    try:
        # construct graph from CONLL data
        response: dict = get_graph_data(title=urn, conll_string_or_urn=conll_string_or_urn, aqls=aqls,
                                        exercise_type=exercise_type, search_phenomena=search_phenomena)
    except ValueError:
        return connexion.problem(500, Config.ERROR_TITLE_INTERNAL_SERVER_ERROR,
                                 Config.ERROR_MESSAGE_INTERNAL_SERVER_ERROR)
    solutions_dict_list: List[Dict] = response["solutions"]
    solutions: List[Solution] = [Solution(json_dict=x) for x in solutions_dict_list]
    ar: AnnisResponse = make_new_exercise(
        conll=response["conll"], correct_feedback=exercise_data.get("correct_feedback", ""),
        exercise_type=exercise_data["type"], general_feedback=exercise_data.get("general_feedback", ""),
        graph_data_raw=response["graph_data_raw"], incorrect_feedback=exercise_data.get("incorrect_feedback", ""),
        instructions=exercise_data["instructions"],
        partially_correct_feedback=exercise_data.get("partially_correct_feedback", ""),
        search_values=exercise_data["search_values"], solutions=solutions,
        type_translation=exercise_data.get("type_translation", ""), urn=urn,
        work_author=exercise_data.get("work_author", ""), work_title=exercise_data.get("work_title", ""))
    return NetworkService.make_json_response(ar.__dict__)