Changeset - 1f92cded1bd2
[Not reviewed]
default
0 5 0
Mads Kiilerich (mads) - 5 years ago 2020-12-29 22:07:41
mads@kiilerich.com
celery: move Whoosh indexing task to whoosh library where it belongs

Avoid bundling everything from many different layers in one big task library.

This is more feasible now when we don't need kallithea.CELERY_APP set at import
time.
5 files changed with 18 insertions and 14 deletions:
0 comments (0 inline, 0 general)
kallithea/controllers/admin/settings.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/>.
 
"""
 
kallithea.controllers.admin.settings
 
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

	
 
settings controller for Kallithea admin
 

	
 
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: Jul 14, 2010
 
:author: marcink
 
:copyright: (c) 2013 RhodeCode GmbH, and others.
 
:license: GPLv3, see LICENSE.md for more details.
 
"""
 

	
 
import logging
 
import traceback
 

	
 
import formencode
 
from formencode import htmlfill
 
from tg import config, request
 
from tg import tmpl_context as c
 
from tg.i18n import ugettext as _
 
from webob.exc import HTTPFound
 

	
 
import kallithea
 
import kallithea.lib.indexers.daemon
 
from kallithea.controllers import base
 
from kallithea.lib import webutils
 
from kallithea.lib.auth import HasPermissionAnyDecorator, LoginRequired
 
from kallithea.lib.utils import repo2db_mapper, set_app_settings
 
from kallithea.lib.utils2 import safe_str
 
from kallithea.lib.vcs import VCSError
 
from kallithea.lib.webutils import url
 
from kallithea.model import async_tasks, db, meta
 
from kallithea.model.forms import ApplicationSettingsForm, ApplicationUiSettingsForm, ApplicationVisualisationForm
 
from kallithea.model.notification import EmailNotificationModel
 
from kallithea.model.scm import ScmModel
 

	
 

	
 
log = logging.getLogger(__name__)
 

	
 

	
 
class SettingsController(base.BaseController):
 

	
 
    @LoginRequired(allow_default_user=True)
 
    def _before(self, *args, **kwargs):
 
        super(SettingsController, self)._before(*args, **kwargs)
 

	
 
    def _get_hg_ui_settings(self):
 
        ret = db.Ui.query().all()
 

	
 
        settings = {}
 
        for each in ret:
 
            k = each.ui_section + '_' + each.ui_key
 
            v = each.ui_value
 
            if k == 'paths_/':
 
                k = 'paths_root_path'
 

	
 
            k = k.replace('.', '_')
 

	
 
            if each.ui_section in ['hooks', 'extensions']:
 
                v = each.ui_active
 

	
 
            settings[k] = v
 
        return settings
 

	
 
    @HasPermissionAnyDecorator('hg.admin')
 
    def settings_vcs(self):
 
        c.active = 'vcs'
 
        if request.POST:
 
            application_form = ApplicationUiSettingsForm()()
 
            try:
 
                form_result = application_form.to_python(dict(request.POST))
 
            except formencode.Invalid as errors:
 
                return htmlfill.render(
 
                     base.render('admin/settings/settings.html'),
 
                     defaults=errors.value,
 
                     errors=errors.error_dict or {},
 
                     prefix_error=False,
 
                     encoding="UTF-8",
 
                     force_defaults=False)
 

	
 
            try:
 
                if c.visual.allow_repo_location_change:
 
                    sett = db.Ui.get_by_key('paths', '/')
 
                    sett.ui_value = form_result['paths_root_path']
 

	
 
                # HOOKS
 
                sett = db.Ui.get_by_key('hooks', db.Ui.HOOK_UPDATE)
 
                sett.ui_active = form_result['hooks_changegroup_update']
 

	
 
                sett = db.Ui.get_by_key('hooks', db.Ui.HOOK_REPO_SIZE)
 
                sett.ui_active = form_result['hooks_changegroup_repo_size']
 

	
 
                ## EXTENSIONS
 
                sett = db.Ui.get_or_create('extensions', 'largefiles')
 
                sett.ui_active = form_result['extensions_largefiles']
 

	
 
#                sett = db.Ui.get_or_create('extensions', 'hggit')
 
#                sett.ui_active = form_result['extensions_hggit']
 

	
 
                meta.Session().commit()
 

	
 
                webutils.flash(_('Updated VCS settings'), category='success')
 

	
 
            except Exception:
 
                log.error(traceback.format_exc())
 
                webutils.flash(_('Error occurred while updating '
 
                          'application settings'), category='error')
 

	
 
        defaults = db.Setting.get_app_settings()
 
        defaults.update(self._get_hg_ui_settings())
 

	
 
        return htmlfill.render(
 
            base.render('admin/settings/settings.html'),
 
            defaults=defaults,
 
            encoding="UTF-8",
 
            force_defaults=False)
 

	
 
    @HasPermissionAnyDecorator('hg.admin')
 
    def settings_mapping(self):
 
        c.active = 'mapping'
 
@@ -285,126 +286,126 @@ class SettingsController(base.BaseContro
 
        if request.POST:
 
            test_email = request.POST.get('test_email')
 
            test_email_subj = 'Kallithea test email'
 
            test_body = ('Kallithea Email test, '
 
                               'Kallithea version: %s' % c.kallithea_version)
 
            if not test_email:
 
                webutils.flash(_('Please enter email address'), category='error')
 
                raise HTTPFound(location=url('admin_settings_email'))
 

	
 
            test_email_txt_body = EmailNotificationModel() \
 
                .get_email_tmpl(EmailNotificationModel.TYPE_DEFAULT,
 
                                'txt', body=test_body)
 
            test_email_html_body = EmailNotificationModel() \
 
                .get_email_tmpl(EmailNotificationModel.TYPE_DEFAULT,
 
                                'html', body=test_body)
 

	
 
            recipients = [test_email] if test_email else None
 

	
 
            async_tasks.send_email(recipients, test_email_subj,
 
                             test_email_txt_body, test_email_html_body)
 

	
 
            webutils.flash(_('Send email task created'), category='success')
 
            raise HTTPFound(location=url('admin_settings_email'))
 

	
 
        defaults = db.Setting.get_app_settings()
 
        defaults.update(self._get_hg_ui_settings())
 

	
 
        c.ini = kallithea.CONFIG
 

	
 
        return htmlfill.render(
 
            base.render('admin/settings/settings.html'),
 
            defaults=defaults,
 
            encoding="UTF-8",
 
            force_defaults=False)
 

	
 
    @HasPermissionAnyDecorator('hg.admin')
 
    def settings_hooks(self):
 
        c.active = 'hooks'
 
        if request.POST:
 
            if c.visual.allow_custom_hooks_settings:
 
                ui_key = request.POST.get('new_hook_ui_key')
 
                ui_value = request.POST.get('new_hook_ui_value')
 

	
 
                hook_id = request.POST.get('hook_id')
 

	
 
                try:
 
                    ui_key = ui_key and ui_key.strip()
 
                    if ui_key in (x.ui_key for x in db.Ui.get_custom_hooks()):
 
                        webutils.flash(_('Hook already exists'), category='error')
 
                    elif ui_key in (x.ui_key for x in db.Ui.get_builtin_hooks()):
 
                        webutils.flash(_('Builtin hooks are read-only. Please use another hook name.'), category='error')
 
                    elif ui_value and ui_key:
 
                        db.Ui.create_or_update_hook(ui_key, ui_value)
 
                        webutils.flash(_('Added new hook'), category='success')
 
                    elif hook_id:
 
                        db.Ui.delete(hook_id)
 
                        meta.Session().commit()
 

	
 
                    # check for edits
 
                    update = False
 
                    _d = request.POST.dict_of_lists()
 
                    for k, v, ov in zip(_d.get('hook_ui_key', []),
 
                                        _d.get('hook_ui_value_new', []),
 
                                        _d.get('hook_ui_value', [])):
 
                        if v != ov:
 
                            db.Ui.create_or_update_hook(k, v)
 
                            update = True
 

	
 
                    if update:
 
                        webutils.flash(_('Updated hooks'), category='success')
 
                    meta.Session().commit()
 
                except Exception:
 
                    log.error(traceback.format_exc())
 
                    webutils.flash(_('Error occurred during hook creation'),
 
                            category='error')
 

	
 
                raise HTTPFound(location=url('admin_settings_hooks'))
 

	
 
        defaults = db.Setting.get_app_settings()
 
        defaults.update(self._get_hg_ui_settings())
 

	
 
        c.hooks = db.Ui.get_builtin_hooks()
 
        c.custom_hooks = db.Ui.get_custom_hooks()
 

	
 
        return htmlfill.render(
 
            base.render('admin/settings/settings.html'),
 
            defaults=defaults,
 
            encoding="UTF-8",
 
            force_defaults=False)
 

	
 
    @HasPermissionAnyDecorator('hg.admin')
 
    def settings_search(self):
 
        c.active = 'search'
 
        if request.POST:
 
            repo_location = self._get_hg_ui_settings()['paths_root_path']
 
            full_index = request.POST.get('full_index', False)
 
            async_tasks.whoosh_index(repo_location, full_index)
 
            kallithea.lib.indexers.daemon.whoosh_index(repo_location, full_index)
 
            webutils.flash(_('Whoosh reindex task scheduled'), category='success')
 
            raise HTTPFound(location=url('admin_settings_search'))
 

	
 
        defaults = db.Setting.get_app_settings()
 
        defaults.update(self._get_hg_ui_settings())
 

	
 
        return htmlfill.render(
 
            base.render('admin/settings/settings.html'),
 
            defaults=defaults,
 
            encoding="UTF-8",
 
            force_defaults=False)
 

	
 
    @HasPermissionAnyDecorator('hg.admin')
 
    def settings_system(self):
 
        c.active = 'system'
 

	
 
        defaults = db.Setting.get_app_settings()
 
        defaults.update(self._get_hg_ui_settings())
 

	
 
        c.ini = kallithea.CONFIG
 
        server_info = db.Setting.get_server_info()
 
        for key, val in server_info.items():
 
            setattr(c, key, val)
 

	
 
        return htmlfill.render(
 
            base.render('admin/settings/settings.html'),
 
            defaults=defaults,
 
            encoding="UTF-8",
 
            force_defaults=False)
kallithea/lib/celery_app.py
Show inline comments
 
# -*- coding: utf-8 -*-
 

	
 
"""
 
Kallithea wrapper of Celery
 

	
 
The Celery configuration is in the Kallithea ini file but must be converted to an
 
entirely different format before Celery can use it.
 

	
 
We read the configuration from tg.config at module import time. This module can
 
thus not be imported in global scope but must be imported on demand in function
 
scope after tg.config has been initialized.
 

	
 
To make sure that the config really has been initialized, we check one of the
 
mandatory settings.
 
"""
 

	
 
import logging
 

	
 

	
 
class CeleryConfig(object):
 
    imports = ['kallithea.model.async_tasks']
 
    imports = [
 
        'kallithea.lib.indexers.daemon',
 
        'kallithea.model.async_tasks',
 
    ]
 
    task_always_eager = False
 

	
 
list_config_names = {'imports', 'accept_content'}
 

	
 

	
 
desupported = set([
 
    'broker.url',
 
    'celery.accept.content',
 
    'celery.always.eager',
 
    'celery.amqp.task.result.expires',
 
    'celeryd.concurrency',
 
    'celeryd.max.tasks.per.child',
 
    'celery.result.backend',  # Note: the .ini template used this instead of 'celery.result_backend' in 0.6
 
    'celery.result.dburi',
 
    'celery.result.serialier',
 
    'celery.result.serializer',
 
    'celery.send.task.error.emails',
 
    'celery.task_always_eager',  # still a valid configuration in celery, but not supported in Kallithea
 
    'celery.task.serializer',
 
])
 

	
 

	
 
log = logging.getLogger(__name__)
 

	
 

	
 
def make_celery_config(config):
 
    """Return Celery config object populated from relevant settings in a config dict, such as tg.config"""
 

	
 
    celery_config = CeleryConfig()
 

	
 
    for config_key, config_value in sorted(config.items()):
 
        if config_key in desupported and config_value:
 
            log.error('Celery configuration setting %r is no longer supported', config_key)
 
        parts = config_key.split('.', 1)
 
        if parts[0] == 'celery' and len(parts) == 2:  # Celery 4 config key
 
            celery_key = parts[1]
 
        else:
 
            continue
 
        if not isinstance(config_value, str):
 
            continue
 
        if celery_key in list_config_names:
 
            celery_value = config_value.split()
 
        elif config_value.isdigit():
 
            celery_value = int(config_value)
 
        elif config_value.lower() in ['true', 'false']:
 
            celery_value = config_value.lower() == 'true'
 
        else:
 
            celery_value = config_value
 
        setattr(celery_config, celery_key, celery_value)
 
    return celery_config
kallithea/lib/indexers/daemon.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/>.
 
"""
 
kallithea.lib.indexers.daemon
 
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

	
 
A daemon will read from task table and run tasks
 

	
 
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 26, 2010
 
:author: marcink
 
:copyright: (c) 2013 RhodeCode GmbH, and others.
 
:license: GPLv3, see LICENSE.md for more details.
 
"""
 

	
 

	
 
import logging
 
import os
 
import traceback
 
from shutil import rmtree
 
from time import mktime
 

	
 
from tg import config
 
from whoosh.index import create_in, exists_in, open_dir
 
from whoosh.qparser import QueryParser
 

	
 
from kallithea.lib import celerylib
 
from kallithea.lib.conf import INDEX_EXTENSIONS, INDEX_FILENAMES
 
from kallithea.lib.indexers import CHGSET_IDX_NAME, CHGSETS_SCHEMA, IDX_NAME, SCHEMA
 
from kallithea.lib.utils2 import safe_str
 
from kallithea.lib.vcs.exceptions import ChangesetDoesNotExistError, ChangesetError, NodeDoesNotExistError, RepositoryError
 
from kallithea.model import db
 
from kallithea.model.scm import ScmModel
 

	
 

	
 
log = logging.getLogger('whoosh_indexer')
 

	
 

	
 
class WhooshIndexingDaemon(object):
 
    """
 
    Daemon for atomic indexing jobs
 
    """
 

	
 
    def __init__(self, indexname=IDX_NAME, index_location=None,
 
                 repo_location=None, repo_list=None,
 
                 repo_update_list=None):
 
        self.indexname = indexname
 

	
 
        self.index_location = index_location
 
        if not index_location:
 
            raise Exception('You have to provide index location')
 

	
 
        self.repo_location = repo_location
 
        if not repo_location:
 
            raise Exception('You have to provide repositories location')
 

	
 
        self.repo_paths = ScmModel().repo_scan(self.repo_location)
 

	
 
        # filter repo list
 
        if repo_list:
 
            repo_list = set(repo_list)
 
            self.filtered_repo_paths = {}
 
            for repo_name, repo in self.repo_paths.items():
 
                if repo_name in repo_list:
 
                    self.filtered_repo_paths[repo_name] = repo
 

	
 
            self.repo_paths = self.filtered_repo_paths
 

	
 
        # filter update repo list
 
        self.filtered_repo_update_paths = {}
 
        if repo_update_list:
 
            self.filtered_repo_update_paths = {}
 
            for repo_name, repo in self.repo_paths.items():
 
                if repo_name in repo_update_list:
 
                    self.filtered_repo_update_paths[repo_name] = repo
 
            self.repo_paths = self.filtered_repo_update_paths
 

	
 
        self.initial = True
 
        if not os.path.isdir(self.index_location):
 
            os.makedirs(self.index_location)
 
            log.info('Cannot run incremental index since it does not '
 
                     'yet exist - running full build')
 
        elif not exists_in(self.index_location, IDX_NAME):
 
            log.info('Running full index build, as the file content '
 
                     'index does not exist')
 
        elif not exists_in(self.index_location, CHGSET_IDX_NAME):
 
            log.info('Running full index build, as the changeset '
 
                     'index does not exist')
 
        else:
 
            self.initial = False
 

	
 
    def _get_index_revision(self, repo):
 
        db_repo = db.Repository.get_by_repo_name(repo.name)
 
        landing_rev = 'tip'
 
        if db_repo:
 
            _rev_type, _rev = db_repo.landing_rev
 
            landing_rev = _rev
 
        return landing_rev
 

	
 
    def _get_index_changeset(self, repo, index_rev=None):
 
        if not index_rev:
 
            index_rev = self._get_index_revision(repo)
 
        cs = repo.get_changeset(index_rev)
 
        return cs
 

	
 
    def get_paths(self, repo):
 
        """
 
        recursive walk in root dir and return a set of all path in that dir
 
        based on repository walk function
 
        """
 
        index_paths_ = set()
 
        try:
 
            cs = self._get_index_changeset(repo)
 
            for _topnode, _dirs, files in cs.walk('/'):
 
                for f in files:
 
                    index_paths_.add(os.path.join(repo.path, f.path))
 

	
 
        except RepositoryError:
 
            log.debug(traceback.format_exc())
 
            pass
 
        return index_paths_
 

	
 
    def get_node(self, repo, path, index_rev=None):
 
@@ -351,96 +353,105 @@ class WhooshIndexingDaemon(object):
 
                        mtime = self.get_node_mtime(node)
 
                        if mtime > indexed_time:
 
                            # The file has changed, delete it and add it to
 
                            # the list of files to reindex
 
                            log.debug(
 
                                'adding to reindex list %s mtime: %s vs %s',
 
                                    indexed_path, mtime, indexed_time
 
                            )
 
                            writer.delete_by_term('fileid', indexed_path)
 
                            writer_is_dirty = True
 

	
 
                            to_index.add(indexed_path)
 
                    except (ChangesetError, NodeDoesNotExistError):
 
                        # This file was deleted since it was indexed
 
                        log.debug('removing from index %s', indexed_path)
 
                        writer.delete_by_term('path', indexed_path)
 
                        writer_is_dirty = True
 

	
 
            # Loop over the files in the filesystem
 
            # Assume we have a function that gathers the filenames of the
 
            # documents to be indexed
 
            ri_cnt_total = 0  # indexed
 
            riwc_cnt_total = 0  # indexed with content
 
            for repo_name, repo in sorted(self.repo_paths.items()):
 
                log.debug('Updating file index for repo %s', repo_name)
 
                # skip indexing if there aren't any revisions
 
                if len(repo) < 1:
 
                    continue
 
                ri_cnt = 0   # indexed
 
                riwc_cnt = 0  # indexed with content
 
                for path in self.get_paths(repo):
 
                    if path in to_index or path not in indexed_paths:
 
                        # This is either a file that's changed, or a new file
 
                        # that wasn't indexed before. So index it!
 
                        i, iwc = self.add_doc(writer, path, repo, repo_name)
 
                        writer_is_dirty = True
 
                        ri_cnt += i
 
                        ri_cnt_total += 1
 
                        riwc_cnt += iwc
 
                        riwc_cnt_total += iwc
 
                log.debug('added %s files %s with content for repo %s',
 
                             ri_cnt + riwc_cnt, riwc_cnt, repo.path
 
                )
 
            log.debug('indexed %s files in total and %s with content',
 
                        ri_cnt_total, riwc_cnt_total
 
            )
 
        finally:
 
            if writer_is_dirty:
 
                log.debug('>> COMMITING CHANGES TO FILE INDEX <<')
 
                writer.commit(merge=True)
 
                log.debug('>>> FINISHED REBUILDING FILE INDEX <<<')
 
            else:
 
                log.debug('>> NOTHING TO COMMIT TO FILE INDEX <<')
 
                writer.cancel()
 

	
 
    def build_indexes(self):
 
        if os.path.exists(self.index_location):
 
            log.debug('removing previous index')
 
            rmtree(self.index_location)
 

	
 
        if not os.path.exists(self.index_location):
 
            os.mkdir(self.index_location)
 

	
 
        chgset_idx = create_in(self.index_location, CHGSETS_SCHEMA,
 
                               indexname=CHGSET_IDX_NAME)
 
        chgset_idx_writer = chgset_idx.writer()
 

	
 
        file_idx = create_in(self.index_location, SCHEMA, indexname=IDX_NAME)
 
        file_idx_writer = file_idx.writer()
 
        log.debug('BUILDING INDEX FOR EXTENSIONS %s '
 
                  'AND REPOS %s', INDEX_EXTENSIONS, ' and '.join(self.repo_paths))
 

	
 
        for repo_name, repo in sorted(self.repo_paths.items()):
 
            log.debug('Updating indices for repo %s', repo_name)
 
            # skip indexing if there aren't any revisions
 
            if len(repo) < 1:
 
                continue
 

	
 
            self.index_files(file_idx_writer, repo_name, repo)
 
            self.index_changesets(chgset_idx_writer, repo_name, repo)
 

	
 
        log.debug('>> COMMITING CHANGES <<')
 
        file_idx_writer.commit(merge=True)
 
        chgset_idx_writer.commit(merge=True)
 
        log.debug('>>> FINISHED BUILDING INDEX <<<')
 

	
 
    def update_indexes(self):
 
        self.update_file_index()
 
        self.update_changeset_index()
 

	
 
    def run(self, full_index=False):
 
        """Run daemon"""
 
        if full_index or self.initial:
 
            self.build_indexes()
 
        else:
 
            self.update_indexes()
 

	
 

	
 
@celerylib.task
 
@celerylib.locked_task
 
def whoosh_index(repo_location, full_index):
 
    index_location = config['index_dir']
 
    WhooshIndexingDaemon(index_location=index_location,
 
                         repo_location=repo_location) \
 
                         .run(full_index=full_index)
kallithea/model/async_tasks.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/>.
 
"""
 
kallithea.model.async_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.message
 
import email.utils
 
import os
 
import smtplib
 
import time
 
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.lib import celerylib, conf, ext_json, hooks
 
from kallithea.lib.indexers.daemon import WhooshIndexingDaemon
 
from kallithea.lib.utils2 import asbool, ascii_bytes
 
from kallithea.lib.vcs.utils import author_email, author_name
 
from kallithea.model import db, meta, repo, userlog
 

	
 

	
 
__all__ = ['whoosh_index', 'get_commits_stats', 'send_email']
 
__all__ = ['get_commits_stats', 'send_email']
 

	
 

	
 
log = celery.utils.log.get_task_logger(__name__)
 

	
 

	
 
@celerylib.task
 
@celerylib.locked_task
 
def whoosh_index(repo_location, full_index):
 
    index_location = config['index_dir']
 
    WhooshIndexingDaemon(index_location=index_location,
 
                         repo_location=repo_location) \
 
                         .run(full_index=full_index)
 

	
 

	
 
def _author_username(author):
 
    """Return the username of the user identified by the email part of the 'author' string,
 
    default to the name or email.
 
    Kind of similar to h.person() ."""
 
    email = author_email(author)
 
    if email:
 
        user = db.User.get_by_email(email)
 
        if user is not None:
 
            return user.username
 
    # Still nothing?  Just pass back the author name if any, else the email
 
    return author_name(author) or email
 

	
 

	
 
@celerylib.task
 
def get_commits_stats(repo_name, ts_min_y, ts_max_y, recurse_limit=100):
 
    lockkey = celerylib.__get_lockkey('get_commits_stats', repo_name, ts_min_y,
 
                            ts_max_y)
 
    log.info('running task with lockkey %s', lockkey)
 
    try:
 
        lock = celerylib.DaemonLock(os.path.join(config['cache_dir'], lockkey))
 

	
 
        co_day_auth_aggr = {}
 
        commits_by_day_aggregate = {}
 
        db_repo = db.Repository.get_by_repo_name(repo_name)
 
        if db_repo is None:
 
            return
 

	
 
        scm_repo = db_repo.scm_instance
 
        repo_size = scm_repo.count()
 
        # return if repo have no revisions
 
        if repo_size < 1:
 
            lock.release()
 
            return
 

	
 
        skip_date_limit = True
 
        parse_limit = int(config.get('commit_parse_limit'))
 
        last_rev = None
 
        last_cs = None
 
        timegetter = itemgetter('time')
 

	
 
        dbrepo = db.Repository.query() \
 
            .filter(db.Repository.repo_name == repo_name).scalar()
 
        cur_stats = db.Statistics.query() \
 
            .filter(db.Statistics.repository == dbrepo).scalar()
 

	
 
        if cur_stats is not None:
 
            last_rev = cur_stats.stat_on_revision
 

	
 
        if last_rev == scm_repo.get_changeset().revision and repo_size > 1:
 
            # pass silently without any work if we're not on first revision or
 
            # current state of parsing revision(from db marker) is the
 
            # last revision
 
            lock.release()
 
            return
 

	
 
        if cur_stats:
 
            commits_by_day_aggregate = OrderedDict(ext_json.loads(
 
                                        cur_stats.commit_activity_combined))
 
            co_day_auth_aggr = ext_json.loads(cur_stats.commit_activity)
 

	
 
        log.debug('starting parsing %s', parse_limit)
 

	
 
        last_rev = last_rev + 1 if last_rev and last_rev >= 0 else 0
 
        log.debug('Getting revisions from %s to %s',
 
             last_rev, last_rev + parse_limit
 
        )
 
        usernames_cache = {}
 
        for cs in scm_repo[last_rev:last_rev + parse_limit]:
 
            log.debug('parsing %s', cs)
 
            last_cs = cs  # remember last parsed changeset
 
            tt = cs.date.timetuple()
 
            k = mktime(tt[:3] + (0, 0, 0, 0, 0, 0))
 

	
 
            # get username from author - similar to what h.person does
 
            username = usernames_cache.get(cs.author)
 
            if username is None:
 
                username = _author_username(cs.author)
 
                usernames_cache[cs.author] = username
 

	
 
            if username in co_day_auth_aggr:
 
                try:
 
                    l = [timegetter(x) for x in
 
                         co_day_auth_aggr[username]['data']]
 
                    time_pos = l.index(k)
 
                except ValueError:
 
                    time_pos = None
 

	
 
                if time_pos is not None and time_pos >= 0:
 
                    datadict = \
 
                        co_day_auth_aggr[username]['data'][time_pos]
 

	
 
                    datadict["commits"] += 1
 
                    datadict["added"] += len(cs.added)
 
                    datadict["changed"] += len(cs.changed)
 
                    datadict["removed"] += len(cs.removed)
 

	
scripts/deps.py
Show inline comments
 
@@ -66,193 +66,192 @@ pkg_resources
 
platform
 
posixpath
 
pprint
 
pwd
 
pyflakes
 
pytest
 
pytest_localserver
 
random
 
re
 
routes
 
setuptools
 
shlex
 
shutil
 
smtplib
 
socket
 
ssl
 
stat
 
string
 
struct
 
subprocess
 
sys
 
tarfile
 
tempfile
 
textwrap
 
tgext
 
threading
 
time
 
traceback
 
traitlets
 
types
 
urllib
 
urlobject
 
uuid
 
warnings
 
webhelpers2
 
webob
 
webtest
 
whoosh
 
win32traceutil
 
zipfile
 
'''.split())
 

	
 
top_modules = set('''
 
kallithea.alembic
 
kallithea.bin
 
kallithea.config
 
kallithea.controllers
 
kallithea.templates.py
 
scripts
 
'''.split())
 

	
 
bottom_external_modules = set('''
 
tg
 
mercurial
 
sqlalchemy
 
alembic
 
formencode
 
pygments
 
dulwich
 
beaker
 
psycopg2
 
docs
 
setup
 
conftest
 
'''.split())
 

	
 
normal_modules = set('''
 
kallithea
 
kallithea.controllers.base
 
kallithea.lib
 
kallithea.lib.auth
 
kallithea.lib.auth_modules
 
kallithea.lib.celerylib
 
kallithea.lib.db_manage
 
kallithea.lib.helpers
 
kallithea.lib.hooks
 
kallithea.lib.indexers
 
kallithea.lib.utils
 
kallithea.lib.utils2
 
kallithea.lib.vcs
 
kallithea.lib.webutils
 
kallithea.model
 
kallithea.model.async_tasks
 
kallithea.model.scm
 
kallithea.templates.py
 
'''.split())
 

	
 
shown_modules = normal_modules | top_modules
 

	
 
# break the chains somehow - this is a cleanup TODO list
 
known_violations = set([
 
('kallithea.lib.auth_modules', 'kallithea.lib.auth'),  # needs base&facade
 
('kallithea.lib.utils', 'kallithea.model'),  # clean up utils
 
('kallithea.lib.utils', 'kallithea.model.db'),
 
('kallithea.lib.utils', 'kallithea.model.scm'),
 
('kallithea.model.async_tasks', 'kallithea.lib.hooks'),
 
('kallithea.model.async_tasks', 'kallithea.lib.indexers'),
 
('kallithea.model.async_tasks', 'kallithea.model'),
 
('kallithea.model', 'kallithea.lib.auth'),  # auth.HasXXX
 
('kallithea.model', 'kallithea.lib.auth_modules'),  # validators
 
('kallithea.model', 'kallithea.lib.hooks'),  # clean up hooks
 
('kallithea.model', 'kallithea.model.scm'),
 
('kallithea.model.scm', 'kallithea.lib.hooks'),
 
])
 

	
 
extra_edges = [
 
('kallithea.config', 'kallithea.controllers'),  # through TG
 
('kallithea.lib.auth', 'kallithea.lib.auth_modules'),  # custom loader
 
]
 

	
 

	
 
def normalize(s):
 
    """Given a string with dot path, return the string it should be shown as."""
 
    parts = s.replace('.__init__', '').split('.')
 
    short_2 = '.'.join(parts[:2])
 
    short_3 = '.'.join(parts[:3])
 
    short_4 = '.'.join(parts[:4])
 
    if parts[0] in ['scripts', 'contributor_data', 'i18n_utils']:
 
        return 'scripts'
 
    if short_3 == 'kallithea.model.meta':
 
        return 'kallithea.model.db'
 
    if parts[:4] == ['kallithea', 'lib', 'vcs', 'ssh']:
 
        return 'kallithea.lib.vcs.ssh'
 
    if short_4 in shown_modules:
 
        return short_4
 
    if short_3 in shown_modules:
 
        return short_3
 
    if short_2 in shown_modules:
 
        return short_2
 
    if short_2 == 'kallithea.tests':
 
        return None
 
    if parts[0] in ignored_modules:
 
        return None
 
    assert parts[0] in bottom_external_modules, parts
 
    return parts[0]
 

	
 

	
 
def main(filenames):
 
    if not filenames or filenames[0].startswith('-'):
 
        print('''\
 
Usage:
 
    hg files 'set:!binary()&grep("^#!.*python")' 'set:**.py' | xargs scripts/deps.py
 
    dot -Tsvg deps.dot > deps.svg
 
        ''')
 
        raise SystemExit(1)
 

	
 
    files_imports = dict()  # map filenames to its imports
 
    import_deps = set()  # set of tuples with module name and its imports
 
    for fn in filenames:
 
        with open(fn) as f:
 
            s = f.read()
 

	
 
        dot_name = (fn[:-3] if fn.endswith('.py') else fn).replace('/', '.')
 
        file_imports = set()
 
        for m in re.finditer(r'^ *(?:from ([^ ]*) import (?:([a-zA-Z].*)|\(([^)]*)\))|import (.*))$', s, re.MULTILINE):
 
            m_from, m_from_import, m_from_import2, m_import = m.groups()
 
            if m_from:
 
                pre = m_from + '.'
 
                if pre.startswith('.'):
 
                    pre = dot_name.rsplit('.', 1)[0] + pre
 
                importlist = m_from_import or m_from_import2
 
            else:
 
                pre = ''
 
                importlist = m_import
 
            for imp in importlist.split('#', 1)[0].split(','):
 
                full_imp = pre + imp.strip().split(' as ', 1)[0]
 
                file_imports.add(full_imp)
 
                import_deps.add((dot_name, full_imp))
 
        files_imports[fn] = file_imports
 

	
 
    # dump out all deps for debugging and analysis
 
    with open('deps.txt', 'w') as f:
 
        for fn, file_imports in sorted(files_imports.items()):
 
            for file_import in sorted(file_imports):
 
                if file_import.split('.', 1)[0] in ignored_modules:
 
                    continue
 
                f.write('%s: %s\n' % (fn, file_import))
 

	
 
    # find leafs that haven't been ignored - they are the important external dependencies and shown in the bottom row
 
    only_imported = set(
 
        set(normalize(b) for a, b in import_deps) -
 
        set(normalize(a) for a, b in import_deps) -
 
        set([None, 'kallithea'])
 
    )
 

	
 
    normalized_dep_edges = set()
 
    for dot_name, full_imp in import_deps:
 
        a = normalize(dot_name)
 
        b = normalize(full_imp)
 
        if a is None or b is None or a == b:
 
            continue
 
        normalized_dep_edges.add((a, b))
 
        #print((dot_name, full_imp, a, b))
0 comments (0 inline, 0 general)