# -*- 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.utils

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

Utilities library 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 18, 2010

:author: marcink

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

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

"""

import os

import re

import logging

import datetime

import traceback

import beaker

from tg import request, response

from tg.i18n import ugettext as _

from beaker.cache import _cache_decorate

from kallithea.lib.vcs.utils.hgcompat import ui, config

from kallithea.lib.vcs.utils.helpers import get_scm

from kallithea.lib.vcs.exceptions import VCSError

from kallithea.lib.exceptions import HgsubversionImportError

from kallithea.model import meta

from kallithea.model.db import Repository, User, Ui, \

    UserLog, RepoGroup, Setting, UserGroup

from kallithea.model.repo_group import RepoGroupModel

from kallithea.lib.utils2 import safe_str, safe_unicode, get_current_authuser

from kallithea.lib.vcs.utils.fakemod import create_module

log = logging.getLogger(__name__)

REMOVED_REPO_PAT = re.compile(r'rm__\d{8}_\d{6}_\d{6}_.*')

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

# PERM DECORATOR HELPERS FOR EXTRACTING NAMES FOR PERM CHECKS

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

def get_repo_slug(request):

    _repo = request.environ['pylons.routes_dict'].get('repo_name')

    if _repo:

        _repo = _repo.rstrip('/')

    return _repo

def get_repo_group_slug(request):

    _group = request.environ['pylons.routes_dict'].get('group_name')

    if _group:

        _group = _group.rstrip('/')

    return _group

def get_user_group_slug(request):

    _group = request.environ['pylons.routes_dict'].get('id')

    _group = UserGroup.get(_group)

    if _group:

        return _group.users_group_name

    return None

def _extract_id_from_repo_name(repo_name):

    if repo_name.startswith('/'):

        repo_name = repo_name.lstrip('/')

    by_id_match = re.match(r'^_(\d{1,})', repo_name)

    if by_id_match:

        return by_id_match.groups()[0]

def get_repo_by_id(repo_name):

    """

    Extracts repo_name by id from special urls. Example url is _11/repo_name

    :param repo_name:

    :return: repo_name if matched else None

    """

    _repo_id = _extract_id_from_repo_name(repo_name)

    if _repo_id:

        from kallithea.model.db import Repository

        repo = Repository.get(_repo_id)

        if repo:

            # TODO: return repo instead of reponame? or would that be a layering violation?

            return repo.repo_name

    return None

def action_logger(user, action, repo, ipaddr='', commit=False):

    """

    Action logger for various actions made by users

    :param user: user that made this action, can be a unique username string or

        object containing user_id attribute

    :param action: action to log, should be on of predefined unique actions for

        easy translations

    :param repo: string name of repository or object containing repo_id,

        that action was made on

    :param ipaddr: optional IP address from what the action was made

    """

    # if we don't get explicit IP address try to get one from registered user

    # in tmpl context var

    if not ipaddr:

        ipaddr = getattr(get_current_authuser(), 'ip_addr', '')

    if getattr(user, 'user_id', None):

        user_obj = User.get(user.user_id)

    elif isinstance(user, basestring):

        user_obj = User.get_by_username(user)

    else:

        raise Exception('You have to provide a user object or a username')

    if getattr(repo, 'repo_id', None):

        repo_obj = Repository.get(repo.repo_id)

        repo_name = repo_obj.repo_name

    elif isinstance(repo, basestring):

        repo_name = repo.lstrip('/')

        repo_obj = Repository.get_by_repo_name(repo_name)

    else:

        repo_obj = None

        repo_name = u''

    user_log = UserLog()

    user_log.user_id = user_obj.user_id

    user_log.username = user_obj.username

    user_log.action = safe_unicode(action)

    user_log.repository = repo_obj

    user_log.repository_name = repo_name

    user_log.action_date = datetime.datetime.now()

    user_log.user_ip = ipaddr

    meta.Session().add(user_log)

    log.info('Logging action:%s on %s by user:%s ip:%s',

             action, safe_unicode(repo), user_obj, ipaddr)

    if commit:

        meta.Session().commit()

def get_filesystem_repos(path):

    """

    Scans given path for repos and return (name,(type,path)) tuple

    :param path: path to scan for repositories

    :param recursive: recursive search and return names with subdirs in front

    """

    # remove ending slash for better results

    path = safe_str(path.rstrip(os.sep))

    log.debug('now scanning in %s', path)

    def isdir(*n):

        return os.path.isdir(os.path.join(*n))

    for root, dirs, _files in os.walk(path):

        recurse_dirs = []

        for subdir in dirs:

            # skip removed repos

            if REMOVED_REPO_PAT.match(subdir):

                continue

            # skip .<something> dirs TODO: rly? then we should prevent creating them ...

            if subdir.startswith('.'):

                continue

            cur_path = os.path.join(root, subdir)

            if isdir(cur_path, '.git'):

                log.warning('ignoring non-bare Git repo: %s', cur_path)

                continue

            if (isdir(cur_path, '.hg') or

                isdir(cur_path, '.svn') or

                isdir(cur_path, 'objects') and (isdir(cur_path, 'refs') or

                                                os.path.isfile(os.path.join(cur_path, 'packed-refs')))):

                if not os.access(cur_path, os.R_OK) or not os.access(cur_path, os.X_OK):

                    log.warning('ignoring repo path without access: %s', cur_path)

                    continue

                if not os.access(cur_path, os.W_OK):

                    log.warning('repo path without write access: %s', cur_path)

                try:

                    scm_info = get_scm(cur_path)

                    assert cur_path.startswith(path)

                    repo_path = cur_path[len(path) + 1:]

                    yield repo_path, scm_info

                    continue # no recursion

                except VCSError:

                    # We should perhaps ignore such broken repos, but especially

                    # the bare git detection is unreliable so we dive into it

                    pass

            recurse_dirs.append(subdir)

        dirs[:] = recurse_dirs

def is_valid_repo_uri(repo_type, url, ui):

    """Check if the url seems like a valid remote repo location - raise an Exception if any problems"""

    if repo_type == 'hg':

        from kallithea.lib.vcs.backends.hg.repository import MercurialRepository

        if url.startswith('http') or url.startswith('ssh'):

            # initially check if it's at least the proper URL

            # or does it pass basic auth

            MercurialRepository._check_url(url, ui)

        elif url.startswith('svn+http'):

            try:

                from hgsubversion.svnrepo import svnremoterepo

            except ImportError:

                raise HgsubversionImportError(_('Unable to activate hgsubversion support. '

                                                'The "hgsubversion" library is missing'))

            svnremoterepo(ui, url).svn.uuid

        elif url.startswith('git+http'):

            raise NotImplementedError()

        else:

            raise Exception('URI %s not allowed' % (url,))

    elif repo_type == 'git':

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

        if url.startswith('http') or url.startswith('git'):

            # initially check if it's at least the proper URL

            # or does it pass basic auth

            GitRepository._check_url(url)

        elif url.startswith('svn+http'):

            raise NotImplementedError()

        elif url.startswith('hg+http'):

            raise NotImplementedError()

        else:

            raise Exception('URI %s not allowed' % (url))

def is_valid_repo(repo_name, base_path, scm=None):

    """

    Returns True if given path is a valid repository False otherwise.

    If scm param is given also compare if given scm is the same as expected

    from scm parameter

    :param repo_name:

    :param base_path:

    :param scm:

    :return True: if given path is a valid repository

    """

    full_path = os.path.join(safe_str(base_path), safe_str(repo_name))

    try:

        scm_ = get_scm(full_path)

        if scm:

            return scm_[0] == scm

        return True

    except VCSError:

        return False

def is_valid_repo_group(repo_group_name, base_path, skip_path_check=False):

    """

    Returns True if given path is a repository group False otherwise

    :param repo_name:

    :param base_path:

    """

    full_path = os.path.join(safe_str(base_path), safe_str(repo_group_name))

    # check if it's not a repo

    if is_valid_repo(repo_group_name, base_path):

        return False

    try:

        # we need to check bare git repos at higher level

        # since we might match branches/hooks/info/objects or possible

        # other things inside bare git repo

        get_scm(os.path.dirname(full_path))

        return False

    except VCSError:

        pass

    # check if it's a valid path

    if skip_path_check or os.path.isdir(full_path):

        return True

    return False

# propagated from mercurial documentation

ui_sections = ['alias', 'auth',

                'decode/encode', 'defaults',

                'diff', 'email',

                'extensions', 'format',

                'merge-patterns', 'merge-tools',

                'hooks', 'http_proxy',

                'smtp', 'patch',

                'paths', 'profiling',

                'server', 'trusted',

                'ui', 'web', ]

def make_ui(read_from='file', path=None, clear_session=True):

    """

    A function that will read python rc files or database

    and make an mercurial ui object from read options

    :param path: path to mercurial config file

    :param read_from: read from 'file' or 'db'

    """

    baseui = ui.ui()

    # clean the baseui object

    baseui._ocfg = config.config()

    baseui._ucfg = config.config()

    baseui._tcfg = config.config()

    if read_from == 'file':

        if not os.path.isfile(path):

            log.debug('hgrc file is not present at %s, skipping...', path)

            return False

        log.debug('reading hgrc from %s', path)

        cfg = config.config()

        cfg.read(path)

        for section in ui_sections:

            for k, v in cfg.items(section):

                log.debug('settings ui from file: [%s] %s=%s', section, k, v)

                baseui.setconfig(safe_str(section), safe_str(k), safe_str(v))

    elif read_from == 'db':

        sa = meta.Session()

        ret = sa.query(Ui).all()

        hg_ui = ret

        for ui_ in hg_ui:

            if ui_.ui_active:

                ui_val = '' if ui_.ui_value is None else safe_str(ui_.ui_value)

                log.debug('settings ui from db: [%s] %s=%r', ui_.ui_section,

                          ui_.ui_key, ui_val)

                baseui.setconfig(safe_str(ui_.ui_section), safe_str(ui_.ui_key),

                                 ui_val)

        if clear_session:

            meta.Session.remove()

        # force set push_ssl requirement to False, Kallithea handles that

        baseui.setconfig('web', 'push_ssl', False)

        baseui.setconfig('web', 'allow_push', '*')

        # prevent interactive questions for ssh password / passphrase

        ssh = baseui.config('ui', 'ssh', default='ssh')

        baseui.setconfig('ui', 'ssh', '%s -oBatchMode=yes -oIdentitiesOnly=yes' % ssh)

    return baseui

def set_app_settings(config):

    """

    Updates app config with new settings from database

    :param config:

    """

    try:

        hgsettings = Setting.get_app_settings()

        for k, v in hgsettings.items():

            config[k] = v

    finally:

        meta.Session.remove()

def set_vcs_config(config):

    """

    Patch VCS config with some Kallithea specific stuff

    :param config: kallithea.CONFIG

    """

    from kallithea.lib.vcs import conf

    from kallithea.lib.utils2 import aslist

    conf.settings.BACKENDS = {

        'hg': 'kallithea.lib.vcs.backends.hg.MercurialRepository',

        'git': 'kallithea.lib.vcs.backends.git.GitRepository',

    }

    conf.settings.GIT_EXECUTABLE_PATH = config.get('git_path', 'git')

    conf.settings.GIT_REV_FILTER = config.get('git_rev_filter', '--all').strip()

    conf.settings.DEFAULT_ENCODINGS = aslist(config.get('default_encoding',

                                                        'utf-8'), sep=',')

def set_indexer_config(config):

    """

    Update Whoosh index mapping

    :param config: kallithea.CONFIG

    """

    from kallithea.config import conf

    log.debug('adding extra into INDEX_EXTENSIONS')

    conf.INDEX_EXTENSIONS.extend(re.split('\s+', config.get('index.extensions', '')))

    log.debug('adding extra into INDEX_FILENAMES')

    conf.INDEX_FILENAMES.extend(re.split('\s+', config.get('index.filenames', '')))

def map_groups(path):

    """

    Given a full path to a repository, create all nested groups that this

    repo is inside. This function creates parent-child relationships between

    groups and creates default perms for all new groups.

    :param paths: full path to repository

    """

    sa = meta.Session()

    groups = path.split(Repository.url_sep())

    parent = None

    group = None

    # last element is repo in nested groups structure

    groups = groups[:-1]

    rgm = RepoGroupModel()

    owner = User.get_first_admin()

    for lvl, group_name in enumerate(groups):

        group_name = u'/'.join(groups[:lvl] + [group_name])

        group = RepoGroup.get_by_group_name(group_name)

        desc = '%s group' % group_name

# -*- 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

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

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 os

import re

import sys

import time

import uuid

import datetime

import urllib

import binascii

import webob

import urlobject

from tg.i18n import ugettext as _, ungettext

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

from kallithea.lib.compat import json

def str2bool(_str):

    """

    returns True/False value from given string, it tries to translate the

    string into boolean

    :param _str: string value to translate into boolean

    :rtype: boolean

    :returns: boolean from given string

    """

    if _str is None:

        return False

    if _str in (True, False):

        return _str

    _str = str(_str).strip().lower()

    return _str in ('t', 'true', 'y', 'yes', 'on', '1')

def aslist(obj, sep=None, strip=True):

    """

    Returns given string separated by sep as list

    :param obj:

    :param sep:

    :param strip:

    """

    if isinstance(obj, (basestring)):

        lst = obj.split(sep)

        if strip:

            lst = [v.strip() for v in lst]

        return lst

    elif isinstance(obj, (list, tuple)):

        return obj

    elif obj is None:

        return []

    else:

        return [obj]

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

    """

    from string import replace

    if mode == 0:

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

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

    elif mode == 1:

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

            line = replace(line, '\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 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 safe_unicode(str_, from_encoding=None):

    """

    safe unicode function. Does few trick to turn str_ into unicode

    In case of UnicodeDecode error we try to return it with encoding detected

    by chardet library if it fails fallback to unicode with errors replaced

    :param str_: string to decode

    :rtype: unicode

    :returns: unicode object

    """

    if isinstance(str_, unicode):

        return str_

    if not from_encoding:

        import kallithea

        DEFAULT_ENCODINGS = aslist(kallithea.CONFIG.get('default_encoding',

                                                        'utf-8'), sep=',')

        from_encoding = DEFAULT_ENCODINGS

    if not isinstance(from_encoding, (list, tuple)):

        from_encoding = [from_encoding]

    try:

        return unicode(str_)

    except UnicodeDecodeError:

        pass

    for enc in from_encoding:

        try:

            return unicode(str_, enc)

        except UnicodeDecodeError:

            pass

    try:

        import chardet

        encoding = chardet.detect(str_)['encoding']

        if encoding is None:

            raise Exception()

        return str_.decode(encoding)

    except (ImportError, UnicodeDecodeError, Exception):

        return unicode(str_, from_encoding[0], 'replace')

def safe_str(unicode_, to_encoding=None):

    """

    safe str function. Does few trick to turn unicode_ into string

    In case of UnicodeEncodeError we try to return it with encoding detected

    by chardet library if it fails fallback to string with errors replaced

    :param unicode_: unicode to encode

    :rtype: str

    :returns: str object

    """

    # if it's not basestr cast to str

    if not isinstance(unicode_, basestring):

        return str(unicode_)

    if isinstance(unicode_, str):

        return unicode_

    if not to_encoding:

        import kallithea

        DEFAULT_ENCODINGS = aslist(kallithea.CONFIG.get('default_encoding',

                                                        'utf-8'), sep=',')

        to_encoding = DEFAULT_ENCODINGS

    if not isinstance(to_encoding, (list, tuple)):

        to_encoding = [to_encoding]

    for enc in to_encoding:

        try:

            return unicode_.encode(enc)

        except UnicodeEncodeError:

            pass

    try:

        import chardet

        encoding = chardet.detect(unicode_)['encoding']

        if encoding is None:

            raise UnicodeEncodeError()

        return unicode_.encode(encoding)

    except (ImportError, UnicodeEncodeError):

        return unicode_.encode(to_encoding[0], 'replace')

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

    :returns: unicode 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

    from dateutil import relativedelta

    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(u'%d year', '%d years', d) % d,

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

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

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

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

        'second': lambda d: ungettext(u'%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:

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

                fmt_funcs[sub_part](sub_value))

    return _('just now')

def uri_filter(uri):

    """

    Removes user:password from given url string

    :param uri:

    :rtype: unicode

    :returns: filtered list of strings

    """

    if not uri:

        return ''

    proto = ''

    for pat in ('https://', 'http://', 'git://'):

        if uri.startswith(pat):

            uri = uri[len(pat):]

            proto = pat

            break

    # remove passwords and username

    uri = uri[uri.find('@') + 1:]

    # get the port

    cred_pos = uri.find(':')