<commit_before># -*- coding: utf-8 -*-
import datetime
import functools
import logging

import markdown
import pytz
from django.db.models.expressions import F
from django.db.models.aggregates import Max
from django.core.exceptions import ValidationError
from django.utils import timezone
from framework.auth.core import Auth
from addons.base.models import BaseNodeSettings
from bleach.callbacks import nofollow
from bleach import Cleaner
from functools import partial
from bleach.linkifier import LinkifyFilter
from django.db import models
from framework.forms.utils import sanitize
from markdown.extensions import codehilite, fenced_code, wikilinks
from osf.models import NodeLog, OSFUser, Comment
from osf.models.base import BaseModel, GuidMixin, ObjectIDMixin
from osf.models.spam import SpamStatus
from osf.utils.fields import NonNaiveDateTimeField
from osf.utils.requests import get_request_and_user_id, string_type_request_headers
from osf.exceptions import NodeStateError
from addons.wiki import utils as wiki_utils
from addons.wiki.exceptions import (
    PageCannotRenameError,
    PageConflictError,
)
from website.util import api_v2_url
from website.files.exceptions import VersionNotFoundError
from website import settings

from .exceptions import (
    NameEmptyError,
    NameInvalidError,
    NameMaximumLengthError,
)

logger = logging.getLogger(__name__)

SHAREJS_HOST = 'localhost'
SHAREJS_PORT = 7007
SHAREJS_URL = '{}:{}'.format(SHAREJS_HOST, SHAREJS_PORT)

SHAREJS_DB_NAME = 'sharejs'
SHAREJS_DB_URL = 'mongodb://{}:{}/{}'.format(settings.DB_HOST, settings.DB_PORT, SHAREJS_DB_NAME)

# TODO: Change to release date for wiki change
WIKI_CHANGE_DATE = datetime.datetime.utcfromtimestamp(1423760098).replace(tzinfo=pytz.utc)

def validate_page_name(value):
    value = (value or '').strip()

    if not value:
        # TODO: determine if this if possible anymore, deprecate if not
        raise NameEmptyError('Page name cannot be blank.')
    if value.find('/') != -1:
        raise NameInvalidError('Page name cannot contain forward slashes.')
    if len(value) > 100:
        raise NameMaximumLengthError('Page name cannot be greater than 100 characters.')
    return True

def build_html_output(content, node):
    return markdown.markdown(
        content,
        extensions=[
            wikilinks.WikiLinkExtension(
                configs=[
                    ('base_url', ''),
                    ('end_url', ''),
                    ('build_url', functools.partial(build_wiki_url, node))
                ]
            ),
            fenced_code.FencedCodeExtension(),
            codehilite.CodeHiliteExtension(
                [('css_class', 'highlight')]
            )
        ]
    )

def render_content(content, node):
    html_output = build_html_output(content, node)

    # linkify gets called after santize, because we're adding rel="nofollow"
    #   to <a> elements - but don't want to allow them for other elements.
    sanitized_content = sanitize(html_output, **settings.WIKI_WHITELIST)
    return sanitized_content


def build_wiki_url(node, label, base, end):
    return '/{pid}/wiki/{wname}/'.format(pid=node._id, wname=label)


class WikiVersionNodeManager(models.Manager):

    def get_for_node(self, node, name=None, version=None, id=None):
        if name:
            wiki_page = WikiPage.objects.get_for_node(node, name)
            if not wiki_page:
                return None

            if version == 'previous':
                version = wiki_page.current_version_number - 1
            elif version == 'current' or version is None:
                version = wiki_page.current_version_number
            elif not ((isinstance(version, int) or version.isdigit())):
                return None

            try:
                return wiki_page.get_version(version=version)
            except WikiVersion.DoesNotExist:
                return None
        return WikiVersion.load(id)


class WikiVersion(ObjectIDMixin, BaseModel):
    objects = WikiVersionNodeManager()

    user = models.ForeignKey('osf.OSFUser', null=True, blank=True, on_delete=models.CASCADE)
    wiki_page = models.ForeignKey('WikiPage', null=True, blank=True, on_delete=models.CASCADE, related_name='versions')
    content = models.TextField(default='', blank=True)
    identifier = models.IntegerField(default=1)

    @property
    def is_current(self):
        return not self.wiki_page.deleted and self.id == self.wiki_page.versions.order_by('-created').first().id

    def html(self, node):
        """The cleaned HTML of the page"""
        html_output = build_html_output(self.content, node=node)
        try:
            cleaner = Cleaner(
                tags=settings.WIKI_WHITELIST['tags'],
                attributes=settings.WIKI_WHITELIST['attributes'],
                styles=settings.WIKI_WHITELIST['styles'],
                filters=[partial(LinkifyFilter, callbacks=[nofollow, ])]
            )
            return cleaner.clean(html_output)
        except TypeError:
            logger.warning('Returning unlinkified content.')
            return render_content(self.content, node=node)

    def raw_text(self, node):
        """ The raw text of the page, suitable for using in a test search"""

        return sanitize(self.content, tags=[], strip=True)

    @property
    def rendered_before_update(self):
        return self.created < WIKI_CHANGE_DATE

    def get_draft(self, node):
        """
        Return most recently edited version of wiki, whether that is the
        last saved version or the most recent sharejs draft.
        """

        db = wiki_utils.share_db()
        sharejs_uuid = wiki_utils.get_sharejs_uuid(node, self.wiki_page.page_name)

        doc_item = db['docs'].find_one({'_id': sharejs_uuid})
        if doc_item:
            sharejs_version = doc_item['_v']
            sharejs_timestamp = doc_item['_m']['mtime']
            sharejs_timestamp /= 1000  # Convert to appropriate units
            sharejs_date = datetime.datetime.utcfromtimestamp(sharejs_timestamp).replace(tzinfo=pytz.utc)

            if sharejs_version > 1 and sharejs_date > self.created:
                return doc_item['_data']

        return self.content

    def save(self, *args, **kwargs):
        rv = super(WikiVersion, self).save(*args, **kwargs)
        if self.wiki_page.node:
            self.wiki_page.node.update_search()
        self.wiki_page.modified = self.created
        self.wiki_page.save()
        self.check_spam()
        return rv

    def check_spam(self):
        request, user_id = get_request_and_user_id()
        user = OSFUser.load(user_id)
        request_headers = string_type_request_headers(request)
        node = self.wiki_page.node

        if not settings.SPAM_CHECK_ENABLED:
            return False
        if settings.SPAM_CHECK_PUBLIC_ONLY and not node.is_public:
            return False
        if user.spam_status == SpamStatus.HAM:
            return False

        content = self._get_spam_content(node)
        if not content:
            return
        is_spam = node.do_check_spam(
            user.fullname,
            user.username,
            content,
            request_headers
        )

        logger.info("Node ({}) '{}' smells like {} (tip: {})".format(
            node._id, node.title.encode('utf-8'), 'SPAM' if is_spam else 'HAM', node.spam_pro_tip
        ))
        if is_spam:
            node._check_spam_user(user)
        return is_spam

    def _get_spam_content(self, node):
        content = []
        content.append(self.raw_text(node))
        if not content:
            return None
        return ' '.join(content)

    def clone_version(self, wiki_page, user):
        """Clone a node wiki page.
        :param wiki_page: The wiki_page you want attached to the clone.
        :return: The cloned wiki version
        """<commit_msg>fix: skip spam checks for trusted wiki edits<commit_after># -*- coding: utf-8 -*-
import datetime
import functools
import logging

import markdown
import pytz
from django.db.models.expressions import F
from django.db.models.aggregates import Max
from django.core.exceptions import ValidationError
from django.utils import timezone
from framework.auth.core import Auth
from addons.base.models import BaseNodeSettings
from bleach.callbacks import nofollow
from bleach import Cleaner
from functools import partial
from bleach.linkifier import LinkifyFilter
from django.db import models
from framework.forms.utils import sanitize
from markdown.extensions import codehilite, fenced_code, wikilinks
from osf.models import NodeLog, OSFUser, Comment
from osf.models.base import BaseModel, GuidMixin, ObjectIDMixin
from osf.utils.fields import NonNaiveDateTimeField
from osf.utils.requests import get_request_and_user_id, string_type_request_headers
from osf.exceptions import NodeStateError
from addons.wiki import utils as wiki_utils
from addons.wiki.exceptions import (
    PageCannotRenameError,
    PageConflictError,
)
from website.util import api_v2_url
from website.files.exceptions import VersionNotFoundError
from website import settings

from .exceptions import (
    NameEmptyError,
    NameInvalidError,
    NameMaximumLengthError,
)

logger = logging.getLogger(__name__)

SHAREJS_HOST = 'localhost'
SHAREJS_PORT = 7007
SHAREJS_URL = '{}:{}'.format(SHAREJS_HOST, SHAREJS_PORT)

SHAREJS_DB_NAME = 'sharejs'
SHAREJS_DB_URL = 'mongodb://{}:{}/{}'.format(settings.DB_HOST, settings.DB_PORT, SHAREJS_DB_NAME)

# TODO: Change to release date for wiki change
WIKI_CHANGE_DATE = datetime.datetime.utcfromtimestamp(1423760098).replace(tzinfo=pytz.utc)

def validate_page_name(value):
    value = (value or '').strip()

    if not value:
        # TODO: determine if this if possible anymore, deprecate if not
        raise NameEmptyError('Page name cannot be blank.')
    if value.find('/') != -1:
        raise NameInvalidError('Page name cannot contain forward slashes.')
    if len(value) > 100:
        raise NameMaximumLengthError('Page name cannot be greater than 100 characters.')
    return True

def build_html_output(content, node):
    return markdown.markdown(
        content,
        extensions=[
            wikilinks.WikiLinkExtension(
                configs=[
                    ('base_url', ''),
                    ('end_url', ''),
                    ('build_url', functools.partial(build_wiki_url, node))
                ]
            ),
            fenced_code.FencedCodeExtension(),
            codehilite.CodeHiliteExtension(
                [('css_class', 'highlight')]
            )
        ]
    )

def render_content(content, node):
    html_output = build_html_output(content, node)

    # linkify gets called after santize, because we're adding rel="nofollow"
    #   to <a> elements - but don't want to allow them for other elements.
    sanitized_content = sanitize(html_output, **settings.WIKI_WHITELIST)
    return sanitized_content


def build_wiki_url(node, label, base, end):
    return '/{pid}/wiki/{wname}/'.format(pid=node._id, wname=label)


class WikiVersionNodeManager(models.Manager):

    def get_for_node(self, node, name=None, version=None, id=None):
        if name:
            wiki_page = WikiPage.objects.get_for_node(node, name)
            if not wiki_page:
                return None

            if version == 'previous':
                version = wiki_page.current_version_number - 1
            elif version == 'current' or version is None:
                version = wiki_page.current_version_number
            elif not ((isinstance(version, int) or version.isdigit())):
                return None

            try:
                return wiki_page.get_version(version=version)
            except WikiVersion.DoesNotExist:
                return None
        return WikiVersion.load(id)


class WikiVersion(ObjectIDMixin, BaseModel):
    objects = WikiVersionNodeManager()

    user = models.ForeignKey('osf.OSFUser', null=True, blank=True, on_delete=models.CASCADE)
    wiki_page = models.ForeignKey('WikiPage', null=True, blank=True, on_delete=models.CASCADE, related_name='versions')
    content = models.TextField(default='', blank=True)
    identifier = models.IntegerField(default=1)

    @property
    def is_current(self):
        return not self.wiki_page.deleted and self.id == self.wiki_page.versions.order_by('-created').first().id

    def html(self, node):
        """The cleaned HTML of the page"""
        html_output = build_html_output(self.content, node=node)
        try:
            cleaner = Cleaner(
                tags=settings.WIKI_WHITELIST['tags'],
                attributes=settings.WIKI_WHITELIST['attributes'],
                styles=settings.WIKI_WHITELIST['styles'],
                filters=[partial(LinkifyFilter, callbacks=[nofollow, ])]
            )
            return cleaner.clean(html_output)
        except TypeError:
            logger.warning('Returning unlinkified content.')
            return render_content(self.content, node=node)

    def raw_text(self, node):
        """ The raw text of the page, suitable for using in a test search"""

        return sanitize(self.content, tags=[], strip=True)

    @property
    def rendered_before_update(self):
        return self.created < WIKI_CHANGE_DATE

    def get_draft(self, node):
        """
        Return most recently edited version of wiki, whether that is the
        last saved version or the most recent sharejs draft.
        """

        db = wiki_utils.share_db()
        sharejs_uuid = wiki_utils.get_sharejs_uuid(node, self.wiki_page.page_name)

        doc_item = db['docs'].find_one({'_id': sharejs_uuid})
        if doc_item:
            sharejs_version = doc_item['_v']
            sharejs_timestamp = doc_item['_m']['mtime']
            sharejs_timestamp /= 1000  # Convert to appropriate units
            sharejs_date = datetime.datetime.utcfromtimestamp(sharejs_timestamp).replace(tzinfo=pytz.utc)

            if sharejs_version > 1 and sharejs_date > self.created:
                return doc_item['_data']

        return self.content

    def save(self, *args, **kwargs):
        rv = super(WikiVersion, self).save(*args, **kwargs)
        if self.wiki_page.node:
            self.wiki_page.node.update_search()
        self.wiki_page.modified = self.created
        self.wiki_page.save()
        self.check_spam()
        return rv

    def check_spam(self):
        request, user_id = get_request_and_user_id()
        user = OSFUser.load(user_id)
        request_headers = string_type_request_headers(request)
        node = self.wiki_page.node

        if not settings.SPAM_CHECK_ENABLED:
            return False
        if settings.SPAM_CHECK_PUBLIC_ONLY and not node.is_public:
            return False
        if user.is_hammy:
            return False

        content = self._get_spam_content(node)
        if not content:
            return
        is_spam = node.do_check_spam(
            user.fullname,
            user.username,
            content,
            request_headers
        )

        logger.info("Node ({}) '{}' smells like {} (tip: {})".format(
            node._id, node.title.encode('utf-8'), 'SPAM' if is_spam else 'HAM', node.spam_pro_tip
        ))
        if is_spam:
            node._check_spam_user(user)
        return is_spam

    def _get_spam_content(self, node):
        content = []
        content.append(self.raw_text(node))
        if not content:
            return None
        return ' '.join(content)

    def clone_version(self, wiki_page, user):
        """Clone a node wiki page.
        :param wiki_page: The wiki_page you want attached to the clone.
        :return: The cloned wiki version
        """
{'fraction_non_alphanumeric': 0.09232728430436167, 'fraction_numerical': 0.003947787328876154, 'mean_word_length': 2.7244486601849656, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6920087', 'n_tokens_mistral': 4716, 'n_tokens_neox': 4521, 'n_words': 1126}