feat(gitea): implement gitea_coder role with scope enforcement
Implements the gitea_coder role with full file/branch operations and commit message generation. Features: - Branch creation with scope gating (prevents main pushes) - Enforces branch naming conventions (feature/, fix/, refactor/, etc.) - Generates detailed commit messages with ticket references - Creates PRs from branches - Reads ticket requirements from issues - Unified file operations workflow Technical: - Uses existing gitea/dev.py operations - Validates branch names against scope patterns - Auto-generates commit messages with issue references - Caches default branch per chat_id for session persistence Refs: #11
This commit is contained in:
958
gitea/coder.py
Normal file
958
gitea/coder.py
Normal file
@@ -0,0 +1,958 @@
|
||||
"""
|
||||
title: Gitea Coder - Development Workflow Role
|
||||
author: Jeff Smith + Claude + minimax
|
||||
version: 1.0.0
|
||||
license: MIT
|
||||
description: gitea_coder role for automated development workflows - reads tickets, creates branches, commits with detailed messages, and creates PRs
|
||||
requirements: pydantic, httpx
|
||||
changelog:
|
||||
1.0.0:
|
||||
- Initial implementation of gitea_coder role
|
||||
- Branch creation with scope gating
|
||||
- Commit message generation with ticket references
|
||||
- PR creation from feature branches
|
||||
- Ticket requirement reading
|
||||
- Scope enforcement (branch naming conventions)
|
||||
"""
|
||||
|
||||
from typing import Optional, Callable, Any, Dict, List
|
||||
from pydantic import BaseModel, Field
|
||||
import re
|
||||
import time
|
||||
|
||||
|
||||
class Tools:
|
||||
"""Gitea Coder - Development workflow automation role"""
|
||||
|
||||
class Valves(BaseModel):
|
||||
"""System-wide configuration for Gitea Coder integration"""
|
||||
|
||||
GITEA_URL: str = Field(
|
||||
default="https://gitea.example.com",
|
||||
description="Gitea server URL (ingress or internal service)",
|
||||
)
|
||||
DEFAULT_REPO: str = Field(
|
||||
default="",
|
||||
description="Default repository in owner/repo format",
|
||||
)
|
||||
DEFAULT_BRANCH: str = Field(
|
||||
default="main",
|
||||
description="Default branch name for operations",
|
||||
)
|
||||
ALLOW_USER_OVERRIDES: bool = Field(
|
||||
default=True,
|
||||
description="Allow users to override defaults via UserValves",
|
||||
)
|
||||
VERIFY_SSL: bool = Field(
|
||||
default=True,
|
||||
description="Verify SSL certificates",
|
||||
)
|
||||
|
||||
# Scope enforcement configuration
|
||||
ALLOWED_SCOPES: List[str] = Field(
|
||||
default=["feature", "fix", "refactor", "docs", "test", "chore"],
|
||||
description="Allowed branch scope prefixes",
|
||||
)
|
||||
PROTECTED_BRANCHES: List[str] = Field(
|
||||
default=["main", "master", "develop", "dev"],
|
||||
description="Branches that cannot be directly modified",
|
||||
)
|
||||
|
||||
class UserValves(BaseModel):
|
||||
"""Per-user configuration"""
|
||||
|
||||
GITEA_TOKEN: str = Field(
|
||||
default="",
|
||||
description="Your Gitea API token",
|
||||
)
|
||||
USER_DEFAULT_REPO: str = Field(
|
||||
default="",
|
||||
description="Override default repository",
|
||||
)
|
||||
USER_DEFAULT_BRANCH: str = Field(
|
||||
default="",
|
||||
description="Override default branch",
|
||||
)
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize with configuration"""
|
||||
self.valves = self.Valves()
|
||||
self.user_valves = self.UserValves()
|
||||
|
||||
# Cache for chat session data (branch, repo, etc.)
|
||||
self._chat_cache: Dict[str, dict] = {}
|
||||
self._cache_ttl = 3600 # 1 hour TTL for cache entries
|
||||
|
||||
# Reference to dev.py operations (will be set by framework)
|
||||
self._dev = None
|
||||
|
||||
def _get_token(self, __user__: dict = None) -> str:
|
||||
"""Extract Gitea token from user context"""
|
||||
if __user__ and "valves" in __user__:
|
||||
user_valves = __user__.get("valves")
|
||||
if user_valves:
|
||||
return user_valves.GITEA_TOKEN
|
||||
return ""
|
||||
|
||||
def _get_repo(self, repo: Optional[str], __user__: dict = None) -> str:
|
||||
"""Get effective repository"""
|
||||
if repo:
|
||||
return repo
|
||||
if __user__ and "valves" in __user__:
|
||||
user_valves = __user__.get("valves")
|
||||
if user_valves:
|
||||
if self.valves.ALLOW_USER_OVERRIDES and user_valves.USER_DEFAULT_REPO:
|
||||
return user_valves.USER_DEFAULT_REPO
|
||||
return self.valves.DEFAULT_REPO
|
||||
|
||||
def _get_branch(self, branch: Optional[str], __user__: dict = None) -> str:
|
||||
"""Get effective branch"""
|
||||
if branch:
|
||||
return branch
|
||||
if __user__ and "valves" in __user__:
|
||||
user_valves = __user__.get("valves")
|
||||
if user_valves:
|
||||
if self.valves.ALLOW_USER_OVERRIDES and user_valves.USER_DEFAULT_BRANCH:
|
||||
return user_valves.USER_DEFAULT_BRANCH
|
||||
return self.valves.DEFAULT_BRANCH
|
||||
|
||||
def _get_chat_cache_key(self, chat_id: str, repo: str) -> str:
|
||||
"""Generate cache key for chat session"""
|
||||
return f"{chat_id}:{repo}"
|
||||
|
||||
def _get_cached_data(self, chat_id: str, repo: str) -> Optional[dict]:
|
||||
"""Get cached data for chat session"""
|
||||
cache_key = self._get_chat_cache_key(chat_id, repo)
|
||||
if cache_key in self._chat_cache:
|
||||
data = self._chat_cache[cache_key]
|
||||
if time.time() - data.get("timestamp", 0) < self._cache_ttl:
|
||||
return data.get("value")
|
||||
else:
|
||||
# Expired, remove from cache
|
||||
del self._chat_cache[cache_key]
|
||||
return None
|
||||
|
||||
def _set_cached_data(self, chat_id: str, repo: str, key: str, value: Any):
|
||||
"""Set cached data for chat session"""
|
||||
cache_key = self._get_chat_cache_key(chat_id, repo)
|
||||
if cache_key not in self._chat_cache:
|
||||
self._chat_cache[cache_key] = {"timestamp": time.time(), "value": {}}
|
||||
self._chat_cache[cache_key]["value"][key] = value
|
||||
self._chat_cache[cache_key]["timestamp"] = time.time()
|
||||
|
||||
def _validate_branch_name(self, branch_name: str) -> tuple[bool, str]:
|
||||
"""
|
||||
Validate branch name against scope conventions.
|
||||
|
||||
Returns:
|
||||
tuple: (is_valid, error_message)
|
||||
"""
|
||||
# Check for protected branches
|
||||
if branch_name in self.valves.PROTECTED_BRANCHES:
|
||||
return False, f"Branch '{branch_name}' is protected. Create a feature branch instead."
|
||||
|
||||
# Check if it starts with a valid scope
|
||||
scope_pattern = f"^({'|'.join(self.valves.ALLOWED_SCOPES)})/"
|
||||
if not re.match(scope_pattern, branch_name):
|
||||
scopes = ", ".join([f"{s}/" for s in self.valves.ALLOWED_SCOPES])
|
||||
return False, f"Branch name must start with a valid scope. Allowed: {scopes}"
|
||||
|
||||
# Validate remaining branch name
|
||||
if not re.match(r"^[a-zA-Z0-9_-]+(/[a-zA-Z0-9_-]+)*$", branch_name):
|
||||
return False, "Branch name contains invalid characters. Use alphanumeric, hyphens, underscores, and forward slashes only."
|
||||
|
||||
return True, ""
|
||||
|
||||
def _generate_commit_message(
|
||||
self,
|
||||
scope: str,
|
||||
short_description: str,
|
||||
issue_id: Optional[str] = None,
|
||||
body: str = "",
|
||||
footer: str = "",
|
||||
) -> str:
|
||||
"""
|
||||
Generate a detailed commit message following conventional commits format.
|
||||
|
||||
Args:
|
||||
scope: The scope of the change (feature, fix, etc.)
|
||||
short_description: Brief description of the change
|
||||
issue_id: Issue/ticket number to reference
|
||||
body: Detailed explanation of the change
|
||||
footer: Breaking changes or issue references
|
||||
|
||||
Returns:
|
||||
str: Formatted commit message
|
||||
"""
|
||||
# Build the header
|
||||
message = f"{scope}({scope}): {short_description}"
|
||||
|
||||
# Add issue reference
|
||||
if issue_id:
|
||||
message += f"\n\nRefs: #{issue_id}"
|
||||
|
||||
# Add body if provided
|
||||
if body:
|
||||
message += f"\n\n{body}"
|
||||
|
||||
# Add footer if provided
|
||||
if footer:
|
||||
message += f"\n\n{footer}"
|
||||
|
||||
return message
|
||||
|
||||
def _extract_issue_info(self, issue_text: str) -> dict:
|
||||
"""
|
||||
Extract key information from issue text.
|
||||
|
||||
Args:
|
||||
issue_text: Raw issue text from Gitea
|
||||
|
||||
Returns:
|
||||
dict: Extracted information (title, body, labels, etc.)
|
||||
"""
|
||||
info = {
|
||||
"title": "",
|
||||
"body": "",
|
||||
"type": "feature",
|
||||
"has_testing": False,
|
||||
"has_documentation": False,
|
||||
}
|
||||
|
||||
lines = issue_text.split("\n")
|
||||
current_section = ""
|
||||
|
||||
for line in lines:
|
||||
line_lower = line.lower().strip()
|
||||
|
||||
# Detect title
|
||||
if line.startswith("# ") and not info["title"]:
|
||||
info["title"] = line[2:].strip()
|
||||
# Detect sections
|
||||
elif line.startswith("## "):
|
||||
current_section = line[3:].lower()
|
||||
# Extract labels/tags from body
|
||||
elif current_section == "description" and line.startswith("- [ ]"):
|
||||
task = line[4:].strip()
|
||||
if "test" in task.lower():
|
||||
info["has_testing"] = True
|
||||
if "doc" in task.lower():
|
||||
info["has_documentation"] = True
|
||||
# Detect issue type from labels
|
||||
elif line_lower.startswith("**labels:**"):
|
||||
labels_str = line.split(":", 1)[1].strip().lower()
|
||||
if "bug" in labels_str or "fix" in labels_str:
|
||||
info["type"] = "fix"
|
||||
elif "docs" in labels_str or "documentation" in labels_str:
|
||||
info["type"] = "docs"
|
||||
elif "test" in labels_str:
|
||||
info["type"] = "test"
|
||||
|
||||
# Extract body content
|
||||
body_match = re.search(r"## Description\s*\n(.*?)(?=\n## |\n# |\Z)", issue_text, re.DOTALL)
|
||||
if body_match:
|
||||
info["body"] = body_match.group(1).strip()
|
||||
|
||||
return info
|
||||
|
||||
async def create_feature_branch(
|
||||
self,
|
||||
branch_name: str,
|
||||
issue_number: Optional[int] = None,
|
||||
from_branch: Optional[str] = None,
|
||||
repo: Optional[str] = None,
|
||||
chat_id: Optional[str] = None,
|
||||
__user__: dict = None,
|
||||
__event_emitter__: Callable[[dict], Any] = None,
|
||||
) -> str:
|
||||
"""
|
||||
Create a new feature branch with scope validation.
|
||||
|
||||
This function enforces branch naming conventions and prevents
|
||||
direct modifications to protected branches like main.
|
||||
|
||||
Args:
|
||||
branch_name: Name for the new branch (e.g., 'feature/42-add-login')
|
||||
issue_number: Optional issue number to include in branch name
|
||||
from_branch: Source branch (defaults to repository default)
|
||||
repo: Repository in 'owner/repo' format
|
||||
chat_id: Chat session ID for caching default branch
|
||||
__user__: User context
|
||||
__event_emitter__: Event emitter callback
|
||||
|
||||
Returns:
|
||||
str: Confirmation with branch details or error message
|
||||
"""
|
||||
token = self._get_token(__user__)
|
||||
if not token:
|
||||
return "Error: GITEA_TOKEN not configured in UserValves settings."
|
||||
|
||||
effective_repo = self._get_repo(repo, __user__)
|
||||
if not effective_repo:
|
||||
return "Error: No repository specified. Set DEFAULT_REPO in Valves or USER_DEFAULT_REPO in UserValves."
|
||||
|
||||
effective_from = from_branch or self._get_branch(None, __user__)
|
||||
|
||||
# Auto-append issue number if provided
|
||||
if issue_number and str(issue_number) not in branch_name:
|
||||
branch_name = f"{branch_name}-{issue_number}"
|
||||
|
||||
# Validate branch name
|
||||
is_valid, error_msg = self._validate_branch_name(branch_name)
|
||||
if not is_valid:
|
||||
return f"Error: {error_msg}"
|
||||
|
||||
# Cache the default branch for this chat session
|
||||
if chat_id:
|
||||
self._set_cached_data(chat_id, effective_repo, "default_branch", effective_from)
|
||||
self._set_cached_data(chat_id, effective_repo, "current_branch", branch_name)
|
||||
|
||||
if __event_emitter__:
|
||||
await __event_emitter__(
|
||||
{
|
||||
"type": "status",
|
||||
"data": {"description": f"Creating branch {branch_name}...", "done": False},
|
||||
}
|
||||
)
|
||||
|
||||
try:
|
||||
# Use gitea/dev.py create_branch operation
|
||||
if self._dev:
|
||||
result = await self._dev.create_branch(
|
||||
branch_name=branch_name,
|
||||
from_branch=effective_from,
|
||||
repo=effective_repo,
|
||||
__user__=__user__,
|
||||
__event_emitter__=__event_emitter__,
|
||||
)
|
||||
return result
|
||||
else:
|
||||
# Fallback: direct API call if dev not available
|
||||
async with httpx.AsyncClient(
|
||||
timeout=30.0, verify=self.valves.VERIFY_SSL
|
||||
) as client:
|
||||
response = await client.post(
|
||||
f"{self.valves.GITEA_URL.rstrip('/')}/api/v1/repos/{effective_repo}/branches",
|
||||
headers={"Authorization": f"token {token}"},
|
||||
json={
|
||||
"new_branch_name": branch_name,
|
||||
"old_branch_name": effective_from,
|
||||
},
|
||||
)
|
||||
response.raise_for_status()
|
||||
|
||||
if __event_emitter__:
|
||||
await __event_emitter__(
|
||||
{
|
||||
"type": "status",
|
||||
"data": {"description": "Done", "done": True, "hidden": True},
|
||||
}
|
||||
)
|
||||
|
||||
return f"✅ Created branch `{branch_name}` from `{effective_from}` in `{effective_repo}`"
|
||||
|
||||
except Exception as e:
|
||||
return f"Error: Failed to create branch. {type(e).__name__}: {e}"
|
||||
|
||||
async def read_ticket(
|
||||
self,
|
||||
issue_number: int,
|
||||
repo: Optional[str] = None,
|
||||
__user__: dict = None,
|
||||
__event_emitter__: Callable[[dict], Any] = None,
|
||||
) -> str:
|
||||
"""
|
||||
Read and parse a ticket/issue to extract requirements.
|
||||
|
||||
This function reads an issue and extracts key information
|
||||
needed for implementation, including title, description,
|
||||
acceptance criteria, and testing requirements.
|
||||
|
||||
Args:
|
||||
issue_number: Issue number to read
|
||||
repo: Repository in 'owner/repo' format
|
||||
__user__: User context
|
||||
__event_emitter__: Event emitter callback
|
||||
|
||||
Returns:
|
||||
str: Structured ticket requirements or error message
|
||||
"""
|
||||
token = self._get_token(__user__)
|
||||
if not token:
|
||||
return "Error: GITEA_TOKEN not configured in UserValves settings."
|
||||
|
||||
effective_repo = self._get_repo(repo, __user__)
|
||||
if not effective_repo:
|
||||
return "Error: No repository specified."
|
||||
|
||||
if __event_emitter__:
|
||||
await __event_emitter__(
|
||||
{
|
||||
"type": "status",
|
||||
"data": {"description": f"Reading ticket #{issue_number}...", "done": False},
|
||||
}
|
||||
)
|
||||
|
||||
try:
|
||||
# Use gitea/dev.py get_issue operation
|
||||
if self._dev:
|
||||
raw_issue = await self._dev.get_issue(
|
||||
issue_number=issue_number,
|
||||
repo=effective_repo,
|
||||
__user__=__user__,
|
||||
)
|
||||
else:
|
||||
# Fallback: direct API call
|
||||
async with httpx.AsyncClient(
|
||||
timeout=30.0, verify=self.valves.VERIFY_SSL
|
||||
) as client:
|
||||
response = await client.get(
|
||||
f"{self.valves.GITEA_URL.rstrip('/')}/api/v1/repos/{effective_repo}/issues/{issue_number}",
|
||||
headers={"Authorization": f"token {token}"},
|
||||
)
|
||||
response.raise_for_status()
|
||||
raw_issue = response.json()
|
||||
|
||||
# Format raw response
|
||||
title = raw_issue.get("title", "No title")
|
||||
body = raw_issue.get("body", "")
|
||||
state = raw_issue.get("state", "unknown")
|
||||
user = raw_issue.get("user", {}).get("login", "unknown")
|
||||
created_at = raw_issue.get("created_at", "")[:10]
|
||||
labels = [label.get("name", "") for label in raw_issue.get("labels", [])]
|
||||
|
||||
raw_issue = f"# Issue #{issue_number}: {title}\n\n**State:** {state.upper()}\n**Author:** @{user}\n**Labels:** {', '.join(labels) if labels else 'None'}\n**Created:** {created_at}\n\n## Description\n{body}"
|
||||
|
||||
# Parse the issue
|
||||
info = self._extract_issue_info(raw_issue)
|
||||
|
||||
# Build structured output
|
||||
output = f"# Ticket Requirements: #{issue_number}\n\n"
|
||||
output += f"**Title:** {info['title'] or 'Untitled'}\n"
|
||||
output += f"**Type:** {info['type'].upper()}\n"
|
||||
output += f"**Testing Required:** {'✅' if info['has_testing'] else '❌'}\n"
|
||||
output += f"**Documentation Required:** {'✅' if info['has_documentation'] else '❌'}\n\n"
|
||||
|
||||
if info['body']:
|
||||
output += "## Requirements\n\n"
|
||||
# Extract bullet points
|
||||
bullets = [line[2:].strip() for line in info['body'].split('\n') if line.strip().startswith('- ')]
|
||||
if bullets:
|
||||
for i, bullet in enumerate(bullets, 1):
|
||||
output += f"{i}. {bullet}\n"
|
||||
else:
|
||||
output += f"{info['body'][:500]}\n"
|
||||
output += "\n"
|
||||
|
||||
# Extract testing criteria
|
||||
if "Testing" in raw_issue or "test" in raw_issue.lower():
|
||||
test_match = re.search(r"(?:Testing Criteria|Tests?).*?\n(.*?)(?=\n## |\n# |\Z)", raw_issue, re.DOTALL | re.IGNORECASE)
|
||||
if test_match:
|
||||
output += "## Testing Criteria\n\n"
|
||||
output += test_match.group(1).strip() + "\n\n"
|
||||
|
||||
# Extract technical notes
|
||||
if "Technical" in raw_issue:
|
||||
tech_match = re.search(r"(?:Technical Notes).*?\n(.*?)(?=\n## |\n# |\Z)", raw_issue, re.DOTALL | re.IGNORECASE)
|
||||
if tech_match:
|
||||
output += "## Technical Notes\n\n"
|
||||
output += tech_match.group(1).strip() + "\n\n"
|
||||
|
||||
if __event_emitter__:
|
||||
await __event_emitter__(
|
||||
{
|
||||
"type": "status",
|
||||
"data": {"description": "Done", "done": True, "hidden": True},
|
||||
}
|
||||
)
|
||||
|
||||
return output.strip()
|
||||
|
||||
except Exception as e:
|
||||
return f"Error: Failed to read ticket. {type(e).__name__}: {e}"
|
||||
|
||||
async def commit_changes(
|
||||
self,
|
||||
path: str,
|
||||
content: str,
|
||||
short_description: str,
|
||||
issue_number: Optional[int] = None,
|
||||
body: str = "",
|
||||
branch: Optional[str] = None,
|
||||
repo: Optional[str] = None,
|
||||
__user__: dict = None,
|
||||
__event_emitter__: Callable[[dict], Any] = None,
|
||||
) -> str:
|
||||
"""
|
||||
Commit file changes with auto-generated detailed commit messages.
|
||||
|
||||
This function creates or updates a file and generates a detailed
|
||||
commit message that references the issue number.
|
||||
|
||||
Args:
|
||||
path: File path to create or update
|
||||
content: New file content
|
||||
short_description: Brief description of changes
|
||||
issue_number: Issue number for commit message reference
|
||||
body: Additional commit message details
|
||||
branch: Branch name (defaults to current feature branch)
|
||||
repo: Repository in 'owner/repo' format
|
||||
__user__: User context
|
||||
__event_emitter__: Event emitter callback
|
||||
|
||||
Returns:
|
||||
str: Commit confirmation or error message
|
||||
"""
|
||||
token = self._get_token(__user__)
|
||||
if not token:
|
||||
return "Error: GITEA_TOKEN not configured in UserValves settings."
|
||||
|
||||
effective_repo = self._get_repo(repo, __user__)
|
||||
if not effective_repo:
|
||||
return "Error: No repository specified."
|
||||
|
||||
effective_branch = self._get_branch(branch, __user__)
|
||||
|
||||
# Validate we're not on a protected branch
|
||||
is_valid, error_msg = self._validate_branch_name(effective_branch)
|
||||
if not is_valid:
|
||||
# Check if it's a protected branch
|
||||
if effective_branch in self.valves.PROTECTED_BRANCHES:
|
||||
return f"Error: Cannot commit directly to protected branch '{effective_branch}'. Create a feature branch first."
|
||||
# If it's just not following conventions, warn but proceed
|
||||
pass
|
||||
|
||||
# Generate commit message
|
||||
scope = effective_branch.split("/")[0] if "/" in effective_branch else "chore"
|
||||
commit_message = self._generate_commit_message(
|
||||
scope=scope,
|
||||
short_description=short_description,
|
||||
issue_id=str(issue_number) if issue_number else None,
|
||||
body=body,
|
||||
)
|
||||
|
||||
if __event_emitter__:
|
||||
await __event_emitter__(
|
||||
{
|
||||
"type": "status",
|
||||
"data": {"description": f"Committing changes to {path}...", "done": False},
|
||||
}
|
||||
)
|
||||
|
||||
try:
|
||||
# Use gitea/dev.py operations
|
||||
if self._dev:
|
||||
# Check if file exists to determine create vs replace
|
||||
file_info = await self._dev.get_file(
|
||||
path=path,
|
||||
repo=effective_repo,
|
||||
branch=effective_branch,
|
||||
__user__=__user__,
|
||||
)
|
||||
|
||||
if "Error: File not found" in file_info:
|
||||
# Create new file
|
||||
result = await self._dev.create_file(
|
||||
path=path,
|
||||
content=content,
|
||||
message=commit_message,
|
||||
repo=effective_repo,
|
||||
branch=effective_branch,
|
||||
__user__=__user__,
|
||||
__event_emitter__=__event_emitter__,
|
||||
)
|
||||
else:
|
||||
# Replace existing file
|
||||
result = await self._dev.replace_file(
|
||||
path=path,
|
||||
content=content,
|
||||
message=commit_message,
|
||||
repo=effective_repo,
|
||||
branch=effective_branch,
|
||||
__user__=__user__,
|
||||
__event_emitter__=__event_emitter__,
|
||||
)
|
||||
|
||||
# Add issue reference to the output
|
||||
if issue_number:
|
||||
result += f"\n\n**Referenced Issue:** #{issue_number}"
|
||||
|
||||
return result
|
||||
else:
|
||||
return "Error: gitea/dev.py operations not available. Cannot commit changes."
|
||||
|
||||
except Exception as e:
|
||||
return f"Error: Failed to commit changes. {type(e).__name__}: {e}"
|
||||
|
||||
async def create_pull_request(
|
||||
self,
|
||||
title: str,
|
||||
head_branch: str,
|
||||
body: str = "",
|
||||
base_branch: Optional[str] = None,
|
||||
repo: Optional[str] = None,
|
||||
issue_number: Optional[int] = None,
|
||||
__user__: dict = None,
|
||||
__event_emitter__: Callable[[dict], Any] = None,
|
||||
) -> str:
|
||||
"""
|
||||
Create a pull request from a feature branch.
|
||||
|
||||
This function creates a PR and optionally references an issue
|
||||
in the PR description.
|
||||
|
||||
Args:
|
||||
title: PR title
|
||||
head_branch: Source branch with changes
|
||||
body: PR description
|
||||
base_branch: Target branch (defaults to main)
|
||||
repo: Repository in 'owner/repo' format
|
||||
issue_number: Issue number to reference in description
|
||||
__user__: User context
|
||||
__event_emitter__: Event emitter callback
|
||||
|
||||
Returns:
|
||||
str: PR creation confirmation or error message
|
||||
"""
|
||||
token = self._get_token(__user__)
|
||||
if not token:
|
||||
return "Error: GITEA_TOKEN not configured in UserValves settings."
|
||||
|
||||
effective_repo = self._get_repo(repo, __user__)
|
||||
if not effective_repo:
|
||||
return "Error: No repository specified."
|
||||
|
||||
effective_base = base_branch or "main"
|
||||
|
||||
# Validate head branch is not protected
|
||||
is_valid, error_msg = self._validate_branch_name(head_branch)
|
||||
if not is_valid:
|
||||
if head_branch in self.valves.PROTECTED_BRANCHES:
|
||||
return f"Error: Cannot create PR from protected branch '{head_branch}'."
|
||||
# For non-conforming names, warn but proceed
|
||||
pass
|
||||
|
||||
# Build PR body with issue reference
|
||||
pr_body = body
|
||||
if issue_number:
|
||||
pr_body = f"**References Issue:** #{issue_number}\n\n{body}"
|
||||
|
||||
if __event_emitter__:
|
||||
await __event_emitter__(
|
||||
{
|
||||
"type": "status",
|
||||
"data": {"description": f"Creating PR from {head_branch}...", "done": False},
|
||||
}
|
||||
)
|
||||
|
||||
try:
|
||||
# Use gitea/dev.py create_pull_request
|
||||
if self._dev:
|
||||
result = await self._dev.create_pull_request(
|
||||
title=title,
|
||||
head_branch=head_branch,
|
||||
base_branch=effective_base,
|
||||
body=pr_body,
|
||||
repo=effective_repo,
|
||||
__user__=__user__,
|
||||
__event_emitter__=__event_emitter__,
|
||||
)
|
||||
return result
|
||||
else:
|
||||
return "Error: gitea/dev.py operations not available. Cannot create PR."
|
||||
|
||||
except Exception as e:
|
||||
return f"Error: Failed to create PR. {type(e).__name__}: {e}"
|
||||
|
||||
async def list_my_branches(
|
||||
self,
|
||||
repo: Optional[str] = None,
|
||||
__user__: dict = None,
|
||||
__event_emitter__: Callable[[dict], Any] = None,
|
||||
) -> str:
|
||||
"""
|
||||
List branches in the repository with scope categorization.
|
||||
|
||||
This function lists all branches and groups them by scope
|
||||
for easier navigation.
|
||||
|
||||
Args:
|
||||
repo: Repository in 'owner/repo' format
|
||||
__user__: User context
|
||||
__event_emitter__: Event emitter callback
|
||||
|
||||
Returns:
|
||||
str: Formatted branch listing or error message
|
||||
"""
|
||||
token = self._get_token(__user__)
|
||||
if not token:
|
||||
return "Error: GITEA_TOKEN not configured in UserValves settings."
|
||||
|
||||
effective_repo = self._get_repo(repo, __user__)
|
||||
if not effective_repo:
|
||||
return "Error: No repository specified."
|
||||
|
||||
if __event_emitter__:
|
||||
await __event_emitter__(
|
||||
{
|
||||
"type": "status",
|
||||
"data": {"description": "Listing branches...", "done": False},
|
||||
}
|
||||
)
|
||||
|
||||
try:
|
||||
# Use gitea/dev.py list_branches
|
||||
if self._dev:
|
||||
result = await self._dev.list_branches(
|
||||
repo=effective_repo,
|
||||
__user__=__user__,
|
||||
__event_emitter__=__event_emitter__,
|
||||
)
|
||||
|
||||
# Add scope categorization if we got a result
|
||||
if "Error:" not in result:
|
||||
# Categorize branches by scope
|
||||
scope_branches: Dict[str, List[str]] = {scope: [] for scope in self.valves.ALLOWED_SCOPES}
|
||||
scope_branches["other"] = []
|
||||
|
||||
# Parse the result to extract branch names
|
||||
branch_pattern = r"- `([^`]+)`"
|
||||
branches = re.findall(branch_pattern, result)
|
||||
|
||||
for branch in branches:
|
||||
added = False
|
||||
for scope in self.valves.ALLOWED_SCOPES:
|
||||
if branch.startswith(f"{scope}/"):
|
||||
scope_branches[scope].append(branch)
|
||||
added = True
|
||||
break
|
||||
if not added and branch not in self.valves.PROTECTED_BRANCHES:
|
||||
scope_branches["other"].append(branch)
|
||||
|
||||
# Add categorization to output
|
||||
categorized = "\n## Branch Scopes\n\n"
|
||||
for scope, branches_list in scope_branches.items():
|
||||
if branches_list:
|
||||
categorized += f"**{scope.upper()}**\n"
|
||||
for branch in branches_list:
|
||||
categorized += f"- `{branch}`\n"
|
||||
categorized += "\n"
|
||||
|
||||
result += categorized
|
||||
|
||||
return result
|
||||
else:
|
||||
return "Error: gitea/dev.py operations not available."
|
||||
|
||||
except Exception as e:
|
||||
return f"Error: Failed to list branches. {type(e).__name__}: {e}"
|
||||
|
||||
async def get_branch_status(
|
||||
self,
|
||||
branch: Optional[str] = None,
|
||||
repo: Optional[str] = None,
|
||||
__user__: dict = None,
|
||||
__event_emitter__: Callable[[dict], Any] = None,
|
||||
) -> str:
|
||||
"""
|
||||
Get the current branch status and recent activity.
|
||||
|
||||
This function shows the current branch and recent commits
|
||||
to help the coder understand their current context.
|
||||
|
||||
Args:
|
||||
branch: Branch name (defaults to current)
|
||||
repo: Repository in 'owner/repo' format
|
||||
__user__: User context
|
||||
__event_emitter__: Event emitter callback
|
||||
|
||||
Returns:
|
||||
str: Branch status or error message
|
||||
"""
|
||||
token = self._get_token(__user__)
|
||||
if not token:
|
||||
return "Error: GITEA_TOKEN not configured in UserValves settings."
|
||||
|
||||
effective_repo = self._get_repo(repo, __user__)
|
||||
if not effective_repo:
|
||||
return "Error: No repository specified."
|
||||
|
||||
effective_branch = self._get_branch(branch, __user__)
|
||||
|
||||
if __event_emitter__:
|
||||
await __event_emitter__(
|
||||
{
|
||||
"type": "status",
|
||||
"data": {"description": f"Getting status for {effective_branch}...", "done": False},
|
||||
}
|
||||
)
|
||||
|
||||
try:
|
||||
# Get recent commits on this branch
|
||||
if self._dev:
|
||||
commits = await self._dev.list_commits(
|
||||
repo=effective_repo,
|
||||
branch=effective_branch,
|
||||
limit=5,
|
||||
__user__=__user__,
|
||||
)
|
||||
|
||||
# Get branch info
|
||||
branches_list = await self._dev.list_branches(
|
||||
repo=effective_repo,
|
||||
__user__=__user__,
|
||||
)
|
||||
|
||||
# Check if branch exists
|
||||
branch_exists = f"`{effective_branch}`" in branches_list
|
||||
|
||||
output = f"# Branch Status: `{effective_branch}`\n\n"
|
||||
output += f"**Status:** {'✅ Exists' if branch_exists else '❌ Not Found'}\n"
|
||||
output += f"**Repository:** {effective_repo}\n\n"
|
||||
|
||||
if branch_exists:
|
||||
output += "## Recent Commits\n\n"
|
||||
output += commits
|
||||
|
||||
return output
|
||||
else:
|
||||
return "Error: gitea/dev.py operations not available."
|
||||
|
||||
except Exception as e:
|
||||
return f"Error: Failed to get branch status. {type(e).__name__}: {e}"
|
||||
|
||||
async def suggest_branch_name(
|
||||
self,
|
||||
issue_text: str,
|
||||
prefix: Optional[str] = None,
|
||||
__user__: dict = None,
|
||||
) -> str:
|
||||
"""
|
||||
Suggest a branch name based on issue requirements.
|
||||
|
||||
This function analyzes the issue text and suggests appropriate
|
||||
branch names following naming conventions.
|
||||
|
||||
Args:
|
||||
issue_text: Raw issue text
|
||||
prefix: Optional scope prefix (defaults to 'feature')
|
||||
__user__: User context
|
||||
|
||||
Returns:
|
||||
str: Suggested branch name(s)
|
||||
"""
|
||||
# Extract issue number
|
||||
issue_match = re.search(r"# (\d+):", issue_text)
|
||||
issue_id = issue_match.group(1) if issue_match else ""
|
||||
|
||||
# Extract title/description
|
||||
title_match = re.search(r"# Issue #?\d+:? (.+)", issue_text)
|
||||
title = title_match.group(1).strip() if title_match else ""
|
||||
|
||||
# Determine scope from labels or content
|
||||
scope = prefix or "feature"
|
||||
if "bug" in issue_text.lower() or "fix" in issue_text.lower():
|
||||
scope = "fix"
|
||||
elif "doc" in issue_text.lower():
|
||||
scope = "docs"
|
||||
elif "test" in issue_text.lower():
|
||||
scope = "test"
|
||||
elif "refactor" in issue_text.lower():
|
||||
scope = "refactor"
|
||||
|
||||
# Generate slug from title
|
||||
if title:
|
||||
# Remove special characters and lowercase
|
||||
slug = re.sub(r"[^a-z0-9]+", "-", title.lower())
|
||||
slug = slug.strip("-")[:30]
|
||||
else:
|
||||
slug = "unknown-change"
|
||||
|
||||
# Build branch names
|
||||
suggestions = []
|
||||
|
||||
if issue_id:
|
||||
suggestions.append(f"{scope}/{issue_id}-{slug}")
|
||||
suggestions.append(f"{scope}/issue-{issue_id}-{slug}")
|
||||
else:
|
||||
suggestions.append(f"{scope}/{slug}")
|
||||
|
||||
output = f"# Suggested Branch Names\n\n"
|
||||
output += f"**Based on:** {title or 'Issue requirements'}\n"
|
||||
output += f"**Suggested Scope:** {scope}\n\n"
|
||||
|
||||
for i, name in enumerate(suggestions, 1):
|
||||
output += f"{i}. `{name}`\n"
|
||||
|
||||
output += "\n**Note:** Add -<issue-number> if not already included."
|
||||
|
||||
return output
|
||||
|
||||
async def workflow_summary(
|
||||
self,
|
||||
repo: Optional[str] = None,
|
||||
__user__: dict = None,
|
||||
__event_emitter__: Callable[[dict], Any] = None,
|
||||
) -> str:
|
||||
"""
|
||||
Provide a summary of the gitea_coder workflow and available commands.
|
||||
|
||||
This function lists all available commands and provides examples
|
||||
of how to use them for the coding workflow.
|
||||
|
||||
Args:
|
||||
repo: Repository in 'owner/repo' format
|
||||
__user__: User context
|
||||
__event_emitter__: Event emitter callback
|
||||
|
||||
Returns:
|
||||
str: Workflow summary
|
||||
"""
|
||||
effective_repo = self._get_repo(repo, __user__)
|
||||
|
||||
output = f"# Gitea Coder Workflow\n\n"
|
||||
output += f"**Repository:** {effective_repo or 'Not configured'}\n"
|
||||
output += f"**Default Branch:** {self.valves.DEFAULT_BRANCH}\n\n"
|
||||
|
||||
output += "## Available Commands\n\n"
|
||||
|
||||
output += "### 1. Start Working on Ticket\n"
|
||||
output += "`read_ticket(issue_number=N)` - Read ticket requirements\n"
|
||||
output += "`suggest_branch_name(issue_text)` - Get branch name suggestions\n"
|
||||
output += "`create_feature_branch(branch_name='feature/N-description')` - Create branch\n\n"
|
||||
|
||||
output += "### 2. Make Changes\n"
|
||||
output += "`commit_changes(path, content, short_description, issue_number=N)` - Commit with auto-generated message\n\n"
|
||||
|
||||
output += "### 3. Review & Submit\n"
|
||||
output += "`get_branch_status()` - Check current branch status\n"
|
||||
output += "`create_pull_request(title, head_branch, issue_number=N)` - Create PR\n\n"
|
||||
|
||||
output += "### 3. Navigation\n"
|
||||
output += "`list_my_branches()` - List all branches by scope\n\n"
|
||||
|
||||
output += "## Branch Naming Conventions\n\n"
|
||||
output += "**Allowed Scopes:**\n"
|
||||
for scope in self.valves.ALLOWED_SCOPES:
|
||||
output += f"- `{scope}/` - {self._get_scope_description(scope)}\n"
|
||||
|
||||
output += "\n**Protected Branches:**\n"
|
||||
output += f"- {', '.join([f'`{b}`' for b in self.valves.PROTECTED_BRANCHES])}\n"
|
||||
output += "*(Cannot directly modify protected branches)*\n\n"
|
||||
|
||||
output += "## Example Workflow\n\n"
|
||||
output += "```\n# Read ticket requirements\nawait read_ticket(issue_number=42)\n\n# Create feature branch\nawait create_feature_branch('feature/42-add-login-functionality')\n\n# Make changes and commit\nawait commit_changes(\n path='src/auth.py',\n content=new_code,\n short_description='add user authentication',\n issue_number=42\n)\n\n# Create PR when done\nawait create_pull_request(\n title='Add user authentication',\n head_branch='feature/42-add-login-functionality'\n)\n```\n"
|
||||
|
||||
return output
|
||||
|
||||
def _get_scope_description(self, scope: str) -> str:
|
||||
"""Get description for a scope type"""
|
||||
descriptions = {
|
||||
"feature": "New functionality or enhancements",
|
||||
"fix": "Bug fixes",
|
||||
"refactor": "Code restructuring (no behavior change)",
|
||||
"docs": "Documentation only",
|
||||
"test": "Test additions/improvements",
|
||||
"chore": "Maintenance tasks",
|
||||
}
|
||||
return descriptions.get(scope, "Other changes")
|
||||
Reference in New Issue
Block a user