Авто ревью кода — это инструмент для автоматической проверки изменений в коде относительно ветки master (или другой указанной ветки). Инструмент использует ruff, mypy и pylint для статического анализа кода и позволяет выявлять замечания только для тех строк, которые были изменены.

auto_code_review.py

"""
Авто ревью кода

Настройка:

1. Добавить путь к файлу `export AUTO_CODE_REVIEW=ПутьФайлу` в `.zshrc`
1. Добавить путь к конфигурации `export RUFF_ROOT_CONFIG=ПутьФайлу` в `.zshrc`
2. Добавить путь к конфигурации `export MYPY_ROOT_CONFIG=ПутьФайлу` в `.zshrc`
3. Добавить путь к конфигурации `export PYLINT_ROOT_CONFIG=ПутьФайлу` в `.zshrc`

Использование:

2. Переходим в нужную папку в котрой есть `.git`
3. Выполняем `python $AUTO_CODE_REVIEW -v=3`

В итоге вы получите замечанию только к тем строкам, которые были изменены от master(можно указать другую в BRANCH_DIFF) 
"""

import argparse
import enum
import json
import os
import re
import subprocess  # noqa: S404
from pathlib import Path
from typing import Any, Callable, Optional

# По умолчанию текущая ветка
BRANCH_SOURCE = ""
# Ветка с которой проверяются изменения
BRANCH_DIFF = "master"
# Команды для запуска ruff, mypy, pylint
COMMAND_RUFF = "python3.11 -m ruff"
COMMAND_MYPY = "python3.11 -m mypy"
COMMAND_PYLINT = "python3.11 -m pylint"
# Конфиги для ruff, mypy, pylint
RUFF_ROOT_CONFIG = os.environ["RUFF_ROOT_CONFIG"]
MYPY_ROOT_CONFIG = os.environ["MYPY_ROOT_CONFIG"]
PYLINT_ROOT_CONFIG = os.environ["PYLINT_ROOT_CONFIG"]


def get_git_diff_array(branch_source: str, branch_diff: str) -> dict[str, list[int]]:
    """
    Получить изменения от ветки branch_diff относительно branch_source
    """
    # Хранятся имена файлов и номера строк с различиями
    git_diff_body: dict[str, list[int]] = {}
    diff_output = subprocess.run(
        f"git diff {branch_source} {branch_diff}".split(),  # noqa: S603
        capture_output=True,
        text=True,
        encoding="utf-8",
        check=False,
    )
    tmp_current_file: str = ""
    for line in diff_output.stdout.splitlines():
        # Игнорировать строки с различиями.
        if line.startswith(("+", "-")):
            continue
        # Когда начало нового файла.
        if line.startswith("diff --git"):
            match = re.search(r"b/(?P<file_name>.*)", line)
            if match:
                tmp_current_file = match.group(1)
                # Если это Python файл.
                if Path(tmp_current_file).suffix == ".py":
                    # Создать пустой список для нового файла.
                    git_diff_body[tmp_current_file] = []
        # Если файл удален то он не нужен.
        elif line.startswith("deleted file mode"):
            del git_diff_body[tmp_current_file]
        # Строки указанием изменений.
        elif tmp_current_file in git_diff_body and line.startswith("@@"):
            # Получить номера строк.
            match = re.search(r"@@ -\d+,\d+ \+(?P<start_row>\d+),(?P<end_row>\d+) @@", line)
            # Добавить строку в список изменений.
            if match:
                git_diff_body[tmp_current_file] = [int(match.group(1)) + i for i in range(int(match.group(2)))]

    # Оставить только не удаленные Python файлы
    return {k: v for k, v in git_diff_body.items() if v}


def run_check(
    git_diff_body: dict[str, list[int]],
    call_loads: Callable[[str], Any],
    command: str,
    *,
    only_diff: bool = True,
) -> dict[str, str]:
    """Выполнить проверку для файлов git_diff_body

    only_diff: True=Проверять файлы и строки только там, где есть изменения от master
    """
    check_warning: dict[str, str] = {
        # ИмяФайлаИСрока: Текст замечания
    }
    if git_diff_body:
        # Выполнить анализ кода для указанных файлов, вернуть ответ в формате json
        check_output = subprocess.run(
            command,
            shell=True,  # noqa: S602
            capture_output=True,
            text=True,
            encoding="utf-8",
            check=False,
        )
        # Если нет ошибок в выполнение команды
        if not check_output.stderr:
            # Парсим ответ в dict
            check_json = call_loads(check_output.stdout)
            for warning in check_json:
                # Относительный путь к файлу с замечанием
                relative_filepath: str = os.path.relpath(Path(warning["filename"]), Path(warning["filename"]).cwd())
                # Список строк которые были изменены от master
                list_rows_diff: Optional[list[int]] = git_diff_body.get(relative_filepath)
                if list_rows_diff:
                    # Начало строки где произошло замечание
                    start_locate: int = warning["location"]["row"]
                    # Конец строки где произошло замечание
                    end_location_warning: int = warning["end_location"]["row"]
                    # Есть ли пересечение заметания с тем что были изменено в MR
                    is_entry_warning: bool = any(
                        x in list_rows_diff for x in range(start_locate, end_location_warning + 1)
                    )
                    if is_entry_warning or not only_diff:
                        key: str = (
                            f"{relative_filepath}:{start_locate}:{warning['location']['column']}: {warning['code']}"
                        )
                        check_warning[key] = warning["message"]
            return check_warning
        msg = f"command output is error: {check_output.stderr}"
        raise ValueError(msg)
    msg = "Нет измененных файлов"
    raise FileNotFoundError(msg)


def mypy_loads(output_mypy: str) -> list[dict[str, Any]]:
    """Перевод ответа mypy в dict"""
    REGEX_PARSE_MYPY_COMPILE = re.compile(r"(?P<filename>[^:]+):(?P<row>\d+): (?P<message>\w+: [^\n]+)")

    return [
        {
            "filename": r["filename"].strip(),
            "location": {"row": int(r["row"]), "column": 0},
            "end_location": {"row": int(r["row"])},
            "message": r["message"],
            "code": "",
        }
        for r in REGEX_PARSE_MYPY_COMPILE.finditer(output_mypy)
    ]


def pylint_loads(output_pylint: str) -> list[dict[str, Any]]:
    """Перевод ответа pylint в dict"""
    return [
        {
            "filename": r["path"].strip(),
            "location": {"row": int(r["line"]), "column": int(r["column"])},
            "end_location": {"row": int(r["endLine"]) if r["endLine"] else int(r["line"])},
            "message": r["message"],
            "code": f'{r["message-id"]} {r["symbol"]}',
        }
        for r in json.loads(output_pylint)
    ]



class VariantRun(enum.Enum):
    """
    Перечисление для вариантов запуска автоматического код ревью.

    - `ruff` - выполнить только проверку кода с помощью инструмента ruff
    - `ruff_mypy` - выполнить проверку кода с помощью инструмента ruff и mypy
    """

    RUFF = 1
    RUFF_MYPY = 2
    RUFF_MYPY_PYLINT = 3


def main(only_diff: bool, variant: int) -> None:
    """
    Главная функция, которая выполняет автоматическое код-ревью.

    Аргументы:
        only_diff (bool): Если True, то проверяются только измененные файлы.
        variant (int): Вариант запуска автоматического код-ревью.

    Возвращает:
        None
    """
    git_diff_body = get_git_diff_array(BRANCH_SOURCE, BRANCH_DIFF)
    print("FILES: ", " ".join(git_diff_body.keys()), "\n")  # noqa: T201

    # Для ruff
    if variant >= VariantRun.RUFF.value:
        print("\nRuff:\n")  # noqa: T201
        # Шаг 2: Выполнить ruff check для измененных файлов
        ruff_warning = run_check(
            git_diff_body,
            only_diff=only_diff,
            command="{} check '{}' --output-format=json --config='{}'".format(
                COMMAND_RUFF,
                "' '".join(git_diff_body.keys()),
                RUFF_ROOT_CONFIG,
            ),
            call_loads=json.loads,
        )
        # Шаг 3: Вывести замечания которые относятся к MR
        for _filename, _warning in ruff_warning.items():
            print(f"{_filename} {_warning}")  # noqa: T201
        print(f"Found {len(ruff_warning)} errors")  # noqa: T201
        print("https://docs.astral.sh/ruff/rules/#refactor-r")  # noqa: T201

    # Для ruff+mypy
    if variant >= VariantRun.RUFF_MYPY.value:
        print("\nMypy:\n")  # noqa: T201
        # Шаг 2: Выполнить mypy для измененных файлов
        mypy_warning = run_check(
            git_diff_body,
            only_diff=only_diff,
            command="{} '{}' --config-file='{}'".format(
                COMMAND_MYPY,
                "' '".join(git_diff_body.keys()),
                MYPY_ROOT_CONFIG,
            ),
            call_loads=mypy_loads,
        )
        # Шаг 3: Вывести замечания которые относятся к MR
        for _filename, _warning in mypy_warning.items():
            print(f"{_filename} {_warning}")  # noqa: T201
        print(f"Found {len(mypy_warning)} errors")  # noqa: T201

    # Для ruff+mypy+pylint
    if variant >= VariantRun.RUFF_MYPY_PYLINT.value:
        print("\nPylint:\n")  # noqa: T201
        # Шаг 2: Выполнить mypy для измененных файлов
        pylint_warning = run_check(
            git_diff_body,
            only_diff=only_diff,
            command="{} '{}' --output-format=json --rcfile='{}'".format(
                COMMAND_PYLINT,
                "' '".join(git_diff_body.keys()),
                PYLINT_ROOT_CONFIG,
            ),
            call_loads=pylint_loads,
        )
        # Шаг 3: Вывести замечания которые относятся к MR
        for _filename, _warning in pylint_warning.items():
            print(f"{_filename} {_warning}")  # noqa: T201
        print(f"Found {len(pylint_warning)} errors")  # noqa: T201


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Пример использования флагов -only_diff и -v")
    parser.add_argument(
        "-only_diff",
        type=int,
        default=1,
        help="Установить флаг для отображения только отличий",
    )
    parser.add_argument(
        "-v",
        type=int,
        default=3,
        help="Целое значение: 1=ruff 2=ruff+mypy 3=ruff+mypy+pylint",
    )

    args = parser.parse_args()

    main(args.only_diff, args.v)