utils.py

                    theme=theme,
                    addon_css=actual_addon_css,
                    addon_js=actual_addon_js,
                    config=actual_config), help_text

# }}}


# {{{ facility processing

def get_facilities_config(request=None):
    # type: (Optional[http.HttpRequest]) -> Optional[Dict[Text, Dict[Text, Any]]]
    from django.conf import settings

    # This is called during offline validation, where Django isn't really set up.
    # The getattr makes this usable.
    facilities = getattr(settings, "RELATE_FACILITIES", None)
    if facilities is None:
        # Only happens during offline validation. Suppresses errors there.
        return None

    if callable(facilities):
        from course.views import get_now_or_fake_time
        now_datetime = get_now_or_fake_time(request)

        result = facilities(now_datetime)
        if not isinstance(result, dict):
            raise RuntimeError("RELATE_FACILITIES must return a dictionary")
        return result
    else:
        return facilities


class FacilityFindingMiddleware(object):
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        pretend_facilities = request.session.get("relate_pretend_facilities")

        if pretend_facilities is not None:
            facilities = pretend_facilities
        else:
            import ipaddress
            remote_address = ipaddress.ip_address(
                    six.text_type(request.META['REMOTE_ADDR']))

            facilities = set()

            for name, props in six.iteritems(get_facilities_config(request)):
                ip_ranges = props.get("ip_ranges", [])
                for ir in ip_ranges:
                    if remote_address in ipaddress.ip_network(six.text_type(ir)):
                        facilities.add(name)

        request.relate_facilities = frozenset(facilities)

        return self.get_response(request)

# }}}


def get_col_contents_or_empty(row, index):
    if index >= len(row):
        return ""
    else:
        return row[index]


def csv_data_importable(file_contents, column_idx_list, header_count):
    import csv
    spamreader = csv.reader(file_contents)
    n_header_row = 0
    try:
        if six.PY2:
            row0 = spamreader.next()
        else:
            row0 = spamreader.__next__()
    except Exception as e:
        err_msg = type(e).__name__
        err_str = str(e)
        if err_msg == "Error":
            err_msg = ""
        else:
            err_msg += ": "
        err_msg += err_str

        if "line contains NULL byte" in err_str:
            err_msg = err_msg.rstrip(".") + ". "
            err_msg += _("Are you sure the file is a CSV file other "
                         "than a Microsoft Excel file?")

        return False, (
            string_concat(
                pgettext_lazy("Starting of Error message", "Error"),
                ": %s" % err_msg))

    from itertools import chain

    for row in chain([row0], spamreader):
        n_header_row += 1
        if n_header_row <= header_count:
            continue
        try:
            for column_idx in column_idx_list:
                if column_idx is not None:
                    six.text_type(get_col_contents_or_empty(row, column_idx-1))
        except UnicodeDecodeError:
            return False, (
                    _("Error: Columns to be imported contain "
                        "non-ASCII characters. "
                        "Please save your CSV file as utf-8 encoded "
                        "and import again.")
            )
        except Exception as e:
            return False, (
                    string_concat(
                        pgettext_lazy("Starting of Error message",
                            "Error"),
                        ": %(err_type)s: %(err_str)s")
                    % {
                        "err_type": type(e).__name__,
                        "err_str": str(e)}
                    )

    return True, ""


def will_use_masked_profile_for_email(recipient_email):
    # type: (Union[None, Text, List[Text]]) -> bool
    if not recipient_email:
        return False
    if not isinstance(recipient_email, list):
        recipient_email = [recipient_email]
    from course.models import Participation  # noqa
    recepient_participations = (
        Participation.objects.filter(
            user__email__in=recipient_email
        ))
    from course.constants import participation_permission as pperm
    for part in recepient_participations:
        if part.has_permission(pperm.view_participant_masked_profile):
            return True
    return False


def get_course_specific_language_choices():
    # type: () -> Tuple[Tuple[str, Any], ...]

    from django.conf import settings
    from collections import OrderedDict

    all_options = ((settings.LANGUAGE_CODE, None),) + tuple(settings.LANGUAGES)
    filtered_options_dict = OrderedDict(all_options)

    def get_default_option():
        # type: () -> Tuple[Text, Text]
        # For the default language used, if USE_I18N is True, display
        # "Disabled". Otherwise display its lang info.
        if not settings.USE_I18N:
            formatted_descr = (
                get_formatted_options(settings.LANGUAGE_CODE, None)[1])
        else:
            formatted_descr = _("disabled (i.e., displayed language is "
                                "determined by user's browser preference)")
        return "", string_concat("%s: " % _("Default"), formatted_descr)

    def get_formatted_options(lang_code, lang_descr):
        # type: (Text, Optional[Text]) -> Tuple[Text, Text]
        if lang_descr is None:
            lang_descr = OrderedDict(settings.LANGUAGES).get(lang_code)
            if lang_descr is None:
                try:
                    lang_info = translation.get_language_info(lang_code)
                    lang_descr = lang_info["name_translated"]
                except KeyError:
                    return (lang_code.strip(), lang_code)

        return (lang_code.strip(),
                string_concat(_(lang_descr), " (%s)" % lang_code))

    filtered_options = (
        [get_default_option()]
        + [get_formatted_options(k, v)
           for k, v in six.iteritems(filtered_options_dict)])

    # filtered_options[1] is the option for settings.LANGUAGE_CODE
    # it's already displayed when settings.USE_I18N is False
    if not settings.USE_I18N:
        filtered_options.pop(1)

    return tuple(filtered_options)


class LanguageOverride(ContextDecorator):
    def __init__(self, course, deactivate=False):
        # type: (Course, bool) -> None
        self.course = course
        self.deactivate = deactivate

        if course.force_lang:
            self.language = course.force_lang
        else:
            from django.conf import settings
            self.language = settings.RELATE_ADMIN_EMAIL_LOCALE

    def __enter__(self):
        # type: () -> None
        self.old_language = translation.get_language()
        if self.language is not None:
            translation.activate(self.language)
        else:
            translation.deactivate_all()

    def __exit__(self, exc_type, exc_value, traceback):
        # type: (Any, Any, Any) -> None
        if self.old_language is None:
            translation.deactivate_all()
        elif self.deactivate:
            translation.deactivate()
        else:
            translation.activate(self.old_language)


class RelateJinjaMacroBase(object):
    def __init__(self, course, repo, commit_sha):
        # type: (Optional[Course], Repo_ish, bytes) -> None
        self.course = course
        self.repo = repo
        self.commit_sha = commit_sha

    @property
    def name(self):
        # The name of the method used in the template
        raise NotImplementedError()

    def __call__(self, *args, **kwargs):
        # type: (*Any, **Any) -> Text
        raise NotImplementedError()


# {{{ ipynb utilities

class IpynbJinjaMacro(RelateJinjaMacroBase):
    name = "render_notebook_cells"

    def _render_notebook_cells(self, ipynb_path, indices=None, clear_output=False,
                 clear_markdown=False, **kwargs):
        # type: (Text, Optional[Any], Optional[bool], Optional[bool], **Any) -> Text
        from course.content import get_repo_blob_data_cached
        try:
            ipynb_source = get_repo_blob_data_cached(self.repo, ipynb_path,
                                                     self.commit_sha).decode()

            return self._render_notebook_from_source(
                ipynb_source,
                indices=indices,
                clear_output=clear_output,
                clear_markdown=clear_markdown,
                **kwargs
            )
        except ObjectDoesNotExist:
            raise

    __call__ = _render_notebook_cells  # type: ignore

    def _render_notebook_from_source(
            self, ipynb_source, indices=None,
            clear_output=False, clear_markdown=False, **kwargs):
        # type: (Text, Optional[Any], Optional[bool], Optional[bool], **Any) -> Text
        """
        Get HTML format of ipython notebook so as to be rendered in RELATE flow
        pages.
        :param ipynb_source: the :class:`text` read from a ipython notebook.
        :param indices: a :class:`list` instance, 0-based indices of notebook cells
        which are expected to be rendered.
        :param clear_output: a :class:`bool` instance, indicating whether existing
        execution output of code cells should be removed.
        :param clear_markdown: a :class:`bool` instance, indicating whether markdown
        cells will be ignored..
        :return:
        """
        import nbformat
        from nbformat.reader import parse_json
        nb_source_dict = parse_json(ipynb_source)

        if indices:
            nb_source_dict.update(
                {"cells": [nb_source_dict["cells"][idx] for idx in indices]})

        if clear_markdown:
            nb_source_dict.update(
                {"cells": [cell for cell in nb_source_dict["cells"]
                           if cell['cell_type'] != "markdown"]})

        nb_source_dict.update({"cells": nb_source_dict["cells"]})

        import json
        ipynb_source = json.dumps(nb_source_dict)
        notebook = nbformat.reads(ipynb_source, as_version=4)

        from traitlets.config import Config
        c = Config()

        # This is to prevent execution of arbitrary code from note book
        c.ExecutePreprocessor.enabled = False
        if clear_output:
            c.ClearOutputPreprocessor.enabled = True

        c.CSSHTMLHeaderPreprocessor.enabled = False
        c.HighlightMagicsPreprocessor.enabled = False

        import os

        # Place the template in course template dir
        import course
        template_path = os.path.join(
                os.path.dirname(course.__file__),
                "templates", "course", "jinja2")
        c.TemplateExporter.template_path.append(template_path)

        from nbconvert import HTMLExporter
        html_exporter = HTMLExporter(
            config=c,
            template_file="nbconvert_template.tpl"
        )

        (body, resources) = html_exporter.from_notebook_node(notebook)

        return body


NBCONVERT_PRE_OPEN_RE = re.compile(r"<pre\s*>\s*<relate_ipynb\s*>")
NBCONVERT_PRE_CLOSE_RE = re.compile(r"</relate_ipynb\s*>\s*</pre\s*>")


class NBConvertHTMLPostprocessor(markdown.postprocessors.Postprocessor):
    def run(self, text):
        text = NBCONVERT_PRE_OPEN_RE.sub("", text)
        text = NBCONVERT_PRE_CLOSE_RE.sub("", text)
        return text


class NBConvertExtension(markdown.Extension):
    def extendMarkdown(self, md, md_globals):  # noqa
        md.postprocessors['relate_nbconvert'] = NBConvertHTMLPostprocessor(md)

# }}}


def get_custom_page_types_stop_support_deadline():
    # type: () -> Optional[datetime.datetime]
    from django.conf import settings
    custom_page_types_removed_deadline = getattr(
        settings, "RELATE_CUSTOM_PAGE_TYPES_REMOVED_DEADLINE", None)

    force_deadline = datetime.datetime(2019, 1, 1, 0, 0, 0, 0)

    if (custom_page_types_removed_deadline is None
            or custom_page_types_removed_deadline > force_deadline):
        custom_page_types_removed_deadline = force_deadline

    from relate.utils import localize_datetime
    return localize_datetime(custom_page_types_removed_deadline)

# vim: foldmethod=marker