Consists of functions to typically be used within templates, but also

available to Controllers. This module is available to both as 'h'.

"""

import hashlib

import json

import logging

import re

import textwrap

import urllib.parse

from beaker.cache import cache_region

from pygments import highlight as code_highlight

from pygments.formatters.html import HtmlFormatter

from tg.i18n import ugettext as _

import kallithea

from kallithea.lib.annotate import annotate_highlight

#==============================================================================

# PERMS

#==============================================================================

from kallithea.lib.auth import HasPermissionAny, HasRepoGroupPermissionLevel, HasRepoPermissionLevel

from kallithea.lib.diffs import BIN_FILENODE, CHMOD_FILENODE, DEL_FILENODE, MOD_FILENODE, NEW_FILENODE, RENAMED_FILENODE

from kallithea.lib.markup_renderer import url_re

from kallithea.lib.pygmentsutils import get_custom_lexer

from kallithea.lib.vcs.backends.base import BaseChangeset, EmptyChangeset

from kallithea.lib.vcs.exceptions import ChangesetDoesNotExistError

#==============================================================================

# SCM FILTERS available via h.

#==============================================================================

from kallithea.lib.vcs.utils import author_email, author_name

from kallithea.lib.webutils import (HTML, Option, canonical_url, checkbox, chop_at, end_form, escape, form, format_byte_size, hidden, html_escape, link_to,

                                    literal, password, pop_flash_messages, radio, reset, safeid, select, session_csrf_secret_name, session_csrf_secret_token,

                                    submit, text, textarea, truncate, url, wrap_paragraphs)

from kallithea.model import db

from kallithea.model.changeset_status import ChangesetStatusModel

# mute pyflakes "imported but unused"

# from webutils

assert Option

assert canonical_url

assert checkbox

assert chop_at

assert end_form

assert form

assert format_byte_size

assert hidden

assert password

assert pop_flash_messages

assert radio

assert reset

assert safeid

assert select

assert session_csrf_secret_name

assert session_csrf_secret_token

assert submit

assert text

assert textarea

assert wrap_paragraphs

# from kallithea.lib.auth

assert HasPermissionAny

assert HasRepoGroupPermissionLevel

assert HasRepoPermissionLevel

# from utils2

assert age

assert fmt_date

assert shorter

assert time_to_datetime

# from vcs

assert EmptyChangeset

log = logging.getLogger(__name__)

def js(value):

    """Convert Python value to the corresponding JavaScript representation.

    This is necessary to safely insert arbitrary values into HTML <script>

    sections e.g. using Mako template expression substitution.

    Note: Rather than using this function, it's preferable to avoid the

    insertion of values into HTML <script> sections altogether. Instead,

    data should (to the extent possible) be passed to JavaScript using

    data attributes or AJAX calls, eliminating the need for JS specific

    escaping.

    Note: This is not safe for use in attributes (e.g. onclick), because

    quotes are not escaped.

                    ) % {

                     'url': issue_url,

                     'text': issue_text,

                    }

            def tmp_urlify_issues_f(s, issue_re=issue_re, issues_replace=issues_replace, chain_f=tmp_urlify_issues_f):

                return issue_re.sub(issues_replace, chain_f(s))

        # Set tmp function globally - atomically

        _urlify_issues_f = tmp_urlify_issues_f

    return _urlify_issues_f(newtext)

def render_w_mentions(source, repo_name=None):

    """

    Render plain text with revision hashes and issue references urlified

    and with @mention highlighting.

    """

    s = safe_str(source)

    s = urlify_text(s, repo_name=repo_name)

    return literal('<div class="formatted-fixed">%s</div>' % s)

def changeset_status(repo, revision):

    return ChangesetStatusModel().get_status(repo, revision)

def changeset_status_lbl(changeset_status):

    return db.ChangesetStatus.get_status_lbl(changeset_status)

def get_permission_name(key):

    return dict(db.Permission.PERMS).get(key)

def journal_filter_help():

    return _(textwrap.dedent('''

        Example filter terms:

            repository:vcs

            username:developer

            action:*push*

            ip:127.0.0.1

            date:20120101

            date:[20120101100000 TO 20120102]

        Generate wildcards using '*' character:

            "repository:vcs*" - search everything starting with 'vcs'

:created_on: Jan 5, 2011

:author: marcink

:copyright: (c) 2013 RhodeCode GmbH, and others.

:license: GPLv3, see LICENSE.md for more details.

"""

import binascii

import datetime

import json

import os

import re

import time

import urllib.parse

import urlobject

from dateutil import relativedelta

from sqlalchemy.engine import url as sa_url

from sqlalchemy.exc import ArgumentError

from tg.i18n import ugettext as _

from tg.i18n import ungettext

from tg.support.converters import asbool, aslist

from webhelpers2.text import collapse, remove_formatting, strip_tags

import kallithea

from kallithea.lib.vcs.backends.base import BaseRepository, EmptyChangeset

from kallithea.lib.vcs.exceptions import RepositoryError

from kallithea.lib.vcs.utils import ascii_bytes, ascii_str, safe_bytes, safe_str  # re-export

from kallithea.lib.vcs.utils.lazy import LazyProperty

try:

    import pwd

except ImportError:

    pass

# mute pyflakes "imported but unused"

assert asbool

assert aslist

assert ascii_bytes

assert ascii_str

assert safe_bytes

assert safe_str

assert LazyProperty

def convert_line_endings(line, mode):

    """

    try:

        system_user = pwd.getpwuid(os.getuid()).pw_name

    except NameError: # TODO: support all systems - especially Windows

        system_user = 'kallithea' # hardcoded default value ...

    args = {

        'scheme': parsed_url.scheme,

        'user': urllib.parse.quote(username or ''),

        'netloc': parsed_url.netloc + prefix,  # like "hostname:port/prefix" (with optional ":port" and "/prefix")

        'prefix': prefix, # undocumented, empty or starting with /

        'repo': repo_name,

        'repoid': str(repo_id),

        'system_user': system_user,

        'hostname': parsed_url.hostname,

    }

    url = re.sub('{([^{}]+)}', lambda m: args.get(m.group(1), m.group(0)), clone_uri_tmpl)

    # remove leading @ sign if it's present. Case of empty user

    url_obj = urlobject.URLObject(url)

    if not url_obj.username:

        url_obj = url_obj.with_username(None)

    return str(url_obj)

def get_changeset_safe(repo, rev):

    """

    Safe version of get_changeset if this changeset doesn't exists for a

    repo it returns a Dummy one instead

    :param repo:

    :param rev:

    """

    if not isinstance(repo, BaseRepository):

        raise Exception('You must pass an Repository '

                        'object as first argument got %s' % type(repo))

    try:

        cs = repo.get_changeset(rev)

    except (RepositoryError, LookupError):

        cs = EmptyChangeset(requested_revision=rev)

    return cs

def datetime_to_time(dt):

    if dt:

        return time.mktime(dt.timetuple())

#

# You should have received a copy of the GNU General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

"""

kallithea.model.pull_request

~~~~~~~~~~~~~~~~~~~~~~~~~~~~

pull request model for Kallithea

This file was forked by the Kallithea project in July 2014.

Original author and date, and relevant copyright and licensing information is below:

:created_on: Jun 6, 2012

:author: marcink

:copyright: (c) 2013 RhodeCode GmbH, and others.

:license: GPLv3, see LICENSE.md for more details.

"""

import datetime

import logging

import re

from tg import request

from tg.i18n import ugettext as _

from kallithea.lib.hooks import log_create_pullrequest

from kallithea.lib.utils import extract_mentioned_users

from kallithea.model import db, meta

from kallithea.model.notification import NotificationModel

log = logging.getLogger(__name__)

def _assert_valid_reviewers(seq):

    """Sanity check: elements are actual User objects, and not the default user."""

    assert not any(user.is_default_user for user in seq)

class PullRequestModel(object):

    def add_reviewers(self, user, pr, reviewers, mention_recipients=None):

        """Add reviewer and send notification to them.

        """

        reviewers = set(reviewers)

        _assert_valid_reviewers(reviewers)

        if mention_recipients is not None:

            mention_recipients = set(mention_recipients) - reviewers

            _assert_valid_reviewers(mention_recipients)

        redundant_reviewers = set(db.User.query() \

    class Unauthorized(ValidationError):

        pass

    @staticmethod

    def is_user_authorized(org_repo, other_repo):

        """Performs authorization check with only the minimum amount of

        information needed for such a check, rather than a full command

        object.

        """

        if (auth.HasRepoPermissionLevel('read')(org_repo.repo_name) and

            auth.HasRepoPermissionLevel('read')(other_repo.repo_name)

        ):

            return True

        return False

    def __init__(self, org_repo, other_repo, org_ref, other_ref, title, description, owner, reviewers):

        reviewers = set(reviewers)

        _assert_valid_reviewers(reviewers)

        (org_ref_type,

         org_ref_name,

         org_rev) = org_ref.split(':')

        if org_ref_type == 'rev':

            cs = org_repo.scm_instance.get_changeset(org_rev)

            org_ref = 'branch:%s:%s' % (cs.branch, cs.raw_id)

        (other_ref_type,

         other_ref_name,

         other_rev) = other_ref.split(':')

        if other_ref_type == 'rev':

            cs = other_repo.scm_instance.get_changeset(other_rev)

            other_ref_name = cs.raw_id[:12]

            other_ref = '%s:%s:%s' % (other_ref_type, other_ref_name, cs.raw_id)

        cs_ranges, _cs_ranges_not, ancestor_revs = \

            org_repo.scm_instance.get_diff_changesets(other_rev, org_repo.scm_instance, org_rev) # org and other "swapped"

        if not cs_ranges:

            raise self.Empty(_('Cannot create empty pull request'))

        if not ancestor_revs:

            ancestor_rev = org_repo.scm_instance.EMPTY_CHANGESET

        elif len(ancestor_revs) == 1:

            ancestor_rev = ancestor_revs[0]

        else:

            raise self.AmbiguousAncestor(

                _('Cannot create pull request - criss cross merge detected, please merge a later %s revision to %s')

                % (other_ref_name, org_ref_name))

        self.revisions = [cs_.raw_id for cs_ in cs_ranges]

        # hack: ancestor_rev is not an other_rev but we want to show the

        # requested destination and have the exact ancestor

        other_ref = '%s:%s:%s' % (other_ref_type, other_ref_name, ancestor_rev)

        if not title:

            if org_repo == other_repo:

                title = '%s to %s' % (org_display, other_display)