path: root/python/fatcat_tools/reviewers/review_common.py

import datetime
import subprocess
import time
from collections import Counter
from typing import Any, List, Optional

import fatcat_openapi_client
from fatcat_openapi_client import ApiClient, Editgroup, EditgroupAnnotation, EntityEdit

"""
checks should return:
- status: pass, fail, warning, skipped
- description

annotation extra should include:
- useragent
- git_rev
- submit timestamp

can apply to individual entity revs, or to entire editgroup

reviewbot takes an editgroup (object) and returns an annotation (object)
"""


class CheckResult:
    status = None
    description = None
    ident = None
    rev = None
    check_type = None

    def __init__(
        self,
        status: str,
        check_type: Optional[str] = None,
        description: Optional[str] = None,
        **kwargs
    ):
        self.status = status
        self.check_type = check_type
        self.description = description
        self.ident = kwargs.get("ident")
        self.rev = kwargs.get("rev")

    def __repr__(self):
        return str(self.__dict__)


class EditCheck:

    scope: List[Any] = []
    name: Optional[str] = None

    def check_editgroup(self, editgroup: fatcat_openapi_client.Editgroup) -> CheckResult:
        raise NotImplementedError

    def check_container(
        self,
        edit: EntityEdit,
        entity: fatcat_openapi_client.ContainerEntity,
    ) -> CheckResult:
        raise NotImplementedError

    def check_creator(
        self,
        edit: EntityEdit,
        entity: fatcat_openapi_client.CreatorEntity,
    ) -> CheckResult:
        raise NotImplementedError

    def check_file(
        self,
        edit: EntityEdit,
        entity: fatcat_openapi_client.FileEntity,
    ) -> CheckResult:
        raise NotImplementedError

    def check_fileset(
        self,
        edit: EntityEdit,
        entity: fatcat_openapi_client.FilesetEntity,
    ) -> CheckResult:
        raise NotImplementedError

    def check_webcapture(
        self,
        edit: EntityEdit,
        entity: fatcat_openapi_client.WebcaptureEntity,
    ) -> CheckResult:
        raise NotImplementedError

    def check_release(
        self,
        edit: EntityEdit,
        entity: fatcat_openapi_client.ReleaseEntity,
    ) -> CheckResult:
        raise NotImplementedError

    def check_work(
        self,
        edit: EntityEdit,
        work: fatcat_openapi_client.WorkEntity,
    ) -> CheckResult:
        raise NotImplementedError


class ReviewBot:
    def __init__(self, api: fatcat_openapi_client.ApiClient, verbose: bool = False, **kwargs):

        self.api = api
        self.checks: List[EditCheck] = []
        self.verbose = verbose
        self.extra = kwargs.get("extra", dict())
        self.extra["git_rev"] = self.extra.get(
            "git_rev", subprocess.check_output(["git", "describe", "--always"]).strip()
        ).decode("utf-8")
        self.extra["agent"] = self.extra.get("agent", "fatcat_tools.ReviewBot")
        self.poll_interval = kwargs.get("poll_interval", 10.0)

    def run_single(self, editgroup_id: str, annotate: bool = True) -> CheckResult:
        eg = self.api.get_editgroup(editgroup_id)
        annotation = self.review_editgroup(eg)
        if annotate:
            self.api.create_editgroup_annotation(eg.editgroup_id, annotation)
        return annotation

    def run(self, start_since: Optional[datetime.datetime] = None) -> None:
        if start_since is None:
            since = datetime.datetime.utcnow()
        else:
            since = start_since
        while True:
            # XXX: better isoformat conversion?
            eg_list = self.api.get_editgroups_reviewable(
                since=since.isoformat()[:19] + "Z", limit=100
            )
            if not eg_list:
                print("Sleeping {} seconds...".format(self.poll_interval))
                time.sleep(self.poll_interval)
                continue
            for eg in eg_list:
                # TODO: fetch annotations to ensure we haven't already annotated
                annotation = self.review_editgroup(eg)
                print(
                    "Reviewed {} disposition:{}".format(
                        eg.editgroup_id, annotation.extra["disposition"]
                    )
                )
                self.api.create_editgroup_annotation(eg.editgroup_id, annotation)
                since = eg.submitted
            # to prevent busy loops (TODO: needs review/rethink; multiple
            # editgroups in the same second)
            since = since + datetime.timedelta(seconds=1)

    def review_editgroup(self, editgroup: Editgroup) -> EditgroupAnnotation:
        results = self.run_checks(editgroup)
        result_counts = self.result_counts(results)
        disposition = self.disposition(results)
        if disposition == "accept":
            comment = "This editgroup looks great! All checks passed."
        elif disposition == "revise":
            comment = "Some issues were found, and changes or close review are recommended before accepting."
        elif disposition == "reject":
            comment = "Serious issues were found; this editgroup should **not** be accepted."
        else:
            raise ValueError

        for (status, title) in (("fail", "Failed check"), ("warning", "Warnings")):
            if result_counts[status] > 0:
                comment += "\n\n### {} ({}):\n".format(status, result_counts[status])
            for result in results:
                if result.status == status and result.check_type == "editgroup":
                    comment += "\n- {description}".format(description=result.description)
                if result.status == status and result.check_type != "editgroup":
                    comment += "\n- {check_type} [{rev}](/{entity_type}/rev/{rev}): {description}".format(
                        check_type=result.check_type,
                        rev=result.rev,
                        entity_type=result.check_type,
                        description=result.description,
                    )

        extra = self.extra.copy()
        extra.update(
            {
                "disposition": disposition,
                "submit_timestamp": editgroup.submitted.isoformat(),
                "checks": [check.name for check in self.checks],
                "result_counts": dict(result_counts),
            }
        )
        annotation = fatcat_openapi_client.EditgroupAnnotation(
            comment_markdown=comment,
            editgroup_id=editgroup.editgroup_id,
            extra=extra,
        )
        return annotation

    def result_counts(self, results: List[CheckResult]) -> Counter:
        counts: Counter = Counter()
        for result in results:
            counts["total"] += 1
            counts[result.status] += 1
        return counts

    def disposition(self, results: List[CheckResult]) -> str:
        """
        Returns one of: accept, revise, reject
        """
        raise NotImplementedError

    def run_checks(self, editgroup: Editgroup) -> List[CheckResult]:

        results = []

        # any full-editgroup checks
        for check in self.checks:
            if "editgroup" in check.scope:
                result = check.check_editgroup(editgroup)
                if self.verbose:
                    print(result)
                results.append(result)

        if not editgroup.edits:
            entity_edits = {}
        else:
            entity_edits = {
                "container": editgroup.edits.containers,
                "creator": editgroup.edits.creators,
                "file": editgroup.edits.files,
                "fileset": editgroup.edits.filesets,
                "webcapture": editgroup.edits.webcaptures,
                "release": editgroup.edits.releases,
                "work": editgroup.edits.works,
            }

        # entity-specific checks
        for entity_type, edits in entity_edits.items():
            for edit in edits:
                entity = None
                for check in self.checks:
                    if entity_type in check.scope:
                        # hack-y python munging
                        get_method = getattr(self.api, "get_{}_rev".format(entity_type))
                        check_method = getattr(check, "check_{}".format(entity_type))
                        entity = get_method(self.api, edit.rev)
                        result = check_method(check, edit, entity)
                        result.rev = edit.rev
                        result.ident = edit.ident
                        if self.verbose:
                            print(result)
                        results.append(result)

        return results


class DummyCheck(EditCheck):

    scope = ["editgroup", "work"]
    name = "DummyCheck"

    def check_editgroup(self, editgroup: Editgroup) -> CheckResult:
        return CheckResult(
            "pass",
            "editgroup",
            "every edit is precious, thanks [editor {editor_id}](/editor/{editor_id})!".format(
                editor_id=editgroup.editor_id
            ),
        )

    def check_work(
        self,
        edit: EntityEdit,
        work: fatcat_openapi_client.WorkEntity,
    ) -> CheckResult:
        return CheckResult("pass", "work", "this work edit is beautiful")


class DummyReviewBot(ReviewBot):
    """
    This bot reviews everything and always passes.
    """

    def __init__(self, api: ApiClient, **kwargs):
        super().__init__(api, **kwargs)
        self.checks = [DummyCheck()]

    def disposition(self, results: List[CheckResult]) -> str:
        return "accept"