diff --git a/pr_agent/git_providers/gitea_provider.py b/pr_agent/git_providers/gitea_provider.py index 0a166ff367..2161d6f22b 100644 --- a/pr_agent/git_providers/gitea_provider.py +++ b/pr_agent/git_providers/gitea_provider.py @@ -1,4 +1,3 @@ -import hashlib import json from typing import Any, Dict, List, Optional, Set, Tuple from urllib.parse import urlparse @@ -31,15 +30,15 @@ def __init__(self, url: Optional[str] = None): self.pr_url = "" self.issue_url = "" - gitea_access_token = get_settings().get("GITEA.PERSONAL_ACCESS_TOKEN", None) - if not gitea_access_token: + self.gitea_access_token = get_settings().get("GITEA.PERSONAL_ACCESS_TOKEN", None) + if not self.gitea_access_token: self.logger.error("Gitea access token not found in settings.") raise ValueError("Gitea access token not found in settings.") self.repo_settings = get_settings().get("GITEA.REPO_SETTING", None) configuration = giteapy.Configuration() configuration.host = "{}/api/v1".format(self.base_url) - configuration.api_key['Authorization'] = f'token {gitea_access_token}' + configuration.api_key['Authorization'] = f'token {self.gitea_access_token}' if get_settings().get("GITEA.SKIP_SSL_VERIFICATION", False): configuration.verify_ssl = False @@ -223,6 +222,19 @@ def get_pr_url(self) -> str: def get_issue_url(self) -> str: return self.issue_url + def get_latest_commit_url(self) -> str: + return self.last_commit.html_url + + def get_comment_url(self, comment) -> str: + return comment.html_url + + def publish_persistent_comment(self, pr_comment: str, + initial_header: str, + update_header: bool = True, + name='review', + final_update_message=True): + self.publish_persistent_comment_full(pr_comment, initial_header, update_header, name, final_update_message) + def publish_comment(self, comment: str,is_temporary: bool = False) -> None: """Publish a comment to the pull request""" if is_temporary and not get_settings().config.publish_output_progress: @@ -308,7 +320,7 @@ def publish_inline_comments(self, comments: List[Dict[str, Any]],body : str = "I if not response: self.logger.error("Failed to publish inline comment") - return None + return self.logger.info("Inline comment published") @@ -515,6 +527,13 @@ def get_line_link(self, relevant_file, relevant_line_start, relevant_line_end = self.logger.info(f"Generated link: {link}") return link + def get_pr_id(self): + try: + pr_id = f"{self.repo}/{self.pr_number}" + return pr_id + except: + return "" + def get_files(self) -> List[Dict[str, Any]]: """Get all files in the PR""" return [file.get("filename","") for file in self.git_files] @@ -551,7 +570,7 @@ def get_pr_branch(self) -> str: if not self.pr: self.logger.error("Failed to get PR branch") return "" - + if not self.pr.head: self.logger.error("PR head not found") return "" @@ -611,6 +630,9 @@ def is_supported(self, capability) -> bool: """Check if the provider is supported""" return True + def get_git_repo_url(self, issues_or_pr_url: str) -> str: + return f"{self.base_url}/{self.owner}/{self.repo}.git" #base_url / /.git + def publish_description(self, pr_title: str, pr_body: str) -> None: """Publish PR description""" response = self.repo_api.edit_pull_request( @@ -685,6 +707,35 @@ def remove_initial_comment(self) -> None: continue self.logger.info(f"Removed initial comment: {comment.get('comment_id')}") + #Clone related + def _prepare_clone_url_with_token(self, repo_url_to_clone: str) -> str | None: + #For example, to clone: + #https://github.com/Codium-ai/pr-agent-pro.git + #Need to embed inside the github token: + #https://@github.com/Codium-ai/pr-agent-pro.git + + gitea_token = self.gitea_access_token + gitea_base_url = self.base_url + scheme = gitea_base_url.split("://")[0] + scheme += "://" + if not all([gitea_token, gitea_base_url]): + get_logger().error("Either missing auth token or missing base url") + return None + base_url = gitea_base_url.split(scheme)[1] + if not base_url: + get_logger().error(f"Base url: {gitea_base_url} has an empty base url") + return None + if base_url not in repo_url_to_clone: + get_logger().error(f"url to clone: {repo_url_to_clone} does not contain {base_url}") + return None + repo_full_name = repo_url_to_clone.split(base_url)[-1] + if not repo_full_name: + get_logger().error(f"url to clone: {repo_url_to_clone} is malformed") + return None + + clone_url = scheme + clone_url += f"{gitea_token}@{base_url}{repo_full_name}" + return clone_url class RepoApi(giteapy.RepositoryApi): def __init__(self, client: giteapy.ApiClient): @@ -693,7 +744,7 @@ def __init__(self, client: giteapy.ApiClient): self.logger = get_logger() super().__init__(client) - def create_inline_comment(self, owner: str, repo: str, pr_number: int, body : str ,commit_id : str, comments: List[Dict[str, Any]]) -> None: + def create_inline_comment(self, owner: str, repo: str, pr_number: int, body : str ,commit_id : str, comments: List[Dict[str, Any]]): body = { "body": body, "comments": comments, diff --git a/pr_agent/servers/gitea_app.py b/pr_agent/servers/gitea_app.py index 018a746df9..4239cf51e2 100644 --- a/pr_agent/servers/gitea_app.py +++ b/pr_agent/servers/gitea_app.py @@ -1,6 +1,6 @@ -import asyncio import copy import os +import re from typing import Any, Dict from fastapi import APIRouter, FastAPI, HTTPException, Request, Response @@ -10,7 +10,9 @@ from starlette_context.middleware import RawContextMiddleware from pr_agent.agent.pr_agent import PRAgent +from pr_agent.algo.utils import update_settings_from_args from pr_agent.config_loader import get_settings, global_settings +from pr_agent.git_providers.utils import apply_repo_settings from pr_agent.log import LoggingFormat, get_logger, setup_logger from pr_agent.servers.utils import verify_signature @@ -50,7 +52,7 @@ async def get_body(request: Request): if not signature_header: get_logger().error("Missing signature header") raise HTTPException(status_code=400, detail="Missing signature header") - + try: verify_signature(body_bytes, webhook_secret, f"sha256={signature_header}") except Exception as ex: @@ -70,6 +72,9 @@ async def handle_request(body: Dict[str, Any], event: str): # Handle different event types if event == "pull_request": + if not should_process_pr_logic(body): + get_logger().debug(f"Request ignored: PR logic filtering") + return {} if action in ["opened", "reopened", "synchronized"]: await handle_pr_event(body, event, action, agent) elif event == "issue_comment": @@ -90,12 +95,21 @@ async def handle_pr_event(body: Dict[str, Any], event: str, action: str, agent: # Handle PR based on action if action in ["opened", "reopened"]: - commands = get_settings().get("gitea.pr_commands", []) - for command in commands: - await agent.handle_request(api_url, command) + # commands = get_settings().get("gitea.pr_commands", []) + await _perform_commands_gitea("pr_commands", agent, body, api_url) + # for command in commands: + # await agent.handle_request(api_url, command) elif action == "synchronized": # Handle push to PR - await agent.handle_request(api_url, "/review --incremental") + commands_on_push = get_settings().get(f"gitea.push_commands", {}) + handle_push_trigger = get_settings().get(f"gitea.handle_push_trigger", False) + if not commands_on_push or not handle_push_trigger: + get_logger().info("Push event, but no push commands found or push trigger is disabled") + return + get_logger().debug(f'A push event has been received: {api_url}') + await _perform_commands_gitea("push_commands", agent, body, api_url) + # for command in commands_on_push: + # await agent.handle_request(api_url, command) async def handle_comment_event(body: Dict[str, Any], event: str, action: str, agent: PRAgent): """Handle comment events""" @@ -113,6 +127,85 @@ async def handle_comment_event(body: Dict[str, Any], event: str, action: str, ag await agent.handle_request(pr_url, comment_body) +async def _perform_commands_gitea(commands_conf: str, agent: PRAgent, body: dict, api_url: str): + apply_repo_settings(api_url) + if commands_conf == "pr_commands" and get_settings().config.disable_auto_feedback: # auto commands for PR, and auto feedback is disabled + get_logger().info(f"Auto feedback is disabled, skipping auto commands for PR {api_url=}") + return + if not should_process_pr_logic(body): # Here we already updated the configuration with the repo settings + return {} + commands = get_settings().get(f"gitea.{commands_conf}") + if not commands: + get_logger().info(f"New PR, but no auto commands configured") + return + get_settings().set("config.is_auto_command", True) + for command in commands: + split_command = command.split(" ") + command = split_command[0] + args = split_command[1:] + other_args = update_settings_from_args(args) + new_command = ' '.join([command] + other_args) + get_logger().info(f"{commands_conf}. Performing auto command '{new_command}', for {api_url=}") + await agent.handle_request(api_url, new_command) + +def should_process_pr_logic(body) -> bool: + try: + pull_request = body.get("pull_request", {}) + title = pull_request.get("title", "") + pr_labels = pull_request.get("labels", []) + source_branch = pull_request.get("head", {}).get("ref", "") + target_branch = pull_request.get("base", {}).get("ref", "") + sender = body.get("sender", {}).get("login") + repo_full_name = body.get("repository", {}).get("full_name", "") + + # logic to ignore PRs from specific repositories + ignore_repos = get_settings().get("CONFIG.IGNORE_REPOSITORIES", []) + if ignore_repos and repo_full_name: + if any(re.search(regex, repo_full_name) for regex in ignore_repos): + get_logger().info(f"Ignoring PR from repository '{repo_full_name}' due to 'config.ignore_repositories' setting") + return False + + # logic to ignore PRs from specific users + ignore_pr_users = get_settings().get("CONFIG.IGNORE_PR_AUTHORS", []) + if ignore_pr_users and sender: + if any(re.search(regex, sender) for regex in ignore_pr_users): + get_logger().info(f"Ignoring PR from user '{sender}' due to 'config.ignore_pr_authors' setting") + return False + + # logic to ignore PRs with specific titles + if title: + ignore_pr_title_re = get_settings().get("CONFIG.IGNORE_PR_TITLE", []) + if not isinstance(ignore_pr_title_re, list): + ignore_pr_title_re = [ignore_pr_title_re] + if ignore_pr_title_re and any(re.search(regex, title) for regex in ignore_pr_title_re): + get_logger().info(f"Ignoring PR with title '{title}' due to config.ignore_pr_title setting") + return False + + # logic to ignore PRs with specific labels or source branches or target branches. + ignore_pr_labels = get_settings().get("CONFIG.IGNORE_PR_LABELS", []) + if pr_labels and ignore_pr_labels: + labels = [label['name'] for label in pr_labels] + if any(label in ignore_pr_labels for label in labels): + labels_str = ", ".join(labels) + get_logger().info(f"Ignoring PR with labels '{labels_str}' due to config.ignore_pr_labels settings") + return False + + # logic to ignore PRs with specific source or target branches + ignore_pr_source_branches = get_settings().get("CONFIG.IGNORE_PR_SOURCE_BRANCHES", []) + ignore_pr_target_branches = get_settings().get("CONFIG.IGNORE_PR_TARGET_BRANCHES", []) + if pull_request and (ignore_pr_source_branches or ignore_pr_target_branches): + if any(re.search(regex, source_branch) for regex in ignore_pr_source_branches): + get_logger().info( + f"Ignoring PR with source branch '{source_branch}' due to config.ignore_pr_source_branches settings") + return False + if any(re.search(regex, target_branch) for regex in ignore_pr_target_branches): + get_logger().info( + f"Ignoring PR with target branch '{target_branch}' due to config.ignore_pr_target_branches settings") + return False + except Exception as e: + get_logger().error(f"Failed 'should_process_pr_logic': {e}") + return True + # FastAPI app setup middleware = [Middleware(RawContextMiddleware)] app = FastAPI(middleware=middleware) diff --git a/pr_agent/settings/configuration.toml b/pr_agent/settings/configuration.toml index 5e21b4f8cc..dd2c3864e4 100644 --- a/pr_agent/settings/configuration.toml +++ b/pr_agent/settings/configuration.toml @@ -290,7 +290,7 @@ push_commands = [ # Configure SSL validation for GitLab. Can be either set to the path of a custom CA or disabled entirely. # ssl_verify = true -[gitea_app] +[gitea] url = "https://gitea.com" handle_push_trigger = false pr_commands = [ @@ -298,6 +298,10 @@ pr_commands = [ "/review", "/improve", ] +push_commands = [ + "/describe", + "/review", +] [bitbucket_app] pr_commands = [