From 2ab87d507faa6d61ef01ce458ee71aa5b05ccc0f Mon Sep 17 00:00:00 2001 From: meeks Date: Thu, 16 Jul 2026 13:54:52 +0200 Subject: [PATCH] refactor: extract hardcoded values to config (section 5.3) - Add AGENT_USERNAMES and GITEA_ORG_FILTER settings to gitea/config.py - Update core/dispatcher.py to use AGENT_USERNAMES from config - Update gitea/client.py to use GITEA_ORG_FILTER in list_all_user_repos() and list_unread_notifications() --- core/dispatcher.py | 4 +- gitea/client.py | 145 ++++++++++++++++++++++++++++++++++----------- gitea/config.py | 7 ++- 3 files changed, 115 insertions(+), 41 deletions(-) diff --git a/core/dispatcher.py b/core/dispatcher.py index af69d02..1911f89 100644 --- a/core/dispatcher.py +++ b/core/dispatcher.py @@ -20,7 +20,7 @@ from gitea.tools.git_tools import GitTools from gitea.client import GiteaClient from core.coordinator_tools import CoordinatorTools from gitea.workspace import WorkspaceManager -from gitea.config import AGENT_MODEL_ID +from gitea.config import AGENT_MODEL_ID, AGENT_USERNAMES from gitea.models import ( CommentModel, PullRequestFileModel, @@ -79,7 +79,7 @@ def _is_awaiting_reply_helper(comments: list[CommentModel], ai_username: str) -> return False # Find the last agent comment index last_agent_idx: int = -1 - agent_usernames = {ai_username, "agent-bot"} + agent_usernames = {ai_username, *AGENT_USERNAMES} for i, c in enumerate(comments): if c.user and c.user.login in agent_usernames: last_agent_idx = i diff --git a/gitea/client.py b/gitea/client.py index 8ea1dba..e3b4174 100644 --- a/gitea/client.py +++ b/gitea/client.py @@ -6,7 +6,7 @@ from typing import Any, Optional logger: logging.Logger = logging.getLogger("gitea.client") -from .config import GITEA_URL, GITEA_TOKEN +from .config import GITEA_URL, GITEA_TOKEN, GITEA_ORG_FILTER from .models import ( UserModel, LabelModel, @@ -17,6 +17,7 @@ from .models import ( PullRequestFileModel, ) + class GiteaClient: """HTTP client for Gitea API v1.""" @@ -59,12 +60,16 @@ class GiteaClient: response = self.client.get(url) response.raise_for_status() repos: list[dict[str, Any]] = response.json() - # Filter to ONLY meeks organization repos, include mirrors + # Filter to ONLY configured organization repos, include mirrors seen: set[str] = set() result: list[RepositoryModel] = [] for r in repos: full_name = r.get("full_name", "") - if full_name and full_name not in seen and (r.get("owner") or {}).get("login") == "meeks": + if ( + full_name + and full_name not in seen + and (r.get("owner") or {}).get("login") == GITEA_ORG_FILTER + ): seen.add(full_name) result.append(RepositoryModel(**r)) return result @@ -72,13 +77,17 @@ class GiteaClient: logger.error(f"Error listing user repos: {e}", exc_info=True) return [] - def list_repo_issues(self, owner: str, repo: str, state: str = "open") -> list[IssueModel]: + def list_repo_issues( + self, owner: str, repo: str, state: str = "open" + ) -> list[IssueModel]: url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues?type=issues&state={state}" response = self.client.get(url) response.raise_for_status() return [IssueModel(**item) for item in response.json()] - def list_repo_pull_requests(self, owner: str, repo: str, state: str = "open") -> list[PullRequestModel]: + def list_repo_pull_requests( + self, owner: str, repo: str, state: str = "open" + ) -> list[PullRequestModel]: url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls?state={state}" response = self.client.get(url) if response.status_code == 404: @@ -86,7 +95,9 @@ class GiteaClient: response.raise_for_status() return [PullRequestModel(**item) for item in response.json()] - def get_pull_request(self, owner: str, repo: str, pull_number: int) -> PullRequestModel: + def get_pull_request( + self, owner: str, repo: str, pull_number: int + ) -> PullRequestModel: url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}" response = self.client.get(url) response.raise_for_status() @@ -105,21 +116,29 @@ class GiteaClient: response.raise_for_status() return IssueModel(**response.json()) - def close_pull_request(self, owner: str, repo: str, pull_number: int) -> PullRequestModel: + def close_pull_request( + self, owner: str, repo: str, pull_number: int + ) -> PullRequestModel: url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}" data: dict[str, str] = {"state": "closed"} response = self.client.patch(url, json=data) response.raise_for_status() return PullRequestModel(**response.json()) - def get_issue_comments(self, owner: str, repo: str, issue_number: int) -> list[CommentModel]: + def get_issue_comments( + self, owner: str, repo: str, issue_number: int + ) -> list[CommentModel]: url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}/comments" response = self.client.get(url) response.raise_for_status() return [CommentModel(**item) for item in response.json()] - def get_pull_request_comments(self, owner: str, repo: str, pull_number: int) -> list[CommentModel]: - url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{pull_number}/comments" + def get_pull_request_comments( + self, owner: str, repo: str, pull_number: int + ) -> list[CommentModel]: + url = ( + f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{pull_number}/comments" + ) response = self.client.get(url) response.raise_for_status() return [CommentModel(**item) for item in response.json()] @@ -136,7 +155,9 @@ class GiteaClient: response.raise_for_status() return response.text - def get_pull_request_files(self, owner: str, repo: str, pull_number: int) -> list[PullRequestFileModel]: + def get_pull_request_files( + self, owner: str, repo: str, pull_number: int + ) -> list[PullRequestFileModel]: url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}/files" response = self.client.get(url) response.raise_for_status() @@ -174,7 +195,9 @@ class GiteaClient: logger.error(f"Error listing assigned issues: {e}", exc_info=True) return [] - def list_assigned_pull_requests(self, owner: str = "", repo: str = "") -> list[PullRequestModel]: + def list_assigned_pull_requests( + self, owner: str = "", repo: str = "" + ) -> list[PullRequestModel]: """List all pull requests assigned to or authored by the authenticated user.""" try: user = self.get_authenticated_user() @@ -188,10 +211,14 @@ class GiteaClient: if response.status_code == 404: return [] response.raise_for_status() - all_prs: list[PullRequestModel] = [PullRequestModel(**pr) for pr in response.json()] + all_prs: list[PullRequestModel] = [ + PullRequestModel(**pr) for pr in response.json() + ] return [ - pr for pr in all_prs - if (pr.assignee and pr.assignee.login == username) or (pr.user and pr.user.login == username) + pr + for pr in all_prs + if (pr.assignee and pr.assignee.login == username) + or (pr.user and pr.user.login == username) ] all_prs: list[PullRequestModel] = [] repos = self.list_all_user_repos() @@ -204,7 +231,9 @@ class GiteaClient: if resp.status_code == 200: for pr_data in resp.json(): pr = PullRequestModel(**pr_data) - if (pr.assignee and pr.assignee.login == username) or (pr.user and pr.user.login == username): + if (pr.assignee and pr.assignee.login == username) or ( + pr.user and pr.user.login == username + ): # Backfill repository if Gitea omitted it if pr.repository is None: pr = pr.model_copy(update={"repository": r}) @@ -215,7 +244,13 @@ class GiteaClient: return [] def create_pull_request( - self, owner: str, repo: str, head: str, base: str, title: str, description: str = "" + self, + owner: str, + repo: str, + head: str, + base: str, + title: str, + description: str = "", ) -> PullRequestModel: try: url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls" @@ -262,21 +297,27 @@ class GiteaClient: ) -> PullRequestModel: return self.create_pull_request(owner, repo, head, base, title, description) - def approve_pr(self, owner: str, repo: str, pr_number: int, comment: str) -> dict[str, Any]: + def approve_pr( + self, owner: str, repo: str, pr_number: int, comment: str + ) -> dict[str, Any]: url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pr_number}/reviews" data: dict[str, Any] = {"event": "APPROVED", "body": comment} response = self.client.post(url, json=data) response.raise_for_status() return response.json() - def request_changes_pr(self, owner: str, repo: str, pr_number: int, comment: str) -> dict[str, Any]: + def request_changes_pr( + self, owner: str, repo: str, pr_number: int, comment: str + ) -> dict[str, Any]: url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pr_number}/reviews" data: dict[str, Any] = {"event": "REQUEST_CHANGES", "body": comment} response = self.client.post(url, json=data) response.raise_for_status() return response.json() - def get_pr_reviews(self, owner: str, repo: str, pr_number: int) -> list[dict[str, Any]]: + def get_pr_reviews( + self, owner: str, repo: str, pr_number: int + ) -> list[dict[str, Any]]: url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pr_number}/reviews" response = self.client.get(url) if response.status_code == 404: @@ -293,7 +334,9 @@ class GiteaClient: response.raise_for_status() return response.json() - def assign_issue(self, owner: str, repo: str, issue_number: int, username: str) -> IssueModel: + def assign_issue( + self, owner: str, repo: str, issue_number: int, username: str + ) -> IssueModel: url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}" data: dict[str, list[str]] = {"assignees": [username]} response = self.client.patch(url, json=data) @@ -319,21 +362,29 @@ class GiteaClient: response.raise_for_status() return IssueModel(**response.json()) - def add_comment(self, owner: str, repo: str, issue_number: int, body: str) -> CommentModel: + def add_comment( + self, owner: str, repo: str, issue_number: int, body: str + ) -> CommentModel: url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}/comments" data: dict[str, str] = {"body": body} response = self.client.post(url, json=data) response.raise_for_status() return CommentModel(**response.json()) - def add_label(self, owner: str, repo: str, issue_number: int, label: str) -> LabelModel: - url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}/labels" + def add_label( + self, owner: str, repo: str, issue_number: int, label: str + ) -> LabelModel: + url = ( + f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}/labels" + ) data: list[str] = [label] response = self.client.post(url, json=data) response.raise_for_status() return LabelModel(**response.json()) - def add_label_pr(self, owner: str, repo: str, pr_number: int, label: str) -> LabelModel: + def add_label_pr( + self, owner: str, repo: str, pr_number: int, label: str + ) -> LabelModel: url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{pr_number}/labels" data: list[str] = [label] response = self.client.post(url, json=data) @@ -368,17 +419,27 @@ class GiteaClient: response.raise_for_status() return response.json() - def get_file_content(self, owner: str, repo: str, path: str, ref: str = "master") -> str | list[str]: + def get_file_content( + self, owner: str, repo: str, path: str, ref: str = "master" + ) -> str | list[str]: url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/contents/{path}" params: dict[str, str] = {"ref": ref} response = self.client.get(url, params=params) response.raise_for_status() data = response.json() if isinstance(data, list): - return [item.get("content", "") for item in data if item.get("type") == "file"] - return base64.b64decode(data.get("content", "")).decode() if data.get("content") else "" + return [ + item.get("content", "") for item in data if item.get("type") == "file" + ] + return ( + base64.b64decode(data.get("content", "")).decode() + if data.get("content") + else "" + ) - def list_unread_notifications(self, since: Optional[str] = None) -> list[dict[str, Any]]: + def list_unread_notifications( + self, since: Optional[str] = None + ) -> list[dict[str, Any]]: try: url = f"{self.base_url}/api/v1/notifications" params: dict[str, str] = {"all": "false"} @@ -387,13 +448,13 @@ class GiteaClient: response = self.client.get(url, params=params) response.raise_for_status() notifications: list[dict[str, Any]] = response.json() - + result: list[dict[str, Any]] = [] for n in notifications: repo_info = n.get("repository") or {} owner_info = repo_info.get("owner") or {} owner_login = owner_info.get("login", "") - if owner_login == "meeks": + if owner_login == GITEA_ORG_FILTER: result.append(n) return result except Exception as e: @@ -407,14 +468,25 @@ class GiteaClient: response.raise_for_status() return True except Exception as e: - logger.error(f"Error marking notification thread {thread_id} as read: {e}", exc_info=True) + logger.error( + f"Error marking notification thread {thread_id} as read: {e}", + exc_info=True, + ) return False def merge_pull_request( - self, owner: str, repo: str, pull_number: int, style: str = "squash", title: str = "", message: str = "" + self, + owner: str, + repo: str, + pull_number: int, + style: str = "squash", + title: str = "", + message: str = "", ) -> bool: try: - url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}/merge" + url = ( + f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}/merge" + ) data: dict[str, Any] = { "Do": style, "MergeTitleField": title, @@ -424,6 +496,7 @@ class GiteaClient: response.raise_for_status() return True except Exception as e: - logger.error(f"Error merging pull request {pull_number}: {e}", exc_info=True) + logger.error( + f"Error merging pull request {pull_number}: {e}", exc_info=True + ) raise - diff --git a/gitea/config.py b/gitea/config.py index dadb996..a2e234b 100644 --- a/gitea/config.py +++ b/gitea/config.py @@ -13,6 +13,8 @@ class AgentSettings(BaseSettings): gitea_repos_root: str = "" agent_model_id: str = "qwen/qwen3.6-35b-a3b" agent_max_retries: int = 2 + agent_usernames: list[str] = ["agent-bot"] + gitea_org_filter: str = "meeks" searxng_url: str = "" searxng_username: str = "" searxng_password: str = "" @@ -31,9 +33,8 @@ GITEA_TOKEN: str = _agent_settings.gitea_token GITEA_REPOS_ROOT: str = _agent_settings.gitea_repos_root AGENT_MODEL_ID: str = _agent_settings.agent_model_id AGENT_MAX_RETRIES: int = _agent_settings.agent_max_retries +AGENT_USERNAMES: list[str] = _agent_settings.agent_usernames +GITEA_ORG_FILTER: str = _agent_settings.gitea_org_filter SEARXNG_URL: str = _agent_settings.searxng_url SEARXNG_USERNAME: str = _agent_settings.searxng_username SEARXNG_PASSWORD: str = _agent_settings.searxng_password - - -