duckling_base_processing_task.py

from abc import abstractmethod, ABC
from pprint import pprint
from typing import Any

import requests

from sziszapangma.integration.repository.experiment_repository import ExperimentRepository
from sziszapangma.integration.task.processing_task import ProcessingTask
from sziszapangma.model.relation_manager import RelationManager


class DucklingBaseProcessingTask(ProcessingTask, ABC):
    _duckling_endpoint: str
    _duckling_language: str
    _input_property_name: str
    _duckling_property_name: str

    def __init__(self, duckling_endpoint: str, duckling_language: str, input_property_name: str,
                 duckling_property_name: str, task_name: str, require_update: bool):
        super().__init__(task_name, require_update)
        self._duckling_endpoint = duckling_endpoint
        self._duckling_language = duckling_language
        self._input_property_name = input_property_name
        self._duckling_property_name = duckling_property_name

    def process_duckling(self, text: str) -> Any:
        return requests.post(self._duckling_endpoint, data={'locale': self._duckling_language, 'text': text}).json()

    def run_single_process(self, record_id: str, experiment_repository: ExperimentRepository,
                           relation_manager: RelationManager):
        test_property = experiment_repository.get_property_for_key(
            record_id, self._input_property_name)
        duckling_result = self.process_duckling(self.get_transcript_to_process(test_property))
        print('duckling_result_len', len(duckling_result))
        experiment_repository.update_property_for_key(record_id, self._duckling_property_name, duckling_result)

    @abstractmethod
    def get_transcript_to_process(self, property_value: Any) -> str:
        pass

    def skip_for_record(self, record_id: str, experiment_repository: ExperimentRepository) -> bool:
        return experiment_repository.property_exists(record_id, self._duckling_property_name)