221 lines
8.3 KiB
Python
221 lines
8.3 KiB
Python
"""Analysis service - AI-powered issue analysis."""
|
|
import httpx
|
|
import json
|
|
from datetime import datetime
|
|
from typing import Optional, Dict, Any, List
|
|
from app.core.config import settings
|
|
|
|
class AnalysisService:
|
|
OPENROUTER_API = "https://openrouter.ai/api/v1/chat/completions"
|
|
MODEL = "meta-llama/llama-3.3-70b-instruct:free"
|
|
|
|
@classmethod
|
|
async def fetch_repository_files(cls, repo: str, path: str = "") -> List[Dict[str, str]]:
|
|
"""Fetch files from Gitea repository."""
|
|
files = []
|
|
async with httpx.AsyncClient() as client:
|
|
try:
|
|
url = f"{settings.GITEA_URL}/api/v1/repos/{repo}/contents/{path}"
|
|
headers = {}
|
|
if settings.GITEA_TOKEN:
|
|
headers["Authorization"] = f"token {settings.GITEA_TOKEN}"
|
|
|
|
response = await client.get(url, headers=headers)
|
|
if response.status_code != 200:
|
|
return files
|
|
|
|
items = response.json()
|
|
for item in items:
|
|
if item["type"] == "file" and item["name"].endswith((".cbl", ".cob", ".py", ".java", ".js", ".ts")):
|
|
content_resp = await client.get(item["download_url"], headers=headers)
|
|
if content_resp.status_code == 200:
|
|
files.append({
|
|
"path": item["path"],
|
|
"content": content_resp.text[:10000] # Limit size
|
|
})
|
|
elif item["type"] == "dir":
|
|
sub_files = await cls.fetch_repository_files(repo, item["path"])
|
|
files.extend(sub_files)
|
|
except Exception as e:
|
|
print(f"Error fetching repo: {e}")
|
|
|
|
return files[:20] # Limit to 20 files
|
|
|
|
@classmethod
|
|
def build_prompt(cls, issue: Dict[str, Any], files: List[Dict[str, str]]) -> str:
|
|
"""Build analysis prompt for LLM."""
|
|
files_context = "\n\n".join([
|
|
f"### {f['path']}\n```\n{f['content']}\n```"
|
|
for f in files
|
|
])
|
|
|
|
return f"""You are an expert software engineer analyzing a support issue.
|
|
|
|
## Issue Details
|
|
**Title:** {issue.get('title', 'N/A')}
|
|
**Description:** {issue.get('description', 'N/A')}
|
|
**Priority:** {issue.get('priority', 'N/A')}
|
|
|
|
## Source Code Files
|
|
{files_context}
|
|
|
|
## Your Task
|
|
Analyze the issue and identify:
|
|
1. Root cause of the problem
|
|
2. Which files are affected
|
|
3. Suggested code fix
|
|
|
|
## Response Format (JSON)
|
|
{{
|
|
"root_cause": "Detailed explanation of what's causing the issue",
|
|
"affected_files": ["file1.py", "file2.py"],
|
|
"suggested_fix": "Code changes needed to fix the issue",
|
|
"confidence": 0.85,
|
|
"explanation": "Step-by-step explanation of the fix"
|
|
}}
|
|
|
|
Respond ONLY with valid JSON."""
|
|
|
|
@classmethod
|
|
async def analyze(cls, issue: Dict[str, Any], repo: Optional[str] = None) -> Dict[str, Any]:
|
|
"""Run AI analysis on an issue."""
|
|
# Fetch code context
|
|
files = []
|
|
if repo:
|
|
files = await cls.fetch_repository_files(repo)
|
|
|
|
# Build prompt
|
|
prompt = cls.build_prompt(issue, files)
|
|
|
|
# Call LLM
|
|
if not settings.OPENROUTER_API_KEY:
|
|
# Mock response for testing
|
|
return {
|
|
"root_cause": "Mock analysis - configure OPENROUTER_API_KEY for real analysis",
|
|
"affected_files": ["example.py"],
|
|
"suggested_fix": "# Mock fix",
|
|
"confidence": 0.5,
|
|
"explanation": "This is a mock response"
|
|
}
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
try:
|
|
response = await client.post(
|
|
cls.OPENROUTER_API,
|
|
headers={
|
|
"Authorization": f"Bearer {settings.OPENROUTER_API_KEY}",
|
|
"Content-Type": "application/json"
|
|
},
|
|
json={
|
|
"model": cls.MODEL,
|
|
"messages": [{"role": "user", "content": prompt}],
|
|
"temperature": 0.2,
|
|
"max_tokens": 2000
|
|
},
|
|
timeout=120
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
content = data["choices"][0]["message"]["content"]
|
|
|
|
# Parse JSON from response
|
|
try:
|
|
# Handle markdown code blocks
|
|
if "```json" in content:
|
|
content = content.split("```json")[1].split("```")[0]
|
|
elif "```" in content:
|
|
content = content.split("```")[1].split("```")[0]
|
|
|
|
return json.loads(content.strip())
|
|
except json.JSONDecodeError:
|
|
return {
|
|
"root_cause": content[:500],
|
|
"affected_files": [],
|
|
"suggested_fix": "",
|
|
"confidence": 0.3,
|
|
"explanation": "Could not parse structured response"
|
|
}
|
|
else:
|
|
return {
|
|
"root_cause": f"API error: {response.status_code}",
|
|
"affected_files": [],
|
|
"suggested_fix": "",
|
|
"confidence": 0,
|
|
"explanation": response.text[:500]
|
|
}
|
|
except Exception as e:
|
|
return {
|
|
"root_cause": f"Analysis error: {str(e)}",
|
|
"affected_files": [],
|
|
"suggested_fix": "",
|
|
"confidence": 0,
|
|
"explanation": str(e)
|
|
}
|
|
|
|
@classmethod
|
|
async def create_pull_request(
|
|
cls,
|
|
repo: str,
|
|
branch: str,
|
|
title: str,
|
|
description: str,
|
|
file_changes: List[Dict[str, str]]
|
|
) -> Optional[str]:
|
|
"""Create a pull request with suggested fix."""
|
|
if not settings.GITEA_TOKEN:
|
|
return None
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
headers = {"Authorization": f"token {settings.GITEA_TOKEN}"}
|
|
|
|
try:
|
|
# 1. Get default branch
|
|
repo_resp = await client.get(
|
|
f"{settings.GITEA_URL}/api/v1/repos/{repo}",
|
|
headers=headers
|
|
)
|
|
if repo_resp.status_code != 200:
|
|
return None
|
|
default_branch = repo_resp.json().get("default_branch", "main")
|
|
|
|
# 2. Get latest commit SHA
|
|
ref_resp = await client.get(
|
|
f"{settings.GITEA_URL}/api/v1/repos/{repo}/git/refs/heads/{default_branch}",
|
|
headers=headers
|
|
)
|
|
if ref_resp.status_code != 200:
|
|
return None
|
|
sha = ref_resp.json()["object"]["sha"]
|
|
|
|
# 3. Create branch
|
|
await client.post(
|
|
f"{settings.GITEA_URL}/api/v1/repos/{repo}/git/refs",
|
|
headers=headers,
|
|
json={"ref": f"refs/heads/{branch}", "sha": sha}
|
|
)
|
|
|
|
# 4. Commit changes (simplified - just description for now)
|
|
# Full implementation would update actual files
|
|
|
|
# 5. Create PR
|
|
pr_resp = await client.post(
|
|
f"{settings.GITEA_URL}/api/v1/repos/{repo}/pulls",
|
|
headers=headers,
|
|
json={
|
|
"title": title,
|
|
"body": description,
|
|
"head": branch,
|
|
"base": default_branch
|
|
}
|
|
)
|
|
|
|
if pr_resp.status_code in (200, 201):
|
|
pr_data = pr_resp.json()
|
|
return pr_data.get("html_url")
|
|
|
|
except Exception as e:
|
|
print(f"PR creation error: {e}")
|
|
|
|
return None
|