owners_finder.py

# Copyright 2013 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.

"""Interactive tool for finding reviewers/owners for a change."""

from __future__ import print_function

import os
import copy


import git_common
import gclient_utils


def first(iterable):
  for element in iterable:
    return element


class OwnersFinder(object):
  COLOR_LINK = '\033[4m'
  COLOR_BOLD = '\033[1;32m'
  COLOR_GREY = '\033[0;37m'
  COLOR_RESET = '\033[0m'

  indentation = 0

  def __init__(self, files, author, reviewers, owners_client,
               email_postfix='@chromium.org',
               disable_color=False,
               ignore_author=False):
    self.email_postfix = email_postfix

    if os.name == 'nt' or disable_color:
      self.COLOR_LINK = ''
      self.COLOR_BOLD = ''
      self.COLOR_GREY = ''
      self.COLOR_RESET = ''

    self.author = author

    filtered_files = files

    reviewers = list(reviewers)
    if author and not ignore_author:
      reviewers.append(author)

    # Eliminate files that existing reviewers can review.
    self.owners_client = owners_client
    approval_status = self.owners_client.GetFilesApprovalStatus(
      filtered_files, reviewers, [])
    filtered_files = [
      f for f in filtered_files
      if approval_status[f] != self.owners_client.APPROVED]

    # If some files are eliminated.
    if len(filtered_files) != len(files):
      files = filtered_files

    self.files_to_owners = self.owners_client.BatchListOwners(files)

    self.owners_to_files = {}
    self._map_owners_to_files()

    self.original_files_to_owners = copy.deepcopy(self.files_to_owners)

    # This is the queue that will be shown in the interactive questions.
    # It is initially sorted by the score in descending order. In the
    # interactive questions a user can choose to "defer" its decision, then the
    # owner will be put to the end of the queue and shown later.
    self.owners_queue = []

    self.unreviewed_files = set()
    self.reviewed_by = {}
    self.selected_owners = set()
    self.deselected_owners = set()
    self.reset()

  def run(self):
    self.reset()
    while self.owners_queue and self.unreviewed_files:
      owner = self.owners_queue[0]

      if (owner in self.selected_owners) or (owner in self.deselected_owners):
        continue

      if not any((file_name in self.unreviewed_files)
                 for file_name in self.owners_to_files[owner]):
        self.deselect_owner(owner)
        continue

      self.print_info(owner)

      while True:
        inp = self.input_command(owner)
        if inp == 'y' or inp == 'yes':
          self.select_owner(owner)
          break
        elif inp == 'n' or inp == 'no':
          self.deselect_owner(owner)
          break
        elif inp == '' or inp == 'd' or inp == 'defer':
          self.owners_queue.append(self.owners_queue.pop(0))
          break
        elif inp == 'f' or inp == 'files':
          self.list_files()
          break
        elif inp == 'o' or inp == 'owners':
          self.list_owners(self.owners_queue)
          break
        elif inp == 'p' or inp == 'pick':
          self.pick_owner(gclient_utils.AskForData('Pick an owner: '))
          break
        elif inp.startswith('p ') or inp.startswith('pick '):
          self.pick_owner(inp.split(' ', 2)[1].strip())
          break
        elif inp == 'r' or inp == 'restart':
          self.reset()
          break
        elif inp == 'q' or inp == 'quit':
          # Exit with error
          return 1

    self.print_result()
    return 0

  def _map_owners_to_files(self):
    for file_name in self.files_to_owners:
      for owner in self.files_to_owners[file_name]:
        self.owners_to_files.setdefault(owner, set())
        self.owners_to_files[owner].add(file_name)

  def reset(self):
    self.files_to_owners = copy.deepcopy(self.original_files_to_owners)
    self.unreviewed_files = set(self.files_to_owners.keys())
    self.reviewed_by = {}
    self.selected_owners = set()
    self.deselected_owners = set()

    # Randomize owners' names so that if many reviewers have identical scores
    # they will be randomly ordered to avoid bias.
    owners = list(self.owners_client.ScoreOwners(self.files_to_owners.keys()))
    if self.author and self.author in owners:
      owners.remove(self.author)
    self.owners_queue = owners
    self.find_mandatory_owners()

  def select_owner(self, owner, findMandatoryOwners=True):
    if owner in self.selected_owners or owner in self.deselected_owners\
        or not (owner in self.owners_queue):
      return
    self.writeln('Selected: ' + owner)
    self.owners_queue.remove(owner)
    self.selected_owners.add(owner)
    for file_name in filter(
        lambda file_name: file_name in self.unreviewed_files,
        self.owners_to_files[owner]):
      self.unreviewed_files.remove(file_name)
      self.reviewed_by[file_name] = owner
    if findMandatoryOwners:
      self.find_mandatory_owners()

  def deselect_owner(self, owner, findMandatoryOwners=True):
    if owner in self.selected_owners or owner in self.deselected_owners\
        or not (owner in self.owners_queue):
      return
    self.writeln('Deselected: ' + owner)
    self.owners_queue.remove(owner)
    self.deselected_owners.add(owner)
    for file_name in self.owners_to_files[owner] & self.unreviewed_files:
      self.files_to_owners[file_name].remove(owner)
    if findMandatoryOwners:
      self.find_mandatory_owners()

  def find_mandatory_owners(self):
    continues = True
    for owner in self.owners_queue:
      if owner in self.selected_owners:
        continue
      if owner in self.deselected_owners:
        continue
      if len(self.owners_to_files[owner] & self.unreviewed_files) == 0:
        self.deselect_owner(owner, False)

    while continues:
      continues = False
      for file_name in filter(
          lambda file_name: len(self.files_to_owners[file_name]) == 1,
          self.unreviewed_files):
        owner = first(self.files_to_owners[file_name])
        self.select_owner(owner, False)
        continues = True
        break

  def print_file_info(self, file_name, except_owner=''):
    if file_name not in self.unreviewed_files:
      self.writeln(self.greyed(file_name +
                               ' (by ' +
                               self.bold_name(self.reviewed_by[file_name]) +
                               ')'))
    else:
      if len(self.files_to_owners[file_name]) <= 3:
        other_owners = []
        for ow in self.files_to_owners[file_name]:
          if ow != except_owner:
            other_owners.append(self.bold_name(ow))
        self.writeln(file_name +
                     ' [' + (', '.join(other_owners)) + ']')
      else:
        self.writeln(file_name + ' [' +
                     self.bold(str(len(self.files_to_owners[file_name]))) +
                     ']')

  def print_file_info_detailed(self, file_name):
    self.writeln(file_name)
    self.indent()
    for ow in sorted(self.files_to_owners[file_name]):
      if ow in self.deselected_owners:
        self.writeln(self.bold_name(self.greyed(ow)))
      elif ow in self.selected_owners:
        self.writeln(self.bold_name(self.greyed(ow)))
      else:
        self.writeln(self.bold_name(ow))
    self.unindent()

  def print_owned_files_for(self, owner):
    # Print owned files
    self.writeln(self.bold_name(owner))
    self.writeln(self.bold_name(owner) + ' owns ' +
                 str(len(self.owners_to_files[owner])) + ' file(s):')
    self.indent()
    for file_name in sorted(self.owners_to_files[owner]):
      self.print_file_info(file_name, owner)
    self.unindent()
    self.writeln()

  def list_owners(self, owners_queue):
    if (len(self.owners_to_files) - len(self.deselected_owners) -
            len(self.selected_owners)) > 3:
      for ow in owners_queue:
        if ow not in self.deselected_owners and ow not in self.selected_owners:
          self.writeln(self.bold_name(ow))
    else:
      for ow in owners_queue:
        if ow not in self.deselected_owners and ow not in self.selected_owners:
          self.writeln()
          self.print_owned_files_for(ow)

  def list_files(self):
    self.indent()
    if len(self.unreviewed_files) > 5:
      for file_name in sorted(self.unreviewed_files):
        self.print_file_info(file_name)
    else:
      for file_name in self.unreviewed_files:
        self.print_file_info_detailed(file_name)
    self.unindent()

  def pick_owner(self, ow):
    # Allowing to omit domain suffixes
    if ow not in self.owners_to_files:
      if ow + self.email_postfix in self.owners_to_files:
        ow += self.email_postfix

    if ow not in self.owners_to_files:
      self.writeln('You cannot pick ' + self.bold_name(ow) + ' manually. ' +
                   'It\'s an invalid name or not related to the change list.')
      return False
    elif ow in self.selected_owners:
      self.writeln('You cannot pick ' + self.bold_name(ow) + ' manually. ' +
                   'It\'s already selected.')
      return False
    elif ow in self.deselected_owners:
      self.writeln('You cannot pick ' + self.bold_name(ow) + ' manually.' +
                   'It\'s already unselected.')
      return False

    self.select_owner(ow)
    return True

  def print_result(self):
    # Print results
    self.writeln()
    self.writeln()
    if len(self.selected_owners) == 0:
      self.writeln('This change list already has owner-reviewers for all '
                   'files.')
      self.writeln('Use --ignore-current if you want to ignore them.')
    else:
      self.writeln('** You selected these owners **')
      self.writeln()
      for owner in self.selected_owners:
        self.writeln(self.bold_name(owner) + ':')
        self.indent()
        for file_name in sorted(self.owners_to_files[owner]):
          self.writeln(file_name)
        self.unindent()

  def bold(self, text):
    return self.COLOR_BOLD + text + self.COLOR_RESET

  def bold_name(self, name):
    return (self.COLOR_BOLD +
            name.replace(self.email_postfix, '') + self.COLOR_RESET)

  def greyed(self, text):
    return self.COLOR_GREY + text + self.COLOR_RESET

  def indent(self):
    self.indentation += 1

  def unindent(self):
    self.indentation -= 1

  def print_indent(self):
    return '  ' * self.indentation

  def writeln(self, text=''):
    print(self.print_indent() + text)

  def hr(self):
    self.writeln('=====================')

  def print_info(self, owner):
    self.hr()
    self.writeln(
        self.bold(str(len(self.unreviewed_files))) + ' file(s) left.')
    self.print_owned_files_for(owner)

  def input_command(self, owner):
    self.writeln('Add ' + self.bold_name(owner) + ' as your reviewer? ')
    return gclient_utils.AskForData(
        '[yes/no/Defer/pick/files/owners/quit/restart]: ').lower()