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: Apr 23, 2010

:author: marcink

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

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

"""

import logging

from beaker.cache import cache_region

from tg import response

from tg import tmpl_context as c

from tg.i18n import ugettext as _

import kallithea

from kallithea.lib import feeds

from kallithea.lib import helpers as h

from kallithea.lib import webutils

from kallithea.lib.auth import HasRepoPermissionLevelDecorator, LoginRequired

from kallithea.lib.base import BaseRepoController

from kallithea.lib.diffs import DiffProcessor

log = logging.getLogger(__name__)

class FeedController(BaseRepoController):

    @LoginRequired(allow_default_user=True)

    @HasRepoPermissionLevelDecorator('read')

    def _before(self, *args, **kwargs):

        super(FeedController, self)._before(*args, **kwargs)

    def _get_title(self, cs):

    def __get_desc(self, cs):

        desc_msg = [(_('%s committed on %s')

                     % (h.person(cs.author), h.fmt_date(cs.date))) + '<br/>']

        # branches, tags, bookmarks

        for branch in cs.branches:

            desc_msg.append('branch: %s<br/>' % branch)

        for book in cs.bookmarks:

            desc_msg.append('bookmark: %s<br/>' % book)

        for tag in cs.tags:

            desc_msg.append('tag: %s<br/>' % tag)

        changes = []

        diff_limit = safe_int(kallithea.CONFIG.get('rss_cut_off_limit', 32 * 1024))

        raw_diff = cs.diff()

        diff_processor = DiffProcessor(raw_diff,

                                       diff_limit=diff_limit,

                                       inline_diff=False)

        for st in diff_processor.parsed:

            st.update({'added': st['stats']['added'],

                       'removed': st['stats']['deleted']})

            changes.append('\n %(operation)s %(filename)s '

                           '(%(added)s lines added, %(removed)s lines removed)'

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 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.

    Because the rules for parsing <script> varies between XHTML (where

    """

    return literal(

        ('(' + json.dumps(value) + ')')

        # In JSON, the following can only appear in string literals.

        .replace('&', r'\x26')

        .replace('<', r'\x3c')

        .replace('>', r'\x3e')

    )

def jshtml(val):

    """HTML escapes a string value, then converts the resulting string

    to its corresponding JavaScript representation (see `js`).

    This is used when a plain-text string (possibly containing special

    HTML characters) will be used by a script in an HTML context (e.g.

    element.innerHTML or jQuery's 'html' method).

    If in doubt, err on the side of using `jshtml` over `js`, since it's

    better to escape too much than too little.

    """

    return js(escape(val))

def FID(raw_id, path):

    """

    Creates a unique ID for filenode based on it's hash of path and revision

    it's safe to use in urls

    """

    return 'C-%s-%s' % (short_id(raw_id), hashlib.md5(safe_bytes(path)).hexdigest()[:12])

def get_ignore_whitespace_diff(GET):

    """Return true if URL requested whitespace to be ignored"""

    return bool(GET.get('ignorews'))

def ignore_whitespace_link(GET, anchor=None):

    """Return snippet with link to current URL with whitespace ignoring toggled"""

    params = dict(GET)  # ignoring duplicates

    if get_ignore_whitespace_diff(GET):

        params.pop('ignorews')

        title = _("Show whitespace changes")

    else:

        params['ignorews'] = '1'

        title = _("Ignore whitespace changes")

    params['anchor'] = anchor

    return link_to(

    instead

    :param val:

    :param default:

    """

    try:

        val = int(val)

    except (ValueError, TypeError):

        val = default

    return val

def remove_suffix(s, suffix):

    if s.endswith(suffix):

        s = s[:-1 * len(suffix)]

    return s

def remove_prefix(s, prefix):

    if s.startswith(prefix):

        s = s[len(prefix):]

    return s

def age(prevdate, show_short_version=False, now=None):

    """

    turns a datetime into an age string.

    If show_short_version is True, then it will generate a not so accurate but shorter string,

    example: 2days ago, instead of 2 days and 23 hours ago.

    :param prevdate: datetime object

    :param show_short_version: if it should approximate the date and return a shorter string

    :rtype: str

    :returns: str words describing age

    """

    now = now or datetime.datetime.now()

    order = ['year', 'month', 'day', 'hour', 'minute', 'second']

    deltas = {}

    future = False

    if prevdate > now:

        now, prevdate = prevdate, now

        future = True

    if future:

        prevdate = prevdate.replace(microsecond=0)

    # Get date parts deltas

    for part in order:

        d = relativedelta.relativedelta(now, prevdate)

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# 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.comment

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

comments 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: Nov 11, 2011

:author: marcink

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

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

"""

import logging

from collections import defaultdict

from tg.i18n import ugettext as _

from kallithea.lib import webutils

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 _list_changeset_commenters(revision):

    return (meta.Session().query(db.User)

        .join(db.ChangesetComment.author)

        .filter(db.ChangesetComment.revision == revision)

        .all())

def _list_pull_request_commenters(pull_request):

    return (meta.Session().query(db.User)

        .join(db.ChangesetComment.author)

        .filter(db.ChangesetComment.pull_request_id == pull_request.pull_request_id)

        .all())

class ChangesetCommentsModel(object):

    def _get_notification_data(self, repo, comment, author, comment_text,

                               line_no=None, revision=None, pull_request=None,

                revision=revision,

                anchor='comment-%s' % comment.comment_id)

            subj = webutils.link_to(

                'Re changeset: %(desc)s %(line)s' %

                          {'desc': desc, 'line': line},

                 comment_url)

            # get the current participants of this changeset

            recipients = _list_changeset_commenters(revision)

            # add changeset author if it's known locally

            cs_author = db.User.get_from_cs_author(cs.author)

            if not cs_author:

                # use repo owner if we cannot extract the author correctly

                # FIXME: just use committer name even if not a user

                cs_author = repo.owner

            recipients.append(cs_author)

            email_kwargs = {

                'status_change': status_change,

                'cs_comment_user': author.full_name_and_username,

                'cs_target_repo': webutils.canonical_url('summary_home', repo_name=repo.repo_name),

                'cs_comment_url': comment_url,

                'cs_url': webutils.canonical_url('changeset_home', repo_name=repo.repo_name, revision=revision),

                'raw_id': revision,

                'message': cs.message,

                'cs_author': cs_author,

                'cs_author_username': cs_author.username,

                'repo_name': repo.repo_name,

                'short_id': revision[:12],

                'branch': cs.branch,

                'comment_username': author.username,

                'threading': threading,

            }

        # pull request

        elif pull_request:

            notification_type = NotificationModel.TYPE_PULL_REQUEST_COMMENT

            desc = comment.pull_request.title

            _org_ref_type, org_ref_name, _org_rev = comment.pull_request.org_ref.split(':')

            _other_ref_type, other_ref_name, _other_rev = comment.pull_request.other_ref.split(':')

            threading = ['%s-pr-%s@%s' % (pull_request.other_repo.repo_name,

                                          pull_request.pull_request_id,

                                          webutils.canonical_hostname())]

            if line_no: # TODO: url to file _and_ line number

                threading.append('%s-pr-%s-line-%s@%s' % (pull_request.other_repo.repo_name,

                                                          pull_request.pull_request_id, line_no,

                                                          webutils.canonical_hostname()))

            comment_url = pull_request.url(canonical=True,

                anchor='comment-%s' % comment.comment_id)

            subj = webutils.link_to(

                'Re pull request %(pr_nice_id)s: %(desc)s %(line)s' %

                          {'desc': desc,

                           'pr_nice_id': comment.pull_request.nice_id(),

                           'line': line},

                comment_url)

            # get the current participants of this pull request

            recipients = _list_pull_request_commenters(pull_request)

            recipients.append(pull_request.owner)

            recipients += pull_request.get_reviewer_users()

            # set some variables for email notification

            email_kwargs = {

                'pr_title': pull_request.title,

                'pr_nice_id': pull_request.nice_id(),

                'status_change': status_change,

                'closing_pr': closing_pr,

                'pr_comment_url': comment_url,

                'pr_url': pull_request.url(canonical=True),

                'pr_comment_user': author.full_name_and_username,

                'pr_target_repo': webutils.canonical_url('summary_home',

                                   repo_name=pull_request.other_repo.repo_name),

                'pr_target_branch': other_ref_name,

                'pr_source_repo': webutils.canonical_url('summary_home',

                                   repo_name=pull_request.org_repo.repo_name),

                'pr_source_branch': org_ref_name,

                'pr_owner': pull_request.owner,

                'pr_owner_username': pull_request.owner.username,

                'repo_name': pull_request.other_repo.repo_name,

                'comment_username': author.username,

                'threading': threading,

            }

        return subj, body, recipients, notification_type, email_kwargs

    def create(self, text, repo, author, revision=None, pull_request=None,

               f_path=None, line_no=None, status_change=None, closing_pr=False,

               send_email=True):

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

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 import auth

from kallithea.lib import helpers as h

from kallithea.lib import webutils

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() \

        log.debug('Adding reviewers to pull request %s: %s', pr.pull_request_id, reviewers)

        for reviewer in reviewers:

            prr = db.PullRequestReviewer(reviewer, pr)

            meta.Session().add(prr)

        # notification to reviewers

        pr_url = pr.url(canonical=True)

        threading = ['%s-pr-%s@%s' % (pr.other_repo.repo_name,

                                      pr.pull_request_id,

                                      webutils.canonical_hostname())]

        subject = webutils.link_to(

            _('%(user)s wants you to review pull request %(pr_nice_id)s: %(pr_title)s') %

                {'user': user.username,

                 'pr_title': pr.title,

                 'pr_nice_id': pr.nice_id()},

            pr_url)

        body = pr.description

        _org_ref_type, org_ref_name, _org_rev = pr.org_ref.split(':')

        _other_ref_type, other_ref_name, _other_rev = pr.other_ref.split(':')

        revision_data = [(x.raw_id, x.message)

                         for x in map(pr.org_repo.get_changeset, pr.revisions)]

        email_kwargs = {

            'pr_title': pr.title,

            'pr_user_created': user.full_name_and_username,

            'pr_repo_url': webutils.canonical_url('summary_home', repo_name=pr.other_repo.repo_name),

            'pr_url': pr_url,

            'pr_revisions': revision_data,

            'repo_name': pr.other_repo.repo_name,

            'org_repo_name': pr.org_repo.repo_name,

            'pr_nice_id': pr.nice_id(),

            'pr_target_repo': webutils.canonical_url('summary_home',

                               repo_name=pr.other_repo.repo_name),

            'pr_target_branch': other_ref_name,

            'pr_source_repo': webutils.canonical_url('summary_home',

                               repo_name=pr.org_repo.repo_name),

            'pr_source_branch': org_ref_name,

            'pr_owner': pr.owner,

            'pr_owner_username': pr.owner.username,

            'pr_username': user.username,

            'threading': threading,

            'is_mention': False,

            }

        if reviewers:

            NotificationModel().create(created_by=user, subject=subject, body=body,

                                       recipients=reviewers,

                                       type_=NotificationModel.TYPE_PULL_REQUEST,

                                       email_kwargs=email_kwargs)

        # Generate complete title/description

        old_revisions = set(old_pull_request.revisions)

        revisions = self.create_action.revisions

        new_revisions = [r for r in revisions if r not in old_revisions]

        lost = old_revisions.difference(revisions)

        infos = ['This is a new iteration of %s "%s".' %

                 (webutils.canonical_url('pullrequest_show', repo_name=old_pull_request.other_repo.repo_name,

                      pull_request_id=old_pull_request.pull_request_id),

                  old_pull_request.title)]

        if lost:

            infos.append(_('Missing changesets since the previous iteration:'))

            for r in old_pull_request.revisions:

                if r in lost:

                    rev_desc = org_repo.get_changeset(r).message.split('\n')[0]

                    infos.append('  %s %s' % (r[:12], rev_desc))

        if new_revisions:

            infos.append(_('New changesets on %s %s since the previous iteration:') % (org_ref_type, org_ref_name))

            for r in reversed(revisions):

                if r in new_revisions:

                    rev_desc = org_repo.get_changeset(r).message.split('\n')[0]

            if self.create_action.other_ref == old_pull_request.other_ref:

                infos.append(_("Ancestor didn't change - diff since previous iteration:"))

                infos.append(webutils.canonical_url('compare_url',

                                 repo_name=org_repo.repo_name, # other_repo is always same as repo_name

                                 org_ref_type='rev', org_ref_name=org_rev[:12], # use old org_rev as base

                                 other_ref_type='rev', other_ref_name=new_org_rev[:12],

                                 )) # note: linear diff, merge or not doesn't matter

            else:

                infos.append(_('This iteration is based on another %s revision and there is no simple diff.') % other_ref_name)

        else:

            infos.append(_('No changes found on %s %s since previous iteration.') % (org_ref_type, org_ref_name))

            # TODO: fail?

        v = 2

        m = re.match(r'(.*)\(v(\d+)\)\s*$', title)

        if m is not None:

            title = m.group(1)

            v = int(m.group(2)) + 1

        self.create_action.title = '%s (v%s)' % (title.strip(), v)

        # using a mail-like separator, insert new iteration info in description with latest first

        descriptions = description.replace('\r\n', '\n').split('\n-- \n', 1)

        description = descriptions[0].strip() + '\n\n-- \n' + '\n'.join(infos)