exerciseAPI.py 10.6 KB
Newer Older
1 2 3 4 5 6 7 8
import uuid
from collections import OrderedDict
from datetime import datetime

import rapidjson as json
from typing import List, Dict

import requests
9 10
from flask_restful import Resource, marshal, abort
from flask_restful.reqparse import RequestParser
11 12 13 14 15 16 17 18 19 20 21 22 23 24

from mcserver.app import db
from mcserver.app.models import ExerciseType, Solution, ExerciseData, Exercise, exercise_fields, AnnisResponse, \
    Phenomenon, TextComplexity, TextComplexityMeasure, UpdateInfo, ResourceType
from mcserver.app.services import AnnotationService, CorpusService, NetworkService, TextComplexityService
from mcserver.config import Config


class ExerciseAPI(Resource):
    """The exercise API resource. It creates exercises for a given text."""

    def __init__(self):
        """Initialize possible arguments for calls to the exercise REST API."""
        # TODO: switch to other request parser, e.g. Marshmallow, because the one used by Flask-RESTful does not allow parsing arguments from different locations, e.g. one argument from 'location=args' and another argument from 'location=form'
25
        self.reqparse: RequestParser = NetworkService.base_request_parser.copy()
26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52
        self.reqparse.add_argument("urn", type=str, required=False, location="form", help="No URN provided")
        self.reqparse.add_argument("type", type=str, required=False, location="form", help="No exercise type provided")
        self.reqparse.add_argument("search_values", type=str, required=False, location="form",
                                   help="No search value provided")
        self.reqparse.add_argument("type_translation", type=str, location="form", required=False,
                                   help="No exercise type translation provided")
        self.reqparse.add_argument("work_author", type=str, location="form", required=False,
                                   help="No work_author provided", default="")
        self.reqparse.add_argument("work_title", type=str, required=False, location="form",
                                   help="No work title provided", default="")
        self.reqparse.add_argument("instructions", type=str, required=False, location="form", default="")
        self.reqparse.add_argument("general_feedback", type=str, required=False, location="form", default=" ")
        self.reqparse.add_argument("correct_feedback", type=str, required=False, location="form", default=" ")
        self.reqparse.add_argument("partially_correct_feedback", type=str, required=False, location="form", default=" ")
        self.reqparse.add_argument("incorrect_feedback", type=str, required=False, location="form", default=" ")
        self.reqparse.add_argument("eid", type=str, required=False, location="args", help="No exercise ID provided")
        super(ExerciseAPI, self).__init__()

    def get(self):
        args: dict = self.reqparse.parse_args()
        eid: str = args["eid"]
        exercise: Exercise = Exercise.query.filter_by(eid=eid).first()
        if exercise is None:
            abort(404)
        ar: AnnisResponse = CorpusService.get_corpus(cts_urn=exercise.urn, is_csm=False)
        if not ar.nodes:
            abort(404)
53 54
        exercise.last_access_time = datetime.utcnow()
        db.session.commit()
55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 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 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164
        exercise_type: ExerciseType = ExerciseType(exercise.exercise_type)
        ar.solutions = json.loads(exercise.solutions)
        ar.uri = exercise.uri
        ar.exercise_id = exercise.eid
        ar.exercise_type = exercise_type.value
        return NetworkService.make_json_response(ar.__dict__)

    def post(self):
        # get request arguments
        args: dict = self.reqparse.parse_args()
        urn: str = args["urn"]
        exercise_type: ExerciseType = ExerciseType(args["type"])
        search_values_json: str = args["search_values"]
        search_values_list: List[str] = json.loads(search_values_json)
        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]
        # 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))
        # 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)
        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(graph_data_raw=response["graph_data_raw"], solutions=solutions, args=args,
                                              conll=response["conll"], search_values=args["search_values"], urn=urn)
        return NetworkService.make_json_response(ar.__dict__)


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


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:
        abort(500)


def make_new_exercise(solutions: List[Solution], args: dict, search_values: str, graph_data_raw: dict,
                      conll: str, urn: str) -> AnnisResponse:
    """ 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)
    exercise_type = args["type"]
    # for markWords exercises, add the maximum number of correct solutions to the description
    instructions: str = args["instructions"] + (
        f"({len(solutions)})" if exercise_type == ExerciseType.markWords.value else "")
    # 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,
                                                           xml_guid=xml_guid, correct_feedback=args["correct_feedback"],
                                                           partially_correct_feedback=args[
                                                               "partially_correct_feedback"],
                                                           incorrect_feedback=args["incorrect_feedback"],
                                                           general_feedback=args["general_feedback"],
                                                           exercise_type_translation=args.get("type_translation", ""),
                                                           conll=conll, work_author=args["work_author"],
                                                           work_title=args["work_title"], search_values=search_values,
                                                           urn=urn)
    # marshal the whole object so we can get the right URI for download purposes
    new_exercise_marshal: OrderedDict = marshal(new_exercise, exercise_fields)
    # create a response
    return AnnisResponse(solutions=json.loads(new_exercise.solutions), uri=new_exercise_marshal["uri"],
                         exercise_id=xml_guid)


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
    solutions = adjust_solutions(exercise_data=exercise_data, solutions=solutions, exercise_type=exercise_type)
    quiz_solutions: str = json.dumps([x.serialize() for x in solutions])
    tc: TextComplexity = TextComplexityService.text_complexity(TextComplexityMeasure.all.name, urn, False,
                                                               exercise_data.graph)
    new_exercise: Exercise = Exercise(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, 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, uri=exercise_data.uri, urn=urn)
    # add the mapped exercise to the database
    db.session.add(new_exercise)
    ui_exercises: UpdateInfo = UpdateInfo.query.filter_by(resource_type=ResourceType.exercise_list.name).first()
    ui_exercises.last_modified_time = datetime.utcnow()
    db.session.commit()
    return new_exercise