kallithea Changeset - 9685f50a69d0

Changeset - 9685f50a69d0

Parent rev.

Child rev.

[Not reviewed]

default

0 13 0

Mads Kiilerich (mads) - 5 years ago 2020-10-11 12:27:26
mads@kiilerich.com

Grafted from: 141906344b85

imports: move more imports to top level

13 files changed with 38 insertions and 63 deletions:

kallithea/lib/annotate.py

kallithea/lib/auth_modules/__init__.py

kallithea/lib/auth_modules/auth_internal.py

kallithea/lib/celerylib/tasks.py

kallithea/lib/db_manage.py

kallithea/lib/helpers.py

kallithea/lib/markup_renderer.py

kallithea/lib/pygmentsutils.py

kallithea/lib/utils2.py

kallithea/lib/vcs/backends/base.py

kallithea/lib/vcs/nodes.py

kallithea/lib/vcs/utils/__init__.py

kallithea/lib/vcs/utils/helpers.py

0 comments (0 inline, 0 general)

kallithea/lib/annotate.py

➞

Show inline comments

@@ @@ -7,69 +7,69 @@ @@
 # 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.annotate
 ~~~~~~~~~~~~~~~~~~~~~~
 Annotation library for usage in Kallithea, previously part of vcs
 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: Dec 4, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 from pygments import highlight
 from pygments.formatters import HtmlFormatter
 from kallithea.lib.pygmentsutils import get_custom_lexer
 from kallithea.lib.vcs.exceptions import VCSError
 from kallithea.lib.vcs.nodes import FileNode
 from kallithea.lib.vcs.utils import safe_str
 def annotate_highlight(filenode, annotate_from_changeset_func,
         order=None, headers=None, **options):
     """
     Returns html portion containing annotated table with 3 columns: line
     numbers, changeset information and pygmentized line of code.
     :param filenode: FileNode object
     :param annotate_from_changeset_func: function taking changeset and
       returning single annotate cell; needs break line at the end
     :param order: ordered sequence of ``ls`` (line numbers column),
       ``annotate`` (annotate column), ``code`` (code column); Default is
       ``['ls', 'annotate', 'code']``
     :param headers: dictionary with headers (keys are whats in ``order``
       parameter)
     """
     from kallithea.lib.pygmentsutils import get_custom_lexer
     options['linenos'] = True
     formatter = AnnotateHtmlFormatter(filenode=filenode,
         annotate_from_changeset_func=annotate_from_changeset_func, order=order,
         headers=headers, **options)
     lexer = get_custom_lexer(filenode.extension) or filenode.lexer
     highlighted = highlight(safe_str(filenode.content), lexer, formatter)
     return highlighted
 class AnnotateHtmlFormatter(HtmlFormatter):
     def __init__(self, filenode, annotate_from_changeset_func,
             order=None, **options):
         """
         ``annotate_from_changeset_func`` must be a function
         which returns string from the given changeset. For example, we may pass
         following function as ``annotate_from_changeset_func``::
             def changeset_to_anchor(changeset):
                 return '<a href="/changesets/%s/">%s</a>\n' % \
                        (changeset.raw_id, changeset.raw_id)
         :param annotate_from_changeset_func: see above
         :param order: (default: ``['ls', 'annotate', 'code']``); order of

kallithea/lib/auth_modules/__init__.py

➞

Show inline comments

 # -*- 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/>.
 """
 Authentication modules
 """
 import importlib
 import logging
 import traceback
 from inspect import isfunction
 from kallithea.lib.auth import AuthUser, PasswordGenerator
 from kallithea.lib.compat import hybrid_property
 from kallithea.lib.utils2 import asbool
 from kallithea.model import validators
 from kallithea.model.db import Setting, User
 from kallithea.model.meta import Session
 from kallithea.model.user import UserModel
 from kallithea.model.user_group import UserGroupModel
 log = logging.getLogger(__name__)
 class LazyFormencode(object):
     def __init__(self, formencode_obj, *args, **kwargs):
         self.formencode_obj = formencode_obj
         self.args = args
         self.kwargs = kwargs
     def __call__(self, *args, **kwargs):
         from inspect import isfunction
         formencode_obj = self.formencode_obj
         if isfunction(formencode_obj):
             # case we wrap validators into functions
             formencode_obj = self.formencode_obj(*args, **kwargs)
         return formencode_obj(*self.args, **self.kwargs)
 class KallitheaAuthPluginBase(object):
     auth_func_attrs = {
         "username": "unique username",
         "firstname": "first name",
         "lastname": "last name",
         "email": "email address",
         "groups": '["list", "of", "groups"]',
         "extern_name": "name in external source of record",
         "admin": 'True|False defines if user should be Kallithea admin',
+    }
     @property
     def validators(self):
         """
         Exposes Kallithea validators modules
         """
         # this is a hack to overcome issues with pylons threadlocals and
         # translator object _() not being registered properly.
         class LazyCaller(object):
             def __init__(self, name):
                 self.validator_name = name
             def __call__(self, *args, **kwargs):
                 from kallithea.model import validators as v
                 obj = getattr(v, self.validator_name)
                 obj = getattr(validators, self.validator_name)
                 #log.debug('Initializing lazy formencode object: %s', obj)
                 return LazyFormencode(obj, *args, **kwargs)
         class ProxyGet(object):
             def __getattribute__(self, name):
                 return LazyCaller(name)
         return ProxyGet()
     @hybrid_property
     def name(self):
         """
         Returns the name of this authentication plugin.
         :returns: string
         """
         raise NotImplementedError("Not implemented in base class")
     @hybrid_property
     def is_container_auth(self):
         """
         Returns bool if this module uses container auth.
         This property will trigger an automatic call to authenticate on

kallithea/lib/auth_modules/auth_internal.py

➞

Show inline comments

@@ @@ -7,49 +7,49 @@ @@
 # 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.auth_modules.auth_internal
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Kallithea authentication plugin for built in internal auth
 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: Created on Nov 17, 2012
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 from kallithea.lib import auth_modules
+from kallithea.lib import auth, auth_modules
 from kallithea.lib.compat import hybrid_property
 log = logging.getLogger(__name__)
 class KallitheaAuthPlugin(auth_modules.KallitheaAuthPluginBase):
     def __init__(self):
         pass
     @hybrid_property
     def name(self):
         # Also found as kallithea.lib.model.db.User.DEFAULT_AUTH_TYPE
         return 'internal'
     def settings(self):
         return []
     def accepts(self, user, accepts_empty=True):
         """
         Custom accepts for this auth that doesn't accept empty users. We
         know that user exists in database.
         """
         return super(KallitheaAuthPlugin, self).accepts(user,
@@ @@ -57,41 +57,40 @@ class KallitheaAuthPlugin(auth_modules.K @@
     def auth(self, userobj, username, password, settings, **kwargs):
         if not userobj:
             log.debug('userobj was:%s skipping', userobj)
             return None
         if userobj.extern_type != self.name:
             log.warning("userobj:%s extern_type mismatch got:`%s` expected:`%s`",
                      userobj, userobj.extern_type, self.name)
             return None
         if not username:
             log.debug('Empty username - skipping...')
             return None
         user_data = {
             "username": userobj.username,
             "firstname": userobj.firstname,
             "lastname": userobj.lastname,
             "groups": [],
             "email": userobj.email,
             "admin": userobj.admin,
             "extern_name": userobj.user_id,
+        }
         log.debug('user data: %s', user_data)
         from kallithea.lib import auth
         password_match = auth.check_password(password, userobj.password)
         if userobj.is_default_user:
             log.info('user %s authenticated correctly as anonymous user',
                      username)
             return user_data
         elif userobj.username == username and password_match:
             log.info('user %s authenticated correctly', user_data['username'])
             return user_data
         log.error("user %s had a bad password", username)
         return None
     def get_managed_fields(self):
         # Note: 'username' should only be editable (at least for user) if self registration is enabled
         return []

kallithea/lib/celerylib/tasks.py

➞

Show inline comments

@@ @@ -16,69 +16,71 @@ kallithea.lib.celerylib.tasks @@
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Kallithea task modules, containing all task that suppose to be run
 by celery daemon
 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: Oct 6, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import email.utils
 import os
 import traceback
 from collections import OrderedDict
 from operator import itemgetter
 from time import mktime
 import celery.utils.log
 from tg import config
 import kallithea
 from kallithea.config import conf
 from kallithea.lib import celerylib, ext_json
 from kallithea.lib.helpers import person
 from kallithea.lib.hooks import log_create_repository
 from kallithea.lib.indexers.daemon import WhooshIndexingDaemon
 from kallithea.lib.rcmail.smtp_mailer import SmtpMailer
 from kallithea.lib.utils import action_logger
 from kallithea.lib.utils2 import asbool, ascii_bytes
 from kallithea.lib.vcs.utils import author_email
 from kallithea.model.db import RepoGroup, Repository, Statistics, User
 from kallithea.model.db import RepoGroup, Repository, Setting, Statistics, User
 from kallithea.model.repo import RepoModel
 __all__ = ['whoosh_index', 'get_commits_stats', 'send_email']
 log = celery.utils.log.get_task_logger(__name__)
 @celerylib.task
 @celerylib.locked_task
 @celerylib.dbsession
 def whoosh_index(repo_location, full_index):
     from kallithea.lib.indexers.daemon import WhooshIndexingDaemon
     celerylib.get_session() # initialize database connection
     index_location = config['index_dir']
     WhooshIndexingDaemon(index_location=index_location,
                          repo_location=repo_location) \
                          .run(full_index=full_index)
 # for js data compatibility cleans the key for person from '
 def akc(k):
     return person(k).replace('"', '')
 @celerylib.task
 @celerylib.dbsession
 def get_commits_stats(repo_name, ts_min_y, ts_max_y, recurse_limit=100):
     DBS = celerylib.get_session()
     lockkey = celerylib.__get_lockkey('get_commits_stats', repo_name, ts_min_y,
                             ts_max_y)
     lockkey_path = config.get('cache_dir') or config['app_conf']['cache_dir']  # Backward compatibility for TurboGears < 2.4
     log.info('running task with lockkey %s', lockkey)
     try:
@@ @@ -302,51 +304,48 @@ def send_email(recipients, subject, body @@
               "html:\n%s\n"
               % (' '.join(recipients), headers, subject, body, html_body))
     if mail_server:
         log.debug("Sending e-mail. " + logmsg)
     else:
         log.error("SMTP mail server not configured - cannot send e-mail.")
         log.warning(logmsg)
         return False
     try:
         m = SmtpMailer(envelope_from, user, passwd, mail_server, smtp_auth,
                        mail_port, ssl, tls, debug=debug)
         m.send(recipients, subject, body, html_body, headers=headers)
     except:
         log.error('Mail sending failed')
         log.error(traceback.format_exc())
         return False
     return True
 @celerylib.task
 @celerylib.dbsession
 def create_repo(form_data, cur_user):
     from kallithea.model.db import Setting
     from kallithea.model.repo import RepoModel
     DBS = celerylib.get_session()
     cur_user = User.guess_instance(cur_user)
     owner = cur_user
     repo_name = form_data['repo_name']
     repo_name_full = form_data['repo_name_full']
     repo_type = form_data['repo_type']
     description = form_data['repo_description']
     private = form_data['repo_private']
     clone_uri = form_data.get('clone_uri')
     repo_group = form_data['repo_group']
     landing_rev = form_data['repo_landing_rev']
     copy_fork_permissions = form_data.get('copy_permissions')
     copy_group_permissions = form_data.get('repo_copy_permissions')
     fork_of = form_data.get('fork_parent_id')
     state = form_data.get('repo_state', Repository.STATE_PENDING)
     # repo creation defaults, private and repo_type are filled in form
     defs = Setting.get_default_repo_settings(strip_prefix=True)
     enable_statistics = defs.get('repo_enable_statistics')
     enable_downloads = defs.get('repo_enable_downloads')
     try:
@@ @@ -389,50 +388,48 @@ def create_repo(form_data, cur_user): @@
         DBS.commit()
     except Exception as e:
         log.warning('Exception %s occurred when forking repository, '
                     'doing cleanup...' % e)
         # rollback things manually !
         repo = Repository.get_by_repo_name(repo_name_full)
         if repo:
             Repository.delete(repo.repo_id)
             DBS.commit()
             RepoModel()._delete_filesystem_repo(repo)
         raise
     return True
 @celerylib.task
 @celerylib.dbsession
 def create_repo_fork(form_data, cur_user):
     """
     Creates a fork of repository using interval VCS methods
     :param form_data:
     :param cur_user:
     """
     from kallithea.model.repo import RepoModel
     DBS = celerylib.get_session()
     base_path = kallithea.CONFIG['base_path']
     cur_user = User.guess_instance(cur_user)
     repo_name = form_data['repo_name']  # fork in this case
     repo_name_full = form_data['repo_name_full']
     repo_type = form_data['repo_type']
     owner = cur_user
     private = form_data['private']
     clone_uri = form_data.get('clone_uri')
     repo_group = form_data['repo_group']
     landing_rev = form_data['landing_rev']
     copy_fork_permissions = form_data.get('copy_permissions')
     try:
         fork_of = Repository.guess_instance(form_data.get('fork_parent_id'))
         RepoModel()._create_repo(
             repo_name=repo_name_full,
             repo_type=repo_type,
             description=form_data['description'],
             owner=owner,
@@ @@ -459,40 +456,39 @@ def create_repo_fork(form_data, cur_user @@
         repo = Repository.get_by_repo_name(repo_name_full)
         log_create_repository(repo.get_dict(), created_by=owner.username)
         # update repo changeset caches initially
         repo.update_changeset_cache()
         # set new created state
         repo.set_state(Repository.STATE_CREATED)
         DBS.commit()
     except Exception as e:
         log.warning('Exception %s occurred when forking repository, '
                     'doing cleanup...' % e)
         # rollback things manually !
         repo = Repository.get_by_repo_name(repo_name_full)
         if repo:
             Repository.delete(repo.repo_id)
             DBS.commit()
             RepoModel()._delete_filesystem_repo(repo)
         raise
     return True
 def __get_codes_stats(repo_name):
     from kallithea.config.conf import LANGUAGES_EXTENSIONS_MAP
     repo = Repository.get_by_repo_name(repo_name).scm_instance
     tip = repo.get_changeset()
     code_stats = {}
     for _topnode, _dirnodes, filenodes in tip.walk('/'):
         for filenode in filenodes:
             ext = filenode.extension.lower()
             if ext in LANGUAGES_EXTENSIONS_MAP and not filenode.is_binary:
+            if ext in conf.LANGUAGES_EXTENSIONS_MAP and not filenode.is_binary:
                 if ext in code_stats:
                     code_stats[ext] += 1
                 else:
                     code_stats[ext] = 1
     return code_stats or {}

kallithea/lib/db_manage.py

➞

Show inline comments

@@ @@ -15,73 +15,73 @@ @@
 kallithea.lib.db_manage
 ~~~~~~~~~~~~~~~~~~~~~~~
 Database creation, and setup module for Kallithea. Used for creation
 of database as well as for migration operations
 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 10, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import os
 import sys
 import uuid
 import alembic.command
 import alembic.config
 import sqlalchemy
 from sqlalchemy.engine import create_engine
 from kallithea.lib.utils2 import ask_ok
 from kallithea.model.base import init_model
 from kallithea.model.db import Repository, Setting, Ui, User
 from kallithea.model.meta import Base, Session
 from kallithea.model.permission import PermissionModel
 from kallithea.model.user import UserModel
 log = logging.getLogger(__name__)
 class DbManage(object):
     def __init__(self, dbconf, root, tests=False, SESSION=None, cli_args=None):
         self.dbname = dbconf.split('/')[-1]
         self.tests = tests
         self.root = root
         self.dburi = dbconf
         self.cli_args = cli_args or {}
         self.init_db(SESSION=SESSION)
     def _ask_ok(self, msg):
         """Invoke ask_ok unless the force_ask option provides the answer"""
         force_ask = self.cli_args.get('force_ask')
         if force_ask is not None:
             return force_ask
         from kallithea.lib.utils2 import ask_ok
         return ask_ok(msg)
     def init_db(self, SESSION=None):
         if SESSION:
             self.sa = SESSION
         else:
             # init new sessions
             engine = create_engine(self.dburi)
             init_model(engine)
             self.sa = Session()
     def create_tables(self, reuse_database=False):
         """
         Create database (optional) and tables.
         If reuse_database is false, the database will be dropped (if it exists)
         and a new one created. If true, the existing database will be reused
         and cleaned for content.
         """
         url = sqlalchemy.engine.url.make_url(self.dburi)
         database = url.database
         if reuse_database:
             log.info("The content of the database %r will be destroyed and new tables created." % database)
         else:
             log.info("The existing database %r will be destroyed and a new one created." % database)

kallithea/lib/helpers.py

➞

Show inline comments

@@ @@ -7,76 +7,80 @@ @@
 # 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/>.
 """
 Helper functions
 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 random
 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 import session
 from tg.i18n import ugettext as _
 from webhelpers2.html import HTML, escape, literal
 from webhelpers2.html.tags import NotGiven, Option, Options, _input, _make_safe_id_component, checkbox, end_form
 from webhelpers2.html.tags import form as insecure_form
 from webhelpers2.html.tags import hidden, link_to, password, radio
 from webhelpers2.html.tags import select as webhelpers2_select
 from webhelpers2.html.tags import submit, text, textarea
 from webhelpers2.number import format_byte_size
 from webhelpers2.text import chop_at, truncate, wrap_paragraphs
 import kallithea
 from kallithea.config.routing import url
 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.utils2 import MENTIONS_REGEX, AttributeDict
 from kallithea.lib.utils2 import age as _age
 from kallithea.lib.utils2 import asbool, credentials_filter, safe_bytes, safe_int, safe_str, time_to_datetime
 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.model.changeset_status import ChangesetStatusModel
 from kallithea.model.db import URL_SEP, ChangesetStatus, Permission, PullRequest, User, UserIpMap
 # mute pyflakes "imported but unused"
 assert Option
 assert checkbox
 assert end_form
 assert password
 assert radio
 assert submit
 assert text
 assert textarea
 assert format_byte_size
 assert chop_at
 assert wrap_paragraphs
 assert HasPermissionAny
 assert HasRepoGroupPermissionLevel
 assert HasRepoPermissionLevel
 assert time_to_datetime
 assert EmptyChangeset
 log = logging.getLogger(__name__)
@@ @@ -464,49 +468,48 @@ def pygmentize_annotation(repo_name, fil @@
         return uri
     return literal(markup_whitespace(annotate_highlight(filenode, url_func, **kwargs)))
 class _Message(object):
     """A message returned by ``pop_flash_messages()``.
     Converting the message to a string returns the message text. Instances
     also have the following attributes:
     * ``category``: the category specified when the message was created.
     * ``message``: the html-safe message text.
     """
     def __init__(self, category, message):
         self.category = category
         self.message = message
 def _session_flash_messages(append=None, clear=False):
     """Manage a message queue in tg.session: return the current message queue
     after appending the given message, and possibly clearing the queue."""
     key = 'flash'
     from tg import session
     if key in session:
         flash_messages = session[key]
     else:
         if append is None:  # common fast path - also used for clearing empty queue
             return []  # don't bother saving
         flash_messages = []
         session[key] = flash_messages
     if append is not None and append not in flash_messages:
         flash_messages.append(append)
     if clear:
         session.pop(key, None)
     session.save()
     return flash_messages
 def flash(message, category, logf=None):
     """
     Show a message to the user _and_ log it through the specified function
     category: notice (default), warning, error, success
     logf: a custom log function - such as log.debug
     logf defaults to log.info, unless category equals 'success', in which
     case logf defaults to log.debug.
@@ @@ -580,93 +583,88 @@ def is_git(repository): @@
         _type = repository.alias
     elif hasattr(repository, 'repo_type'):
         _type = repository.repo_type
     else:
         _type = repository
     return _type == 'git'
 def is_hg(repository):
     if hasattr(repository, 'alias'):
         _type = repository.alias
     elif hasattr(repository, 'repo_type'):
         _type = repository.repo_type
     else:
         _type = repository
     return _type == 'hg'
 @cache_region('long_term', 'user_attr_or_none')
 def user_attr_or_none(author, show_attr):
     """Try to match email part of VCS committer string with a local user and return show_attr
     - or return None if user not found"""
     email = author_email(author)
     if email:
         from kallithea.model.db import User
         user = User.get_by_email(email)
         if user is not None:
             return getattr(user, show_attr)
     return None
 def email_or_none(author):
     """Try to match email part of VCS committer string with a local user.
     Return primary email of user, email part of the specified author name, or None."""
     if not author:
         return None
     email = user_attr_or_none(author, 'email')
     if email is not None:
         return email # always use user's main email address - not necessarily the one used to find user
     # extract email from the commit string
     email = author_email(author)
     if email:
         return email
     # No valid email, not a valid user in the system, none!
     return None
 def person(author, show_attr="username"):
     """Find the user identified by 'author', return one of the users attributes,
     default to the username attribute, None if there is no user"""
     from kallithea.model.db import User
     # if author is already an instance use it for extraction
     if isinstance(author, User):
         return getattr(author, show_attr)
     value = user_attr_or_none(author, show_attr)
     if value is not None:
         return value
     # Still nothing?  Just pass back the author name if any, else the email
     return author_name(author) or email(author)
 def person_by_id(id_, show_attr="username"):
     from kallithea.model.db import User
     # maybe it's an ID ?
     if str(id_).isdigit() or isinstance(id_, int):
         id_ = int(id_)
         user = User.get(id_)
         if user is not None:
             return getattr(user, show_attr)
     return id_
 def boolicon(value):
     """Returns boolean value of a value, represented as small html image of true/false
     icons
     :param value: value
     """
     if value:
         return HTML.tag('i', class_="icon-ok")
     else:
         return HTML.tag('i', class_="icon-minus-circled")
 def action_parser(user_log, feed=False, parse_cs=False):
     """
@@ @@ -806,49 +804,48 @@ def action_parser(user_log, feed=False, @@
+            )
             if len(revs_ids) > revs_top_limit:
                 morelinks += ', ...'
             cs_links.append(html_tmpl % (uniq_id, morelinks))
         if len(revs) > 1:
             cs_links.append(compare_view)
         return ''.join(cs_links)
     def get_fork_name():
         repo_name = action_params
         url_ = url('summary_home', repo_name=repo_name)
         return _('Fork name %s') % link_to(action_params, url_)
     def get_user_name():
         user_name = action_params
         return user_name
     def get_users_group():
         group_name = action_params
         return group_name
     def get_pull_request():
         from kallithea.model.db import PullRequest
         pull_request_id = action_params
         nice_id = PullRequest.make_nice_id(pull_request_id)
         deleted = user_log.repository is None
         if deleted:
             repo_name = user_log.repository_name
         else:
             repo_name = user_log.repository.repo_name
         return link_to(_('Pull request %s') % nice_id,
                     url('pullrequest_show', repo_name=repo_name,
                     pull_request_id=pull_request_id))
     def get_archive_name():
         archive_name = action_params
         return archive_name
     # action : translated str, callback(extractor), icon
     action_map = {
         'user_deleted_repo':           (_('[deleted] repository'),
                                         None, 'icon-trashcan'),
         'user_created_repo':           (_('[created] repository'),
                                         None, 'icon-plus'),
         'user_created_fork':           (_('[created] repository as fork'),
@@ @@ -964,85 +961,82 @@ def gravatar(email_address, cls='', size @@
         html = ('<i class="icon-gravatar {cls}"'
                 ' style="font-size: {size}px;background-size: {size}px;background-image: url(\'{src}\')"'
                 '></i>').format(cls=cls, size=size, src=src)
     else:
         # if src is empty then there was no gravatar, so we use a font icon
         html = ("""<i class="icon-user {cls}" style="font-size: {size}px;"></i>"""
             .format(cls=cls, size=size))
     return literal(html)
 def gravatar_url(email_address, size=30, default=''):
     from tg import tmpl_context as c
     if not c.visual.use_gravatar:
         return ""
     _def = 'anonymous@kallithea-scm.org'  # default gravatar
     email_address = email_address or _def
     if email_address == _def:
         return default
     from kallithea.model.db import User
     parsed_url = urllib.parse.urlparse(url.current(qualified=True))
     return (c.visual.gravatar_url or User.DEFAULT_GRAVATAR_URL) \
                .replace('{email}', email_address) \
                .replace('{md5email}', hashlib.md5(safe_bytes(email_address).lower()).hexdigest()) \
                .replace('{netloc}', parsed_url.netloc) \
                .replace('{scheme}', parsed_url.scheme) \
                .replace('{size}', str(size))
 def changed_tooltip(nodes):
     """
     Generates a html string for changed nodes in changeset page.
     It limits the output to 30 entries
     :param nodes: LazyNodesGenerator
     """
     if nodes:
         pref = ': <br/> '
         suf = ''
         if len(nodes) > 30:
             suf = '<br/>' + _(' and %s more') % (len(nodes) - 30)
         return literal(pref + '<br/> '.join([x.path
                                              for x in nodes[:30]]) + suf)
     else:
         return ': ' + _('No files')
 def fancy_file_stats(stats):
     """
     Displays a fancy two colored bar for number of added/deleted
     lines of code on file
     :param stats: two element list of added/deleted lines of code
     """
     from kallithea.lib.diffs import BIN_FILENODE, CHMOD_FILENODE, DEL_FILENODE, MOD_FILENODE, NEW_FILENODE, RENAMED_FILENODE
     a, d = stats['added'], stats['deleted']
     width = 100
     if stats['binary']:
         # binary mode
         lbl = ''
         bin_op = 1
         if BIN_FILENODE in stats['ops']:
             lbl = 'bin+'
         if NEW_FILENODE in stats['ops']:
             lbl += _('new file')
             bin_op = NEW_FILENODE
         elif MOD_FILENODE in stats['ops']:
             lbl += _('mod')
             bin_op = MOD_FILENODE
         elif DEL_FILENODE in stats['ops']:
             lbl += _('del')
             bin_op = DEL_FILENODE
         elif RENAMED_FILENODE in stats['ops']:
             lbl += _('rename')
             bin_op = RENAMED_FILENODE
@@ @@ -1185,49 +1179,48 @@ def linkify_others(t, l): @@
     """Add a default link to html with links.
     HTML doesn't allow nesting of links, so the outer link must be broken up
     in pieces and give space for other links.
     """
     urls = re.compile(r'(\<a.*?\<\/a\>)',)
     links = []
     for e in urls.split(t):
         if e.strip() and not urls.match(e):
             links.append('<a class="message-link" href="%s">%s</a>' % (l, e))
         else:
             links.append(e)
     return ''.join(links)
 # Global variable that will hold the actual urlify_issues function body.
 # Will be set on first use when the global configuration has been read.
 _urlify_issues_f = None
 def urlify_issues(newtext, repo_name):
     """Urlify issue references according to .ini configuration"""
     global _urlify_issues_f
     if _urlify_issues_f is None:
         from kallithea.model.db import URL_SEP
         assert kallithea.CONFIG['sqlalchemy.url'] # make sure config has been loaded
         # Build chain of urlify functions, starting with not doing any transformation
         def tmp_urlify_issues_f(s):
             return s
         issue_pat_re = re.compile(r'issue_pat(.*)')
         for k in kallithea.CONFIG:
             # Find all issue_pat* settings that also have corresponding server_link and prefix configuration
             m = issue_pat_re.match(k)
             if m is None:
                 continue
             suffix = m.group(1)
             issue_pat = kallithea.CONFIG.get(k)
             issue_server_link = kallithea.CONFIG.get('issue_server_link%s' % suffix)
             issue_sub = kallithea.CONFIG.get('issue_sub%s' % suffix)
             issue_prefix = kallithea.CONFIG.get('issue_prefix%s' % suffix)
             if issue_prefix:
                 log.error('found unsupported issue_prefix%s = %r - use issue_sub%s instead', suffix, issue_prefix, suffix)
             if not issue_pat:
                 log.error('skipping incomplete issue pattern %r: it needs a regexp', k)
                 continue
             if not issue_server_link:
                 log.error('skipping incomplete issue pattern %r: it needs issue_server_link%s', k, suffix)
@@ @@ -1296,90 +1289,85 @@ def short_ref(ref_type, ref_name): @@
     if ref_type == 'rev':
         return short_id(ref_name)
     return ref_name
 def link_to_ref(repo_name, ref_type, ref_name, rev=None):
     """
     Return full markup for a href to changeset_home for a changeset.
     If ref_type is branch it will link to changelog.
     ref_name is shortened if ref_type is 'rev'.
     if rev is specified show it too, explicitly linking to that revision.
     """
     txt = short_ref(ref_type, ref_name)
     if ref_type == 'branch':
         u = url('changelog_home', repo_name=repo_name, branch=ref_name)
     else:
         u = url('changeset_home', repo_name=repo_name, revision=ref_name)
     l = link_to(repo_name + '#' + txt, u)
     if rev and ref_type != 'rev':
         l = literal('%s (%s)' % (l, link_to(short_id(rev), url('changeset_home', repo_name=repo_name, revision=rev))))
     return l
 def changeset_status(repo, revision):
     from kallithea.model.changeset_status import ChangesetStatusModel
     return ChangesetStatusModel().get_status(repo, revision)
 def changeset_status_lbl(changeset_status):
     from kallithea.model.db import ChangesetStatus
     return ChangesetStatus.get_status_lbl(changeset_status)
 def get_permission_name(key):
     from kallithea.model.db import Permission
     return dict(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'
             "repository:*vcs*" - search for repository containing 'vcs'
         Optional AND / OR operators in queries
             "repository:vcs OR repository:test"
             "username:test AND repository:test*"
     '''))
 def not_mapped_error(repo_name):
     flash(_('%s repository is not mapped to db perhaps'
             ' it was created or renamed from the filesystem'
             ' please run the application again'
             ' in order to rescan repositories') % repo_name, category='error')
 def ip_range(ip_addr):
     from kallithea.model.db import UserIpMap
     s, e = UserIpMap._get_ip_range(ip_addr)
     return '%s - %s' % (s, e)
 session_csrf_secret_name = "_session_csrf_secret_token"
 def session_csrf_secret_token():
     """Return (and create) the current session's CSRF protection token."""
     from tg import session
     if not session_csrf_secret_name in session:
         session[session_csrf_secret_name] = str(random.getrandbits(128))
         session.save()
     return session[session_csrf_secret_name]
 def form(url, method="post", **attrs):
     """Like webhelpers.html.tags.form , but automatically adding
     session_csrf_secret_token for POST. The secret is thus never leaked in GET
     URLs.
     """
     form = insecure_form(url, method, **attrs)
     if method.lower() == 'get':
         return form
     return form + HTML.div(hidden(session_csrf_secret_name, session_csrf_secret_token()), style="display: none;")

kallithea/lib/markup_renderer.py

➞

Show inline comments

@@ @@ -5,103 +5,105 @@ @@
 # (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.markup_renderer
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Renderer for markup languages with ability to parse using rst or markdown
 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: Oct 27, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import hashlib
 import logging
 import re
 import traceback
 import bleach
 import markdown as markdown_mod
 from docutils.core import publish_parts
 from docutils.parsers.rst import directives
 from kallithea.lib.utils2 import MENTIONS_REGEX, safe_str
 log = logging.getLogger(__name__)
 url_re = re.compile(r'''\bhttps?://(?:[\da-zA-Z0-9@:.-]+)'''
                     r'''(?:[/a-zA-Z0-9_=@#~&+%.,:;?!*()-]*[/a-zA-Z0-9_=@#~])?''')
 class MarkupRenderer(object):
     RESTRUCTUREDTEXT_DISALLOWED_DIRECTIVES = ['include', 'meta', 'raw']
     MARKDOWN_PAT = re.compile(r'md|mkdn?|mdown|markdown', re.IGNORECASE)
     RST_PAT = re.compile(r're?st', re.IGNORECASE)
     PLAIN_PAT = re.compile(r'readme', re.IGNORECASE)
     @classmethod
     def _detect_renderer(cls, source, filename):
         """
         runs detection of what renderer should be used for generating html
         from a markup language
         filename can be also explicitly a renderer name
         """
         if cls.MARKDOWN_PAT.findall(filename):
             return cls.markdown
         elif cls.RST_PAT.findall(filename):
             return cls.rst
         elif cls.PLAIN_PAT.findall(filename):
             return cls.rst
         return cls.plain
     @classmethod
     def _flavored_markdown(cls, text):
         """
         Github style flavored markdown
         :param text:
         """
         from hashlib import sha1
         # Extract pre blocks.
         extractions = {}
         def pre_extraction_callback(matchobj):
             digest = sha1(matchobj.group(0)).hexdigest()
+            digest = hashlib.sha1(matchobj.group(0)).hexdigest()
             extractions[digest] = matchobj.group(0)
             return "{gfm-extraction-%s}" % digest
         pattern = re.compile(r'<pre>.*?</pre>', re.MULTILINE | re.DOTALL)
         text = re.sub(pattern, pre_extraction_callback, text)
         # Prevent foo_bar_baz from ending up with an italic word in the middle.
         def italic_callback(matchobj):
             s = matchobj.group(0)
             if list(s).count('_') >= 2:
                 return s.replace('_', r'\_')
             return s
         text = re.sub(r'^(?! {4}|\t)\w+_\w+_\w[\w_]*', italic_callback, text)
         # In very clear cases, let newlines become <br /> tags.
         def newline_callback(matchobj):
             if len(matchobj.group(1)) == 1:
                 return matchobj.group(0).rstrip() + '  \n'
             else:
                 return matchobj.group(0)
         pattern = re.compile(r'^[\w\<][^\n]*(\n+)', re.MULTILINE)
         text = re.sub(pattern, newline_callback, text)
         # Insert pre block extractions.
         def pre_insert_callback(matchobj):
@@ @@ -194,59 +196,54 @@ class MarkupRenderer(object): @@
         <span class="nb">echo</span> <span class="s2">&quot;hello&quot;</span>
         </pre></div>
         </td></tr></table>
         """
         source = safe_str(source)
         try:
             if flavored:
                 source = cls._flavored_markdown(source)
             return markdown_mod.markdown(
                 source,
                 extensions=['markdown.extensions.codehilite', 'markdown.extensions.extra'],
                 extension_configs={'markdown.extensions.codehilite': {'css_class': 'code-highlight'}})
         except Exception:
             log.error(traceback.format_exc())
             if safe:
                 log.debug('Falling back to render in plain mode')
                 return cls.plain(source)
             else:
                 raise
     @classmethod
     def rst(cls, source, safe=True):
         source = safe_str(source)
         try:
             from docutils.core import publish_parts
             from docutils.parsers.rst import directives
             docutils_settings = dict([(alias, None) for alias in
                                 cls.RESTRUCTUREDTEXT_DISALLOWED_DIRECTIVES])
             docutils_settings.update({'input_encoding': 'unicode',
                                       'report_level': 4})
             for k, v in docutils_settings.items():
                 directives.register_directive(k, v)
             parts = publish_parts(source=source,
                                   writer_name="html4css1",
                                   settings_overrides=docutils_settings)
             return parts['html_title'] + parts["fragment"]
         except ImportError:
             log.warning('Install docutils to use this function')
             return cls.plain(source)
         except Exception:
             log.error(traceback.format_exc())
             if safe:
                 log.debug('Falling back to render in plain mode')
                 return cls.plain(source)
             else:
                 raise
     @classmethod
     def rst_with_mentions(cls, source):
         def wrapp(match_obj):
             uname = match_obj.groups()[0]
             return r'\ **@%(uname)s**\ ' % {'uname': uname}
         mention_hl = MENTIONS_REGEX.sub(wrapp, source).strip()
         return cls.rst(mention_hl)

kallithea/lib/pygmentsutils.py

➞

Show inline comments

@@ @@ -8,72 +8,73 @@ @@
 # 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.pygmentsutils
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Functions for extracting internal Pygments data.
 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.
 """
 from collections import defaultdict
 from pygments import lexers
 import kallithea
 def get_extension_descriptions():
     """
     Based on what's inside pygments lexers, return a mapping from lowercase
     extensions to lists of very brief descriptions.
     """
     ext_descs = defaultdict(list)
     for lx, t in sorted(lexers.LEXERS.items()):
         desc = lx.replace('Lexer', '')
         for glob in t[-2]:
             s = glob.lstrip('*').lstrip('.').lower()
             start = s.find('[')
             if start > -1 and s.endswith(']'):
                 # expand trailing [] range
                 prefix = s[:start]
                 for char in s[start + 1:-1]:
                     ext_descs[prefix + char].append(desc)
             else:
                 # use stripped glob as extension
                 ext_descs[s].append(desc)
     return dict(ext_descs)
 def get_index_filenames():
     """
     Get list of known indexable filenames from pygment lexer internals
     """
     filenames = []
     for lx, t in sorted(lexers.LEXERS.items()):
         for f in t[-2]:
             if '*' not in f and '[' not in f:
                 filenames.append(f)
     return filenames
 def get_custom_lexer(extension):
     """
     returns a custom lexer if it's defined in the extensions module, or None
     if there's no custom lexer defined
     """
     import kallithea
     lexer_name = getattr(kallithea.EXTENSIONS, 'EXTRA_LEXERS', {}).get(extension)
     if lexer_name is None:
         return None
     return lexers.get_lexer_by_name(lexer_name)

kallithea/lib/utils2.py

➞

Show inline comments

@@ @@ -15,54 +15,59 @@ @@
 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 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):
     """
     Converts a given line  "line end" according to given mode
@@ @@ -149,49 +154,48 @@ def remove_prefix(s, prefix): @@
 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
     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]
@@ @@ -311,50 +315,48 @@ def get_clone_url(clone_uri_tmpl, 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:
     """
     from kallithea.lib.vcs.backends.base import BaseRepository, EmptyChangeset
     from kallithea.lib.vcs.exceptions import RepositoryError
     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())
 def time_to_datetime(tm):
     if tm:
         if isinstance(tm, str):
             try:
                 tm = float(tm)
             except ValueError:
                 return
         return datetime.datetime.fromtimestamp(tm)
@@ @@ -374,50 +376,48 @@ def extract_mentioned_usernames(text): @@
     ['1-2.a_X', '1234', 'ddd', 'ee', 'ff', 'gg', 'gg', 'hh', 'zz']
     """
     return MENTIONS_REGEX.findall(text)
 def extract_mentioned_users(text):
     """ Returns set of actual database Users @mentioned in given text. """
     from kallithea.model.db import User
     result = set()
     for name in extract_mentioned_usernames(text):
         user = User.get_by_username(name, case_insensitive=True)
         if user is not None and not user.is_default_user:
             result.add(user)
     return result
 class AttributeDict(dict):
     def __getattr__(self, attr):
         return self.get(attr, None)
     __setattr__ = dict.__setitem__
     __delattr__ = dict.__delitem__
 def obfuscate_url_pw(engine):
     from sqlalchemy.engine import url as sa_url
     from sqlalchemy.exc import ArgumentError
     try:
         _url = sa_url.make_url(engine or '')
     except ArgumentError:
         return engine
     if _url.password:
         _url.password = 'XXXXX'
     return str(_url)
 class HookEnvironmentError(Exception): pass
 def get_hook_environment():
     """
     Get hook context by deserializing the global KALLITHEA_EXTRAS environment
     variable.
     Called early in Git out-of-process hooks to get .ini config path so the
     basic environment can be configured properly. Also used in all hooks to get
     information about the action that triggered it.
     """
     try:
         kallithea_extras = os.environ['KALLITHEA_EXTRAS']

kallithea/lib/vcs/backends/base.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 """
     vcs.backends.base
     ~~~~~~~~~~~~~~~~~
     Base for all available scm backends
     :created_on: Apr 8, 2010
     :copyright: (c) 2010-2011 by Marcin Kuzminski, Lukasz Balcerzak.
 """
 import datetime
 import itertools
 from kallithea.lib.vcs.backends import get_backend
 from kallithea.lib.vcs.conf import settings
 from kallithea.lib.vcs.exceptions import (ChangesetError, EmptyRepositoryError, NodeAlreadyAddedError, NodeAlreadyChangedError, NodeAlreadyExistsError,
                                           NodeAlreadyRemovedError, NodeDoesNotExistError, NodeNotChangedError, RepositoryError)
 from kallithea.lib.vcs.utils import author_email, author_name
 from kallithea.lib.vcs.utils.helpers import get_dict_for_attrs
 from kallithea.lib.vcs.utils.lazy import LazyProperty
 class BaseRepository(object):
     """
     Base Repository for final backends
     **Attributes**
         ``DEFAULT_BRANCH_NAME``
             name of default branch (i.e. "master" for git etc.
         ``scm``
             alias of scm, i.e. *git* or *hg*
         ``repo``
             object from external api
         ``revisions``
@@ @@ -986,54 +987,52 @@ class EmptyChangeset(BaseChangeset): @@
     """
     def __init__(self, cs='0' * 40, repo=None, requested_revision=None,
                  alias=None, revision=-1, message='', author='', date=None):
         self._empty_cs = cs
         self.revision = revision
         self.message = message
         self.author = author
         self.date = date or datetime.datetime.fromtimestamp(0)
         self.repository = repo
         self.requested_revision = requested_revision
         self.alias = alias
     @LazyProperty
     def raw_id(self):
         """
         Returns raw string identifying this changeset, useful for web
         representation.
         """
         return self._empty_cs
     @LazyProperty
     def branch(self):
         from kallithea.lib.vcs.backends import get_backend
         return get_backend(self.alias).DEFAULT_BRANCH_NAME
     @LazyProperty
     def branches(self):
         from kallithea.lib.vcs.backends import get_backend
         return [get_backend(self.alias).DEFAULT_BRANCH_NAME]
     @LazyProperty
     def short_id(self):
         return self.raw_id[:12]
     def get_file_changeset(self, path):
         return self
     def get_file_content(self, path):
         return b''
     def get_file_size(self, path):
         return 0
 class CollectionGenerator(object):
     def __init__(self, repo, revs):
         self.repo = repo
         self.revs = revs
     def __len__(self):
         return len(self.revs)

kallithea/lib/vcs/nodes.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 """
     vcs.nodes
     ~~~~~~~~~
     Module holding everything related to vcs nodes.
     :created_on: Apr 8, 2010
     :copyright: (c) 2010-2011 by Marcin Kuzminski, Lukasz Balcerzak.
 """
 import functools
 import mimetypes
 import posixpath
 import stat
 from pygments import lexers
 from kallithea.lib.vcs.backends.base import EmptyChangeset
 from kallithea.lib.vcs.exceptions import NodeError, RemovedFileNodeError
 from kallithea.lib.vcs.utils import safe_bytes, safe_str
 from kallithea.lib.vcs.utils.lazy import LazyProperty
 class NodeKind:
     SUBMODULE = -1
     DIR = 1
     FILE = 2
 class NodeState:
     ADDED = 'added'
     CHANGED = 'changed'
     NOT_CHANGED = 'not changed'
     REMOVED = 'removed'
 class NodeGeneratorBase(object):
     """
     Base class for removed added and changed filenodes, it's a lazy generator
     class that will create filenodes only on iteration or call
@@ @@ -284,79 +286,77 @@ class FileNode(Node): @@
     @LazyProperty
     def last_changeset(self):
         if self.changeset:
             return self.changeset.get_file_changeset(self.path)
         raise NodeError("Cannot retrieve last changeset of the file without "
             "related changeset attribute")
     def get_mimetype(self):
         """
         Mimetype is calculated based on the file's content.
         """
         mtype, encoding = mimetypes.guess_type(self.name)
         if mtype is None:
             if self.is_binary:
                 mtype = 'application/octet-stream'
                 encoding = None
             else:
                 mtype = 'text/plain'
                 encoding = None
                 # try with pygments
                 from pygments import lexers
                 try:
                     mt = lexers.get_lexer_for_filename(self.name).mimetypes
                 except lexers.ClassNotFound:
                     mt = None
                 if mt:
                     mtype = mt[0]
         return mtype, encoding
     @LazyProperty
     def mimetype(self):
         """
         Wrapper around full mimetype info. It returns only type of fetched
         mimetype without the encoding part. use get_mimetype function to fetch
         full set of (type,encoding)
         """
         return self.get_mimetype()[0]
     @LazyProperty
     def mimetype_main(self):
         return self.mimetype.split('/')[0]
     @LazyProperty
     def lexer(self):
         """
         Returns pygment's lexer class. Would try to guess lexer taking file's
         content, name and mimetype.
         """
         from pygments import lexers
         try:
             lexer = lexers.guess_lexer_for_filename(self.name, safe_str(self.content), stripnl=False)
         except lexers.ClassNotFound:
             lexer = lexers.TextLexer(stripnl=False)
         # returns first alias
         return lexer
     @LazyProperty
     def lexer_alias(self):
         """
         Returns first alias of the lexer guessed for this file.
         """
         return self.lexer.aliases[0]
     @LazyProperty
     def history(self):
         """
         Returns a list of changeset for this file in which the file was changed
         """
         if self.changeset is None:
             raise NodeError('Unable to get changeset for this FileNode')
         return self.changeset.get_file_history(self.path)
     @LazyProperty

kallithea/lib/vcs/utils/__init__.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 """
 This module provides some useful tools for ``vcs`` like annotate/diff html
 output. It also includes some internal helpers.
 """
 import datetime
 import re
 import time
 import chardet
 from kallithea.lib.vcs.conf import settings
 def makedate():
     lt = time.localtime()
     if lt[8] == 1 and time.daylight:
         tz = time.altzone
     else:
         tz = time.timezone
     return time.mktime(lt), tz
 def aslist(obj, sep=None, strip=True):
     """
     Returns given string separated by sep as list
     :param obj:
     :param sep:
     :param strip:
     """
     if isinstance(obj, str):
         lst = obj.split(sep)
         if strip:
             lst = [v.strip() for v in lst]
         return lst
     elif isinstance(obj, (list, tuple)):
@@ @@ -60,78 +64,75 @@ def safe_int(val, default=None): @@
     :param default:
     """
     try:
         val = int(val)
     except (ValueError, TypeError):
         val = default
     return val
 def safe_str(s):
     """
     Safe unicode str function. Use a few tricks to turn s into str:
     In case of UnicodeDecodeError with configured default encodings, try to
     detect encoding with chardet library, then fall back to first encoding with
     errors replaced.
     """
     if isinstance(s, str):
         return s
     if not isinstance(s, bytes):  # use __str__ and don't expect UnicodeDecodeError
         return str(s)
     from kallithea.lib.vcs.conf import settings
     for enc in settings.DEFAULT_ENCODINGS:
         try:
             return str(s, enc)
         except UnicodeDecodeError:
             pass
     try:
         import chardet
         encoding = chardet.detect(s)['encoding']
         if encoding is not None:
             return s.decode(encoding)
-    except (ImportError, UnicodeDecodeError):
     except UnicodeDecodeError:
         pass
     return str(s, settings.DEFAULT_ENCODINGS[0], 'replace')
 def safe_bytes(s):
     """
     Safe bytes function. Use a few tricks to turn s into bytes string:
     In case of UnicodeEncodeError with configured default encodings, fall back
     to first configured encoding with errors replaced.
     """
     if isinstance(s, bytes):
         return s
     assert isinstance(s, str), repr(s)  # bytes cannot coerse with __str__ or handle None or int
     from kallithea.lib.vcs.conf import settings
     for enc in settings.DEFAULT_ENCODINGS:
         try:
             return s.encode(enc)
         except UnicodeEncodeError:
             pass
     return s.encode(settings.DEFAULT_ENCODINGS[0], 'replace')
 def ascii_bytes(s):
     """
     Simple conversion from str to bytes, *assuming* all codepoints are
 -bit and it thus is pure ASCII.
     Will fail badly with UnicodeError on invalid input.
     This should be used where enocding and "safe" ambiguity should be avoided.
     Where strings already have been encoded in other ways but still are unicode
     string - for example to hex, base64, json, urlencoding, or are known to be
     identifiers.
     >>> ascii_bytes('a')
     b'a'
     >>> ascii_bytes(u'a')
     b'a'
     >>> ascii_bytes('å')

kallithea/lib/vcs/utils/helpers.py

➞

Show inline comments

 """
 Utilities aimed to help achieve mostly basic tasks.
 """
 import datetime
 import logging
 import os
 import re
 import time
 import urllib.request
 import mercurial.url
 from pygments import highlight
 from pygments.formatters import TerminalFormatter
 from pygments.lexers import ClassNotFound, guess_lexer_for_filename
 from kallithea.lib.vcs.exceptions import RepositoryError, VCSError
 from kallithea.lib.vcs.utils import safe_str
 from kallithea.lib.vcs.utils.paths import abspath
 ALIASES = ['hg', 'git']
 def get_scm(path, search_up=False, explicit_alias=None):
     """
     Returns one of alias from ``ALIASES`` (in order of precedence same as
     shortcuts given in ``ALIASES``) and top working dir path for the given
     argument. If no scm-specific directory is found or more than one scm is
     found at that directory, ``VCSError`` is raised.
     :param search_up: if set to ``True``, this function would try to
       move up to parent directory every time no scm is recognized for the
       currently checked path. Default: ``False``.
     :param explicit_alias: can be one of available backend aliases, when given
       it will return given explicit alias in repositories under more than one
       version control, if explicit_alias is different than found it will raise
       VCSError
     """
@@ @@ -84,58 +88,48 @@ def get_scms_for_path(path): @@
         dirname = os.path.join(path, 'rm__.' + key)
         if os.path.isdir(dirname):
             return result
         # We still need to check if it's not bare repository as
         # bare repos don't have working directories
         try:
             get_backend(key)(path)
             result.append(key)
             continue
         except RepositoryError:
             # Wrong backend
             pass
         except VCSError:
             # No backend at all
             pass
     return result
 def get_highlighted_code(name, code, type='terminal'):
     """
     If pygments are available on the system
     then returned output is colored. Otherwise
     unchanged content is returned.
     """
     import logging
     try:
         import pygments
         pygments
     except ImportError:
         return code
     from pygments import highlight
     from pygments.formatters import TerminalFormatter
     from pygments.lexers import ClassNotFound, guess_lexer_for_filename
     try:
         lexer = guess_lexer_for_filename(name, code)
         formatter = TerminalFormatter()
         content = highlight(code, lexer, formatter)
     except ClassNotFound:
         logging.debug("Couldn't guess Lexer, will not use pygments.")
         content = code
     return content
 def parse_changesets(text):
     """
     Returns dictionary with *start*, *main* and *end* ids.
     Examples::
         >>> parse_changesets('aaabbb')
         {'start': None, 'main': 'aaabbb', 'end': None}
         >>> parse_changesets('aaabbb..cccddd')
         {'start': 'aaabbb', 'end': 'cccddd', 'main': None}
     """
     text = text.strip()
     CID_RE = r'[a-zA-Z0-9]+'

0 comments (0 inline, 0 general)