# -*- coding: utf-8 -*-

# This program is free software: you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# 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.lib.utils2

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

Some simple helper functions.

Note: all these functions should be independent of Kallithea classes, i.e.

models, controllers, etc.  to prevent import cycles.

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: 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 hashlib

import json

import logging

import os

import re

import string

import sys

import time

import urllib.parse

from distutils.version import StrictVersion

import bcrypt

import urlobject

from dateutil import relativedelta

from sqlalchemy.engine import url as sa_url

from sqlalchemy.exc import ArgumentError

from tg import tmpl_context

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

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

from kallithea.lib.vcs.backends.git.repository import GitRepository

from kallithea.lib.vcs.conf import settings

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

log = logging.getLogger(__name__)

# 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):

    """

    Converts a given line  "line end" according to given mode

    Available modes are::

        0 - Unix

        1 - Mac

        2 - DOS

    :param line: given line to convert

    :param mode: mode to convert to

    :rtype: str

    :return: converted line according to mode

    """

    if mode == 0:

        line = line.replace('\r\n', '\n')

        line = line.replace('\r', '\n')

    elif mode == 1:

        line = line.replace('\r\n', '\r')

        line = line.replace('\n', '\r')

    elif mode == 2:

        line = re.sub("\r(?!\n)|(?<!\r)\n", "\r\n", line)

    return line

def detect_mode(line, default):

    """

    Detects line break for given line, if line break couldn't be found

    given default value is returned

    :param line: str line

    :param default: default

    :rtype: int

    :return: value of line end on of 0 - Unix, 1 - Mac, 2 - DOS

    """

    if line.endswith('\r\n'):

        return 2

    elif line.endswith('\n'):

        return 0

    elif line.endswith('\r'):

        return 1

    else:

        return default

def generate_api_key():

    """

    Generates a random (presumably unique) API key.

    This value is used in URLs and "Bearer" HTTP Authorization headers,

    which in practice means it should only contain URL-safe characters

    (RFC 3986):

        unreserved = ALPHA / DIGIT / "-" / "." / "_" / "~"

    """

    # Hexadecimal certainly qualifies as URL-safe.

    return ascii_str(binascii.hexlify(os.urandom(20)))

def safe_int(val, default=None):

    """

    Returns int() of val if val is not convertable to int use default

    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 shorter(s, size=20, firstline=False, postfix='...'):

    """Truncate s to size, including the postfix string if truncating.

    If firstline, truncate at newline.

    """

    if firstline:

        s = s.split('\n', 1)[0].rstrip()

    if len(s) > size:

        return s[:size - len(postfix)] + postfix

    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)

        deltas[part] = getattr(d, part + 's')

    # Fix negative offsets (there is 1 second between 10:59:59 and 11:00:00,

    # not 1 hour, -59 minutes and -59 seconds)

    for num, length in [(5, 60), (4, 60), (3, 24)]:  # seconds, minutes, hours

        part = order[num]

        carry_part = order[num - 1]

        if deltas[part] < 0:

            deltas[part] += length

            deltas[carry_part] -= 1

    # Same thing for days except that the increment depends on the (variable)

    # number of days in the month

    month_lengths = [31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]

    if deltas['day'] < 0:

        if prevdate.month == 2 and (prevdate.year % 4 == 0 and

            (prevdate.year % 100 != 0 or prevdate.year % 400 == 0)

        ):

            deltas['day'] += 29

        else:

            deltas['day'] += month_lengths[prevdate.month - 1]

        deltas['month'] -= 1

    if deltas['month'] < 0:

        deltas['month'] += 12

        deltas['year'] -= 1

    # In short version, we want nicer handling of ages of more than a year

    if show_short_version:

        if deltas['year'] == 1:

            # ages between 1 and 2 years: show as months

            deltas['month'] += 12

            deltas['year'] = 0

        if deltas['year'] >= 2:

            # ages 2+ years: round

            if deltas['month'] > 6:

                deltas['year'] += 1

                deltas['month'] = 0

    # Format the result

    fmt_funcs = {

        'year': lambda d: ungettext('%d year', '%d years', d) % d,

        'month': lambda d: ungettext('%d month', '%d months', d) % d,

        'day': lambda d: ungettext('%d day', '%d days', d) % d,

        'hour': lambda d: ungettext('%d hour', '%d hours', d) % d,

        'minute': lambda d: ungettext('%d minute', '%d minutes', d) % d,

        'second': lambda d: ungettext('%d second', '%d seconds', d) % d,

    }

    for i, part in enumerate(order):

        value = deltas[part]

        if value == 0:

            continue

        if i < 5:

            sub_part = order[i + 1]

            sub_value = deltas[sub_part]

        else:

            sub_value = 0

        if sub_value == 0 or show_short_version:

            if future:

                return _('in %s') % fmt_funcs[part](value)

            else:

                return _('%s ago') % fmt_funcs[part](value)

        if future:

            return _('in %s and %s') % (fmt_funcs[part](value),

                fmt_funcs[sub_part](sub_value))

        else:

# -*- coding: utf-8 -*-

# This program is free software: you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

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

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

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

:author: marcink

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

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

"""

import logging

import os

import posixpath

import re

import sys

import tempfile

import traceback

import pkg_resources

from tg.i18n import ugettext as _

import kallithea

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

from kallithea.lib.exceptions import IMCCommitError, NonRelativePathError

from kallithea.lib.hooks import process_pushed_raw_ids

from kallithea.lib.utils import action_logger, get_filesystem_repos, make_ui

from kallithea.lib.vcs import get_repo

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

from kallithea.lib.vcs.exceptions import RepositoryError, VCSError

from kallithea.lib.vcs.nodes import FileNode

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

from kallithea.model import db, meta

log = logging.getLogger(__name__)

class UserTemp(object):

    def __init__(self, user_id):

        self.user_id = user_id

    def __repr__(self):

        return "<%s('id:%s')>" % (self.__class__.__name__, self.user_id)

class RepoTemp(object):

    def __init__(self, repo_id):

        self.repo_id = repo_id

    def __repr__(self):

        return "<%s('id:%s')>" % (self.__class__.__name__, self.repo_id)

class _PermCheckIterator(object):

    def __init__(self, obj_list, obj_attr, perm_set, perm_checker, extra_kwargs=None):

        """

        Creates iterator from given list of objects, additionally

        checking permission for them from perm_set var

        :param obj_list: list of db objects

        :param obj_attr: attribute of object to pass into perm_checker

        :param perm_set: list of permissions to check

        :param perm_checker: callable to check permissions against

        """

        self.obj_list = obj_list

        self.obj_attr = obj_attr

        self.perm_set = perm_set

        self.perm_checker = perm_checker

        self.extra_kwargs = extra_kwargs or {}

    def __len__(self):

        return len(self.obj_list)

    def __repr__(self):

        return '<%s (%s)>' % (self.__class__.__name__, self.__len__())

    def __iter__(self):

        for db_obj in self.obj_list:

            # check permission at this level

            name = getattr(db_obj, self.obj_attr, None)

            if not self.perm_checker(*self.perm_set)(

                    name, self.__class__.__name__, **self.extra_kwargs):

                continue

            yield db_obj

class RepoList(_PermCheckIterator):

    def __init__(self, db_repo_list, perm_level, extra_kwargs=None):

        super(RepoList, self).__init__(obj_list=db_repo_list,

                    obj_attr='repo_name', perm_set=[perm_level],

                    perm_checker=HasRepoPermissionLevel,

                    extra_kwargs=extra_kwargs)

class RepoGroupList(_PermCheckIterator):

    def __init__(self, db_repo_group_list, perm_level, extra_kwargs=None):

        super(RepoGroupList, self).__init__(obj_list=db_repo_group_list,

                    obj_attr='group_name', perm_set=[perm_level],

                    perm_checker=HasRepoGroupPermissionLevel,

                    extra_kwargs=extra_kwargs)

class UserGroupList(_PermCheckIterator):

    def __init__(self, db_user_group_list, perm_level, extra_kwargs=None):

        super(UserGroupList, self).__init__(obj_list=db_user_group_list,

                    obj_attr='users_group_name', perm_set=[perm_level],

                    perm_checker=HasUserGroupPermissionLevel,

                    extra_kwargs=extra_kwargs)

class ScmModel(object):

    """

    Generic Scm Model

    """

    def __get_repo(self, instance):

        cls = db.Repository

        if isinstance(instance, cls):

            return instance

        elif isinstance(instance, int):

            return cls.get(instance)

        elif isinstance(instance, str):

            if instance.isdigit():

                return cls.get(int(instance))

            return cls.get_by_repo_name(instance)

        raise Exception('given object must be int, basestr or Instance'

                        ' of %s got %s' % (type(cls), type(instance)))

    @LazyProperty

    def repos_path(self):

        """

        Gets the repositories root path from database

        """

        q = db.Ui.query().filter(db.Ui.ui_key == '/').one()

        return q.ui_value

    def repo_scan(self, repos_path=None):

        """

        Listing of repositories in given path. This path should not be a

        repository itself. Return a dictionary of repository objects mapping to

        vcs instances.

        :param repos_path: path to directory containing repositories

        """

        if repos_path is None:

            repos_path = self.repos_path

        log.info('scanning for repositories in %s', repos_path)

        baseui = make_ui()

        repos = {}

        for name, path in get_filesystem_repos(repos_path):

            # name need to be decomposed and put back together using the /

            # since this is internal storage separator for kallithea

            name = db.Repository.normalize_repo_name(name)

            try:

                if name in repos:

                    raise RepositoryError('Duplicate repository name %s '

                                          'found in %s' % (name, path))

                else:

                    repos[name] = get_repo(path[1], baseui=baseui)

            except (OSError, VCSError):

                continue

        log.debug('found %s paths with repositories', len(repos))

        return repos

    def get_repos(self, repos):

        """Return the repos the user has access to"""

        return RepoList(repos, perm_level='read')

    def get_repo_groups(self, groups=None):

        """Return the repo groups the user has access to

        If no groups are specified, use top level groups.

        """

        if groups is None:

            groups = db.RepoGroup.query() \

                .filter(db.RepoGroup.parent_group_id == None).all()

        return RepoGroupList(groups, perm_level='read')

    def mark_for_invalidation(self, repo_name):

        """

        Mark caches of this repo invalid in the database.

        :param repo_name: the repo for which caches should be marked invalid

        """

        log.debug("Marking %s as invalidated and update cache", repo_name)

        repo = db.Repository.get_by_repo_name(repo_name)

        if repo is not None:

            repo.set_invalidate()

            repo.update_changeset_cache()

    def toggle_following_repo(self, follow_repo_id, user_id):

        f = db.UserFollowing.query() \

            .filter(db.UserFollowing.follows_repository_id == follow_repo_id) \

            .filter(db.UserFollowing.user_id == user_id).scalar()

        if f is not None:

            try:

                meta.Session().delete(f)

                action_logger(UserTemp(user_id),

                              'stopped_following_repo',

                              RepoTemp(follow_repo_id))

                return

            except Exception:

                log.error(traceback.format_exc())

                raise

        try:

            elif op == 'del':

                imc.remove(filenode)

            elif op == 'mod':

                if filename != old_filename:

                    # TODO: handle renames, needs vcs lib changes

                    imc.remove(filenode)

                    imc.add(FileNode(filename, content=content))

                else:

                    imc.change(filenode)

        # commit changes

        tip = imc.commit(message=message,

                         author=author,

                         parents=parents,

                         branch=parent_cs.branch)

        if trigger_push_hook:

            self._handle_push(scm_instance,

                              username=user.username,

                              ip_addr=ip_addr,

                              action='push_local',

                              repo_name=repo.repo_name,

                              revisions=[tip.raw_id])

        else:

            self.mark_for_invalidation(repo.repo_name)

    def delete_nodes(self, user, ip_addr, repo, message, nodes, parent_cs=None,

                     author=None, trigger_push_hook=True):

        """

        Deletes specified nodes from repo.

        :param user: Kallithea User object or user_id, the committer

        :param repo: Kallithea Repository object

        :param message: commit message

        :param nodes: mapping {filename:{'content':content},...}

        :param parent_cs: parent changeset, can be empty than it's initial commit

        :param author: author of commit, cna be different that committer only for git

        :param trigger_push_hook: trigger push hooks

        :returns: new committed changeset after deletion

        """

        user = db.User.guess_instance(user)

        scm_instance = repo.scm_instance_no_cache()

        processed_nodes = []

        for f_path in nodes:

            f_path = self._sanitize_path(f_path)

            # content can be empty but for compatibility it allows same dicts

            # structure as add_nodes

            content = nodes[f_path].get('content')

            processed_nodes.append((f_path, content))

        message = message

        committer = user.full_contact

        if not author:

            author = committer

        if not parent_cs:

            parent_cs = EmptyChangeset(alias=scm_instance.alias)

        if isinstance(parent_cs, EmptyChangeset):

            # EmptyChangeset means we we're editing empty repository

            parents = None

        else:

            parents = [parent_cs]

        # add multiple nodes

        imc = scm_instance.in_memory_changeset

        for path, content in processed_nodes:

            imc.remove(FileNode(path, content=content))

        tip = imc.commit(message=message,

                         author=author,

                         parents=parents,

                         branch=parent_cs.branch)

        if trigger_push_hook:

            self._handle_push(scm_instance,

                              username=user.username,

                              ip_addr=ip_addr,

                              action='push_local',

                              repo_name=repo.repo_name,

                              revisions=[tip.raw_id])

        else:

            self.mark_for_invalidation(repo.repo_name)

        return tip

    def get_unread_journal(self):

        return db.UserLog.query().count()

    def get_repo_landing_revs(self, repo=None):

        """

        Generates select option with tags branches and bookmarks (for hg only)

        grouped by type

        :param repo:

        """

        hist_l = []

        choices = []

        hist_l.append(('rev:tip', _('latest tip')))

        choices.append('rev:tip')

        if repo is None:

            return choices, hist_l

        repo = self.__get_repo(repo)

        repo = repo.scm_instance

        branches_group = ([('branch:%s' % k, k) for k, v in

                           repo.branches.items()], _("Branches"))

        hist_l.append(branches_group)

        choices.extend([x[0] for x in branches_group[0]])

        if repo.alias == 'hg':

            bookmarks_group = ([('book:%s' % k, k) for k, v in

                                repo.bookmarks.items()], _("Bookmarks"))

            hist_l.append(bookmarks_group)

            choices.extend([x[0] for x in bookmarks_group[0]])

        tags_group = ([('tag:%s' % k, k) for k, v in

                       repo.tags.items()], _("Tags"))

        hist_l.append(tags_group)

        choices.extend([x[0] for x in tags_group[0]])

        return choices, hist_l

    def _get_git_hook_interpreter(self):

        """Return a suitable interpreter for Git hooks.

        Return a suitable string to be written in the POSIX #! shebang line for

        Git hook scripts so they invoke Kallithea code with the right Python

        interpreter and in the right environment.

        """

        # Note: sys.executable might not point at a usable Python interpreter. For

        # example, when using uwsgi, it will point at the uwsgi program itself.

        # FIXME This may not work on Windows and may need a shell wrapper script.

        return (kallithea.CONFIG.get('git_hook_interpreter')

                or sys.executable

                or '/usr/bin/env python3')

    def install_git_hooks(self, repo, force=False):

        """

        Creates a kallithea hook inside a git repository

        :param repo: Instance of VCS repo

        :param force: Overwrite existing non-Kallithea hooks

        """

        hooks_path = os.path.join(repo.path, 'hooks')

        if not repo.bare:

            hooks_path = os.path.join(repo.path, '.git', 'hooks')

        if not os.path.isdir(hooks_path):

            os.makedirs(hooks_path)

        tmpl_post = b"#!%s\n" % safe_bytes(self._get_git_hook_interpreter())

        tmpl_post += pkg_resources.resource_string(

            'kallithea', os.path.join('templates', 'py', 'git_post_receive_hook.py')

        )

        tmpl_pre = b"#!%s\n" % safe_bytes(self._get_git_hook_interpreter())

        tmpl_pre += pkg_resources.resource_string(

            'kallithea', os.path.join('templates', 'py', 'git_pre_receive_hook.py')

        )

        for h_type, tmpl in [('pre-receive', tmpl_pre), ('post-receive', tmpl_post)]:

            hook_file = os.path.join(hooks_path, h_type)

            other_hook = False

            log.debug('Installing git hook in repo %s', repo)

            if os.path.islink(hook_file):

                log.debug("Found symlink hook at %s", hook_file)

                other_hook = True

            elif os.path.isfile(hook_file):

                log.debug('hook exists, checking if it is from kallithea')

                with open(hook_file, 'rb') as f:

                    data = f.read()

                    matches = re.search(br'^KALLITHEA_HOOK_VER\s*=\s*(.*)$', data, flags=re.MULTILINE)

                    if matches:

                        ver = safe_str(matches.group(1))

                        log.debug('Found Kallithea hook - it has KALLITHEA_HOOK_VER %s', ver)

                    else:

                        log.debug('Found non-Kallithea hook at %s', hook_file)

                        other_hook = True

            elif os.path.exists(hook_file):

                log.debug("Found hook that isn't a regular file at %s", hook_file)

                other_hook = True

            if other_hook and not force:

                log.warning('skipping overwriting hook file %s', hook_file)

            else:

                log.debug('writing %s hook file !', h_type)

                try:

                    fh, fn = tempfile.mkstemp(prefix=hook_file + '.tmp.')

                    os.write(fh, tmpl.replace(b'_TMPL_', safe_bytes(kallithea.__version__)))

                    os.close(fh)

                    os.rename(fn, hook_file)

                except (OSError, IOError) as e:

                    log.error('error writing hook %s: %s', hook_file, e)

def AvailableRepoGroupChoices(repo_group_perm_level, extras=()):

    """Return group_id,string tuples with choices for all the repo groups where

    the user has the necessary permissions.

    Top level is -1.

    """

    groups = db.RepoGroup.query().all()

    if HasPermissionAny('hg.admin')('available repo groups'):

        groups.append(None)

    else:

        groups = list(RepoGroupList(groups, perm_level=repo_group_perm_level))

        if HasPermissionAny('hg.create.repository')('available repo groups'):

            groups.append(None)

        for extra in extras:

            if not any(rg == extra for rg in groups):

                groups.append(extra)

    return db.RepoGroup.groups_choices(groups=groups)