110 lines
4.2 KiB
Python
110 lines
4.2 KiB
Python
import json
|
|
import re
|
|
import requests
|
|
from sentry_sdk import add_breadcrumb
|
|
|
|
from core.logger import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
# from spellcheck.client import TaskbaseClient
|
|
# client = TaskbaseClient('info@iterativ.ch', 'myverysafepassword1234', 'https://dev-iterativ.taskbase.com')
|
|
# client.spellcheck('aOciP9H7tNu7pLsR4ohllk', 'Dies ist ein Sats mit filen Felern')
|
|
|
|
class TaskbaseException(Exception):
|
|
pass
|
|
|
|
|
|
class TaskbaseClient:
|
|
# def __init__(self, resource_url):
|
|
# # resource url should be in the form https://username:password@baseurl
|
|
# pattern = re.compile(r'(\w+)://(\w+):(\w+)@([a-zA-Z0-9.]+)')
|
|
# scheme, username, password, url = pattern.match(resource_url).groups()
|
|
# self.username = username
|
|
# self.password = password
|
|
# self.token = None
|
|
# self.base_url = '{}://{}'.format(scheme, url)
|
|
def __init__(self, username, password, base_url):
|
|
self.username = username
|
|
self.password = password
|
|
self.base_url = base_url
|
|
|
|
self.token = None
|
|
|
|
def login(self):
|
|
payload = {
|
|
'username': self.username,
|
|
'password': self.password
|
|
}
|
|
response = requests.post('{}/api/login'.format(self.base_url), json=payload)
|
|
if response.ok:
|
|
data = response.json()
|
|
self.token = data['accessToken']
|
|
else:
|
|
if response.status_code == 401:
|
|
raise TaskbaseException('Unauthorized')
|
|
else:
|
|
raise TaskbaseException(f'Error Code: {response.status_code}')
|
|
|
|
def register_assignment(self, assignment):
|
|
if self.token is None:
|
|
self.login()
|
|
|
|
headers = {'Authorization': 'Bearer {}'.format(self.token), 'Content-Type': 'application/json'}
|
|
payload = {'type': 'SPELL_CHECK', 'tenantId': 4}
|
|
# we first need to register the assignment via API, so we get an ID
|
|
response = requests.post('{}/api/task'.format(self.base_url), json=payload, headers=headers)
|
|
if response.ok:
|
|
data = response.json()
|
|
assignment.taskbase_id = data['id']
|
|
assignment.save()
|
|
|
|
payload = {
|
|
"id": assignment.taskbase_id,
|
|
"type": "SPELL_CHECK",
|
|
"documentType": "SPELL_CHECK",
|
|
"title": assignment.title,
|
|
"description": assignment.assignment,
|
|
"solutionSteps": [assignment.solution] if assignment.solution is not None else []
|
|
}
|
|
|
|
# we can then update the task via API with the description, title and sample solution
|
|
response = requests.post('{}/api/task/{}'.format(self.base_url, assignment.taskbase_id), json=payload,
|
|
headers=headers)
|
|
|
|
if not response.ok:
|
|
logger.warning(f'Something went wrong - {response.status_code}: {response.reason} - {response.text}')
|
|
raise TaskbaseException(
|
|
f'Something went wrong - {response.status_code}: {response.reason} - {response.text}')
|
|
else: # todo: define what to do here
|
|
logger.warning(f'Something went wrong - {response.status_code}: {response.reason} - {response.text}')
|
|
raise TaskbaseException(
|
|
f'Something went wrong - {response.status_code}: {response.reason} - {response.text}')
|
|
|
|
def spellcheck(self, task, text):
|
|
if self.token is None:
|
|
self.login()
|
|
|
|
payload = {
|
|
"taskId": task,
|
|
"input": {
|
|
"text": text,
|
|
"type": "SPELL_CHECK"
|
|
}
|
|
}
|
|
|
|
headers = {'Authorization': 'Bearer {}'.format(self.token), 'Content-Type': 'application/json'}
|
|
|
|
response = requests.post('{}/api/grade'.format(self.base_url), json=payload, headers=headers)
|
|
if response.status_code == 200:
|
|
return response.content
|
|
else: # todo: define what to do here
|
|
add_breadcrumb(
|
|
category='taskbase',
|
|
message=f'Error:{response.reason} - {response.text}',
|
|
|
|
)
|
|
logger.warning(f'Something went wrong - {response.status_code}: {response.reason} - {response.text}')
|
|
raise Exception('Something went wrong')
|