363 lines
11 KiB
Python
363 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""AI Code Review Agent - Main Entry Point
|
|
|
|
This is the main CLI for running AI code review agents.
|
|
Can be invoked directly or through CI/CD workflows.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import logging
|
|
import os
|
|
import sys
|
|
|
|
import yaml
|
|
|
|
# Add the package to path
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from agents.issue_agent import IssueAgent
|
|
from agents.pr_agent import PRAgent
|
|
from agents.codebase_agent import CodebaseAgent
|
|
from agents.chat_agent import ChatAgent
|
|
from dispatcher import Dispatcher, get_dispatcher
|
|
|
|
|
|
def setup_logging(verbose: bool = False):
|
|
"""Configure logging."""
|
|
level = logging.DEBUG if verbose else logging.INFO
|
|
logging.basicConfig(
|
|
level=level,
|
|
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
|
)
|
|
|
|
|
|
def load_config(config_path: str | None = None) -> dict:
|
|
"""Load configuration from file."""
|
|
if config_path and os.path.exists(config_path):
|
|
with open(config_path) as f:
|
|
return yaml.safe_load(f)
|
|
|
|
default_path = os.path.join(os.path.dirname(__file__), "config.yml")
|
|
if os.path.exists(default_path):
|
|
with open(default_path) as f:
|
|
return yaml.safe_load(f)
|
|
|
|
return {}
|
|
|
|
|
|
def run_pr_review(args, config: dict):
|
|
"""Run PR review agent."""
|
|
from agents.base_agent import AgentContext
|
|
|
|
agent = PRAgent(config=config)
|
|
|
|
# Build context from environment or arguments
|
|
owner, repo = args.repo.split("/")
|
|
pr_number = args.pr_number
|
|
|
|
context = AgentContext(
|
|
owner=owner,
|
|
repo=repo,
|
|
event_type="pull_request",
|
|
event_data={
|
|
"action": "opened",
|
|
"pull_request": {
|
|
"number": pr_number,
|
|
"title": args.title or f"PR #{pr_number}",
|
|
},
|
|
},
|
|
config=config,
|
|
)
|
|
|
|
result = agent.run(context)
|
|
|
|
if result.success:
|
|
print(f"✅ PR Review Complete: {result.message}")
|
|
print(f" Actions: {', '.join(result.actions_taken)}")
|
|
else:
|
|
print(f"❌ PR Review Failed: {result.message}")
|
|
if result.error:
|
|
print(f" Error: {result.error}")
|
|
sys.exit(1)
|
|
|
|
|
|
def run_issue_triage(args, config: dict):
|
|
"""Run issue triage agent."""
|
|
from agents.base_agent import AgentContext
|
|
from clients.gitea_client import GiteaClient
|
|
|
|
agent = IssueAgent(config=config)
|
|
|
|
owner, repo = args.repo.split("/")
|
|
issue_number = args.issue_number
|
|
|
|
# Fetch the actual issue data from Gitea API to get the complete body
|
|
gitea = GiteaClient()
|
|
|
|
try:
|
|
issue_data = gitea.get_issue(owner, repo, issue_number)
|
|
except Exception as e:
|
|
print(f"❌ Failed to fetch issue: {e}")
|
|
sys.exit(1)
|
|
|
|
context = AgentContext(
|
|
owner=owner,
|
|
repo=repo,
|
|
event_type="issues",
|
|
event_data={
|
|
"action": "opened",
|
|
"issue": issue_data,
|
|
},
|
|
config=config,
|
|
)
|
|
|
|
result = agent.run(context)
|
|
|
|
if result.success:
|
|
print(f"✅ Issue Triage Complete: {result.message}")
|
|
print(f" Actions: {', '.join(result.actions_taken)}")
|
|
else:
|
|
print(f"❌ Issue Triage Failed: {result.message}")
|
|
if result.error:
|
|
print(f" Error: {result.error}")
|
|
sys.exit(1)
|
|
|
|
|
|
def run_issue_comment(args, config: dict):
|
|
"""Handle @ai-bot command in issue comment."""
|
|
from agents.base_agent import AgentContext
|
|
|
|
agent = IssueAgent(config=config)
|
|
|
|
owner, repo = args.repo.split("/")
|
|
issue_number = args.issue_number
|
|
|
|
# Fetch the actual issue data from Gitea API
|
|
from clients.gitea_client import GiteaClient
|
|
gitea = GiteaClient()
|
|
|
|
try:
|
|
issue_data = gitea.get_issue(owner, repo, issue_number)
|
|
except Exception as e:
|
|
print(f"❌ Failed to fetch issue: {e}")
|
|
sys.exit(1)
|
|
|
|
context = AgentContext(
|
|
owner=owner,
|
|
repo=repo,
|
|
event_type="issue_comment",
|
|
event_data={
|
|
"action": "created",
|
|
"issue": issue_data,
|
|
"comment": {
|
|
"body": args.comment_body,
|
|
},
|
|
},
|
|
config=config,
|
|
)
|
|
|
|
result = agent.run(context)
|
|
|
|
if result.success:
|
|
print(f"✅ Comment Response Complete: {result.message}")
|
|
print(f" Actions: {', '.join(result.actions_taken)}")
|
|
else:
|
|
print(f"❌ Comment Response Failed: {result.message}")
|
|
if result.error:
|
|
print(f" Error: {result.error}")
|
|
sys.exit(1)
|
|
|
|
|
|
def run_codebase_analysis(args, config: dict):
|
|
"""Run codebase analysis agent."""
|
|
from agents.base_agent import AgentContext
|
|
|
|
agent = CodebaseAgent(config=config)
|
|
|
|
owner, repo = args.repo.split("/")
|
|
|
|
context = AgentContext(
|
|
owner=owner,
|
|
repo=repo,
|
|
event_type="workflow_dispatch",
|
|
event_data={},
|
|
config=config,
|
|
)
|
|
|
|
result = agent.run(context)
|
|
|
|
if result.success:
|
|
print(f"✅ Codebase Analysis Complete: {result.message}")
|
|
print(f" Health Score: {result.data.get('health_score', 'N/A')}")
|
|
print(f" Actions: {', '.join(result.actions_taken)}")
|
|
else:
|
|
print(f"❌ Codebase Analysis Failed: {result.message}")
|
|
if result.error:
|
|
print(f" Error: {result.error}")
|
|
sys.exit(1)
|
|
|
|
|
|
def run_chat(args, config: dict):
|
|
"""Run interactive chat with the Bartender bot."""
|
|
from agents.base_agent import AgentContext
|
|
from clients.gitea_client import GiteaClient
|
|
|
|
agent = ChatAgent(config=config)
|
|
|
|
owner, repo = args.repo.split("/")
|
|
|
|
# Build context
|
|
event_data = {"message": args.message}
|
|
|
|
# If issue number provided, add issue context
|
|
if args.issue_number:
|
|
gitea = GiteaClient()
|
|
try:
|
|
issue_data = gitea.get_issue(owner, repo, args.issue_number)
|
|
event_data["issue"] = issue_data
|
|
event_data["issue_number"] = args.issue_number
|
|
except Exception as e:
|
|
print(f"Warning: Could not fetch issue #{args.issue_number}: {e}")
|
|
|
|
context = AgentContext(
|
|
owner=owner,
|
|
repo=repo,
|
|
event_type="chat",
|
|
event_data=event_data,
|
|
config=config,
|
|
)
|
|
|
|
result = agent.run(context)
|
|
|
|
if result.success:
|
|
print(f"\n🍸 Bartender says:\n")
|
|
print(result.data.get("response", ""))
|
|
print()
|
|
if result.data.get("tools_used"):
|
|
print(f" [Tools used: {', '.join(result.data['tools_used'])}]")
|
|
else:
|
|
print(f"❌ Chat Failed: {result.message}")
|
|
if result.error:
|
|
print(f" Error: {result.error}")
|
|
sys.exit(1)
|
|
|
|
|
|
def run_webhook_dispatch(args, config: dict):
|
|
"""Dispatch a webhook event."""
|
|
dispatcher = get_dispatcher()
|
|
|
|
# Register all agents
|
|
dispatcher.register_agent(IssueAgent(config=config))
|
|
dispatcher.register_agent(PRAgent(config=config))
|
|
dispatcher.register_agent(CodebaseAgent(config=config))
|
|
dispatcher.register_agent(ChatAgent(config=config))
|
|
|
|
# Parse event data
|
|
event_data = json.loads(args.event_data)
|
|
owner, repo = args.repo.split("/")
|
|
|
|
result = dispatcher.dispatch(
|
|
event_type=args.event_type,
|
|
event_data=event_data,
|
|
owner=owner,
|
|
repo=repo,
|
|
)
|
|
|
|
print(f"Dispatched event: {result.event_type}")
|
|
print(f"Agents run: {result.agents_run}")
|
|
for i, agent_result in enumerate(result.results):
|
|
status = "✅" if agent_result.success else "❌"
|
|
print(f" {status} {result.agents_run[i]}: {agent_result.message}")
|
|
|
|
if result.errors:
|
|
sys.exit(1)
|
|
|
|
|
|
def main():
|
|
"""Main entry point."""
|
|
parser = argparse.ArgumentParser(
|
|
description="AI Code Review Agent",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
)
|
|
parser.add_argument("-v", "--verbose", action="store_true", help="Verbose output")
|
|
parser.add_argument("-c", "--config", help="Path to config file")
|
|
|
|
subparsers = parser.add_subparsers(dest="command", help="Available commands")
|
|
|
|
# PR review command
|
|
pr_parser = subparsers.add_parser("pr", help="Review a pull request")
|
|
pr_parser.add_argument("repo", help="Repository (owner/repo)")
|
|
pr_parser.add_argument("pr_number", type=int, help="PR number")
|
|
pr_parser.add_argument("--title", help="PR title (optional)")
|
|
|
|
# Issue triage command
|
|
issue_parser = subparsers.add_parser("issue", help="Triage an issue")
|
|
issue_parser.add_argument("repo", help="Repository (owner/repo)")
|
|
issue_parser.add_argument("issue_number", type=int, help="Issue number")
|
|
issue_parser.add_argument("--title", help="Issue title")
|
|
issue_parser.add_argument("--body", help="Issue body")
|
|
|
|
# Issue comment command (for @ai-bot mentions)
|
|
comment_parser = subparsers.add_parser("comment", help="Respond to @ai-bot command")
|
|
comment_parser.add_argument("repo", help="Repository (owner/repo)")
|
|
comment_parser.add_argument("issue_number", type=int, help="Issue number")
|
|
comment_parser.add_argument("comment_body", help="Comment body with @ai-bot command")
|
|
|
|
# Codebase analysis command
|
|
codebase_parser = subparsers.add_parser("codebase", help="Analyze codebase")
|
|
codebase_parser.add_argument("repo", help="Repository (owner/repo)")
|
|
|
|
# Chat command (Bartender)
|
|
chat_parser = subparsers.add_parser("chat", help="Chat with Bartender bot")
|
|
chat_parser.add_argument("repo", help="Repository (owner/repo)")
|
|
chat_parser.add_argument("message", help="Message to send to Bartender")
|
|
chat_parser.add_argument(
|
|
"--issue", dest="issue_number", type=int,
|
|
help="Optional issue number to post response to"
|
|
)
|
|
|
|
# Webhook dispatch command
|
|
webhook_parser = subparsers.add_parser("dispatch", help="Dispatch webhook event")
|
|
webhook_parser.add_argument("repo", help="Repository (owner/repo)")
|
|
webhook_parser.add_argument("event_type", help="Event type")
|
|
webhook_parser.add_argument("event_data", help="Event data (JSON)")
|
|
|
|
args = parser.parse_args()
|
|
|
|
if not args.command:
|
|
parser.print_help()
|
|
sys.exit(1)
|
|
|
|
setup_logging(args.verbose)
|
|
config = load_config(args.config)
|
|
|
|
# Allow overriding the provider via a Gitea/CI secret (AI_PROVIDER env var)
|
|
ai_provider = os.environ.get("AI_PROVIDER")
|
|
if ai_provider:
|
|
config["provider"] = ai_provider
|
|
|
|
# Allow overriding the model via a Gitea/CI secret (AI_MODEL env var)
|
|
# Overrides the model for whichever provider is active.
|
|
ai_model = os.environ.get("AI_MODEL")
|
|
if ai_model:
|
|
provider = config.get("provider", "openai")
|
|
config.setdefault("model", {})[provider] = ai_model
|
|
|
|
if args.command == "pr":
|
|
run_pr_review(args, config)
|
|
elif args.command == "issue":
|
|
run_issue_triage(args, config)
|
|
elif args.command == "comment":
|
|
run_issue_comment(args, config)
|
|
elif args.command == "codebase":
|
|
run_codebase_analysis(args, config)
|
|
elif args.command == "chat":
|
|
run_chat(args, config)
|
|
elif args.command == "dispatch":
|
|
run_webhook_dispatch(args, config)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|