| #!/usr/bin/env python |
| |
| import sys, re, subprocess, os |
| |
| def usage(): |
| print("""Usage: cat <issues> | triage-guesser.py |
| triage-guesser.py attempts to guess the assignee based on the title of the bug |
| |
| triage-guesser reads issues from stdin (issues can be copy-pasted from the hotlist) |
| """) |
| sys.exit(1) |
| |
| class Issue(object): |
| def __init__(self, issueId, description): |
| self.issueId = issueId |
| self.description = description |
| |
| class IssueComponent(object): |
| def __init__(self, name): |
| self.name = name |
| def __str__(self): |
| return "Component: '" + self.name + "'" |
| def __repr__(self): |
| return str(self) |
| |
| components = {} |
| components["navigation"] = IssueComponent("Navigation") |
| |
| class AssigneeRecommendation(object): |
| def __init__(self, usernames, justification): |
| self.usernames = usernames |
| self.justification = justification |
| |
| def intersect(self, other): |
| names = [] |
| for name in self.usernames: |
| if name in other.usernames: |
| names.append(name) |
| justification = self.justification + ", " + other.justification |
| return AssigneeRecommendation(names, justification) |
| |
| class RecommenderRule(object): |
| def __init__(self): |
| return |
| |
| def recommend(self, bug): |
| return |
| |
| class ShellRunner(object): |
| def __init__(self): |
| return |
| |
| def runAndGetOutput(self, args): |
| return subprocess.check_output(args) |
| shellRunner = ShellRunner() |
| |
| class WordRule(RecommenderRule): |
| def __init__(self, word, assignees): |
| super(WordRule, self).__init__() |
| self.word = word |
| self.assignees = assignees |
| |
| def recommend(self, bug): |
| if self.word.lower() in bug.description.lower(): |
| return AssigneeRecommendation(self.assignees, '"' + self.word + '"') |
| return None |
| |
| class FileFinder(object): |
| def __init__(self, rootPath): |
| self.rootPath = rootPath |
| self.resultsCache = {} |
| |
| def findIname(self, name): |
| if name not in self.resultsCache: |
| text = shellRunner.runAndGetOutput(["find", self.rootPath , "-type", "f", "-iname", name]) |
| filePaths = [path.strip() for path in text.split("\n")] |
| filePaths = [path for path in filePaths if path != ""] |
| self.resultsCache[name] = filePaths |
| return self.resultsCache[name] |
| |
| def tryToIdentifyFile(self, nameComponent): |
| if len(nameComponent) < 1: |
| return [] |
| queries = [nameComponent + ".*", "nameComponent*"] |
| if len(nameComponent) >= 10: |
| # For a sufficiently specific query, allow it to match the middle of a filename too |
| queries.append("*" + nameComponent + ".*") |
| for query in queries: |
| matches = self.findIname(query) |
| if len(matches) > 0 and len(matches) <= 4: |
| # We found a small enough number of matches to have |
| # reasonable confidence in having found the right file |
| return matches |
| return [] |
| |
| class InterestingWordChooser(object): |
| def __init__(self): |
| return |
| |
| def findInterestingWords(self, text): |
| words = re.split("#| |\.", text) |
| words = [word for word in words if len(word) >= 4] |
| words.sort(key=len, reverse=True) |
| return words |
| interestingWordChooser = InterestingWordChooser() |
| |
| class GitLogger(object): |
| def __init__(self): |
| return |
| |
| def gitLog1Author(self, filePath): |
| text = shellRunner.runAndGetOutput(["bash", "-c", "cd " + os.path.dirname(filePath) + " && git log --no-merges -1 --format='%ae' -- " + os.path.basename(filePath)]).strip().replace("@google.com", "") |
| return text |
| gitLogger = GitLogger() |
| |
| class LastTouchedBy_Rule(RecommenderRule): |
| def __init__(self, fileFinder): |
| super(LastTouchedBy_Rule, self).__init__() |
| self.fileFinder = fileFinder |
| |
| def recommend(self, bug): |
| interestingWords = interestingWordChooser.findInterestingWords(bug.description) |
| for word in interestingWords: |
| filePaths = self.fileFinder.tryToIdentifyFile(word) |
| if len(filePaths) > 0: |
| candidateAuthors = [] |
| for path in filePaths: |
| thisAuthor = gitLogger.gitLog1Author(path) |
| if len(candidateAuthors) == 0 or thisAuthor != candidateAuthors[-1]: |
| candidateAuthors.append(thisAuthor) |
| if len(candidateAuthors) == 1: |
| return AssigneeRecommendation(candidateAuthors, "last touched " + os.path.basename(filePaths[0])) |
| return None |
| |
| class OwnersRule(RecommenderRule): |
| def __init__(self, fileFinder): |
| super(OwnersRule, self).__init__() |
| self.fileFinder = fileFinder |
| |
| def recommend(self, bug): |
| interestingWords = interestingWordChooser.findInterestingWords(bug.description) |
| for word in interestingWords: |
| filePaths = self.fileFinder.tryToIdentifyFile(word) |
| if len(filePaths) > 0: |
| commonPrefix = os.path.commonprefix(filePaths) |
| dirToCheck = commonPrefix |
| if len(dirToCheck) < 1: |
| continue |
| while True: |
| if dirToCheck[-1] == "/": |
| dirToCheck = dirToCheck[:-1] |
| if len(dirToCheck) <= len(self.fileFinder.rootPath): |
| break |
| ownerFilePath = os.path.join(dirToCheck, "OWNERS") |
| if os.path.isfile(ownerFilePath): |
| with open(ownerFilePath) as ownerFile: |
| lines = ownerFile.readlines() |
| names = [line.replace("@google.com", "").strip() for line in lines] |
| relOwnersPath = os.path.relpath(ownerFilePath, self.fileFinder.rootPath) |
| justification = relOwnersPath + " (" + os.path.basename(filePaths[0] + ' ("' + word + '")') |
| if len(filePaths) > 1: |
| justification += "..." |
| justification += ")" |
| return AssigneeRecommendation(names, justification) |
| else: |
| parent = os.path.dirname(dirToCheck) |
| if len(parent) >= len(dirToCheck): |
| break |
| dirToCheck = parent |
| |
| |
| class Triager(object): |
| def __init__(self, fileFinder): |
| self.recommenderRules = self.parseKnownOwners({ |
| "fragment": ["ilake", "mount", "adamp"], |
| "animation": ["mount", "tianliu"], |
| "transition": ["mount"], |
| "theme": ["alanv"], |
| "style": ["alanv"], |
| "preferences": ["pavlis", "lpf"], |
| "ViewPager": ["jgielzak", "jellefresen"], |
| "DrawerLayout": ["sjgilbert"], |
| "RecyclerView": ["shepshapard", "ryanmentley"], |
| "Loaders": ["ilake"], |
| "VectorDrawableCompat": ["tianliu"], |
| "AppCompat": ["kirillg"], |
| "Design Library": ["material-android-firehose"], |
| "android.support.design": ["material-android-firehose"], |
| "NavigationView": ["material-android-firehose"], # not to be confused with Navigation |
| "RenderThread": ["jreck"], |
| "VectorDrawable": ["tianliu"], |
| "Vector Drawable": ["tianliu"], |
| "drawable": ["alanv"], |
| "colorstatelist": ["alanv"], |
| "multilocale": ["nona", "mnita"], |
| "TextView": ["siyamed", "clarabayarri"], |
| "text": ["android-text"], |
| "emoji": ["android-text", "siyamed"], |
| "Linkify": ["android-text", "siyamed", "toki"], |
| "Spannable": ["android-text", "siyamed"], |
| "Minikin": ["android-text", "nona"], |
| "Fonts": ["android-text", "nona", "dougfelt"], |
| "freetype": ["android-text", "nona", "junkshik"], |
| "harfbuzz": ["android-text", "nona", "junkshik"], |
| "slice": ["madym"], |
| "checkApi": ["jeffrygaston", "aurimas"], |
| "compose": ["chuckj", "jsproch", "lelandr"], |
| "jetifier": ["pavlis", "jeffrygaston"], |
| "navigat": [components["navigation"]], # "navigation", "navigate", etc, |
| "room": ["danysantiago", "sergeyv", "yboyar"] |
| }) |
| self.recommenderRules.append(OwnersRule(fileFinder)) |
| self.recommenderRules.append(LastTouchedBy_Rule(fileFinder)) |
| |
| def parseKnownOwners(self, ownersDict): |
| rules = [] |
| keywords = sorted(ownersDict.keys()) |
| for keyword in keywords: |
| assignees = ownersDict[keyword] |
| rules.append(WordRule(keyword, assignees)) |
| return rules |
| |
| def process(self, lines): |
| issues = self.parseIssues(lines) |
| recognizedTriages = [] |
| unrecognizedTriages = [] |
| print("Analyzing " + str(len(issues)) + " issues") |
| for issue in issues: |
| print(".") |
| assigneeRecommendation = self.recommendAssignees(issue) |
| recommendationText = "?" |
| if assigneeRecommendation is not None: |
| usernames = assigneeRecommendation.usernames |
| if len(usernames) > 2: |
| usernames = usernames[:2] |
| recommendationText = str(usernames) + " (" + assigneeRecommendation.justification + ")" |
| recognizedTriages.append(("(" + issue.issueId + ") " + issue.description.replace("\t", "...."), recommendationText, )) |
| else: |
| unrecognizedTriages.append(("(" + issue.issueId + ") " + issue.description.replace("\t", "...."), recommendationText, )) |
| maxColumnWidth = 0 |
| allTriages = recognizedTriages + unrecognizedTriages |
| for item in allTriages: |
| maxColumnWidth = max(maxColumnWidth, len(item[0])) |
| for item in allTriages: |
| print(str(item[0]) + (" " * (maxColumnWidth - len(item[0]))) + " -> " + str(item[1])) |
| |
| def parseIssues(self, lines): |
| priority = "" |
| issueType = "" |
| description = "" |
| when = "" |
| |
| lines = [line.strip() for line in lines] |
| fields = [line for line in lines if line != ""] |
| linesPerIssue = 5 |
| if len(fields) % linesPerIssue != 0: |
| raise Exception("Parse error, number of lines must be divisible by " + str(linesPerIssue) + ", not " + str(len(fields)) + ". Last line: " + fields[-1]) |
| issues = [] |
| while len(fields) > 0: |
| priority = fields[0] |
| issueType = fields[1] |
| |
| middle = fields[2].split("\t") |
| expectedNumTabComponents = 3 |
| if len(middle) != expectedNumTabComponents: |
| raise Exception("Parse error: wrong number of tabs in " + str(middle) + ", got " + str(len(middle) - 1) + ", expected " + str(expectedNumTabComponents - 1)) |
| description = middle[0] |
| currentAssignee = middle[1] |
| status = middle[2] |
| |
| bottom = fields[4] |
| bottomSplit = bottom.split("\t") |
| expectedNumTabComponents = 2 |
| if len(bottomSplit) != expectedNumTabComponents: |
| raise Exception("Parse error: wrong number of tabs in " + str(bottomSplit) + ", got " + str(len(bottomSplit)) + ", expected " + str(expectedNumTabComponents - 1)) |
| issueId = bottomSplit[0] |
| when = bottomSplit[1] |
| |
| issues.append(Issue(issueId, description)) |
| fields = fields[linesPerIssue:] |
| return issues |
| |
| def recommendAssignees(self, issue): |
| overallRecommendation = None |
| for rule in self.recommenderRules: |
| thisRecommendation = rule.recommend(issue) |
| if thisRecommendation is not None: |
| if overallRecommendation is None: |
| overallRecommendation = thisRecommendation |
| else: |
| newRecommendation = overallRecommendation.intersect(thisRecommendation) |
| count = len(newRecommendation.usernames) |
| if count > 0 and count < len(overallRecommendation.usernames): |
| overallRecommendation = newRecommendation |
| return overallRecommendation |
| |
| |
| |
| def main(args): |
| if len(args) != 1: |
| usage() |
| fileFinder = FileFinder(os.path.dirname(os.path.dirname(args[0]))) |
| print("Reading issues (copy-paste from the hotlist) from stdin") |
| lines = sys.stdin.readlines() |
| triager = Triager(fileFinder) |
| triager.process(lines) |
| |
| |
| |
| |
| main(sys.argv) |
