tasks.py
495 lines
| 16.8 KiB
| text/x-python
|
PythonLexer
r903 | # -*- coding: utf-8 -*- | |||
r1206 | # 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. | ||||
r1203 | # | |||
r903 | # 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. | ||||
r1203 | # | |||
r903 | # You should have received a copy of the GNU General Public License | |||
r1206 | # along with this program. If not, see <http://www.gnu.org/licenses/>. | |||
|
r4116 | """ | ||
rhodecode.lib.celerylib.tasks | ||||
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ | ||||
RhodeCode task modules, containing all task that suppose to be run | ||||
by celery daemon | ||||
:created_on: Oct 6, 2010 | ||||
:author: marcink | ||||
:copyright: (c) 2013 RhodeCode GmbH. | ||||
:license: GPLv3, see LICENSE for more details. | ||||
""" | ||||
r547 | from celery.decorators import task | |||
r555 | ||||
r692 | import os | |||
import traceback | ||||
r1002 | import logging | |||
r1717 | from os.path import join as jn | |||
r1002 | ||||
r692 | from time import mktime | |||
r547 | from operator import itemgetter | |||
r1244 | from string import lower | |||
r776 | ||||
r1417 | from pylons import config, url | |||
r547 | from pylons.i18n.translation import _ | |||
r776 | ||||
r2007 | from rhodecode.lib.vcs import get_backend | |||
r1717 | ||||
r2299 | from rhodecode import CELERY_ON, CELERY_EAGER | |||
r2109 | from rhodecode.lib.utils2 import safe_str | |||
r1929 | from rhodecode.lib.celerylib import run_task, locked_task, dbsession, \ | |||
str2bool, __get_lockkey, LockHeld, DaemonLock, get_session | ||||
r547 | from rhodecode.lib.helpers import person | |||
r1717 | from rhodecode.lib.rcmail.smtp_mailer import SmtpMailer | |||
r1722 | from rhodecode.lib.utils import add_cache, action_logger | |||
r1514 | from rhodecode.lib.compat import json, OrderedDict | |||
r2185 | from rhodecode.lib.hooks import log_create_repository | |||
r1514 | ||||
r1722 | from rhodecode.model.db import Statistics, Repository, User | |||
r2991 | from rhodecode.model.scm import ScmModel | |||
r776 | ||||
r692 | ||||
|
r4116 | add_cache(config) # pragma: no cover | ||
r777 | ||||
r547 | __all__ = ['whoosh_index', 'get_commits_stats', | |||
'reset_user_password', 'send_email'] | ||||
r1722 | ||||
r1717 | def get_logger(cls): | |||
if CELERY_ON: | ||||
try: | ||||
log = cls.get_logger() | ||||
|
r4116 | except Exception: | ||
r1717 | log = logging.getLogger(__name__) | |||
else: | ||||
log = logging.getLogger(__name__) | ||||
return log | ||||
r1244 | ||||
r1807 | ||||
r1002 | @task(ignore_result=True) | |||
r547 | @locked_task | |||
r1929 | @dbsession | |||
r547 | def whoosh_index(repo_location, full_index): | |||
r1722 | from rhodecode.lib.indexers.daemon import WhooshIndexingDaemon | |||
r2344 | log = get_logger(whoosh_index) | |||
r1929 | DBS = get_session() | |||
r1722 | ||||
r776 | index_location = config['index_dir'] | |||
r662 | WhooshIndexingDaemon(index_location=index_location, | |||
r1929 | repo_location=repo_location, sa=DBS)\ | |||
r777 | .run(full_index=full_index) | |||
r547 | ||||
r1244 | ||||
r1002 | @task(ignore_result=True) | |||
r1929 | @dbsession | |||
r3276 | def get_commits_stats(repo_name, ts_min_y, ts_max_y, recurse_limit=100): | |||
r1717 | log = get_logger(get_commits_stats) | |||
r1929 | DBS = get_session() | |||
r1264 | lockkey = __get_lockkey('get_commits_stats', repo_name, ts_min_y, | |||
ts_max_y) | ||||
r3055 | lockkey_path = config['app_conf']['cache_dir'] | |||
r1540 | ||||
r1976 | log.info('running task with lockkey %s' % lockkey) | |||
r1807 | ||||
r1264 | try: | |||
r1540 | lock = l = DaemonLock(file_=jn(lockkey_path, lockkey)) | |||
r699 | ||||
r1976 | # for js data compatibility cleans the key for person from ' | |||
r1264 | akc = lambda k: person(k).replace('"', "") | |||
r629 | ||||
r1264 | co_day_auth_aggr = {} | |||
commits_by_day_aggregate = {} | ||||
r1749 | repo = Repository.get_by_repo_name(repo_name) | |||
if repo is None: | ||||
return True | ||||
r1818 | ||||
r1749 | repo = repo.scm_instance | |||
r1807 | repo_size = repo.count() | |||
# return if repo have no revisions | ||||
r1264 | if repo_size < 1: | |||
lock.release() | ||||
return True | ||||
r547 | ||||
r1264 | skip_date_limit = True | |||
parse_limit = int(config['app_conf'].get('commit_parse_limit')) | ||||
r1807 | last_rev = None | |||
r1264 | last_cs = None | |||
timegetter = itemgetter('time') | ||||
r629 | ||||
r1929 | dbrepo = DBS.query(Repository)\ | |||
r1264 | .filter(Repository.repo_name == repo_name).scalar() | |||
r1929 | cur_stats = DBS.query(Statistics)\ | |||
r1264 | .filter(Statistics.repository == dbrepo).scalar() | |||
r1105 | ||||
r1264 | if cur_stats is not None: | |||
last_rev = cur_stats.stat_on_revision | ||||
r1105 | ||||
r1264 | if last_rev == repo.get_changeset().revision and repo_size > 1: | |||
r1722 | # 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 | ||||
r1264 | lock.release() | |||
return True | ||||
r629 | ||||
r1264 | if cur_stats: | |||
commits_by_day_aggregate = OrderedDict(json.loads( | ||||
cur_stats.commit_activity_combined)) | ||||
co_day_auth_aggr = json.loads(cur_stats.commit_activity) | ||||
r629 | ||||
r1976 | log.debug('starting parsing %s' % parse_limit) | |||
r1264 | lmktime = mktime | |||
r1807 | last_rev = last_rev + 1 if last_rev >= 0 else 0 | |||
log.debug('Getting revisions from %s to %s' % ( | ||||
last_rev, last_rev + parse_limit) | ||||
) | ||||
r1264 | for cs in repo[last_rev:last_rev + parse_limit]: | |||
r2109 | log.debug('parsing %s' % cs) | |||
r1264 | last_cs = cs # remember last parsed changeset | |||
k = lmktime([cs.date.timetuple()[0], cs.date.timetuple()[1], | ||||
cs.date.timetuple()[2], 0, 0, 0, 0, 0, 0]) | ||||
r1076 | ||||
r1264 | if akc(cs.author) in co_day_auth_aggr: | |||
try: | ||||
l = [timegetter(x) for x in | ||||
co_day_auth_aggr[akc(cs.author)]['data']] | ||||
time_pos = l.index(k) | ||||
except ValueError: | ||||
|
r3626 | time_pos = None | ||
r1264 | ||||
|
r3626 | if time_pos >= 0 and time_pos is not None: | ||
r1264 | ||||
datadict = \ | ||||
co_day_auth_aggr[akc(cs.author)]['data'][time_pos] | ||||
r804 | ||||
r1264 | datadict["commits"] += 1 | |||
datadict["added"] += len(cs.added) | ||||
datadict["changed"] += len(cs.changed) | ||||
datadict["removed"] += len(cs.removed) | ||||
else: | ||||
if k >= ts_min_y and k <= ts_max_y or skip_date_limit: | ||||
r629 | ||||
r1264 | datadict = {"time": k, | |||
"commits": 1, | ||||
"added": len(cs.added), | ||||
"changed": len(cs.changed), | ||||
"removed": len(cs.removed), | ||||
} | ||||
co_day_auth_aggr[akc(cs.author)]['data']\ | ||||
.append(datadict) | ||||
r629 | ||||
r547 | else: | |||
if k >= ts_min_y and k <= ts_max_y or skip_date_limit: | ||||
r1264 | co_day_auth_aggr[akc(cs.author)] = { | |||
"label": akc(cs.author), | ||||
"data": [{"time":k, | ||||
"commits":1, | ||||
"added":len(cs.added), | ||||
"changed":len(cs.changed), | ||||
"removed":len(cs.removed), | ||||
}], | ||||
"schema": ["commits"], | ||||
} | ||||
r629 | ||||
r1264 | #gather all data by day | |||
if k in commits_by_day_aggregate: | ||||
commits_by_day_aggregate[k] += 1 | ||||
else: | ||||
commits_by_day_aggregate[k] = 1 | ||||
r629 | ||||
r1264 | overview_data = sorted(commits_by_day_aggregate.items(), | |||
key=itemgetter(0)) | ||||
if not co_day_auth_aggr: | ||||
co_day_auth_aggr[akc(repo.contact)] = { | ||||
"label": akc(repo.contact), | ||||
"data": [0, 1], | ||||
"schema": ["commits"], | ||||
} | ||||
r629 | ||||
r1264 | stats = cur_stats if cur_stats else Statistics() | |||
stats.commit_activity = json.dumps(co_day_auth_aggr) | ||||
stats.commit_activity_combined = json.dumps(overview_data) | ||||
r547 | ||||
r1976 | log.debug('last revison %s' % last_rev) | |||
r1264 | leftovers = len(repo.revisions[last_rev:]) | |||
r1976 | log.debug('revisions to parse %s' % leftovers) | |||
r547 | ||||
r1264 | if last_rev == 0 or leftovers < parse_limit: | |||
log.debug('getting code trending stats') | ||||
stats.languages = json.dumps(__get_codes_stats(repo_name)) | ||||
r629 | ||||
r1264 | try: | |||
stats.repository = dbrepo | ||||
stats.stat_on_revision = last_cs.revision if last_cs else 0 | ||||
r1929 | DBS.add(stats) | |||
DBS.commit() | ||||
r1264 | except: | |||
log.error(traceback.format_exc()) | ||||
r1929 | DBS.rollback() | |||
r1264 | lock.release() | |||
return False | ||||
r629 | ||||
r2109 | # final release | |||
r1264 | lock.release() | |||
r629 | ||||
r2109 | # execute another task if celery is enabled | |||
r3276 | if len(repo.revisions) > 1 and CELERY_ON and recurse_limit > 0: | |||
recurse_limit -= 1 | ||||
run_task(get_commits_stats, repo_name, ts_min_y, ts_max_y, | ||||
recurse_limit) | ||||
if recurse_limit <= 0: | ||||
log.debug('Breaking recursive mode due to reach of recurse limit') | ||||
r1264 | return True | |||
except LockHeld: | ||||
log.info('LockHeld') | ||||
return 'Task with key %s already running' % lockkey | ||||
r547 | ||||
r1244 | ||||
r1002 | @task(ignore_result=True) | |||
r1929 | @dbsession | |||
r3448 | def send_email(recipients, subject, body='', html_body=''): | |||
r689 | """ | |||
Sends an email with defined parameters from the .ini files. | ||||
r1203 | ||||
r689 | :param recipients: list of recipients, it this is empty the defined email | |||
address from field 'email_to' is used instead | ||||
:param subject: subject of the mail | ||||
:param body: body of the mail | ||||
r1717 | :param html_body: html version of body | |||
r689 | """ | |||
r1717 | log = get_logger(send_email) | |||
r1929 | DBS = get_session() | |||
r1944 | ||||
r776 | email_config = config | |||
r2154 | subject = "%s %s" % (email_config.get('email_prefix', ''), subject) | |||
r689 | if not recipients: | |||
r1612 | # if recipients are not defined we send to email_config + all admins | |||
r1717 | admins = [u.email for u in User.query() | |||
.filter(User.admin == True).all()] | ||||
r1612 | recipients = [email_config.get('email_to')] + admins | |||
r689 | ||||
r1717 | mail_from = email_config.get('app_email_from', 'RhodeCode') | |||
r547 | user = email_config.get('smtp_username') | |||
passwd = email_config.get('smtp_password') | ||||
mail_server = email_config.get('smtp_server') | ||||
mail_port = email_config.get('smtp_port') | ||||
r689 | tls = str2bool(email_config.get('smtp_use_tls')) | |||
ssl = str2bool(email_config.get('smtp_use_ssl')) | ||||
r3448 | debug = str2bool(email_config.get('debug')) | |||
|
r1581 | smtp_auth = email_config.get('smtp_auth') | ||
r629 | ||||
|
r3139 | if not mail_server: | ||
log.error("SMTP mail server not configured - cannot send mail") | ||||
return False | ||||
r547 | try: | |||
r1673 | m = SmtpMailer(mail_from, user, passwd, mail_server, smtp_auth, | |||
r1169 | mail_port, ssl, tls, debug=debug) | |||
r1717 | m.send(recipients, subject, body, html_body) | |||
r547 | except: | |||
log.error('Mail sending failed') | ||||
log.error(traceback.format_exc()) | ||||
return False | ||||
return True | ||||
|
r4116 | @task(ignore_result=False) | ||
@dbsession | ||||
def create_repo(form_data, cur_user): | ||||
from rhodecode.model.repo import RepoModel | ||||
from rhodecode.model.user import UserModel | ||||
from rhodecode.model.db import RhodeCodeSetting | ||||
r1244 | ||||
|
r4116 | log = get_logger(create_repo) | ||
DBS = get_session() | ||||
cur_user = UserModel(DBS)._get_user(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 = RhodeCodeSetting.get_default_repo_settings(strip_prefix=True) | ||||
enable_statistics = defs.get('repo_enable_statistics') | ||||
enable_locking = defs.get('repo_enable_locking') | ||||
enable_downloads = defs.get('repo_enable_downloads') | ||||
try: | ||||
repo = RepoModel(DBS)._create_repo( | ||||
repo_name=repo_name_full, | ||||
repo_type=repo_type, | ||||
description=description, | ||||
owner=owner, | ||||
private=private, | ||||
clone_uri=clone_uri, | ||||
repo_group=repo_group, | ||||
landing_rev=landing_rev, | ||||
fork_of=fork_of, | ||||
copy_fork_permissions=copy_fork_permissions, | ||||
copy_group_permissions=copy_group_permissions, | ||||
enable_statistics=enable_statistics, | ||||
enable_locking=enable_locking, | ||||
enable_downloads=enable_downloads, | ||||
state=state | ||||
) | ||||
action_logger(cur_user, 'user_created_repo', | ||||
form_data['repo_name_full'], '', DBS) | ||||
DBS.commit() | ||||
# now create this repo on Filesystem | ||||
RepoModel(DBS)._create_filesystem_repo( | ||||
repo_name=repo_name, | ||||
repo_type=repo_type, | ||||
repo_group=RepoModel(DBS)._get_repo_group(repo_group), | ||||
clone_uri=clone_uri, | ||||
) | ||||
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, e: | ||||
log.warning('Exception %s occured 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(DBS)._delete_filesystem_repo(repo) | ||||
raise | ||||
# it's an odd fix to make celery fail task when exception occurs | ||||
def on_failure(self, *args, **kwargs): | ||||
pass | ||||
return True | ||||
@task(ignore_result=False) | ||||
r1929 | @dbsession | |||
r547 | def create_repo_fork(form_data, cur_user): | |||
r1722 | """ | |||
Creates a fork of repository using interval VCS methods | ||||
r1818 | ||||
r1722 | :param form_data: | |||
:param cur_user: | ||||
""" | ||||
from rhodecode.model.repo import RepoModel | ||||
r2652 | from rhodecode.model.user import UserModel | |||
r1722 | ||||
r1717 | log = get_logger(create_repo_fork) | |||
r1941 | DBS = get_session() | |||
r1722 | base_path = Repository.base_path() | |||
r2652 | cur_user = UserModel(DBS)._get_user(cur_user) | |||
r1722 | ||||
|
r4116 | repo_name = form_data['repo_name'] # fork in this case | ||
repo_name_full = form_data['repo_name_full'] | ||||
r2652 | repo_type = form_data['repo_type'] | |||
owner = cur_user | ||||
private = form_data['private'] | ||||
clone_uri = form_data.get('clone_uri') | ||||
|
r4116 | repo_group = form_data['repo_group'] | ||
r2652 | landing_rev = form_data['landing_rev'] | |||
copy_fork_permissions = form_data.get('copy_permissions') | ||||
|
r4116 | try: | ||
fork_of = RepoModel(DBS)._get_repo(form_data.get('fork_parent_id')) | ||||
r2652 | ||||
|
r4116 | fork_repo = RepoModel(DBS)._create_repo( | ||
repo_name=repo_name_full, | ||||
repo_type=repo_type, | ||||
description=form_data['description'], | ||||
owner=owner, | ||||
private=private, | ||||
clone_uri=clone_uri, | ||||
repo_group=repo_group, | ||||
landing_rev=landing_rev, | ||||
fork_of=fork_of, | ||||
copy_fork_permissions=copy_fork_permissions | ||||
) | ||||
action_logger(cur_user, 'user_forked_repo:%s' % repo_name_full, | ||||
fork_of.repo_name, '', DBS) | ||||
DBS.commit() | ||||
r2652 | ||||
|
r4116 | update_after_clone = form_data['update_after_clone'] | ||
source_repo_path = os.path.join(base_path, fork_of.repo_name) | ||||
r2813 | ||||
|
r4116 | # now create this repo on Filesystem | ||
RepoModel(DBS)._create_filesystem_repo( | ||||
repo_name=repo_name, | ||||
repo_type=repo_type, | ||||
repo_group=RepoModel(DBS)._get_repo_group(repo_group), | ||||
clone_uri=source_repo_path, | ||||
) | ||||
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() | ||||
r2813 | ||||
|
r4116 | # set new created state | ||
repo.set_state(Repository.STATE_CREATED) | ||||
DBS.commit() | ||||
except Exception, e: | ||||
log.warning('Exception %s occured 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(DBS)._delete_filesystem_repo(repo) | ||||
raise | ||||
r1818 | ||||
|
r4116 | # it's an odd fix to make celery fail task when exception occurs | ||
def on_failure(self, *args, **kwargs): | ||||
pass | ||||
return True | ||||
r1244 | ||||
r2185 | ||||
r547 | def __get_codes_stats(repo_name): | |||
|
r4116 | from rhodecode.config.conf import LANGUAGES_EXTENSIONS_MAP | ||
r1722 | repo = Repository.get_by_repo_name(repo_name).scm_instance | |||
r603 | tip = repo.get_changeset() | |||
r547 | code_stats = {} | |||
r630 | ||||
def aggregate(cs): | ||||
for f in cs[2]: | ||||
r1244 | ext = lower(f.extension) | |||
r789 | if ext in LANGUAGES_EXTENSIONS_MAP.keys() and not f.is_binary: | |||
r1244 | if ext in code_stats: | |||
code_stats[ext] += 1 | ||||
r547 | else: | |||
r1244 | code_stats[ext] = 1 | |||
r629 | ||||
r630 | map(aggregate, tip.walk('/')) | |||
r547 | return code_stats or {} | |||