195 lines
7.2 KiB
Python
195 lines
7.2 KiB
Python
import json
|
|
|
|
import bleach
|
|
import graphene
|
|
import re
|
|
from django.core.exceptions import ValidationError
|
|
from graphene import relay
|
|
|
|
from api.utils import get_object, get_errors
|
|
from book.models import ContentBlock, Chapter, UserGroup
|
|
from book.schema.inputs import ContentBlockInput
|
|
from book.schema.queries import ContentBlockNode
|
|
|
|
|
|
def newlines_to_paragraphs(text):
|
|
parts = re.split(r'[\r\n]+', text)
|
|
paragraphs = ['<p>{}</p>'.format(p.strip()) for p in parts]
|
|
return '\n'.join(paragraphs)
|
|
|
|
|
|
def handle_content_blocks(content_data):
|
|
new_contents = []
|
|
|
|
for content in content_data:
|
|
# todo: add all the content blocks
|
|
# todo: sanitize user inputs!
|
|
if content['type'] == 'text_block':
|
|
new_contents.append({
|
|
'type': 'text_block',
|
|
'value': {
|
|
'text': newlines_to_paragraphs(bleach.clean(content['value']['text'], strip=True))
|
|
}})
|
|
elif content['type'] == 'student_entry':
|
|
pass
|
|
elif content['type'] == 'image_url_block':
|
|
new_contents.append({
|
|
'type': 'image_url_block',
|
|
'value': {
|
|
'url': bleach.clean(content['value']['url'])
|
|
}})
|
|
elif content['type'] == 'link_block':
|
|
new_contents.append({
|
|
'type': 'link_block',
|
|
'value': {
|
|
'text': bleach.clean(content['value']['text']),
|
|
'url': bleach.clean(content['value']['url'])
|
|
}
|
|
})
|
|
elif content['type'] == 'task':
|
|
pass
|
|
elif content['type'] == 'video_block':
|
|
new_contents.append({
|
|
'type': 'video_block',
|
|
'value': {
|
|
'url': bleach.clean(content['value']['url'])
|
|
}})
|
|
elif content['type'] == 'document_block':
|
|
new_contents.append({
|
|
'type': 'document_block',
|
|
'value': {
|
|
'url': bleach.clean(content['value']['url'])
|
|
}})
|
|
|
|
return new_contents
|
|
|
|
|
|
class MutateContentBlock(relay.ClientIDMutation):
|
|
class Input:
|
|
id = graphene.ID(required=True)
|
|
content_block = graphene.Argument(ContentBlockInput)
|
|
|
|
errors = graphene.List(graphene.String)
|
|
content_block = graphene.Field(ContentBlockNode)
|
|
|
|
@classmethod
|
|
def mutate_and_get_payload(cls, *args, **kwargs):
|
|
try:
|
|
id_param = kwargs['id']
|
|
content_block_data = kwargs.get('content_block')
|
|
# type_param = content_block_data.get('type')
|
|
title = content_block_data.get('title', None)
|
|
contents = content_block_data.get('contents', None)
|
|
visibility_list = content_block_data.get('visibility', None)
|
|
|
|
content_block = get_object(ContentBlock, id_param)
|
|
|
|
if visibility_list is not None:
|
|
for v in visibility_list:
|
|
user_group = get_object(UserGroup, v.user_group_id)
|
|
if v.hidden:
|
|
content_block.hidden_for.add(user_group)
|
|
else:
|
|
content_block.hidden_for.remove(user_group)
|
|
|
|
if title is not None:
|
|
content_block.title = title
|
|
|
|
if contents is not None:
|
|
content_block.contents = json.dumps(handle_content_blocks(contents))
|
|
|
|
content_block.save()
|
|
|
|
# content_block.add_sibling(instance=new_content_block, pos='right')
|
|
|
|
# ContentBlock.objects.get()
|
|
# cb.add_sibling()
|
|
|
|
# new_content_block.saveI()
|
|
|
|
# image_instance = get_object(Image, kwargs['id'])
|
|
# if image_instance:
|
|
# image_data = kwargs.get('image')
|
|
# updated_image = update_create_instance(image_instance, image_data, exception=['id', 'tags'])
|
|
# tag_slugs = image_data.get('tag_slugs', '')
|
|
# if tag_slugs is not None:
|
|
# tag_slugs = [t.strip() for t in tag_slugs.split(',') if t.strip()]
|
|
# tags = list(Tag.objects.filter(slug__in=tag_slugs))
|
|
# tag_slugs = [t for t in tag_slugs if t not in [e.slug for e in tags]]
|
|
# updated_image.tags.set(*(tags + tag_slugs))
|
|
|
|
return cls(content_block=content_block)
|
|
|
|
except ValidationError as e:
|
|
errors = get_errors(e)
|
|
except Exception as e:
|
|
errors = ['Error: {}'.format(e)]
|
|
|
|
return cls(content_block=None, errors=errors)
|
|
|
|
|
|
# todo: handle mutation for add and for edit (with the ID of the content block instead of a sibling)
|
|
# todo: merge with above class, for error handling and sibling assignment
|
|
class AddContentBlock(relay.ClientIDMutation):
|
|
class Input:
|
|
content_block = graphene.Argument(ContentBlockInput)
|
|
parent = graphene.ID() # ID of chapter node; new content block will be inserted at the start of it
|
|
after = graphene.ID() # ID of content block node; new content block will be inserted after this content block node
|
|
|
|
new_content_block = graphene.Field(ContentBlockNode)
|
|
errors = graphene.List(graphene.String)
|
|
|
|
@classmethod
|
|
def create_content_block(cls, content_block_data, parent=None, after=None):
|
|
if after is None and parent is None:
|
|
raise Exception('Define either a parent or a sibling id')
|
|
|
|
title = content_block_data.get('title')
|
|
contents = content_block_data.get('contents')
|
|
|
|
new_content_block = ContentBlock(title=title)
|
|
|
|
if parent is not None:
|
|
parent_chapter = get_object(Chapter, parent).specific
|
|
first_sibling = parent_chapter.get_first_child()
|
|
if first_sibling is not None:
|
|
first_sibling.add_sibling(instance=new_content_block, pos='left')
|
|
else:
|
|
parent_chapter.add_child(instance=new_content_block)
|
|
elif after is not None:
|
|
sibling = get_object(ContentBlock, after).specific
|
|
sibling.add_sibling(instance=new_content_block, pos='right')
|
|
|
|
revision = new_content_block.save_revision()
|
|
revision.publish()
|
|
new_content_block.save()
|
|
|
|
new_contents = handle_content_blocks(contents) # can only do this after the content block has been saved
|
|
new_content_block.contents = json.dumps(new_contents)
|
|
new_content_block.save()
|
|
|
|
return new_content_block
|
|
|
|
@classmethod
|
|
def mutate_and_get_payload(cls, root, info, **args):
|
|
try:
|
|
parent = args.get('parent', None)
|
|
after = args.get('after', None)
|
|
new_content_block = cls.create_content_block(content_block_data=args.get('content_block'),
|
|
parent=parent,
|
|
after=after)
|
|
|
|
return cls(new_content_block=new_content_block)
|
|
|
|
except ValidationError as e:
|
|
errors = get_errors(e)
|
|
except Exception as e:
|
|
errors = ['Error: {}'.format(e)]
|
|
|
|
return cls(new_content_block=None, errors=errors)
|
|
|
|
|
|
class BookMutations(object):
|
|
mutate_content_block = MutateContentBlock.Field()
|
|
add_content_block = AddContentBlock.Field()
|