Project import generated by Copybara.
GitOrigin-RevId: d9e9e3fb4e31372ec1fb43b178994ca78fa8fe70
diff --git a/services/tracker_fulltext.py b/services/tracker_fulltext.py
new file mode 100644
index 0000000..ecbfc44
--- /dev/null
+++ b/services/tracker_fulltext.py
@@ -0,0 +1,320 @@
+# Copyright 2016 The Chromium Authors. All rights reserved.
+# Use of this source code is governed by a BSD-style
+# license that can be found in the LICENSE file or at
+# https://developers.google.com/open-source/licenses/bsd
+
+"""A set of functions that provide fulltext search for issues."""
+from __future__ import print_function
+from __future__ import division
+from __future__ import absolute_import
+
+import collections
+import logging
+import time
+
+from six import string_types
+
+from google.appengine.api import search
+
+import settings
+from framework import framework_constants
+from framework import framework_helpers
+from framework import framework_views
+from services import fulltext_helpers
+from tracker import tracker_bizobj
+
+
+# When updating and re-indexing all issues in a project, work in batches
+# of this size to manage memory usage and avoid rpc timeouts.
+_INDEX_BATCH_SIZE = 40
+
+
+# The user can search for text that occurs specifically in these
+# parts of an issue.
+ISSUE_FULLTEXT_FIELDS = ['summary', 'description', 'comment']
+# Note: issue documents also contain a "metadata" field, but we do not
+# expose that to users. Issue metadata can be searched in a structured way
+# by giving a specific field name such as "owner:" or "status:". The metadata
+# search field exists only for fulltext queries that do not specify any field.
+
+
+def IndexIssues(cnxn, issues, user_service, issue_service, config_service):
+ """(Re)index all the given issues.
+
+ Args:
+ cnxn: connection to SQL database.
+ issues: list of Issue PBs to index.
+ user_service: interface to user data storage.
+ issue_service: interface to issue data storage.
+ config_service: interface to configuration data storage.
+ """
+ issues = list(issues)
+ config_dict = config_service.GetProjectConfigs(
+ cnxn, {issue.project_id for issue in issues})
+ for start in range(0, len(issues), _INDEX_BATCH_SIZE):
+ logging.info('indexing issues: %d remaining', len(issues) - start)
+ _IndexIssueBatch(
+ cnxn, issues[start:start + _INDEX_BATCH_SIZE], user_service,
+ issue_service, config_dict)
+
+
+def _IndexIssueBatch(cnxn, issues, user_service, issue_service, config_dict):
+ """Internal method to (re)index the given batch of issues.
+
+ Args:
+ cnxn: connection to SQL database.
+ issues: list of Issue PBs to index.
+ user_service: interface to user data storage.
+ issue_service: interface to issue data storage.
+ config_dict: dict {project_id: config} for all the projects that
+ the given issues are in.
+ """
+ user_ids = tracker_bizobj.UsersInvolvedInIssues(issues)
+ comments_dict = issue_service.GetCommentsForIssues(
+ cnxn, [issue.issue_id for issue in issues])
+ for comments in comments_dict.values():
+ user_ids.update([ic.user_id for ic in comments])
+
+ users_by_id = framework_views.MakeAllUserViews(
+ cnxn, user_service, user_ids)
+ _CreateIssueSearchDocuments(issues, comments_dict, users_by_id, config_dict)
+
+
+def _CreateIssueSearchDocuments(
+ issues, comments_dict, users_by_id, config_dict):
+ """Make the GAE search index documents for the given issue batch.
+
+ Args:
+ issues: list of issues to index.
+ comments_dict: prefetched dictionary of comments on those issues.
+ users_by_id: dictionary {user_id: UserView} so that the email
+ addresses of users who left comments can be found via search.
+ config_dict: dict {project_id: config} for all the projects that
+ the given issues are in.
+ """
+ documents_by_shard = collections.defaultdict(list)
+ for issue in issues:
+ summary = issue.summary
+ # TODO(jrobbins): allow search specifically on explicit vs derived
+ # fields.
+ owner_id = tracker_bizobj.GetOwnerId(issue)
+ owner_email = users_by_id[owner_id].email
+ config = config_dict[issue.project_id]
+ component_paths = []
+ for component_id in issue.component_ids:
+ cd = tracker_bizobj.FindComponentDefByID(component_id, config)
+ if cd:
+ component_paths.append(cd.path)
+
+ field_values = [tracker_bizobj.GetFieldValue(fv, users_by_id)
+ for fv in issue.field_values]
+ # Convert to string only the values that are not strings already.
+ # This is done because the default encoding in appengine seems to be 'ascii'
+ # and string values might contain unicode characters, so str will fail to
+ # encode them.
+ field_values = [value if isinstance(value, string_types) else str(value)
+ for value in field_values]
+
+ metadata = '%s %s %s %s %s %s' % (
+ tracker_bizobj.GetStatus(issue),
+ owner_email,
+ [users_by_id[cc_id].email for cc_id in
+ tracker_bizobj.GetCcIds(issue)],
+ ' '.join(component_paths),
+ ' '.join(field_values),
+ ' '.join(tracker_bizobj.GetLabels(issue)))
+ custom_fields = _BuildCustomFTSFields(issue)
+
+ comments = comments_dict.get(issue.issue_id, [])
+ room_for_comments = (framework_constants.MAX_FTS_FIELD_SIZE -
+ len(summary) -
+ len(metadata) -
+ sum(len(cf.value) for cf in custom_fields))
+ comments = _IndexableComments(
+ comments, users_by_id, remaining_chars=room_for_comments)
+ logging.info('len(comments) is %r', len(comments))
+ if comments:
+ description = _ExtractCommentText(comments[0], users_by_id)
+ description = description[:framework_constants.MAX_FTS_FIELD_SIZE]
+ all_comments = ' '. join(
+ _ExtractCommentText(c, users_by_id) for c in comments[1:])
+ all_comments = all_comments[:framework_constants.MAX_FTS_FIELD_SIZE]
+ else:
+ description = ''
+ all_comments = ''
+ logging.info(
+ 'Issue %s:%r has zero indexable comments',
+ issue.project_name, issue.local_id)
+
+ logging.info('Building document for %s:%d',
+ issue.project_name, issue.local_id)
+ logging.info('len(summary) = %d', len(summary))
+ logging.info('len(metadata) = %d', len(metadata))
+ logging.info('len(description) = %d', len(description))
+ logging.info('len(comment) = %d', len(all_comments))
+ for cf in custom_fields:
+ logging.info('len(%s) = %d', cf.name, len(cf.value))
+
+ doc = search.Document(
+ doc_id=str(issue.issue_id),
+ fields=[
+ search.NumberField(name='project_id', value=issue.project_id),
+ search.TextField(name='summary', value=summary),
+ search.TextField(name='metadata', value=metadata),
+ search.TextField(name='description', value=description),
+ search.TextField(name='comment', value=all_comments),
+ ] + custom_fields)
+
+ shard_id = issue.issue_id % settings.num_logical_shards
+ documents_by_shard[shard_id].append(doc)
+
+ start_time = time.time()
+ promises = []
+ for shard_id, documents in documents_by_shard.items():
+ if documents:
+ promises.append(framework_helpers.Promise(
+ _IndexDocsInShard, shard_id, documents))
+
+ for promise in promises:
+ promise.WaitAndGetValue()
+
+ logging.info('Finished %d indexing in shards in %d ms',
+ len(documents_by_shard), int((time.time() - start_time) * 1000))
+
+
+def _IndexableComments(comments, users_by_id, remaining_chars=None):
+ """We only index the comments that are not deleted or banned.
+
+ Args:
+ comments: list of Comment PBs for one issue.
+ users_by_id: Dict of (user_id -> UserView) for all users.
+ remaining_chars: number of characters available for comment text
+ without hitting the GAE search index max document size.
+
+ Returns:
+ A list of comments filtered to not have any deleted comments or
+ comments from banned users. If the issue has a huge number of
+ comments, only a certain number of the first and last comments
+ are actually indexed.
+ """
+ if remaining_chars is None:
+ remaining_chars = framework_constants.MAX_FTS_FIELD_SIZE
+ allowed_comments = []
+ for comment in comments:
+ user_view = users_by_id.get(comment.user_id)
+ if not (comment.deleted_by or (user_view and user_view.banned)):
+ if comment.is_description and allowed_comments:
+ # index the latest description, but not older descriptions
+ allowed_comments[0] = comment
+ else:
+ allowed_comments.append(comment)
+
+ reasonable_size = (framework_constants.INITIAL_COMMENTS_TO_INDEX +
+ framework_constants.FINAL_COMMENTS_TO_INDEX)
+ if len(allowed_comments) <= reasonable_size:
+ candidates = allowed_comments
+ else:
+ candidates = ( # Prioritize the description and recent comments.
+ allowed_comments[0:1] +
+ allowed_comments[-framework_constants.FINAL_COMMENTS_TO_INDEX:] +
+ allowed_comments[1:framework_constants.INITIAL_COMMENTS_TO_INDEX])
+
+ total_length = 0
+ result = []
+ for comment in candidates:
+ total_length += len(comment.content)
+ if total_length > remaining_chars:
+ break
+ result.append(comment)
+
+ return result
+
+
+def _IndexDocsInShard(shard_id, documents):
+ search_index = search.Index(
+ name=settings.search_index_name_format % shard_id)
+ search_index.put(documents)
+ logging.info('FTS indexed %d docs in shard %d', len(documents), shard_id)
+ # TODO(jrobbins): catch OverQuotaError and add the issues to the
+ # ReindexQueue table instead.
+
+
+def _ExtractCommentText(comment, users_by_id):
+ """Return a string with all the searchable text of the given Comment PB."""
+ commenter_email = users_by_id[comment.user_id].email
+ return '%s %s %s' % (
+ commenter_email,
+ comment.content,
+ ' '.join(attach.filename
+ for attach in comment.attachments
+ if not attach.deleted))
+
+
+def _BuildCustomFTSFields(issue):
+ """Return a list of FTS Fields to index string-valued custom fields."""
+ fts_fields = []
+ for fv in issue.field_values:
+ if fv.str_value:
+ # TODO(jrobbins): also indicate which were derived vs. explicit.
+ # TODO(jrobbins): also toss in the email addresses of any users in
+ # user-valued custom fields, ints for int-valued fields, etc.
+ fts_field = search.TextField(
+ name='custom_%d' % fv.field_id, value=fv.str_value)
+ fts_fields.append(fts_field)
+
+ return fts_fields
+
+
+def UnindexIssues(issue_ids):
+ """Remove many issues from the sharded search indexes."""
+ iids_by_shard = {}
+ for issue_id in issue_ids:
+ shard_id = issue_id % settings.num_logical_shards
+ iids_by_shard.setdefault(shard_id, [])
+ iids_by_shard[shard_id].append(issue_id)
+
+ for shard_id, iids_in_shard in iids_by_shard.items():
+ try:
+ logging.info(
+ 'unindexing %r issue_ids in %r', len(iids_in_shard), shard_id)
+ search_index = search.Index(
+ name=settings.search_index_name_format % shard_id)
+ search_index.delete([str(iid) for iid in iids_in_shard])
+ except search.Error:
+ logging.exception('FTS deletion failed')
+
+
+def SearchIssueFullText(project_ids, query_ast_conj, shard_id):
+ """Do full-text search in GAE FTS.
+
+ Args:
+ project_ids: list of project ID numbers to consider.
+ query_ast_conj: One conjuctive clause from the AST parsed
+ from the user's query.
+ shard_id: int shard ID for the shard to consider.
+
+ Returns:
+ (issue_ids, capped) where issue_ids is a list of issue issue_ids that match
+ the full-text query. And, capped is True if the results were capped due to
+ an implementation limitation. Or, return (None, False) if the given AST
+ conjunction contains no full-text conditions.
+ """
+ fulltext_query = fulltext_helpers.BuildFTSQuery(
+ query_ast_conj, ISSUE_FULLTEXT_FIELDS)
+ if fulltext_query is None:
+ return None, False
+
+ if project_ids:
+ project_clause = ' OR '.join(
+ 'project_id:%d' % pid for pid in project_ids)
+ fulltext_query = '(%s) %s' % (project_clause, fulltext_query)
+
+ # TODO(jrobbins): it would be good to also include some other
+ # structured search terms to narrow down the set of index
+ # documents considered. E.g., most queries are only over the
+ # open issues.
+ logging.info('FTS query is %r', fulltext_query)
+ issue_ids = fulltext_helpers.ComprehensiveSearch(
+ fulltext_query, settings.search_index_name_format % shard_id)
+ capped = len(issue_ids) >= settings.fulltext_limit_per_shard
+ return issue_ids, capped