AI Code Reviewer

review.py

from typing import List, Any

import gitlab
import os
from itertools import dropwhile
import openai
from dataclasses import dataclass
import logging
logging.basicConfig(encoding='utf-8', level=logging.INFO)

@dataclass
class Diff:
    path: str
    diff: str


gl = gitlab.Gitlab(private_token=os.environ["PAT"])
openai.api_key = os.environ["OPENAI_API_KEY"]


def main():
    diffs, mr = get_diffs_from_mr()
    response = get_review(diffs)
    logging.info(response)
    mr.discussions.create({'body': response})


def get_review(diffs):
    user_message_line = ["Review the following code:"]
    for d in diffs:
        user_message_line.append(f"PATH: {d.path}; DIFF: {d.diff}")
    user_message = "\n".join(user_message_line)
    message = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[
            {
                "role": "system",
                "content": "You are a code reviewer on a Merge Request on Gitlab. Your responsibility is to review "
                           "the provided code and offer"
                           "recommendations for enhancement. Identify any problematic code snippets, "
                           "highlight potential issues, and evaluate the overall quality of the code you review. "
                           "You will be given input in the format PATH: <path of the file changed>; DIFF: <diff>. "
                           "In diffs, plus signs (+) will mean the line has been added and minus signs (-) will "
                           "mean that the line has been removed. Lines will be separated by \\n."
            },
            {
                "role": "user",
                "content": user_message
            }
        ],
    )
    response = message['choices'][0]['message']['content']
    return response


def get_diffs_from_mr() -> (List[Diff], Any):
    project = gl.projects.get(os.environ["CI_PROJECT_PATH"])
    mr = project.mergerequests.get(id=os.environ["CI_MERGE_REQUEST_IID"])
    changes = mr.changes()
    diffs = [Diff(c['new_path'], sanitize_diff_content(c['diff'])) for c in changes['changes']]
    return diffs, mr


def sanitize_diff_content(diff: str):
    return "".join(list(dropwhile(lambda x: x != "@", diff[2:]))[2:])


if __name__ == "__main__":
    main()