|
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 |
|
|
|
|
|
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: |
|
|
|
task = { |
|
"id": f"github-issue-{issue['id']}", |
|
"title": issue["title"], |
|
"description": issue["body"] or "", |
|
"status": "todo" if issue["state"] == "open" else "done", |
|
"priority": "medium", |
|
"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", []) |
|
} |
|
} |
|
|
|
|
|
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: |
|
|
|
if task.get("source") == "github": |
|
synced_tasks.append(task) |
|
continue |
|
|
|
|
|
issue_data = { |
|
"title": task["title"], |
|
"body": task.get("description", ""), |
|
"labels": [] |
|
} |
|
|
|
|
|
if task.get("priority"): |
|
issue_data["labels"].append(f"priority:{task['priority']}") |
|
|
|
|
|
issue = self.create_issue( |
|
repo_full_name=repo_full_name, |
|
title=issue_data["title"], |
|
body=issue_data["body"], |
|
labels=issue_data["labels"] |
|
) |
|
|
|
|
|
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 |