pr-agent/pr_agent/servers/github_action_runner.py

import asyncio
import json
import os

from pr_agent.agent.pr_agent import PRAgent
from pr_agent.config_loader import get_settings
from pr_agent.git_providers import get_git_provider
from pr_agent.git_providers.utils import apply_repo_settings
from pr_agent.log import get_logger
from pr_agent.tools.pr_code_suggestions import PRCodeSuggestions
from pr_agent.tools.pr_description import PRDescription
from pr_agent.tools.pr_reviewer import PRReviewer


async def run_action():
    # Get environment variables
    GITHUB_EVENT_NAME = os.environ.get('GITHUB_EVENT_NAME')
    GITHUB_EVENT_PATH = os.environ.get('GITHUB_EVENT_PATH')
    OPENAI_KEY = os.environ.get('OPENAI_KEY') or os.environ.get('OPENAI.KEY')
    OPENAI_ORG = os.environ.get('OPENAI_ORG') or os.environ.get('OPENAI.ORG')
    GITHUB_TOKEN = os.environ.get('GITHUB_TOKEN')
    CUSTOM_LABELS = os.environ.get('CUSTOM_LABELS')
    CUSTOM_LABELS_DESCRIPTIONS = os.environ.get('CUSTOM_LABELS_DESCRIPTIONS')
    # CUSTOM_LABELS is a comma separated list of labels (string), convert to list and strip spaces

    get_settings().set("CONFIG.PUBLISH_OUTPUT_PROGRESS", False)

    # Check if required environment variables are set
    if not GITHUB_EVENT_NAME:
        print("GITHUB_EVENT_NAME not set")
        return
    if not GITHUB_EVENT_PATH:
        print("GITHUB_EVENT_PATH not set")
        return
    if not OPENAI_KEY:
        print("OPENAI_KEY not set")
        return
    if not GITHUB_TOKEN:
        print("GITHUB_TOKEN not set")
        return
    CUSTOM_LABELS_DICT = handle_custom_labels(CUSTOM_LABELS, CUSTOM_LABELS_DESCRIPTIONS)

    # Set the environment variables in the settings
    get_settings().set("OPENAI.KEY", OPENAI_KEY)
    if OPENAI_ORG:
        get_settings().set("OPENAI.ORG", OPENAI_ORG)
    get_settings().set("GITHUB.USER_TOKEN", GITHUB_TOKEN)
    get_settings().set("GITHUB.DEPLOYMENT_TYPE", "user")
    get_settings().set("CUSTOM_LABELS", CUSTOM_LABELS_DICT)

    # Load the event payload
    try:
        with open(GITHUB_EVENT_PATH, 'r') as f:
            event_payload = json.load(f)
    except json.decoder.JSONDecodeError as e:
        print(f"Failed to parse JSON: {e}")
        return

    # Handle pull request event
    if GITHUB_EVENT_NAME == "pull_request":
        action = event_payload.get("action")

        try:
            get_logger().info("Applying repo settings")
            pr_url = event_payload.get("pull_request", {}).get("html_url")
            if pr_url:
                apply_repo_settings(pr_url)
        except Exception as e:
            get_logger().info(f"github action: failed to apply repo settings: {e}")

        if action in ["opened", "reopened"]:
            pr_url = event_payload.get("pull_request", {}).get("url")
            if pr_url:
                auto_review = os.environ.get('github_action.auto_review', None)
                if auto_review is None or (isinstance(auto_review, str) and auto_review.lower() == 'true'):
                    await PRReviewer(pr_url).run()
                auto_describe = os.environ.get('github_action.auto_describe', None)
                if isinstance(auto_describe, str) and auto_describe.lower() == 'true':
                    await PRDescription(pr_url).run()
                auto_improve = os.environ.get('github_action.auto_improve', None)
                if isinstance(auto_improve, str) and auto_improve.lower() == 'true':
                    await PRCodeSuggestions(pr_url).run()

    # Handle issue comment event
    elif GITHUB_EVENT_NAME == "issue_comment":
        action = event_payload.get("action")
        if action in ["created", "edited"]:
            comment_body = event_payload.get("comment", {}).get("body")
            if comment_body:
                is_pr = False
                # check if issue is pull request
                if event_payload.get("issue", {}).get("pull_request"):
                    url = event_payload.get("issue", {}).get("pull_request", {}).get("url")
                    is_pr = True
                else:
                    url = event_payload.get("issue", {}).get("url")
                if url:
                    body = comment_body.strip().lower()
                    comment_id = event_payload.get("comment", {}).get("id")
                    provider = get_git_provider()(pr_url=url)
                    if is_pr:
                        await PRAgent().handle_request(url, body, notify=lambda: provider.add_eyes_reaction(comment_id))
                    else:
                        await PRAgent().handle_request(url, body)


def handle_custom_labels(CUSTOM_LABELS, CUSTOM_LABELS_DESCRIPTIONS):
    if CUSTOM_LABELS:
        CUSTOM_LABELS = [x.strip() for x in CUSTOM_LABELS.split(',')]
    else:
        # Set default labels
        CUSTOM_LABELS = ['Bug fix', 'Tests', 'Bug fix with tests', 'Refactoring', 'Enhancement', 'Documentation',
                         'Other']
        print(f"Using default labels: {CUSTOM_LABELS}")
    if CUSTOM_LABELS_DESCRIPTIONS:
        CUSTOM_LABELS_DESCRIPTIONS = [x.strip() for x in CUSTOM_LABELS_DESCRIPTIONS.split(',')]
    else:
        # Set default labels
        CUSTOM_LABELS_DESCRIPTIONS = ['Fixes a bug in the code', 'Adds or modifies tests',
                                     'Fixes a bug in the code and adds or modifies tests',
                                     'Refactors the code without changing its functionality',
                                     'Adds new features or functionality',
                                     'Adds or modifies documentation',
                                     'Other changes that do not fit in any of the above categories']
        print(f"Using default labels: {CUSTOM_LABELS_DESCRIPTIONS}")
    # create a dictionary of labels and descriptions
    CUSTOM_LABELS_DICT = dict()
    for i in range(len(CUSTOM_LABELS)):
        CUSTOM_LABELS_DICT[CUSTOM_LABELS[i]] = {'description': CUSTOM_LABELS_DESCRIPTIONS[i]}
    return CUSTOM_LABELS_DICT


if __name__ == '__main__':
    asyncio.run(run_action())