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()
This commit is contained in:
meeks
2026-07-16 13:54:52 +02:00
parent 9b63fbbcfc
commit 2ab87d507f
3 changed files with 115 additions and 41 deletions
+2 -2
View File
@@ -20,7 +20,7 @@ from gitea.tools.git_tools import GitTools
from gitea.client import GiteaClient from gitea.client import GiteaClient
from core.coordinator_tools import CoordinatorTools from core.coordinator_tools import CoordinatorTools
from gitea.workspace import WorkspaceManager 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 ( from gitea.models import (
CommentModel, CommentModel,
PullRequestFileModel, PullRequestFileModel,
@@ -79,7 +79,7 @@ def _is_awaiting_reply_helper(comments: list[CommentModel], ai_username: str) ->
return False return False
# Find the last agent comment index # Find the last agent comment index
last_agent_idx: int = -1 last_agent_idx: int = -1
agent_usernames = {ai_username, "agent-bot"} agent_usernames = {ai_username, *AGENT_USERNAMES}
for i, c in enumerate(comments): for i, c in enumerate(comments):
if c.user and c.user.login in agent_usernames: if c.user and c.user.login in agent_usernames:
last_agent_idx = i last_agent_idx = i
+109 -36
View File
@@ -6,7 +6,7 @@ from typing import Any, Optional
logger: logging.Logger = logging.getLogger("gitea.client") 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 ( from .models import (
UserModel, UserModel,
LabelModel, LabelModel,
@@ -17,6 +17,7 @@ from .models import (
PullRequestFileModel, PullRequestFileModel,
) )
class GiteaClient: class GiteaClient:
"""HTTP client for Gitea API v1.""" """HTTP client for Gitea API v1."""
@@ -59,12 +60,16 @@ class GiteaClient:
response = self.client.get(url) response = self.client.get(url)
response.raise_for_status() response.raise_for_status()
repos: list[dict[str, Any]] = response.json() 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() seen: set[str] = set()
result: list[RepositoryModel] = [] result: list[RepositoryModel] = []
for r in repos: for r in repos:
full_name = r.get("full_name", "") 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) seen.add(full_name)
result.append(RepositoryModel(**r)) result.append(RepositoryModel(**r))
return result return result
@@ -72,13 +77,17 @@ class GiteaClient:
logger.error(f"Error listing user repos: {e}", exc_info=True) logger.error(f"Error listing user repos: {e}", exc_info=True)
return [] 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}" url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues?type=issues&state={state}"
response = self.client.get(url) response = self.client.get(url)
response.raise_for_status() response.raise_for_status()
return [IssueModel(**item) for item in response.json()] 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}" url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls?state={state}"
response = self.client.get(url) response = self.client.get(url)
if response.status_code == 404: if response.status_code == 404:
@@ -86,7 +95,9 @@ class GiteaClient:
response.raise_for_status() response.raise_for_status()
return [PullRequestModel(**item) for item in response.json()] 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}" url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}"
response = self.client.get(url) response = self.client.get(url)
response.raise_for_status() response.raise_for_status()
@@ -105,21 +116,29 @@ class GiteaClient:
response.raise_for_status() response.raise_for_status()
return IssueModel(**response.json()) 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}" url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}"
data: dict[str, str] = {"state": "closed"} data: dict[str, str] = {"state": "closed"}
response = self.client.patch(url, json=data) response = self.client.patch(url, json=data)
response.raise_for_status() response.raise_for_status()
return PullRequestModel(**response.json()) 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" url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}/comments"
response = self.client.get(url) response = self.client.get(url)
response.raise_for_status() response.raise_for_status()
return [CommentModel(**item) for item in response.json()] return [CommentModel(**item) for item in response.json()]
def get_pull_request_comments(self, owner: str, repo: str, pull_number: int) -> list[CommentModel]: def get_pull_request_comments(
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{pull_number}/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 = self.client.get(url)
response.raise_for_status() response.raise_for_status()
return [CommentModel(**item) for item in response.json()] return [CommentModel(**item) for item in response.json()]
@@ -136,7 +155,9 @@ class GiteaClient:
response.raise_for_status() response.raise_for_status()
return response.text 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" url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}/files"
response = self.client.get(url) response = self.client.get(url)
response.raise_for_status() response.raise_for_status()
@@ -174,7 +195,9 @@ class GiteaClient:
logger.error(f"Error listing assigned issues: {e}", exc_info=True) logger.error(f"Error listing assigned issues: {e}", exc_info=True)
return [] 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.""" """List all pull requests assigned to or authored by the authenticated user."""
try: try:
user = self.get_authenticated_user() user = self.get_authenticated_user()
@@ -188,10 +211,14 @@ class GiteaClient:
if response.status_code == 404: if response.status_code == 404:
return [] return []
response.raise_for_status() 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 [ return [
pr for pr in all_prs pr
if (pr.assignee and pr.assignee.login == username) or (pr.user and pr.user.login == username) 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] = [] all_prs: list[PullRequestModel] = []
repos = self.list_all_user_repos() repos = self.list_all_user_repos()
@@ -204,7 +231,9 @@ class GiteaClient:
if resp.status_code == 200: if resp.status_code == 200:
for pr_data in resp.json(): for pr_data in resp.json():
pr = PullRequestModel(**pr_data) 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 # Backfill repository if Gitea omitted it
if pr.repository is None: if pr.repository is None:
pr = pr.model_copy(update={"repository": r}) pr = pr.model_copy(update={"repository": r})
@@ -215,7 +244,13 @@ class GiteaClient:
return [] return []
def create_pull_request( 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: ) -> PullRequestModel:
try: try:
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls" url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls"
@@ -262,21 +297,27 @@ class GiteaClient:
) -> PullRequestModel: ) -> PullRequestModel:
return self.create_pull_request(owner, repo, head, base, title, description) 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" url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pr_number}/reviews"
data: dict[str, Any] = {"event": "APPROVED", "body": comment} data: dict[str, Any] = {"event": "APPROVED", "body": comment}
response = self.client.post(url, json=data) response = self.client.post(url, json=data)
response.raise_for_status() response.raise_for_status()
return response.json() 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" url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pr_number}/reviews"
data: dict[str, Any] = {"event": "REQUEST_CHANGES", "body": comment} data: dict[str, Any] = {"event": "REQUEST_CHANGES", "body": comment}
response = self.client.post(url, json=data) response = self.client.post(url, json=data)
response.raise_for_status() response.raise_for_status()
return response.json() 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" url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pr_number}/reviews"
response = self.client.get(url) response = self.client.get(url)
if response.status_code == 404: if response.status_code == 404:
@@ -293,7 +334,9 @@ class GiteaClient:
response.raise_for_status() response.raise_for_status()
return response.json() 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}" url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}"
data: dict[str, list[str]] = {"assignees": [username]} data: dict[str, list[str]] = {"assignees": [username]}
response = self.client.patch(url, json=data) response = self.client.patch(url, json=data)
@@ -319,21 +362,29 @@ class GiteaClient:
response.raise_for_status() response.raise_for_status()
return IssueModel(**response.json()) 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" url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}/comments"
data: dict[str, str] = {"body": body} data: dict[str, str] = {"body": body}
response = self.client.post(url, json=data) response = self.client.post(url, json=data)
response.raise_for_status() response.raise_for_status()
return CommentModel(**response.json()) return CommentModel(**response.json())
def add_label(self, owner: str, repo: str, issue_number: int, label: str) -> LabelModel: def add_label(
url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}/labels" 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] data: list[str] = [label]
response = self.client.post(url, json=data) response = self.client.post(url, json=data)
response.raise_for_status() response.raise_for_status()
return LabelModel(**response.json()) 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" url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{pr_number}/labels"
data: list[str] = [label] data: list[str] = [label]
response = self.client.post(url, json=data) response = self.client.post(url, json=data)
@@ -368,17 +419,27 @@ class GiteaClient:
response.raise_for_status() response.raise_for_status()
return response.json() 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}" url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/contents/{path}"
params: dict[str, str] = {"ref": ref} params: dict[str, str] = {"ref": ref}
response = self.client.get(url, params=params) response = self.client.get(url, params=params)
response.raise_for_status() response.raise_for_status()
data = response.json() data = response.json()
if isinstance(data, list): if isinstance(data, list):
return [item.get("content", "") for item in data if item.get("type") == "file"] return [
return base64.b64decode(data.get("content", "")).decode() if data.get("content") else "" 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: try:
url = f"{self.base_url}/api/v1/notifications" url = f"{self.base_url}/api/v1/notifications"
params: dict[str, str] = {"all": "false"} params: dict[str, str] = {"all": "false"}
@@ -387,13 +448,13 @@ class GiteaClient:
response = self.client.get(url, params=params) response = self.client.get(url, params=params)
response.raise_for_status() response.raise_for_status()
notifications: list[dict[str, Any]] = response.json() notifications: list[dict[str, Any]] = response.json()
result: list[dict[str, Any]] = [] result: list[dict[str, Any]] = []
for n in notifications: for n in notifications:
repo_info = n.get("repository") or {} repo_info = n.get("repository") or {}
owner_info = repo_info.get("owner") or {} owner_info = repo_info.get("owner") or {}
owner_login = owner_info.get("login", "") owner_login = owner_info.get("login", "")
if owner_login == "meeks": if owner_login == GITEA_ORG_FILTER:
result.append(n) result.append(n)
return result return result
except Exception as e: except Exception as e:
@@ -407,14 +468,25 @@ class GiteaClient:
response.raise_for_status() response.raise_for_status()
return True return True
except Exception as e: 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 return False
def merge_pull_request( 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: ) -> bool:
try: 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] = { data: dict[str, Any] = {
"Do": style, "Do": style,
"MergeTitleField": title, "MergeTitleField": title,
@@ -424,6 +496,7 @@ class GiteaClient:
response.raise_for_status() response.raise_for_status()
return True return True
except Exception as e: 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 raise
+4 -3
View File
@@ -13,6 +13,8 @@ class AgentSettings(BaseSettings):
gitea_repos_root: str = "" gitea_repos_root: str = ""
agent_model_id: str = "qwen/qwen3.6-35b-a3b" agent_model_id: str = "qwen/qwen3.6-35b-a3b"
agent_max_retries: int = 2 agent_max_retries: int = 2
agent_usernames: list[str] = ["agent-bot"]
gitea_org_filter: str = "meeks"
searxng_url: str = "" searxng_url: str = ""
searxng_username: str = "" searxng_username: str = ""
searxng_password: str = "" searxng_password: str = ""
@@ -31,9 +33,8 @@ GITEA_TOKEN: str = _agent_settings.gitea_token
GITEA_REPOS_ROOT: str = _agent_settings.gitea_repos_root GITEA_REPOS_ROOT: str = _agent_settings.gitea_repos_root
AGENT_MODEL_ID: str = _agent_settings.agent_model_id AGENT_MODEL_ID: str = _agent_settings.agent_model_id
AGENT_MAX_RETRIES: int = _agent_settings.agent_max_retries 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_URL: str = _agent_settings.searxng_url
SEARXNG_USERNAME: str = _agent_settings.searxng_username SEARXNG_USERNAME: str = _agent_settings.searxng_username
SEARXNG_PASSWORD: str = _agent_settings.searxng_password SEARXNG_PASSWORD: str = _agent_settings.searxng_password