Spaces:
Runtime error
Runtime error
import re | |
from collections.abc import Mapping | |
from logging import getLogger | |
import datetime as dt | |
from dateutil.parser import isoparse | |
from fuzzywuzzy import fuzz | |
from fuzzywuzzy import process | |
from mathtext_fastapi.intent_classification import predict_message_intent | |
from mathtext_fastapi.logging import prepare_message_data_for_logging | |
from mathtext.sentiment import sentiment | |
from mathtext.text2int import text2int, TOKENS2INT_ERROR_INT | |
log = getLogger(__name__) | |
PAYLOAD_VALUE_TYPES = { | |
'author_id': str, | |
'author_type': str, | |
'contact_uuid': str, | |
'message_body': str, | |
'message_direction': str, | |
'message_id': str, | |
'message_inserted_at': str, | |
'message_updated_at': str, | |
} | |
def build_nlu_response_object(nlu_type, data, confidence): | |
""" Turns nlu results into an object to send back to Turn.io | |
Inputs | |
- nlu_type: str - the type of nlu run (integer or sentiment-analysis) | |
- data: str/int - the student message | |
- confidence: - the nlu confidence score (sentiment) or '' (integer) | |
>>> build_nlu_response_object('integer', 8, 0) | |
{'type': 'integer', 'data': 8, 'confidence': 0} | |
>>> build_nlu_response_object('sentiment', 'POSITIVE', 0.99) | |
{'type': 'sentiment', 'data': 'POSITIVE', 'confidence': 0.99} | |
""" | |
return { | |
'type': nlu_type, | |
'data': data, | |
'confidence': confidence | |
} | |
# def test_for_float_or_int(message_data, message_text): | |
# nlu_response = {} | |
# if type(message_text) == int or type(message_text) == float: | |
# nlu_response = build_nlu_response_object('integer', message_text, '') | |
# prepare_message_data_for_logging(message_data, nlu_response) | |
# return nlu_response | |
def test_for_number_sequence(message_text_arr, message_data, message_text): | |
""" Determines if the student's message is a sequence of numbers | |
>>> test_for_number_sequence(['1','2','3'], {"author_id": "57787919091", "author_type": "OWNER", "contact_uuid": "df78gsdf78df", "message_body": "I am tired", "message_direction": "inbound", "message_id": "dfgha789789ag9ga", "message_inserted_at": "2023-01-10T02:37:28.487319Z", "message_updated_at": "2023-01-10T02:37:28.487319Z"}, '1, 2, 3') | |
{'type': 'integer', 'data': '1,2,3', 'confidence': 0} | |
>>> test_for_number_sequence(['a','b','c'], {"author_id": "57787919091", "author_type": "OWNER", "contact_uuid": "df78gsdf78df", "message_body": "I am tired", "message_direction": "inbound", "message_id": "dfgha789789ag9ga", "message_inserted_at": "2023-01-10T02:37:28.487319Z", "message_updated_at": "2023-01-10T02:37:28.487319Z"}, 'a, b, c') | |
{} | |
""" | |
nlu_response = {} | |
if all(ele.isdigit() for ele in message_text_arr): | |
nlu_response = build_nlu_response_object( | |
'integer', | |
','.join(message_text_arr), | |
0 | |
) | |
prepare_message_data_for_logging(message_data, nlu_response) | |
return nlu_response | |
def run_text2int_on_each_list_item(message_text_arr): | |
""" Attempts to convert each list item to an integer | |
Input | |
- message_text_arr: list - a set of text extracted from the student message | |
Output | |
- student_response_arr: list - a set of integers (32202 for error code) | |
>>> run_text2int_on_each_list_item(['1','2','3']) | |
[1, 2, 3] | |
""" | |
student_response_arr = [] | |
for student_response in message_text_arr: | |
int_api_resp = text2int(student_response.lower()) | |
student_response_arr.append(int_api_resp) | |
return student_response_arr | |
def run_sentiment_analysis(message_text): | |
""" Evaluates the sentiment of a student message | |
>>> run_sentiment_analysis("I am tired") | |
[{'label': 'NEGATIVE', 'score': 0.9997807145118713}] | |
>>> run_sentiment_analysis("I am full of joy") | |
[{'label': 'POSITIVE', 'score': 0.999882698059082}] | |
""" | |
# TODO: Add intent labelling here | |
# TODO: Add logic to determine whether intent labeling or sentiment analysis is more appropriate (probably default to intent labeling) | |
return sentiment(message_text) | |
def run_intent_classification(message_text): | |
""" Process a student's message using basic fuzzy text comparison | |
>>> run_intent_classification("exit") | |
{'type': 'intent', 'data': 'exit', 'confidence': 1.0} | |
>>> run_intent_classification("exi") | |
{'type': 'intent', 'data': 'exit', 'confidence': 0.86} | |
>>> run_intent_classification("eas") | |
{'type': 'intent', 'data': '', 'confidence': 0} | |
>>> run_intent_classification("hard") | |
{'type': 'intent', 'data': '', 'confidence': 0} | |
>>> run_intent_classification("hardier") | |
{'type': 'intent', 'data': 'harder', 'confidence': 0.92} | |
""" | |
label = '' | |
ratio = 0 | |
nlu_response = {'type': 'intent', 'data': label, 'confidence': ratio} | |
keywords = [ | |
'easier', | |
'exit', | |
'harder', | |
'hint', | |
'next', | |
'stop', | |
'tired', | |
'tomorrow', | |
'finished', | |
'help', | |
'easier', | |
'easy', | |
'support', | |
'skip', | |
'menu' | |
] | |
try: | |
tokens = re.findall(r"[-a-zA-Z'_]+", message_text.lower()) | |
except AttributeError: | |
tokens = '' | |
for keyword in keywords: | |
try: | |
tok, score = process.extractOne(keyword, tokens, scorer=fuzz.ratio) | |
except: | |
score = 0 | |
if score > 80: | |
nlu_response['data'] = keyword | |
nlu_response['confidence'] = score | |
return nlu_response | |
def payload_is_valid(payload_object): | |
""" | |
>>> payload_is_valid({'author_id': '+5555555', 'author_type': 'OWNER', 'contact_uuid': '3246-43ad-faf7qw-zsdhg-dgGdg', 'message_body': 'thirty one', 'message_direction': 'inbound', 'message_id': 'SDFGGwafada-DFASHA4aDGA', 'message_inserted_at': '2022-07-05T04:00:34.03352Z', 'message_updated_at': '2023-04-06T10:08:23.745072Z'}) | |
True | |
>>> payload_is_valid({"author_id": "@event.message._vnd.v1.chat.owner", "author_type": "@event.message._vnd.v1.author.type", "contact_uuid": "@event.message._vnd.v1.chat.contact_uuid", "message_body": "@event.message.text.body", "message_direction": "@event.message._vnd.v1.direction", "message_id": "@event.message.id", "message_inserted_at": "@event.message._vnd.v1.chat.inserted_at", "message_updated_at": "@event.message._vnd.v1.chat.updated_at"}) | |
False | |
""" | |
try: | |
isinstance( | |
isoparse(payload_object.get('message_inserted_at','')), | |
dt.datetime | |
) | |
isinstance( | |
isoparse(payload_object.get('message_updated_at','')), | |
dt.datetime | |
) | |
except ValueError: | |
return False | |
return ( | |
isinstance(payload_object, Mapping) and | |
isinstance(payload_object.get('author_id'), str) and | |
isinstance(payload_object.get('author_type'), str) and | |
isinstance(payload_object.get('contact_uuid'), str) and | |
isinstance(payload_object.get('message_body'), str) and | |
isinstance(payload_object.get('message_direction'), str) and | |
isinstance(payload_object.get('message_id'), str) and | |
isinstance(payload_object.get('message_inserted_at'), str) and | |
isinstance(payload_object.get('message_updated_at'), str) | |
) | |
def log_payload_errors(payload_object): | |
errors = [] | |
try: | |
assert isinstance(payload_object, Mapping) | |
except Exception as e: | |
log.error(f'Invalid HTTP request payload object: {e}') | |
errors.append(e) | |
for k, typ in PAYLOAD_VALUE_TYPES.items(): | |
try: | |
assert isinstance(payload_object.get(k), typ) | |
except Exception as e: | |
log.error(f'Invalid HTTP request payload object: {e}') | |
errors.append(e) | |
try: | |
assert isinstance( | |
dt.datetime.fromisoformat(payload_object.get('message_inserted_at')), | |
dt.datetime | |
) | |
except Exception as e: | |
log.error(f'Invalid HTTP request payload object: {e}') | |
errors.append(e) | |
try: | |
isinstance( | |
dt.datetime.fromisoformat(payload_object.get('message_updated_at')), | |
dt.datetime | |
) | |
except Exception as e: | |
log.error(f'Invalid HTTP request payload object: {e}') | |
errors.append(e) | |
return errors | |
def evaluate_message_with_nlu(message_data): | |
""" Process a student's message using NLU functions and send the result | |
>>> evaluate_message_with_nlu({"author_id": "57787919091", "author_type": "OWNER", "contact_uuid": "df78gsdf78df", "message_body": "8", "message_direction": "inbound", "message_id": "dfgha789789ag9ga", "message_inserted_at": "2023-01-10T02:37:28.487319Z", "message_updated_at": "2023-01-10T02:37:28.487319Z"}) | |
{'type': 'integer', 'data': 8, 'confidence': 0} | |
>>> evaluate_message_with_nlu({"author_id": "57787919091", "author_type": "OWNER", "contact_uuid": "df78gsdf78df", "message_body": "I am tired", "message_direction": "inbound", "message_id": "dfgha789789ag9ga", "message_inserted_at": "2023-01-10T02:37:28.487319Z", "message_updated_at": "2023-01-10T02:37:28.487319Z"}) | |
{'type': 'sentiment', 'data': 'NEGATIVE', 'confidence': 0.9997807145118713} | |
""" | |
# Keeps system working with two different inputs - full and filtered @event object | |
# Call validate payload | |
log.info(f'Starting evaluate message: {message_data}') | |
if not payload_is_valid(message_data): | |
log_payload_errors(message_data) | |
return {'type': 'error', 'data': TOKENS2INT_ERROR_INT, 'confidence': 0} | |
try: | |
message_text = str(message_data.get('message_body', '')) | |
except: | |
log.error(f'Invalid request payload: {message_data}') | |
# use python logging system to do this// | |
return {'type': 'error', 'data': TOKENS2INT_ERROR_INT, 'confidence': 0} | |
# Run intent classification only for keywords | |
intent_api_response = run_intent_classification(message_text) | |
if intent_api_response['data']: | |
prepare_message_data_for_logging(message_data, intent_api_response) | |
return intent_api_response | |
number_api_resp = text2int(message_text.lower()) | |
if number_api_resp == TOKENS2INT_ERROR_INT: | |
# Run intent classification with logistic regression model | |
predicted_label = predict_message_intent(message_text) | |
if predicted_label['confidence'] > 0.01: | |
nlu_response = predicted_label | |
else: | |
# Run sentiment analysis | |
sentiment_api_resp = sentiment(message_text) | |
nlu_response = build_nlu_response_object( | |
'sentiment', | |
sentiment_api_resp[0]['label'], | |
sentiment_api_resp[0]['score'] | |
) | |
else: | |
nlu_response = build_nlu_response_object( | |
'integer', | |
number_api_resp, | |
0 | |
) | |
prepare_message_data_for_logging(message_data, nlu_response) | |
return nlu_response | |