feat: improvements for Gitea integration

siccous · siccous · commit bf1cc50ece66 · 2025-10-19T16:00:01.000+02:00
diff --git 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 / <OWNER>/<REPO>.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://<token>@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
@@ -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
@@ -290,14 +290,18 @@ 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 = [
     "/describe",
     "/review",
     "/improve",
 ]
+push_commands = [
+    "/describe",
+    "/review",
+]
 
 [bitbucket_app]
 pr_commands = [
