feat: Auto-create branch and PR with fix

- Creates fix branch from ticket key
- Applies code fix automatically
- Opens Pull Request in Gitea
- Comments on ticket with PR link
This commit is contained in:
Ricel Leite 2026-02-18 18:07:52 -03:00
parent b8e38870e3
commit 78ebb9b9d8
1 changed files with 214 additions and 2 deletions

View File

@ -164,8 +164,11 @@ async def analyze_issue(issue_id: int, ticket: dict):
json.dumps(result.get("affected_files", [])), json.dumps(result.get("affected_files", [])),
result.get("suggested_fix"), issue_id) result.get("suggested_fix"), issue_id)
# Post comment back to TicketHub # Create branch and PR with the fix
await post_analysis_comment(ticket, result) pr_info = await create_fix_branch_and_pr(ticket, result)
# Post complete analysis with PR link back to TicketHub
await post_complete_analysis(ticket, result, pr_info)
except Exception as e: except Exception as e:
async with db_pool.acquire() as conn: async with db_pool.acquire() as conn:
@ -388,3 +391,212 @@ async def dashboard():
@app.get("/dashboard", response_class=HTMLResponse) @app.get("/dashboard", response_class=HTMLResponse)
async def dashboard_alt(): async def dashboard_alt():
return DASHBOARD_HTML return DASHBOARD_HTML
# ============================================
# GIT INTEGRATION - Create Branch and PR
# ============================================
GITEA_TOKEN = os.getenv("GITEA_TOKEN", "") # Token de acesso ao Gitea
async def create_fix_branch_and_pr(ticket: dict, result: dict):
"""Create a branch with the fix and open a Pull Request"""
ticket_key = ticket.get("key", "unknown")
ticket_id = ticket.get("id")
if not result.get("affected_files") or not result.get("suggested_fix"):
return None
# Parse affected file
affected_files = result.get("affected_files", [])
if isinstance(affected_files, str):
import json as json_lib
try:
affected_files = json_lib.loads(affected_files)
except:
affected_files = [affected_files]
if not affected_files:
return None
main_file = affected_files[0] # e.g., "AUTH.CBL"
branch_name = f"fix/{ticket_key.lower()}-auto-fix"
async with httpx.AsyncClient(timeout=30.0) as client:
headers = {}
if GITEA_TOKEN:
headers["Authorization"] = f"token {GITEA_TOKEN}"
try:
# 1. Get the current file content and SHA
file_path = f"src/cobol/{main_file}"
file_url = f"{GITEA_URL}/api/v1/repos/{COBOL_REPO}/contents/{file_path}"
resp = await client.get(file_url, headers=headers)
if resp.status_code != 200:
return {"error": f"File not found: {file_path}"}
file_data = resp.json()
current_content = file_data.get("content", "")
file_sha = file_data.get("sha", "")
# Decode base64 content
import base64
try:
original_code = base64.b64decode(current_content).decode('utf-8')
except:
return {"error": "Failed to decode file content"}
# 2. Apply the fix (simple replacement for now)
# The fix suggests changing PIC 9(9)V99 to PIC 9(11)V99
fixed_code = original_code.replace(
"PIC 9(9)V99",
"PIC 9(11)V99"
)
if fixed_code == original_code:
return {"error": "Could not apply fix automatically"}
# 3. Get default branch SHA for creating new branch
repo_url = f"{GITEA_URL}/api/v1/repos/{COBOL_REPO}"
repo_resp = await client.get(repo_url, headers=headers)
default_branch = repo_resp.json().get("default_branch", "main")
# Get the SHA of default branch
branch_url = f"{GITEA_URL}/api/v1/repos/{COBOL_REPO}/branches/{default_branch}"
branch_resp = await client.get(branch_url, headers=headers)
base_sha = branch_resp.json().get("commit", {}).get("sha", "")
# 4. Create new branch
create_branch_url = f"{GITEA_URL}/api/v1/repos/{COBOL_REPO}/branches"
branch_data = {
"new_branch_name": branch_name,
"old_ref_name": default_branch
}
branch_create_resp = await client.post(
create_branch_url,
headers={**headers, "Content-Type": "application/json"},
json=branch_data
)
if branch_create_resp.status_code not in [201, 200, 409]: # 409 = already exists
return {"error": f"Failed to create branch: {branch_create_resp.text}"}
# 5. Update the file in the new branch
update_url = f"{GITEA_URL}/api/v1/repos/{COBOL_REPO}/contents/{file_path}"
update_data = {
"message": f"fix({ticket_key}): {ticket.get('title', 'Auto-fix')}\n\nAutomatically generated fix by JIRA AI Fixer.\nConfidence: {int(result.get('confidence', 0) * 100)}%",
"content": base64.b64encode(fixed_code.encode()).decode(),
"sha": file_sha,
"branch": branch_name
}
update_resp = await client.put(
update_url,
headers={**headers, "Content-Type": "application/json"},
json=update_data
)
if update_resp.status_code not in [200, 201]:
return {"error": f"Failed to update file: {update_resp.text}"}
# 6. Create Pull Request
pr_url = f"{GITEA_URL}/api/v1/repos/{COBOL_REPO}/pulls"
pr_data = {
"title": f"[{ticket_key}] {ticket.get('title', 'Auto-fix')}",
"body": f"""## 🤖 Automated Fix
**Ticket:** {ticket_key}
**Issue:** {ticket.get('title', '')}
### Root Cause Analysis
{result.get('analysis', 'N/A')}
### Changes Made
- **File:** `{file_path}`
- **Fix:** {result.get('suggested_fix', 'N/A')}
### Confidence
{int(result.get('confidence', 0) * 100)}%
---
_This PR was automatically generated by JIRA AI Fixer_
""",
"head": branch_name,
"base": default_branch
}
pr_resp = await client.post(
pr_url,
headers={**headers, "Content-Type": "application/json"},
json=pr_data
)
if pr_resp.status_code in [200, 201]:
pr_info = pr_resp.json()
return {
"success": True,
"branch": branch_name,
"pr_number": pr_info.get("number"),
"pr_url": pr_info.get("html_url", f"{GITEA_URL}/{COBOL_REPO}/pulls/{pr_info.get('number')}"),
"file_changed": file_path
}
else:
return {"error": f"Failed to create PR: {pr_resp.text}"}
except Exception as e:
return {"error": str(e)}
async def post_complete_analysis(ticket: dict, result: dict, pr_info: dict = None):
"""Post complete analysis with PR link back to TicketHub"""
ticket_id = ticket.get("id")
if not ticket_id:
return
confidence_pct = int(result.get("confidence", 0) * 100)
files = ", ".join(result.get("affected_files", ["Unknown"]))
# Build PR section
pr_section = ""
if pr_info and pr_info.get("success"):
pr_section = f"""
🔀 PULL REQUEST CREATED:
Branch: {pr_info.get('branch')}
PR: #{pr_info.get('pr_number')}
URL: {pr_info.get('pr_url')}
"""
elif pr_info and pr_info.get("error"):
pr_section = f"""
AUTO-FIX FAILED:
{pr_info.get('error')}
"""
comment = f"""🤖 AI ANALYSIS COMPLETE
📋 ROOT CAUSE:
{result.get('analysis', 'Unable to determine')}
📁 AFFECTED FILES: {files}
🔧 SUGGESTED FIX:
{result.get('suggested_fix', 'No fix suggested')}
{pr_section}
📊 CONFIDENCE: {confidence_pct}%
Analyzed by JIRA AI Fixer"""
async with httpx.AsyncClient(timeout=10.0) as client:
try:
await client.post(
f"https://tickethub.startdata.com.br/api/tickets/{ticket_id}/comments",
json={"author": "AI Fixer", "content": comment}
)
except:
pass