"""
Helper methods related to EdxNotes.
"""
import json
import logging
import requests
from requests.exceptions import RequestException
from uuid import uuid4
from json import JSONEncoder
from datetime import datetime
from courseware.access import has_access
from courseware.views import get_current_child
from django.conf import settings
from django.core.urlresolvers import reverse
from django.core.exceptions import ImproperlyConfigured
from django.utils.translation import ugettext as _
from capa.util import sanitize_html
from student.models import anonymous_id_for_user
from xmodule.modulestore.django import modulestore
from xmodule.modulestore.exceptions import ItemNotFoundError
from util.date_utils import get_default_time_display
from dateutil.parser import parse as dateutil_parse
from provider.oauth2.models import AccessToken, Client
import oauth2_provider.oidc as oidc
from provider.utils import now
from opaque_keys.edx.keys import UsageKey
from .exceptions import EdxNotesParseError, EdxNotesServiceUnavailable
log = logging.getLogger(__name__)
HIGHLIGHT_TAG = "span"
HIGHLIGHT_CLASS = "note-highlight"
class NoteJSONEncoder(JSONEncoder):
"""
Custom JSON encoder that encode datetime objects to appropriate time strings.
"""
# pylint: disable=method-hidden
def default(self, obj):
if isinstance(obj, datetime):
return get_default_time_display(obj)
return json.JSONEncoder.default(self, obj)
def get_id_token(user):
"""
Generates JWT ID-Token, using or creating user's OAuth access token.
"""
try:
client = Client.objects.get(name="edx-notes")
except Client.DoesNotExist:
raise ImproperlyConfigured("OAuth2 Client with name 'edx-notes' is not present in the DB")
try:
access_token = AccessToken.objects.get(
client=client,
user=user,
expires__gt=now()
)
except AccessToken.DoesNotExist:
access_token = AccessToken(client=client, user=user)
access_token.save()
id_token = oidc.id_token(access_token)
secret = id_token.access_token.client.client_secret
return id_token.encode(secret)
def get_token_url(course_id):
"""
Returns token url for the course.
"""
return reverse("get_token", kwargs={
"course_id": unicode(course_id),
})
def send_request(user, course_id, path="", query_string=None):
"""
Sends a request with appropriate parameters and headers.
"""
url = get_internal_endpoint(path)
params = {
"user": anonymous_id_for_user(user, None),
"course_id": unicode(course_id).encode("utf-8"),
}
if query_string:
params.update({
"text": query_string,
"highlight": True,
"highlight_tag": HIGHLIGHT_TAG,
"highlight_class": HIGHLIGHT_CLASS,
})
try:
response = requests.get(
url,
headers={
"x-annotator-auth-token": get_id_token(user)
},
params=params
)
except RequestException:
raise EdxNotesServiceUnavailable(_("EdxNotes Service is unavailable. Please try again in a few minutes."))
return response
def get_parent_unit(xblock):
"""
Find vertical that is a unit, not just some container.
"""
while xblock:
xblock = xblock.get_parent()
if xblock is None:
return None
parent = xblock.get_parent()
if parent is None:
return None
if parent.category == 'sequential':
return xblock
def preprocess_collection(user, course, collection):
"""
Prepare `collection(notes_list)` provided by edx-notes-api
for rendering in a template:
add information about ancestor blocks,
convert "updated" to date
Raises:
ItemNotFoundError - when appropriate module is not found.
"""
# pylint: disable=too-many-statements
store = modulestore()
filtered_collection = list()
cache = {}
with store.bulk_operations(course.id):
for model in collection:
update = {
u"text": sanitize_html(model["text"]),
u"quote": sanitize_html(model["quote"]),
u"updated": dateutil_parse(model["updated"]),
}
if "tags" in model:
update[u"tags"] = [sanitize_html(tag) for tag in model["tags"]]
model.update(update)
usage_id = model["usage_id"]
if usage_id in cache:
model.update(cache[usage_id])
filtered_collection.append(model)
continue
usage_key = UsageKey.from_string(usage_id)
# Add a course run if necessary.
usage_key = usage_key.replace(course_key=store.fill_in_run(usage_key.course_key))
try:
item = store.get_item(usage_key)
except ItemNotFoundError:
log.debug("Module not found: %s", usage_key)
continue
if not has_access(user, "load", item, course_key=course.id):
log.debug("User %s does not have an access to %s", user, item)
continue
unit = get_parent_unit(item)
if unit is None:
log.debug("Unit not found: %s", usage_key)
continue
section = unit.get_parent()
if not section:
log.debug("Section not found: %s", usage_key)
continue
if section in cache:
usage_context = cache[section]
usage_context.update({
"unit": get_module_context(course, unit),
})
model.update(usage_context)
cache[usage_id] = cache[unit] = usage_context
filtered_collection.append(model)
continue
chapter = section.get_parent()
if not chapter:
log.debug("Chapter not found: %s", usage_key)
continue
if chapter in cache:
usage_context = cache[chapter]
usage_context.update({
"unit": get_module_context(course, unit),
"section": get_module_context(course, section),
})
model.update(usage_context)
cache[usage_id] = cache[unit] = cache[section] = usage_context
filtered_collection.append(model)
continue
usage_context = {
"unit": get_module_context(course, unit),
"section": get_module_context(course, section),
"chapter": get_module_context(course, chapter),
}
model.update(usage_context)
cache[usage_id] = cache[unit] = cache[section] = cache[chapter] = usage_context
filtered_collection.append(model)
return filtered_collection
def get_module_context(course, item):
"""
Returns dispay_name and url for the parent module.
"""
item_dict = {
'location': unicode(item.location),
'display_name': item.display_name_with_default,
}
if item.category == 'chapter' and item.get_parent():
# course is a locator w/o branch and version
# so for uniformity we replace it with one that has them
course = item.get_parent()
item_dict['index'] = get_index(item_dict['location'], course.children)
elif item.category == 'vertical':
section = item.get_parent()
chapter = section.get_parent()
# Position starts from 1, that's why we add 1.
position = get_index(unicode(item.location), section.children) + 1
item_dict['url'] = reverse('courseware_position', kwargs={
'course_id': unicode(course.id),
'chapter': chapter.url_name,
'section': section.url_name,
'position': position,
})
if item.category in ('chapter', 'sequential'):
item_dict['children'] = [unicode(child) for child in item.children]
return item_dict
def get_index(usage_key, children):
"""
Returns an index of the child with `usage_key`.
"""
children = [unicode(child) for child in children]
return children.index(usage_key)
def search(user, course, query_string):
"""
Returns search results for the `query_string(str)`.
"""
response = send_request(user, course.id, "search", query_string)
try:
content = json.loads(response.content)
collection = content["rows"]
except (ValueError, KeyError):
log.warning("invalid JSON: %s", response.content)
raise EdxNotesParseError(_("Server error. Please try again in a few minutes."))
content.update({
"rows": preprocess_collection(user, course, collection)
})
return json.dumps(content, cls=NoteJSONEncoder)
def get_notes(user, course):
"""
Returns all notes for the user.
"""
response = send_request(user, course.id, "annotations")
try:
collection = json.loads(response.content)
except ValueError:
return None
if not collection:
return None
return json.dumps(preprocess_collection(user, course, collection), cls=NoteJSONEncoder)
def get_endpoint(api_url, path=""):
"""
Returns edx-notes-api endpoint.
Arguments:
api_url (str): base url to the notes api
path (str): path to the resource
Returns:
str: full endpoint to the notes api
"""
try:
if not api_url.endswith("/"):
api_url += "/"
if path:
if path.startswith("/"):
path = path.lstrip("/")
if not path.endswith("/"):
path += "/"
return api_url + path
except (AttributeError, KeyError):
raise ImproperlyConfigured(_("No endpoint was provided for EdxNotes."))
def get_public_endpoint(path=""):
"""Get the full path to a resource on the public notes API."""
return get_endpoint(settings.EDXNOTES_PUBLIC_API, path)
def get_internal_endpoint(path=""):
"""Get the full path to a resource on the private notes API."""
return get_endpoint(settings.EDXNOTES_INTERNAL_API, path)
def get_course_position(course_module):
"""
Return the user's current place in the course.
If this is the user's first time, leads to COURSE/CHAPTER/SECTION.
If this isn't the users's first time, leads to COURSE/CHAPTER.
If there is no current position in the course or chapter, then selects
the first child.
"""
urlargs = {'course_id': unicode(course_module.id)}
chapter = get_current_child(course_module, min_depth=1)
if chapter is None:
log.debug("No chapter found when loading current position in course")
return None
urlargs['chapter'] = chapter.url_name
if course_module.position is not None:
return {
'display_name': chapter.display_name_with_default,
'url': reverse('courseware_chapter', kwargs=urlargs),
}
# Relying on default of returning first child
section = get_current_child(chapter, min_depth=1)
if section is None:
log.debug("No section found when loading current position in course")
return None
urlargs['section'] = section.url_name
return {
'display_name': section.display_name_with_default,
'url': reverse('courseware_section', kwargs=urlargs)
}
def generate_uid():
"""
Generates unique id.
"""
return uuid4().int # pylint: disable=no-member
def is_feature_enabled(course):
"""
Returns True if Student Notes feature is enabled for the course,
False otherwise.
In order for the application to be enabled it must be:
1) enabled globally via FEATURES.
2) present in the course tab configuration.
3) Harvard Annotation Tool must be disabled for the course.
"""
return (settings.FEATURES.get("ENABLE_EDXNOTES")
and [t for t in course.tabs if t["type"] == "edxnotes"] # tab found
and not is_harvard_notes_enabled(course))
def is_harvard_notes_enabled(course):
"""
Returns True if Harvard Annotation Tool is enabled for the course,
False otherwise.
Checks for 'textannotation', 'imageannotation', 'videoannotation' in the list
of advanced modules of the course.
"""
modules = set(['textannotation', 'imageannotation', 'videoannotation'])
return bool(modules.intersection(course.advanced_modules))