Mercurial > kallithea
view kallithea/model/comment.py @ 4208:ad38f9f93b3b kallithea-2.2.5-rebrand
Correct licensing information in individual files.
The top-level license file is now LICENSE.md.
Also, in various places where there should have been joint copyright holders
listed, a single copyright holder was listed. It does not appear easy to add
a link to a large list of copyright holders in these places, so it simply
refers to the fact that various authors hold copyright.
In future, if an easy method is discovered to link to a list from those
places, we should do so.
Finally, text is added to LICENSE.md to point to where the full list of
copyright holders is, and that Kallithea as a whole is GPLv3'd.
author | Bradley M. Kuhn <bkuhn@sfconservancy.org> |
---|---|
date | Wed, 21 May 2014 16:59:37 -0400 |
parents | d1addaf7a91e |
children | 1948ede028ef |
line wrap: on
line source
# -*- 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.comment ~~~~~~~~~~~~~~~~~~~~~~~ comments model for RhodeCode :created_on: Nov 11, 2011 :author: marcink :copyright: (c) 2013 RhodeCode GmbH. :license: GPLv3, see LICENSE.md for more details. """ import logging import traceback from pylons.i18n.translation import _ from sqlalchemy.util.compat import defaultdict from kallithea.lib.utils2 import extract_mentioned_users, safe_unicode from kallithea.lib import helpers as h from kallithea.model import BaseModel from kallithea.model.db import ChangesetComment, User, Repository, \ Notification, PullRequest from kallithea.model.notification import NotificationModel from kallithea.model.meta import Session log = logging.getLogger(__name__) class ChangesetCommentsModel(BaseModel): cls = ChangesetComment def __get_changeset_comment(self, changeset_comment): return self._get_instance(ChangesetComment, changeset_comment) def __get_pull_request(self, pull_request): return self._get_instance(PullRequest, pull_request) def _extract_mentions(self, s): user_objects = [] for username in extract_mentioned_users(s): user_obj = User.get_by_username(username, case_insensitive=True) if user_obj: user_objects.append(user_obj) return user_objects def _get_notification_data(self, repo, comment, user, comment_text, line_no=None, revision=None, pull_request=None, status_change=None, closing_pr=False): """ Get notification data :param comment_text: :param line: :returns: tuple (subj,body,recipients,notification_type,email_kwargs) """ # make notification body = comment_text # text of the comment line = '' if line_no: line = _('on line %s') % line_no #changeset if revision: notification_type = Notification.TYPE_CHANGESET_COMMENT cs = repo.scm_instance.get_changeset(revision) desc = "%s" % (cs.short_id) _url = h.url('changeset_home', repo_name=repo.repo_name, revision=revision, anchor='comment-%s' % comment.comment_id, qualified=True, ) subj = safe_unicode( h.link_to('Re changeset: %(desc)s %(line)s' % \ {'desc': desc, 'line': line}, _url) ) email_subject = '%s commented on changeset %s' % \ (user.username, h.short_id(revision)) # get the current participants of this changeset recipients = ChangesetComment.get_users(revision=revision) # add changeset author if it's in rhodecode system cs_author = User.get_from_cs_author(cs.author) if not cs_author: #use repo owner if we cannot extract the author correctly cs_author = repo.user recipients += [cs_author] email_kwargs = { 'status_change': status_change, 'cs_comment_user': h.person(user.email), 'cs_target_repo': h.url('summary_home', repo_name=repo.repo_name, qualified=True), 'cs_comment_url': _url, 'raw_id': revision, 'message': cs.message } #pull request elif pull_request: notification_type = Notification.TYPE_PULL_REQUEST_COMMENT desc = comment.pull_request.title _url = h.url('pullrequest_show', repo_name=pull_request.other_repo.repo_name, pull_request_id=pull_request.pull_request_id, anchor='comment-%s' % comment.comment_id, qualified=True, ) subj = safe_unicode( h.link_to('Re pull request #%(pr_id)s: %(desc)s %(line)s' % \ {'desc': desc, 'pr_id': comment.pull_request.pull_request_id, 'line': line}, _url) ) email_subject = '%s commented on pull request #%s' % \ (user.username, comment.pull_request.pull_request_id) # get the current participants of this pull request recipients = ChangesetComment.get_users(pull_request_id= pull_request.pull_request_id) # add pull request author recipients += [pull_request.author] # add the reviewers to notification recipients += [x.user for x in pull_request.reviewers] #set some variables for email notification email_kwargs = { 'pr_title': pull_request.title, 'pr_id': pull_request.pull_request_id, 'status_change': status_change, 'closing_pr': closing_pr, 'pr_comment_url': _url, 'pr_comment_user': h.person(user.email), 'pr_target_repo': h.url('summary_home', repo_name=pull_request.other_repo.repo_name, qualified=True) } return subj, body, recipients, notification_type, email_kwargs, email_subject def create(self, text, repo, user, revision=None, pull_request=None, f_path=None, line_no=None, status_change=None, closing_pr=False, send_email=True): """ Creates new comment for changeset or pull request. IF status_change is not none this comment is associated with a status change of changeset or changesets associated with pull request :param text: :param repo: :param user: :param revision: :param pull_request: :param f_path: :param line_no: :param status_change: :param closing_pr: :param send_email: """ if not text: log.warning('Missing text for comment, skipping...') return repo = self._get_repo(repo) user = self._get_user(user) comment = ChangesetComment() comment.repo = repo comment.author = user comment.text = text comment.f_path = f_path comment.line_no = line_no if revision: comment.revision = revision elif pull_request: pull_request = self.__get_pull_request(pull_request) comment.pull_request = pull_request else: raise Exception('Please specify revision or pull_request_id') Session().add(comment) Session().flush() if send_email: (subj, body, recipients, notification_type, email_kwargs, email_subject) = self._get_notification_data( repo, comment, user, comment_text=text, line_no=line_no, revision=revision, pull_request=pull_request, status_change=status_change, closing_pr=closing_pr) # create notification objects, and emails NotificationModel().create( created_by=user, subject=subj, body=body, recipients=recipients, type_=notification_type, email_kwargs=email_kwargs, email_subject=email_subject ) mention_recipients = set(self._extract_mentions(body))\ .difference(recipients) if mention_recipients: email_kwargs.update({'pr_mention': True}) subj = _('[Mention]') + ' ' + subj NotificationModel().create( created_by=user, subject=subj, body=body, recipients=mention_recipients, type_=notification_type, email_kwargs=email_kwargs ) return comment def delete(self, comment): """ Deletes given comment :param comment_id: """ comment = self.__get_changeset_comment(comment) Session().delete(comment) return comment def get_comments(self, repo_id, revision=None, pull_request=None): """ Gets main comments based on revision or pull_request_id :param repo_id: :param revision: :param pull_request: """ q = ChangesetComment.query()\ .filter(ChangesetComment.repo_id == repo_id)\ .filter(ChangesetComment.line_no == None)\ .filter(ChangesetComment.f_path == None) if revision: q = q.filter(ChangesetComment.revision == revision) elif pull_request: pull_request = self.__get_pull_request(pull_request) q = q.filter(ChangesetComment.pull_request == pull_request) else: raise Exception('Please specify revision or pull_request') q = q.order_by(ChangesetComment.created_on) return q.all() def get_inline_comments(self, repo_id, revision=None, pull_request=None): q = Session().query(ChangesetComment)\ .filter(ChangesetComment.repo_id == repo_id)\ .filter(ChangesetComment.line_no != None)\ .filter(ChangesetComment.f_path != None)\ .order_by(ChangesetComment.comment_id.asc())\ if revision: q = q.filter(ChangesetComment.revision == revision) elif pull_request: pull_request = self.__get_pull_request(pull_request) q = q.filter(ChangesetComment.pull_request == pull_request) else: raise Exception('Please specify revision or pull_request_id') comments = q.all() paths = defaultdict(lambda: defaultdict(list)) for co in comments: paths[co.f_path][co.line_no].append(co) return paths.items()