mrradix's picture
Upload 48 files
8e4018d verified
import requests
import json
import os
from typing import Dict, List, Any, Optional
from datetime import datetime
from utils.logging import setup_logger
from utils.error_handling import handle_exceptions, IntegrationError
from utils.storage import load_data, save_data
# Initialize logger
logger = setup_logger(__name__)
class GitHubIntegration:
"""GitHub API integration for issues and repositories"""
def __init__(self, token: Optional[str] = None):
"""Initialize GitHub integration
Args:
token: GitHub API token (optional)
"""
self.base_url = "https://api.github.com"
self.token = token
self.headers = {
"Accept": "application/vnd.github.v3+json"
}
if token:
self.headers["Authorization"] = f"token {token}"
@handle_exceptions
def set_token(self, token: str) -> None:
"""Set GitHub API token
Args:
token: GitHub API token
"""
self.token = token
self.headers["Authorization"] = f"token {token}"
@handle_exceptions
def test_connection(self) -> bool:
"""Test GitHub API connection
Returns:
True if connection is successful, False otherwise
"""
try:
response = requests.get(f"{self.base_url}/user", headers=self.headers)
return response.status_code == 200
except Exception as e:
logger.error(f"GitHub connection test failed: {str(e)}")
return False
@handle_exceptions
def get_user_repos(self) -> List[Dict[str, Any]]:
"""Get user repositories
Returns:
List of repositories
"""
response = requests.get(f"{self.base_url}/user/repos", headers=self.headers)
if response.status_code != 200:
raise IntegrationError(f"Failed to get repositories: {response.text}")
repos = response.json()
return [{
"id": repo.get("id"),
"name": repo.get("name"),
"full_name": repo.get("full_name"),
"description": repo.get("description"),
"url": repo.get("html_url"),
"stars": repo.get("stargazers_count"),
"forks": repo.get("forks_count"),
"open_issues": repo.get("open_issues_count"),
"created_at": repo.get("created_at"),
"updated_at": repo.get("updated_at")
} for repo in repos]
@handle_exceptions
def get_repo_issues(self, repo_full_name: str, state: str = "open") -> List[Dict[str, Any]]:
"""Get repository issues
Args:
repo_full_name: Repository full name (e.g., "username/repo")
state: Issue state ("open", "closed", "all")
Returns:
List of issues
"""
response = requests.get(
f"{self.base_url}/repos/{repo_full_name}/issues",
headers=self.headers,
params={"state": state}
)
if response.status_code != 200:
raise IntegrationError(f"Failed to get issues: {response.text}")
issues = response.json()
return [{
"id": issue.get("id"),
"number": issue.get("number"),
"title": issue.get("title"),
"body": issue.get("body"),
"state": issue.get("state"),
"url": issue.get("html_url"),
"created_at": issue.get("created_at"),
"updated_at": issue.get("updated_at"),
"user": issue.get("user", {}).get("login"),
"labels": [label.get("name") for label in issue.get("labels", [])],
"assignees": [assignee.get("login") for assignee in issue.get("assignees", [])]
} for issue in issues]
@handle_exceptions
def create_issue(self, repo_full_name: str, title: str, body: str, labels: List[str] = None) -> Dict[str, Any]:
"""Create a new issue
Args:
repo_full_name: Repository full name (e.g., "username/repo")
title: Issue title
body: Issue body
labels: Issue labels
Returns:
Created issue data
"""
data = {
"title": title,
"body": body
}
if labels:
data["labels"] = labels
response = requests.post(
f"{self.base_url}/repos/{repo_full_name}/issues",
headers=self.headers,
json=data
)
if response.status_code != 201:
raise IntegrationError(f"Failed to create issue: {response.text}")
issue = response.json()
return {
"id": issue.get("id"),
"number": issue.get("number"),
"title": issue.get("title"),
"body": issue.get("body"),
"state": issue.get("state"),
"url": issue.get("html_url"),
"created_at": issue.get("created_at"),
"updated_at": issue.get("updated_at")
}
@handle_exceptions
def update_issue(self, repo_full_name: str, issue_number: int, title: str = None,
body: str = None, state: str = None, labels: List[str] = None) -> Dict[str, Any]:
"""Update an existing issue
Args:
repo_full_name: Repository full name (e.g., "username/repo")
issue_number: Issue number
title: Issue title (optional)
body: Issue body (optional)
state: Issue state ("open" or "closed") (optional)
labels: Issue labels (optional)
Returns:
Updated issue data
"""
data = {}
if title is not None:
data["title"] = title
if body is not None:
data["body"] = body
if state is not None:
data["state"] = state
if labels is not None:
data["labels"] = labels
response = requests.patch(
f"{self.base_url}/repos/{repo_full_name}/issues/{issue_number}",
headers=self.headers,
json=data
)
if response.status_code != 200:
raise IntegrationError(f"Failed to update issue: {response.text}")
issue = response.json()
return {
"id": issue.get("id"),
"number": issue.get("number"),
"title": issue.get("title"),
"body": issue.get("body"),
"state": issue.get("state"),
"url": issue.get("html_url"),
"created_at": issue.get("created_at"),
"updated_at": issue.get("updated_at")
}
@handle_exceptions
def convert_issues_to_tasks(self, repo_full_name: str, state: str = "open") -> List[Dict[str, Any]]:
"""Convert GitHub issues to MONA tasks
Args:
repo_full_name: Repository full name (e.g., "username/repo")
state: Issue state ("open", "closed", "all")
Returns:
List of tasks
"""
issues = self.get_repo_issues(repo_full_name, state)
tasks = []
for issue in issues:
# Convert issue to task format
task = {
"id": f"github-issue-{issue['id']}",
"title": issue["title"],
"description": issue["body"] or "",
"status": "todo" if issue["state"] == "open" else "done",
"priority": "medium", # Default priority
"created_at": issue["created_at"],
"updated_at": issue["updated_at"],
"source": "github",
"source_id": str(issue["id"]),
"source_url": issue["url"],
"metadata": {
"repo": repo_full_name,
"issue_number": issue["number"],
"labels": issue.get("labels", []),
"assignees": issue.get("assignees", [])
}
}
# Set priority based on labels if available
if "priority:high" in issue.get("labels", []):
task["priority"] = "high"
elif "priority:low" in issue.get("labels", []):
task["priority"] = "low"
tasks.append(task)
return tasks
@handle_exceptions
def sync_tasks_to_issues(self, tasks: List[Dict[str, Any]], repo_full_name: str) -> List[Dict[str, Any]]:
"""Sync MONA tasks to GitHub issues
Args:
tasks: List of tasks to sync
repo_full_name: Repository full name (e.g., "username/repo")
Returns:
List of synced tasks with updated metadata
"""
synced_tasks = []
for task in tasks:
# Skip tasks that are already synced with GitHub
if task.get("source") == "github":
synced_tasks.append(task)
continue
# Create new issue from task
issue_data = {
"title": task["title"],
"body": task.get("description", ""),
"labels": []
}
# Add priority label
if task.get("priority"):
issue_data["labels"].append(f"priority:{task['priority']}")
# Create issue
issue = self.create_issue(
repo_full_name=repo_full_name,
title=issue_data["title"],
body=issue_data["body"],
labels=issue_data["labels"]
)
# Update task with GitHub metadata
task.update({
"source": "github",
"source_id": str(issue["id"]),
"source_url": issue["url"],
"metadata": {
"repo": repo_full_name,
"issue_number": issue["number"]
}
})
synced_tasks.append(task)
return synced_tasks