tools/owners

#!/bin/env python3
import argparse
import json
import os
import re
import yaml

from sh import git  # type: ignore
from typing import Dict, List, Set, TypedDict, Optional
from yaml.loader import SafeLoader

# A list of Gerrit users (email addresses).
#   Some OWNERS files have empty lists for 'owners' or 'reviewers', which
#   results in a None type for the value.
UsersList = Optional[List[str]]

# A YAML node with an extra line number.
class NumberedNode(TypedDict):
    line_number: int


class MatchEntry(TypedDict, total=False):
    suffix: str
    regex: str
    partial_regex: str
    exact: str
    owners: UsersList
    reviewers: UsersList


# The root YAML node of an OWNERS file
class OwnersData(NumberedNode, TypedDict, total=False):
    owners: UsersList
    reviewers: UsersList
    matchers: List[MatchEntry]


# A YAML loader that adds the start line number onto each node (for
# later linting support)
class YamlLoader(SafeLoader):
    def construct_mapping(
        self, node: yaml.nodes.Node, deep: bool = False
    ) -> NumberedNode:
        mapping: NumberedNode = super(YamlLoader, self).construct_mapping(
            node, deep=deep
        )  # type: ignore
        mapping["line_number"] = node.start_mark.line + 1
        return mapping

    # Load a file and return the OwnersData.
    @staticmethod
    def load(file: str) -> OwnersData:
        data: OwnersData
        with open(file, "r") as f:
            data = yaml.load(f, Loader=YamlLoader)
        return data


# Class to match commit information with OWNERS files.
class CommitMatch:
    def __init__(
        self, args: argparse.Namespace, owners: Dict[str, OwnersData]
    ):
        files: Set[str] = set(
            git.bake("-C", args.path)
            .show(args.commit, pretty="", name_only=True, _tty_out=False)
            .splitlines()
        )

        root_owners = owners[""]

        self.owners: Set[str] = set()
        self.reviewers: Set[str] = set()

        for f in files:
            path = f

            while True:
                path = os.path.dirname(path)

                if path not in owners:
                    if not path:
                        break
                    continue

                local_owners = owners[path]

                self.owners = self.owners.union(
                    local_owners.get("owners") or []
                )
                self.reviewers = self.reviewers.union(
                    local_owners.get("reviewers") or []
                )

                rel_file = os.path.relpath(f, path)

                for e in local_owners.get("matchers", None) or []:
                    if "exact" in e:
                        self.__exact(rel_file, e)
                    elif "partial_regex" in e:
                        self.__partial_regex(rel_file, e)
                    elif "regex" in e:
                        self.__regex(rel_file, e)
                    elif "suffix" in e:
                        self.__suffix(rel_file, e)

                if not path:
                    break

        self.reviewers = self.reviewers.difference(self.owners)

    def __add_entry(self, entry: MatchEntry) -> None:
        self.owners = self.owners.union(entry.get("owners") or [])
        self.reviewers = self.reviewers.union(entry.get("reviewers") or [])

    def __exact(self, file: str, entry: MatchEntry) -> None:
        if file == entry["exact"]:
            self.__add_entry(entry)

    def __partial_regex(self, file: str, entry: MatchEntry) -> None:
        if re.search(entry["partial_regex"], file):
            self.__add_entry(entry)

    def __regex(self, file: str, entry: MatchEntry) -> None:
        if re.fullmatch(entry["regex"], file):
            self.__add_entry(entry)

    def __suffix(self, file: str, entry: MatchEntry) -> None:
        if os.path.splitext(file)[1] == entry["suffix"]:
            self.__add_entry(entry)


# The subcommand to get the reviewers.
def subcmd_reviewers(
    args: argparse.Namespace, data: Dict[str, OwnersData]
) -> None:
    matcher = CommitMatch(args, data)

    # Print in `git push refs/for/branch%<reviewers>` format.
    if args.push_args:
        result = []
        for o in sorted(matcher.owners):
            # Gerrit uses 'r' for the required reviewers (owners).
            result.append(f"r={o}")
        for r in sorted(matcher.reviewers):
            # Gerrit uses 'cc' for the optional reviewers.
            result.append(f"cc={r}")
        print(",".join(result))
    # Print as Gerrit Add Reviewers POST format.
    # https://gerrit.openbmc.org/Documentation/rest-api-changes.html#add-reviewer
    else:

        def review_js(reviewer: str, state: str) -> str:
            return json.dumps(
                {
                    "reviewer": reviewer,
                    "state": state,
                    "notify": "NONE",
                    "notify_details": {"TO": {"accounts": [reviewer]}},
                }
            )

        for o in sorted(matcher.owners):
            print(review_js(o, "REVIEWER"))
        for r in sorted(matcher.reviewers):
            print(review_js(r, "CC"))


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "-p", "--path", default=".", help="Root path to analyse"
    )
    subparsers = parser.add_subparsers()

    parser_reviewers = subparsers.add_parser(
        "reviewers", help="Generate List of Reviewers"
    )
    parser_reviewers.add_argument(
        "--push-args",
        default=False,
        action="store_true",
        help="Format as git push options",
    )
    parser_reviewers.add_argument(
        "--commit",
        default="HEAD",
        help="Commit(s) to match against",
    )
    parser_reviewers.set_defaults(func=subcmd_reviewers)

    args = parser.parse_args()

    owners_files = git.bake("-C", args.path)(
        "ls-files", "OWNERS", "**/OWNERS"
    ).splitlines()

    files = {}
    for f in owners_files:
        file = YamlLoader.load(os.path.join(args.path, f))
        dirpath = os.path.dirname(f)
        files[dirpath] = file

    args.func(args, files)


if __name__ == "__main__":
    main()