An interactive tool to help find owners covering current change list.

owners_finder.py

+#!/usr/bin/python

[M-A Ruel, 2013/04/24 01:32:20]

Remove, this file is not executable.

[Bei Zhang, 2013/04/24 03:29:02]

Done.

+# Copyright (c) 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.
+
+"""A tool to help picking owner_to_files for reviewing."""

[M-A Ruel, 2013/04/24 01:32:20]

"""Helps picking ..

[Bei Zhang, 2013/04/24 03:29:02]

Done.

+
+import os.path
+import copy
+import owners as owners_module
+
+
+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, local_root,
+               fopen, os_path, glob,
+               email_postfix="@chromium.org",
+               disable_color=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 = ''
+
+    db = OwnersFinder._read_from_database(files, local_root, fopen, os_path,
+                                          glob)
+
+    self.file_to_owners = OwnersFinder._map_files_to_owners(files, db)
+    self.owner_to_files = OwnersFinder._map_owners_to_files(self.file_to_owners,
+                                                            db)
+    self.original_files_to_owners = copy.deepcopy(self.file_to_owners)
+    self.comments = db.comments
+    self.owners_queue = []
+    self.unreviewed_files = set()
+    self.reviewed_by = {}
+    self.selected_owners = set()
+    self.deselected_owners = set()
+    self.reset()
+
+  @staticmethod
+  def _read_from_database(files, local_root, fopen, os_path, glob):
+    db = owners_module.Database(local_root, fopen, os_path, glob)
+    db.load_data_needed_for(files)
+    return db
+
+  @staticmethod
+  def _map_files_to_owners(files, db):
+    files_to_owners = {}
+    for file_name in files:
+      owners_set = set()
+      if file_name in db.owners_for:
+        owners_set = owners_set | db.owners_for[file_name]
+      else:
+        dir_name = file_name
+        while dir_name != '':
+          if dir_name in db.stop_looking:
+            break
+          dir_name = os.path.dirname(dir_name)
+          if dir_name in db.owners_for:
+            owners_set = owners_set | db.owners_for[dir_name]
+            if owners_module.EVERYONE in owners_set:
+              break
+
+      if len(owners_set) == 0:
+        raise Exception("File '%s' has no owner" % file_name)
+
+      # Eliminate files that EVERYONE can review
+      if owners_module.EVERYONE in owners_set:
+        continue
+      files_to_owners[file_name] = owners_set
+    return files_to_owners
+
+  @staticmethod
+  def _map_owners_to_files(files_to_owners, db):
+    owner_to_files = {}
+    for owner_name in db.owned_by:
+      if owner_name == owners_module.EVERYONE:
+        continue
+      files_set = set()
+      for file_name in files_to_owners:
+        if owner_name in files_to_owners[file_name]:
+          files_set.add(file_name)
+      if len(files_set) > 0:
+        owner_to_files[owner_name] = files_set
+    return owner_to_files
+
+  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 reset(self):
+    self.file_to_owners = copy.deepcopy(self.original_files_to_owners)
+    self.unreviewed_files = set(self.file_to_owners.keys())
+    self.reviewed_by = {}
+    self.selected_owners = set()
+    self.deselected_owners = set()
+
+    # Initialize owners queue, sort it by the number of files
+    # each owns
+    self.owners_queue = list(sorted(self.owner_to_files.keys(),
+                                    key=lambda owner: len(
+                                      self.owner_to_files[owner]),
+                                    reverse=True))
+    self.find_mandatory_owners()
+
+  def select_owner(self, owner, findMandatoryOwners=True):
+    if owner in self.selected_owners:
+      return
+    if owner in self.deselected_owners:
+      return
+    if 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.owner_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:
+      return
+    if owner in self.deselected_owners:
+      return
+    if 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.owner_to_files[owner] & self.unreviewed_files:
+      self.file_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.owner_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.file_to_owners[file_name]) == 1,
+                              self.unreviewed_files):
+        self.select_owner(first(self.file_to_owners[file_name]), False)
+        continues = True
+        break
+
+  def print_comments(self, owner):
+    if owner not in self.comments:
+      self.writeln(self.bold_name(owner))
+    else:
+      self.writeln(self.bold_name(owner) + " is commented as:")
+      self.indent()
+      for path in self.comments[owner]:
+        if len(self.comments[owner][path]) > 0:
+          self.writeln(self.greyed(self.comments[owner][path]) +
+                       " (at " + self.bold(path or "<root>") + ")")
+        else:
+          self.writeln(self.greyed("[No comment] ") + " (at " +
+                       self.bold(path or "<root>") + ")")
+      self.unindent()
+
+  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:
+      self.indent()
+      if len(self.file_to_owners[file_name]) <= 3:
+        other_owners = []
+        for ow in self.file_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.file_to_owners[file_name]))) +
+                     "]")
+        self.unindent()
+
+  def print_file_info_detailed(self, file_name):
+    self.writeln(file_name)
+    self.indent()
+    for ow in sorted(self.file_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.print_comments(owner)
+    self.writeln(self.bold_name(owner) + " owns " +
+                 str(len(self.owner_to_files[owner])) + " file(s):")
+    for file_name in sorted(self.owner_to_files[owner]):
+      self.print_file_info(file_name, owner)
+    self.writeln()
+
+  def list_owners(self, owners_queue):
+    if (len(self.owner_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.print_comments(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):
+    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)
+
+  def pick_owner(self, ow):
+    # Allowing to omit domain suffixes
+    if ow not in self.owner_to_files:
+      if ow + self.email_postfix in self.owner_to_files:
+        ow += self.email_postfix
+
+    if ow not in self.owner_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("** 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.owner_to_files[owner]):
+        self.writeln(file_name)
+      self.unindent()
+
+  def hr(self):
+    self.writeln("=====================")
+
+  def run(self):
+    self.reset()
+    while len(self.owners_queue) > 0 and len(self.unreviewed_files) > 0:
+      owner = self.owners_queue[0]
+      try:
+        if owner in self.selected_owners:
+          continue
+        if len(self.unreviewed_files) == 0:
+          self.writeln("Finished.\n\n")
+          break
+        if owner in self.deselected_owners:
+          # If this owner is already deseleted.
+          continue
+        if not any((file_name in self.unreviewed_files)
+                   for file_name in self.owner_to_files[owner]):
+          self.deselect_owner(owner)
+          continue
+        self.hr()
+        self.writeln(
+          self.bold(str(len(self.unreviewed_files))) + " file(s) left.")
+        self.print_owned_files_for(owner)
+        while True:
+          self.writeln("Add " + self.bold_name(owner) + " as your reviewer? ")
+          inp = raw_input(
+            "[yes/no/Defer/pick/files/owners/quit/restart]: ").lower()
+          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(owner)
+            break
+          elif inp == "f" or inp == "files":
+            self.list_files()
+          elif inp == "o" or inp == "owners":
+            self.list_owners(self.owners_queue)
+          elif inp == "p" or inp == "pick":
+            self.pick_owner(raw_input("Pick an owner:"))
+            self.owners_queue.insert(0, owner)
+            break
+          elif inp.startswith("p ") or inp.startswith("pick "):
+            self.pick_owner(inp.split(' ', 2)[1])
+            self.owners_queue.insert(0, owner)
+            break
+          elif inp == 'r' or inp == 'restart':
+            self.reset()
+            break
+          elif inp == "q" or inp == "quit":
+            # Exit with error
+            return 1
+      finally:
+        if len(self.owners_queue):
+          self.owners_queue.pop(0)
+
+    self.print_result()
+    return 0